feat: allow Multi-Link on Input Ports

common/config/src/main/resources/storage.conf

@@ -43,10 +43,10 @@ storage {
                 uri-without-scheme = "localhost:5432/texera_iceberg_catalog"
                 uri-without-scheme = ${?STORAGE_ICEBERG_CATALOG_POSTGRES_URI_WITHOUT_SCHEME}
 
-                username = "texera"
+                username = "postgres"
                 username = ${?STORAGE_ICEBERG_CATALOG_POSTGRES_USERNAME}
 
-                password = "password"
+                password = "postgres"
                 password = ${?STORAGE_ICEBERG_CATALOG_POSTGRES_PASSWORD}
             }
         }

common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/workflow.proto

@@ -42,12 +42,11 @@ message GlobalPortIdentity{
 message InputPort {
   PortIdentity id = 1 [(scalapb.field).no_box = true];
   string displayName = 2;
-  bool allowMultiLinks = 3;
+  bool disallowMultiLinks = 3;
   repeated PortIdentity dependencies = 4;
 }
 
 
-
 message OutputPort {
   enum OutputMode {
     // outputs complete result set snapshot for each update

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/PortDescriptor.scala

@@ -25,7 +25,7 @@ import org.apache.texera.amber.core.workflow.PartitionInfo
 case class PortDescription(
     portID: String,
     displayName: String,
-    allowMultiInputs: Boolean,
+    disallowMultiInputs: Boolean,
     isDynamicPort: Boolean,
     partitionRequirement: PartitionInfo,
     dependencies: List[Int] = List.empty

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dummy/DummyOpDesc.scala

@@ -39,12 +39,12 @@ class DummyOpDesc extends LogicalOp with PortDescriptor {
           InputPort(
             PortIdentity(idx),
             displayName = portDesc.displayName,
-            allowMultiLinks = portDesc.allowMultiInputs,
+            disallowMultiLinks = portDesc.disallowMultiInputs,
             dependencies = portDesc.dependencies.map(idx => PortIdentity(idx))
           )
       }
     } else {
-      List(InputPort(PortIdentity(), allowMultiLinks = true))
+      List(InputPort())
     }
     val outputPortInfo = if (outputPorts != null) {
       outputPorts.zipWithIndex.map {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala

@@ -25,6 +25,7 @@ import com.fasterxml.jackson.databind.jsontype.NamedType
 import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode}
 import com.kjetland.jackson.jsonSchema.JsonSchemaConfig.html5EnabledSchema
 import com.kjetland.jackson.jsonSchema.{JsonSchemaConfig, JsonSchemaDraft, JsonSchemaGenerator}
+import org.apache.texera.amber.core.workflow.OutputPort.OutputMode
 import org.apache.texera.amber.core.workflow.{InputPort, OutputPort}
 import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc
@@ -45,6 +46,21 @@ case class OperatorInfo(
     allowPortCustomization: Boolean = false
 )
 
+object OperatorInfo {
+  def forVisualization(
+      userFriendlyName: String,
+      operatorDescription: String,
+      operatorGroupName: String
+  ): OperatorInfo =
+    OperatorInfo(
+      userFriendlyName,
+      operatorDescription,
+      operatorGroupName,
+      inputPorts = List(InputPort(disallowMultiLinks = true)),
+      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+    )
+}
+
 case class OperatorMetadata(
     operatorType: String,
     jsonSchema: JsonNode,

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala

@@ -91,8 +91,7 @@ class SklearnTestingOpDesc extends PythonOperatorDescriptor {
         InputPort(
           PortIdentity(),
           "model",
-          dependencies = List(PortIdentity(1)),
-          allowMultiLinks = true
+          dependencies = List(PortIdentity(1))
         ),
         InputPort(PortIdentity(1), "data")
       ),

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/java/JavaUDFOpDesc.scala

@@ -143,12 +143,12 @@ class JavaUDFOpDesc extends LogicalOp {
           InputPort(
             PortIdentity(idx),
             displayName = portDesc.displayName,
-            allowMultiLinks = portDesc.allowMultiInputs,
+            disallowMultiLinks = portDesc.disallowMultiInputs,
             dependencies = portDesc.dependencies.map(idx => PortIdentity(idx))
           )
       }
     } else {
-      List(InputPort(PortIdentity(), allowMultiLinks = true))
+      List(InputPort())
     }
     val outputPortInfo = if (outputPorts != null) {
       outputPorts.zipWithIndex.map {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala

@@ -137,11 +137,10 @@ class DualInputPortsPythonUDFOpDescV2 extends LogicalOp {
       "User-defined function operator in Python script",
       OperatorGroupConstants.PYTHON_GROUP,
       inputPorts = List(
-        InputPort(PortIdentity(), displayName = "model", allowMultiLinks = true),
+        InputPort(PortIdentity(), displayName = "model"),
         InputPort(
           PortIdentity(1),
           displayName = "tuples",
-          allowMultiLinks = true,
           dependencies = List(PortIdentity(0))
         )
       ),

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala

@@ -149,12 +149,12 @@ class PythonUDFOpDescV2 extends LogicalOp {
           InputPort(
             PortIdentity(idx),
             displayName = portDesc.displayName,
-            allowMultiLinks = portDesc.allowMultiInputs,
+            disallowMultiLinks = portDesc.disallowMultiInputs,
             dependencies = portDesc.dependencies.map(idx => PortIdentity(idx))
           )
       }
     } else {
-      List(InputPort(PortIdentity(), allowMultiLinks = true))
+      List(InputPort())
     }
     val outputPortInfo = if (outputPorts != null) {
       outputPorts.zipWithIndex.map {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/r/RUDFOpDesc.scala

@@ -141,12 +141,12 @@ class RUDFOpDesc extends LogicalOp {
           InputPort(
             PortIdentity(idx),
             displayName = portDesc.displayName,
-            allowMultiLinks = portDesc.allowMultiInputs,
+            disallowMultiLinks = portDesc.disallowMultiInputs,
             dependencies = portDesc.dependencies.map(idx => PortIdentity(idx))
           )
       }
     } else {
-      List(InputPort(PortIdentity(), allowMultiLinks = true))
+      List(InputPort())
     }
     val outputPortInfo = if (outputPorts != null) {
       outputPorts.zipWithIndex.map {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala

@@ -47,7 +47,7 @@ class UnionOpDesc extends LogicalOp {
       "Union",
       "Unions the output rows from multiple input operators",
       OperatorGroupConstants.SET_GROUP,
-      inputPorts = List(InputPort(PortIdentity(0), allowMultiLinks = true)),
+      inputPorts = List(InputPort()),
       outputPorts = List(OutputPort())
     )
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/DotPlot/DotPlotOpDesc.scala

@@ -52,12 +52,10 @@ class DotPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Dot Plot",
       "Visualize data using a dot plot",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/IcicleChart/IcicleChartOpDesc.scala

@@ -70,12 +70,10 @@ class IcicleChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Icicle Chart",
       "Visualize hierarchical data from root to leaves",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   private def getIcicleAttributesInPython: String =

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ImageViz/ImageVisualizerOpDesc.scala

@@ -48,12 +48,10 @@ class ImageVisualizerOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Image Visualizer",
       "visualize image content",
-      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP
     )
 
   def createBinaryData(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ScatterMatrixChart/ScatterMatrixChartOpDesc.scala

@@ -66,12 +66,10 @@ class ScatterMatrixChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Scatter Matrix Chart",
       "Visualize datasets in a Scatter Matrix",
-      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/barChart/BarChartOpDesc.scala

@@ -86,12 +86,10 @@ class BarChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Bar Chart",
       "Visualize data in a Bar Chart",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/boxViolinPlot/BoxViolinPlotOpDesc.scala

@@ -78,12 +78,10 @@ class BoxViolinPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Box/Violin Plot",
       "Visualize data using either a Box Plot or a Violin Plot. Box plots are drawn as a box with a vertical line down the middle which is mean value, and has horizontal lines attached to each side (known as “whiskers”). Violin plots provide more detail by showing a smoothed density curve on each side, and also include a box plot inside for comparison.",
-      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bubbleChart/BubbleChartOpDesc.scala

@@ -85,12 +85,10 @@ class BubbleChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Bubble Chart",
       "a 3D Scatter Plot; Bubbles are graphed using x and y labels, and their sizes determined by a z-value.",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bulletChart/BulletChartOpDesc.scala

@@ -30,6 +30,7 @@ import org.apache.texera.amber.operator.PythonOperatorDescriptor
 import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
+import java.util
 import java.util.{ArrayList, List => JList}
 import scala.jdk.CollectionConverters._
 
@@ -57,7 +58,7 @@ class BulletChartOpDesc extends PythonOperatorDescriptor {
   @JsonProperty(value = "steps", required = false)
   @JsonSchemaTitle("Steps")
   @JsonPropertyDescription("Optional: Each step includes a start and end value e.g., 0, 100.")
-  var steps: JList[BulletChartStepDefinition] = new ArrayList[BulletChartStepDefinition]()
+  var steps: JList[BulletChartStepDefinition] = new util.ArrayList[BulletChartStepDefinition]()
 
   override def getOutputSchemas(
       inputSchemas: Map[PortIdentity, Schema]
@@ -67,13 +68,11 @@ class BulletChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Bullet Chart",
       """Visualize data using a Bullet Chart that shows a primary quantitative bar and delta indicator.
         |Optional elements such as qualitative ranges (steps) and a performance threshold are displayed only when provided.""".stripMargin,
-      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP
     )
 
   override def generatePythonCode(): String = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/candlestickChart/CandlestickChartOpDesc.scala

@@ -72,12 +72,10 @@ class CandlestickChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Candlestick Chart",
       "Visualize data in a Candlestick Chart",
-      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP
     )
 
   override def generatePythonCode(): String = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/choroplethMap/ChoroplethMapOpDesc.scala

@@ -70,12 +70,10 @@ class ChoroplethMapOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Choropleth Map",
       "Visualize data using a Choropleth Map that uses shades of colors to show differences in properties or quantities between regions",
-      OperatorGroupConstants.VISUALIZATION_ADVANCED_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_ADVANCED_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/continuousErrorBands/ContinuousErrorBandsOpDesc.scala

@@ -57,12 +57,10 @@ class ContinuousErrorBandsOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Continuous Error Bands",
       "Visualize error or uncertainty along a continuous line",
-      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/contourPlot/ContourPlotOpDesc.scala

@@ -77,12 +77,10 @@ class ContourPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Contour Plot",
       "Displays terrain or gradient variations in a Contour Plot",
-      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   override def generatePythonCode(): String = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dendrogram/DendrogramOpDesc.scala

@@ -64,12 +64,10 @@ class DendrogramOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Dendrogram",
       "Visualize data in a Dendrogram",
-      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   private def createDendrogram(): PythonTemplateBuilder = {