diff --git a/amber/src/main/python/proto/org/apache/texera/amber/core/__init__.py b/amber/src/main/python/proto/org/apache/texera/amber/core/__init__.py
index 6dbf282e193..2d21638c263 100644
--- a/amber/src/main/python/proto/org/apache/texera/amber/core/__init__.py
+++ b/amber/src/main/python/proto/org/apache/texera/amber/core/__init__.py
@@ -87,7 +87,7 @@ class GlobalPortIdentity(betterproto.Message):
 class InputPort(betterproto.Message):
     id: "PortIdentity" = betterproto.message_field(1)
     display_name: str = betterproto.string_field(2)
-    allow_multi_links: bool = betterproto.bool_field(3)
+    disallow_multi_links: bool = betterproto.bool_field(3)
     dependencies: List["PortIdentity"] = betterproto.message_field(4)
 
 
diff --git a/common/config/src/main/resources/storage.conf b/common/config/src/main/resources/storage.conf
index d1ae5941a5e..1f39359155c 100644
--- a/common/config/src/main/resources/storage.conf
+++ b/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}
             }
         }
diff --git a/common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/workflow.proto b/common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/workflow.proto
index 984536fca72..c5b2cb248ff 100644
--- a/common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/workflow.proto
+++ b/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
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/PortDescriptor.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/PortDescriptor.scala
index 73f1ef26007..2d6763b1025 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/PortDescriptor.scala
+++ b/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
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dummy/DummyOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dummy/DummyOpDesc.scala
index c1ae1dd05fc..4a2c4c48d54 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dummy/DummyOpDesc.scala
+++ b/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 {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala
index 6899f06a2d5..fdfcbcf27dc 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala
+++ b/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,
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala
index df7d9336650..e262bb29531 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala
+++ b/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")
       ),
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/java/JavaUDFOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/java/JavaUDFOpDesc.scala
index 5c452a6c1d6..44cc54f3865 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/java/JavaUDFOpDesc.scala
+++ b/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 {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
index 3f056c96055..4c4e3dc0098 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
+++ b/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))
         )
       ),
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala
index ef4da06cef9..88e8846b994 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala
+++ b/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 {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/r/RUDFOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/r/RUDFOpDesc.scala
index 1a7aa938f77..17eb2570815 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/r/RUDFOpDesc.scala
+++ b/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 {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala
index a10a4e43004..4294732f608 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala
+++ b/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())
     )
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/DotPlot/DotPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/DotPlot/DotPlotOpDesc.scala
index 33069a89212..29b1fb080cd 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/DotPlot/DotPlotOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/IcicleChart/IcicleChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/IcicleChart/IcicleChartOpDesc.scala
index a39e9b2681c..7287c036ba6 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/IcicleChart/IcicleChartOpDesc.scala
+++ b/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 =
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ImageViz/ImageVisualizerOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ImageViz/ImageVisualizerOpDesc.scala
index dc9eec9a8ee..9999251239a 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ImageViz/ImageVisualizerOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ScatterMatrixChart/ScatterMatrixChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ScatterMatrixChart/ScatterMatrixChartOpDesc.scala
index 3bfc5eb6b68..def972dac91 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ScatterMatrixChart/ScatterMatrixChartOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/barChart/BarChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/barChart/BarChartOpDesc.scala
index 723b64bd3d0..2912003c460 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/barChart/BarChartOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/boxViolinPlot/BoxViolinPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/boxViolinPlot/BoxViolinPlotOpDesc.scala
index 9f3a2a1f31e..70a5f97facc 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/boxViolinPlot/BoxViolinPlotOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bubbleChart/BubbleChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bubbleChart/BubbleChartOpDesc.scala
index 59a8cf5cc89..da2a6fe17a3 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bubbleChart/BubbleChartOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bulletChart/BulletChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bulletChart/BulletChartOpDesc.scala
index a5e19bca9a5..c2360e30717 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/bulletChart/BulletChartOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/candlestickChart/CandlestickChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/candlestickChart/CandlestickChartOpDesc.scala
index f0cc02a9b7d..33b47501b14 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/candlestickChart/CandlestickChartOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/choroplethMap/ChoroplethMapOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/choroplethMap/ChoroplethMapOpDesc.scala
index f9774ce80f4..678633bb959 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/choroplethMap/ChoroplethMapOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/continuousErrorBands/ContinuousErrorBandsOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/continuousErrorBands/ContinuousErrorBandsOpDesc.scala
index b21c882ed35..abd0bb8ddb5 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/continuousErrorBands/ContinuousErrorBandsOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/contourPlot/ContourPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/contourPlot/ContourPlotOpDesc.scala
index dd0f41b0faa..56d7bb9f0ce 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/contourPlot/ContourPlotOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dendrogram/DendrogramOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dendrogram/DendrogramOpDesc.scala
index d33ff2708f1..82356a6e2a0 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dendrogram/DendrogramOpDesc.scala
+++ b/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 = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dumbbellPlot/DumbbellPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dumbbellPlot/DumbbellPlotOpDesc.scala
index 88b6caae614..65c23468237 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dumbbellPlot/DumbbellPlotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/dumbbellPlot/DumbbellPlotOpDesc.scala
@@ -85,7 +85,7 @@ class DumbbellPlotOpDesc extends PythonOperatorDescriptor {
   @JsonProperty(value = "showLegends", required = false)
   @JsonSchemaTitle("Show Legends?")
   @JsonPropertyDescription("whether show legends in the graph")
-  var showLegends: Boolean = false;
+  var showLegends: Boolean = false
 
   override def getOutputSchemas(
       inputSchemas: Map[PortIdentity, Schema]
@@ -97,12 +97,10 @@ class DumbbellPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Dumbbell Plot",
       "Visualize data in a Dumbbell Plots. A dumbbell plots (also known as a lollipop chart) is typically used to compare two distinct values or time points for the same entity.",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def createPlotlyDumbbellLineFigure(): PythonTemplateBuilder = {
@@ -113,27 +111,27 @@ class DumbbellPlotOpDesc extends PythonOperatorDescriptor {
     }
     pyb"""
        |
-       |        entityNames = list(table[${comparedColumnName}].unique())
+       |        entityNames = list(table[$comparedColumnName].unique())
        |        entityNames = sorted(entityNames, reverse=True)
-       |        categoryValues = [${dumbbellValues}]
-       |        filtered_table = table[(table[${comparedColumnName}].isin(entityNames)) &
-       |                    (table[${categoryColumnName}].isin(categoryValues))]
+       |        categoryValues = [$dumbbellValues]
+       |        filtered_table = table[(table[$comparedColumnName].isin(entityNames)) &
+       |                    (table[$categoryColumnName].isin(categoryValues))]
        |
        |        # Create the dumbbell line using Plotly
        |        fig = go.Figure()
        |        color = 'black'
        |        for entity in entityNames:
-       |          entity_data = filtered_table[filtered_table[${comparedColumnName}] == entity]
-       |          fig.add_trace(go.Scatter(x=entity_data[${measurementColumnName}],
+       |          entity_data = filtered_table[filtered_table[$comparedColumnName] == entity]
+       |          fig.add_trace(go.Scatter(x=entity_data[$measurementColumnName],
        |                             y=[entity]*len(entity_data),
        |                             mode='lines',
        |                             name=entity,
        |                             line=dict(color=color)))
        |
-       |          fig.update_layout(xaxis_title=${measurementColumnName},
-       |                  yaxis_title=${comparedColumnName},
+       |          fig.update_layout(xaxis_title=$measurementColumnName,
+       |                  yaxis_title=$comparedColumnName,
        |                  yaxis=dict(categoryorder='array', categoryarray=entityNames),
-       |                  ${showLegendsOption}
+       |                  $showLegendsOption
        |                  )
        |"""
   }
@@ -150,12 +148,12 @@ class DumbbellPlotOpDesc extends PythonOperatorDescriptor {
     }
 
     pyb"""
-       |        dotColumnNames = [${dotColumnNames}]
+       |        dotColumnNames = [$dotColumnNames]
        |        if len(dotColumnNames) > 0:
        |          for dotColumn in dotColumnNames:
        |              # Extract dot data for each entity
        |              for entity in entityNames:
-       |                  entity_dot_data = filtered_table[filtered_table[${comparedColumnName}] == entity]
+       |                  entity_dot_data = filtered_table[filtered_table[$comparedColumnName] == entity]
        |                  # Extract X and Y values for the dot
        |                  x_values = entity_dot_data[dotColumn].values
        |                  y_values = [entity] * len(x_values)
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/figureFactoryTable/FigureFactoryTableOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/figureFactoryTable/FigureFactoryTableOpDesc.scala
index 11168488b74..14d4e2bf6d6 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/figureFactoryTable/FigureFactoryTableOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/figureFactoryTable/FigureFactoryTableOpDesc.scala
@@ -117,12 +117,10 @@ class FigureFactoryTableOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo = {
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Figure Factory Table",
       "Visualize data in a figure factory table",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
   }
 
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala
index d8d47696157..2eaf12d369c 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala
@@ -82,12 +82,10 @@ class FilledAreaPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Filled Area Plot",
       "Visualize data in filled area plot",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/funnelPlot/FunnelPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/funnelPlot/FunnelPlotOpDesc.scala
index 89c1cb0b104..5ca280b4dfa 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/funnelPlot/FunnelPlotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/funnelPlot/FunnelPlotOpDesc.scala
@@ -67,12 +67,10 @@ class FunnelPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Funnel Plot",
       "Visualize data in a Funnel Plot",
-      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP
     )
 
   private def createPlotlyFigure(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ganttChart/GanttChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ganttChart/GanttChartOpDesc.scala
index 9a1f32e43ef..a30e73cbb18 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ganttChart/GanttChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ganttChart/GanttChartOpDesc.scala
@@ -90,12 +90,10 @@ class GanttChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Gantt Chart",
       "A Gantt chart is a type of bar chart that illustrates a project schedule. The chart lists the tasks to be performed on the vertical axis, and time intervals on the horizontal axis. The width of the horizontal bars in the graph shows the duration of each activity.",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/gaugeChart/GaugeChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/gaugeChart/GaugeChartOpDesc.scala
index c890786eeff..4f771719e3c 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/gaugeChart/GaugeChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/gaugeChart/GaugeChartOpDesc.scala
@@ -61,12 +61,10 @@ class GaugeChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Gauge Chart",
       "Visualize a single value with a radial gauge chart, showing progress towards a goal with optional steps, threshold, and delta.",
-      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP
     )
 
   private val mapper = new ObjectMapper()
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/heatMap/HeatMapOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/heatMap/HeatMapOpDesc.scala
index d38dfdf4c90..ed0b13736d9 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/heatMap/HeatMapOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/heatMap/HeatMapOpDesc.scala
@@ -60,12 +60,10 @@ class HeatMapOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Heatmap",
       "Visualize data in a HeatMap Chart",
-      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   private def createHeatMap(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/hierarchychart/HierarchyChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/hierarchychart/HierarchyChartOpDesc.scala
index d46549111c2..49fc7875209 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/hierarchychart/HierarchyChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/hierarchychart/HierarchyChartOpDesc.scala
@@ -75,12 +75,10 @@ class HierarchyChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Hierarchy Chart",
       "Visualize data in hierarchy",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   private def getHierarchyAttributesInPython: String =
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram/HistogramChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram/HistogramChartOpDesc.scala
index 0c1a29b781b..ed7f74733b6 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram/HistogramChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram/HistogramChartOpDesc.scala
@@ -61,12 +61,10 @@ class HistogramChartOpDesc extends PythonOperatorDescriptor {
   var pattern: EncodableString = ""
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Histogram",
       "Visualize data in a Histogram Chart",
-      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram2d/Histogram2DOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram2d/Histogram2DOpDesc.scala
index 88167a8353f..8756acd982b 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram2d/Histogram2DOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/histogram2d/Histogram2DOpDesc.scala
@@ -61,12 +61,10 @@ class Histogram2DOpDesc extends PythonOperatorDescriptor {
   var normalize: NormalizationType = NormalizationType.DENSITY
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Histogram2D",
       "Displays a bivariate histogram as a density heatmap",
-      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP
     )
 
   override def getOutputSchemas(
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala
index f5b9bb81e7c..f8d78028a50 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/htmlviz/HtmlVizOpDesc.scala
@@ -70,12 +70,10 @@ class HtmlVizOpDesc extends LogicalOp {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "HTML Visualizer",
       "Render the result of HTML content",
-      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP
     )
 
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/lineChart/LineChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/lineChart/LineChartOpDesc.scala
index 2400b53e11b..2fcadd43c00 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/lineChart/LineChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/lineChart/LineChartOpDesc.scala
@@ -58,12 +58,10 @@ class LineChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Line Chart",
       "View the result in line chart",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/nestedTable/NestedTableOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/nestedTable/NestedTableOpDesc.scala
index aaaf4cdc95b..23719728298 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/nestedTable/NestedTableOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/nestedTable/NestedTableOpDesc.scala
@@ -47,12 +47,10 @@ class NestedTableOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Nested Table",
       "Visualize Data in a Depth Two Nested Table",
-      OperatorGroupConstants.VISUALIZATION_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_GROUP
     )
 
   private def createNestedTable(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala
index 58ae0c00cbc..2c5bd2a3eed 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala
@@ -58,12 +58,10 @@ class NetworkGraphOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Network Graph",
       "Visualize data in a network graph",
-      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/pieChart/PieChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/pieChart/PieChartOpDesc.scala
index 75e532e2d89..977d50edfc3 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/pieChart/PieChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/pieChart/PieChartOpDesc.scala
@@ -69,12 +69,10 @@ class PieChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Pie Chart",
       "Visualize data in a Pie Chart",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/quiverPlot/QuiverPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/quiverPlot/QuiverPlotOpDesc.scala
index 8246131fc2d..fc5ed3b6699 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/quiverPlot/QuiverPlotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/quiverPlot/QuiverPlotOpDesc.scala
@@ -74,12 +74,10 @@ class QuiverPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Quiver Plot",
       "Visualize vector data in a Quiver Plot",
-      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   //data cleaning for missing value
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/radarChart/RadarChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/radarChart/RadarChartOpDesc.scala
index ae4c383ea1a..84547296578 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/radarChart/RadarChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/radarChart/RadarChartOpDesc.scala
@@ -70,12 +70,10 @@ class RadarChartOpDesc extends PythonOperatorDescriptor {
   var fillOpacity: Double = 0.5
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Radar Chart",
       "Visualize data in a Radar Chart",
-      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   override def getOutputSchemas(
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/rangeSlider/RangeSliderOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/rangeSlider/RangeSliderOpDesc.scala
index 2a13db3e035..a6c1623d3a1 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/rangeSlider/RangeSliderOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/rangeSlider/RangeSliderOpDesc.scala
@@ -72,12 +72,10 @@ class RangeSliderOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Range Slider",
       "Visualize data in a Range Slider",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/sankeyDiagram/SankeyDiagramOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/sankeyDiagram/SankeyDiagramOpDesc.scala
index 0261baf741c..15d4d88f255 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/sankeyDiagram/SankeyDiagramOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/sankeyDiagram/SankeyDiagramOpDesc.scala
@@ -66,12 +66,10 @@ class SankeyDiagramOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Sankey Diagram",
       "Visualize data using a Sankey diagram",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatter3DChart/Scatter3dChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatter3DChart/Scatter3dChartOpDesc.scala
index e20ad4a8d1e..4ef718c7e13 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatter3DChart/Scatter3dChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatter3DChart/Scatter3dChartOpDesc.scala
@@ -66,12 +66,10 @@ class Scatter3dChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Scatter3D Chart",
       "Visualize data in a Scatter3D Plot",
-      OperatorGroupConstants.VISUALIZATION_ADVANCED_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_ADVANCED_GROUP
     )
 
   private def createPlotlyFigure(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatterplot/ScatterplotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatterplot/ScatterplotOpDesc.scala
index 92cf4845993..31a711622ec 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatterplot/ScatterplotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/scatterplot/ScatterplotOpDesc.scala
@@ -102,12 +102,10 @@ class ScatterplotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Scatter Plot",
       "View the result in a scatterplot",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/stripChart/StripChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/stripChart/StripChartOpDesc.scala
index aea4d4afb4f..fabaccaa7c6 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/stripChart/StripChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/stripChart/StripChartOpDesc.scala
@@ -65,12 +65,10 @@ class StripChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Strip Chart",
       "Visualize distribution of data points as a strip plot",
-      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP
     )
 
   override def generatePythonCode(): String = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/tablesChart/TablesPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/tablesChart/TablesPlotOpDesc.scala
index 0f71be4ab94..382c2604357 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/tablesChart/TablesPlotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/tablesChart/TablesPlotOpDesc.scala
@@ -95,12 +95,10 @@ class TablesPlotOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo = {
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Tables Plot",
       "Visualize data in a table chart.",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
   }
 
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ternaryPlot/TernaryPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ternaryPlot/TernaryPlotOpDesc.scala
index 14db98ee20b..8df0381387e 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ternaryPlot/TernaryPlotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/ternaryPlot/TernaryPlotOpDesc.scala
@@ -71,12 +71,10 @@ class TernaryPlotOpDesc extends PythonOperatorDescriptor {
 
   // OperatorInfo instance describing ternary plot
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       userFriendlyName = "Ternary Plot",
       operatorDescription = "Points are graphed on a Ternary Plot using 3 specified data fields",
-      operatorGroupName = OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      operatorGroupName = OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   override def getOutputSchemas(
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/timeSeriesPlot/TimeSeriesPlot.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/timeSeriesplot/TimeSeriesOpDesc.scala
similarity index 97%
rename from common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/timeSeriesPlot/TimeSeriesPlot.scala
rename to common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/timeSeriesplot/TimeSeriesOpDesc.scala
index 0fdcb09a5cb..9290b5e26d8 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/timeSeriesPlot/TimeSeriesPlot.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/timeSeriesplot/TimeSeriesOpDesc.scala
@@ -79,12 +79,10 @@ class TimeSeriesOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Time Series Plot",
       "Visualize trends and patterns over time.",
-      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_BASIC_GROUP
     )
 
   override def generatePythonCode(): String = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreeplotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDesc.scala
similarity index 98%
rename from common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreeplotOpDesc.scala
rename to common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDesc.scala
index 60829fe4915..2b718b6e17a 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreeplotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDesc.scala
@@ -46,13 +46,11 @@ class TreePlotOpDesc extends PythonOperatorDescriptor {
   var edgeListColumn: EncodableString = ""
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       userFriendlyName = "Tree Plot",
       operatorDescription =
         "Visualize hierarchical data as a top-down, interactive, auto-sizing tree",
-      operatorGroupName = OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      operatorGroupName = OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP
     )
 
   override def getOutputSchemas(
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/urlviz/UrlVizOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/urlviz/UrlVizOpDesc.scala
index e367cbc3012..22924c3a4e1 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/urlviz/UrlVizOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/urlviz/UrlVizOpDesc.scala
@@ -81,12 +81,10 @@ class UrlVizOpDesc extends LogicalOp {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "URL Visualizer",
       "Render the content of URL",
-      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP
     )
 
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/volcanoPlot/VolcanoPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/volcanoPlot/VolcanoPlotOpDesc.scala
index e4ac94b178a..b1b6101145b 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/volcanoPlot/VolcanoPlotOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/volcanoPlot/VolcanoPlotOpDesc.scala
@@ -51,12 +51,10 @@ class VolcanoPlotOpDesc extends PythonOperatorDescriptor {
   @AutofillAttributeName var pvalueColumn: EncodableString = ""
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       userFriendlyName = "Volcano Plot",
       operatorDescription = "Displays statistical significance versus effect size",
-      operatorGroupName = OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      operatorGroupName = OperatorGroupConstants.VISUALIZATION_SCIENTIFIC_GROUP
     )
 
   override def getOutputSchemas(
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/waterfallChart/WaterfallChartOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/waterfallChart/WaterfallChartOpDesc.scala
index 8586b1868c4..5bbc1a33126 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/waterfallChart/WaterfallChartOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/waterfallChart/WaterfallChartOpDesc.scala
@@ -55,12 +55,10 @@ class WaterfallChartOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Waterfall Chart",
       "Visualize data as a waterfall chart",
-      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_FINANCIAL_GROUP
     )
 
   def createPlotlyFigure(): PythonTemplateBuilder = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/wordCloud/WordCloudOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/wordCloud/WordCloudOpDesc.scala
index 9daf6dd67b3..d484bf56ab2 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/wordCloud/WordCloudOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/wordCloud/WordCloudOpDesc.scala
@@ -55,12 +55,10 @@ class WordCloudOpDesc extends PythonOperatorDescriptor {
   }
 
   override def operatorInfo: OperatorInfo =
-    OperatorInfo(
+    OperatorInfo.forVisualization(
       "Word Cloud",
       "Generate word cloud for   texts",
-      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP,
-      inputPorts = List(InputPort()),
-      outputPorts = List(OutputPort(mode = OutputMode.SINGLE_SNAPSHOT))
+      OperatorGroupConstants.VISUALIZATION_MEDIA_GROUP
     )
 
   def manipulateTable(): PythonTemplateBuilder = {
diff --git a/frontend/src/app/common/type/proto/org/apache/texera/amber/core/workflow.ts b/frontend/src/app/common/type/proto/org/apache/texera/amber/core/workflow.ts
index 04371d42bf4..5b63451356b 100644
--- a/frontend/src/app/common/type/proto/org/apache/texera/amber/core/workflow.ts
+++ b/frontend/src/app/common/type/proto/org/apache/texera/amber/core/workflow.ts
@@ -24,7 +24,7 @@ export interface GlobalPortIdentity {
 export interface InputPort {
   id: PortIdentity | undefined;
   displayName: string;
-  allowMultiLinks: boolean;
+  disallowMultiLinks: boolean;
   dependencies: PortIdentity[];
 }
 
@@ -256,7 +256,7 @@ export const GlobalPortIdentity: MessageFns<GlobalPortIdentity> = {
 };
 
 function createBaseInputPort(): InputPort {
-  return { id: undefined, displayName: "", allowMultiLinks: false, dependencies: [] };
+  return { id: undefined, displayName: "", disallowMultiLinks: false, dependencies: [] };
 }
 
 export const InputPort: MessageFns<InputPort> = {
@@ -267,8 +267,8 @@ export const InputPort: MessageFns<InputPort> = {
     if (message.displayName !== "") {
       writer.uint32(18).string(message.displayName);
     }
-    if (message.allowMultiLinks !== false) {
-      writer.uint32(24).bool(message.allowMultiLinks);
+    if (message.disallowMultiLinks !== false) {
+      writer.uint32(24).bool(message.disallowMultiLinks);
     }
     for (const v of message.dependencies) {
       PortIdentity.encode(v!, writer.uint32(34).fork()).join();
@@ -302,7 +302,7 @@ export const InputPort: MessageFns<InputPort> = {
             break;
           }
 
-          message.allowMultiLinks = reader.bool();
+          message.disallowMultiLinks = reader.bool();
           continue;
         case 4:
           if (tag !== 34) {
@@ -324,7 +324,7 @@ export const InputPort: MessageFns<InputPort> = {
     return {
       id: isSet(object.id) ? PortIdentity.fromJSON(object.id) : undefined,
       displayName: isSet(object.displayName) ? globalThis.String(object.displayName) : "",
-      allowMultiLinks: isSet(object.allowMultiLinks) ? globalThis.Boolean(object.allowMultiLinks) : false,
+      disallowMultiLinks: isSet(object.disallowMultiLinks) ? globalThis.Boolean(object.disallowMultiLinks) : false,
       dependencies: globalThis.Array.isArray(object?.dependencies)
         ? object.dependencies.map((e: any) => PortIdentity.fromJSON(e))
         : [],
@@ -339,8 +339,8 @@ export const InputPort: MessageFns<InputPort> = {
     if (message.displayName !== "") {
       obj.displayName = message.displayName;
     }
-    if (message.allowMultiLinks !== false) {
-      obj.allowMultiLinks = message.allowMultiLinks;
+    if (message.disallowMultiLinks !== false) {
+      obj.disallowMultiLinks = message.disallowMultiLinks;
     }
     if (message.dependencies?.length) {
       obj.dependencies = message.dependencies.map((e) => PortIdentity.toJSON(e));
@@ -355,7 +355,7 @@ export const InputPort: MessageFns<InputPort> = {
     const message = createBaseInputPort();
     message.id = (object.id !== undefined && object.id !== null) ? PortIdentity.fromPartial(object.id) : undefined;
     message.displayName = object.displayName ?? "";
-    message.allowMultiLinks = object.allowMultiLinks ?? false;
+    message.disallowMultiLinks = object.disallowMultiLinks ?? false;
     message.dependencies = object.dependencies?.map((e) => PortIdentity.fromPartial(e)) || [];
     return message;
   },
diff --git a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts
index 161d217fd86..8048d3a487f 100644
--- a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts
+++ b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts
@@ -406,7 +406,7 @@ describe("WorkflowEditorComponent", () => {
           mockSentimentPredicate.operatorID,
           "input-0"
         )
-      ).toBeFalse();
+      ).toBeTrue();
 
       // should still allow a link from scan to view result
       expect(
@@ -439,7 +439,7 @@ describe("WorkflowEditorComponent", () => {
           mockResultPredicate.operatorID,
           "input-0"
         )
-      ).toBeFalse();
+      ).toBeTrue();
     });
 
     it("should validate operator connections with ports that allow multi-inputs correctly", () => {
diff --git a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts
index f1532bf9a9d..ef92eb8e1b9 100644
--- a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts
+++ b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts
@@ -1046,7 +1046,7 @@ export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy
       return false;
     }
 
-    let allowMultiInput = false;
+    let disallowMultiInput = false;
     if (this.workflowActionService.getTexeraGraph().hasOperator(targetCellID)) {
       const portIndex = this.workflowActionService
         .getTexeraGraph()
@@ -1055,10 +1055,10 @@ export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy
       if (portIndex >= 0) {
         const portInfo =
           this.dynamicSchemaService.getDynamicSchema(targetCellID).additionalMetadata.inputPorts[portIndex];
-        allowMultiInput = portInfo?.allowMultiLinks ?? false;
+        disallowMultiInput = portInfo?.disallowMultiLinks ?? false;
       }
     }
-    return !(connectedLinksToTargetPort.length > 0 && !allowMultiInput);
+    return !(connectedLinksToTargetPort.length > 0 && disallowMultiInput);
   }
 
   /**
diff --git a/frontend/src/app/workspace/service/operator-metadata/mock-operator-metadata.data.ts b/frontend/src/app/workspace/service/operator-metadata/mock-operator-metadata.data.ts
index 3d698b84d17..b90d578e351 100644
--- a/frontend/src/app/workspace/service/operator-metadata/mock-operator-metadata.data.ts
+++ b/frontend/src/app/workspace/service/operator-metadata/mock-operator-metadata.data.ts
@@ -264,7 +264,7 @@ export const mockUnionSchema: OperatorSchema = {
     userFriendlyName: "Union",
     operatorDescription: "Union multiple inputs",
     operatorGroupName: "Analysis",
-    inputPorts: [{ allowMultiLinks: true }],
+    inputPorts: [{}],
     outputPorts: [{}],
   },
   operatorVersion: "union1",
diff --git a/frontend/src/app/workspace/service/validation/validation-workflow.service.ts b/frontend/src/app/workspace/service/validation/validation-workflow.service.ts
index 93099888b21..92ebb88bb22 100644
--- a/frontend/src/app/workspace/service/validation/validation-workflow.service.ts
+++ b/frontend/src/app/workspace/service/validation/validation-workflow.service.ts
@@ -325,15 +325,15 @@ export class ValidationWorkflowService {
     for (let i = 0; i < operator.inputPorts.length; i++) {
       const port = operator.inputPorts[i];
       const portNumInputs = numInputLinksByPort.get(port.portID) ?? 0;
-      if (port.allowMultiInputs) {
-        if (portNumInputs < 1) {
+      if (port.disallowMultiInputs) {
+        if (portNumInputs !== 1) {
           satisfyInput = false;
-          inputPortsViolationMessage += `${port.displayName ?? ""} requires at least 1 inputs, has ${portNumInputs}`;
+          inputPortsViolationMessage += `${port.displayName ?? ""} requires 1 input, has ${portNumInputs}`;
         }
       } else {
-        if (portNumInputs !== 1) {
+        if (portNumInputs < 1) {
           satisfyInput = false;
-          inputPortsViolationMessage += `${port.displayName ?? ""} requires 1 input, has ${portNumInputs}`;
+          inputPortsViolationMessage += `${port.displayName ?? ""} requires at least 1 inputs, has ${portNumInputs}`;
         }
       }
     }
diff --git a/frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts b/frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts
index 32a5e3a2db4..9dd1ac8e2ef 100644
--- a/frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts
+++ b/frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts
@@ -236,7 +236,7 @@ export class WorkflowActionService {
     });
   }
 
-  public addPort(operatorID: string, isInput: boolean, allowMultiInputs?: boolean): void {
+  public addPort(operatorID: string, isInput: boolean, disallowMultiInputs?: boolean): void {
     const operator = this.texeraGraph.getOperator(operatorID);
     // TODO: use uniform serde to calculate the portID
     const prefix = isInput ? "input-" : "output-";
@@ -251,7 +251,7 @@ export class WorkflowActionService {
     const port: PortDescription = {
       portID,
       displayName: "",
-      allowMultiInputs,
+      disallowMultiInputs,
       isDynamicPort: true,
       dependencies: [],
     };
@@ -262,8 +262,8 @@ export class WorkflowActionService {
     if (!operator.dynamicOutputPorts && !isInput) {
       throw new Error(`operator ${operatorID} does not have dynamic output ports`);
     }
-    if (!isInput && allowMultiInputs !== undefined) {
-      throw new Error("error: allowMultiInputs property of an output port should not be specified");
+    if (!isInput && disallowMultiInputs !== undefined) {
+      throw new Error("error: disallowMultiInputs property of an output port should not be specified");
     }
 
     this.texeraGraph.bundleActions(() => {
diff --git a/frontend/src/app/workspace/service/workflow-graph/util/workflow-util.service.ts b/frontend/src/app/workspace/service/workflow-graph/util/workflow-util.service.ts
index 8bca014062a..64681965b42 100644
--- a/frontend/src/app/workspace/service/workflow-graph/util/workflow-util.service.ts
+++ b/frontend/src/app/workspace/service/workflow-graph/util/workflow-util.service.ts
@@ -189,7 +189,7 @@ export class WorkflowUtilService {
     return {
       portID,
       displayName: inputPortInfo.displayName ?? "",
-      allowMultiInputs: inputPortInfo.allowMultiLinks ?? false,
+      disallowMultiInputs: inputPortInfo.disallowMultiLinks ?? false,
       isDynamicPort: false,
       dependencies: inputPortInfo.dependencies ?? [],
     };
@@ -199,7 +199,7 @@ export class WorkflowUtilService {
     return {
       portID,
       displayName: outputPortInfo.displayName ?? "",
-      allowMultiInputs: false,
+      disallowMultiInputs: false,
       isDynamicPort: false,
     };
   }
diff --git a/frontend/src/app/workspace/types/execute-workflow.interface.ts b/frontend/src/app/workspace/types/execute-workflow.interface.ts
index 1633e4bfdf7..9b6edb00b29 100644
--- a/frontend/src/app/workspace/types/execute-workflow.interface.ts
+++ b/frontend/src/app/workspace/types/execute-workflow.interface.ts
@@ -35,7 +35,7 @@ export interface InputPort
   extends Readonly<{
     id: PortIdentity;
     displayName: string;
-    allowMultiLinks: boolean;
+    disallowMultiLinks: boolean;
     dependencies: ReadonlyArray<PortIdentity>;
   }> {}
 
diff --git a/frontend/src/app/workspace/types/operator-schema.interface.ts b/frontend/src/app/workspace/types/operator-schema.interface.ts
index c02911568af..c0356c1eaba 100644
--- a/frontend/src/app/workspace/types/operator-schema.interface.ts
+++ b/frontend/src/app/workspace/types/operator-schema.interface.ts
@@ -31,7 +31,7 @@ import { CustomJSONSchema7 } from "./custom-json-schema.interface";
 export interface InputPortInfo
   extends Readonly<{
     displayName?: string;
-    allowMultiLinks?: boolean;
+    disallowMultiLinks?: boolean;
     dependencies?: { id: number; internal: boolean }[];
   }> {}
 
@@ -107,7 +107,7 @@ export function areOperatorSchemasEqual(schema1: OperatorSchema, schema2: Operat
     const port1 = meta1.inputPorts[i];
     const port2 = meta2.inputPorts[i];
 
-    if (port1.displayName !== port2.displayName || port1.allowMultiLinks !== port2.allowMultiLinks) {
+    if (port1.displayName !== port2.displayName || port1.disallowMultiLinks !== port2.disallowMultiLinks) {
       return false;
     }
   }
diff --git a/frontend/src/app/workspace/types/workflow-common.interface.ts b/frontend/src/app/workspace/types/workflow-common.interface.ts
index 74cfeac5755..3fb3aaa3d4b 100644
--- a/frontend/src/app/workspace/types/workflow-common.interface.ts
+++ b/frontend/src/app/workspace/types/workflow-common.interface.ts
@@ -55,7 +55,7 @@ export interface PortDescription
   extends Readonly<{
     portID: string;
     displayName?: string;
-    allowMultiInputs?: boolean;
+    disallowMultiInputs?: boolean;
     isDynamicPort?: boolean;
     partitionRequirement?: PartitionInfo;
     dependencies?: { id: number; internal: boolean }[];