Add Waterfall Chart Visualization Operator (#2825)

This pull request introduces a new plot visualizer, which is the
**Waterfall Chart** operator.

The Waterfall Chart visualizes sequential data where each category
(step) contributes to a cumulative effect, showing increases or
decreases. The waterfall chart helps users analyze how different factors
contribute to the total value. Users can specify an X attribute
(categories or steps), Y attribute (numerical values for each step).

**Screenshot:**

![Waterfall](https://github.com/user-attachments/assets/5ad96e7b-e50b-4995-89b6-2181e1feeb28)

---------

Authored-by: seongjinyoon <[email protected]>

---------

Co-authored-by: Seongjin <[email protected]>

core/amber/src/main/scala/edu/uci/ics/texera/workflow/common/operators/LogicalOp.scala

@@ -129,6 +129,7 @@ import edu.uci.ics.texera.workflow.operators.visualization.dumbbellPlot.Dumbbell
 import edu.uci.ics.texera.workflow.operators.visualization.histogram.HistogramChartOpDesc
 import edu.uci.ics.texera.workflow.operators.visualization.heatMap.HeatMapOpDesc
 import edu.uci.ics.texera.workflow.operators.visualization.lineChart.LineChartOpDesc
+import edu.uci.ics.texera.workflow.operators.visualization.waterfallChart.WaterfallChartOpDesc
 import edu.uci.ics.texera.workflow.operators.visualization.scatter3DChart.Scatter3dChartOpDesc
 import edu.uci.ics.texera.workflow.operators.visualization.ScatterMatrixChart.ScatterMatrixChartOpDesc
 import edu.uci.ics.texera.workflow.operators.visualization.funnelPlot.FunnelPlotOpDesc
@@ -187,6 +188,7 @@ trait StateTransferFunc
     new Type(value = classOf[KeywordSearchOpDesc], name = "KeywordSearch"),
     new Type(value = classOf[AggregateOpDesc], name = "Aggregate"),
     new Type(value = classOf[LineChartOpDesc], name = "LineChart"),
+    new Type(value = classOf[WaterfallChartOpDesc], name = "WaterfallChart"),
     new Type(value = classOf[BarChartOpDesc], name = "BarChart"),
     new Type(value = classOf[PieChartOpDesc], name = "PieChart"),
     new Type(value = classOf[QuiverPlotOpDesc], name = "QuiverPlot"),

core/amber/src/main/scala/edu/uci/ics/texera/workflow/operators/visualization/waterfallChart/WaterfallOpDesc.scala

@@ -0,0 +1,91 @@
+package edu.uci.ics.texera.workflow.operators.visualization.waterfallChart
+
+import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
+import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import edu.uci.ics.texera.workflow.common.metadata.{OperatorGroupConstants, OperatorInfo}
+import edu.uci.ics.texera.workflow.common.operators.PythonOperatorDescriptor
+import edu.uci.ics.texera.workflow.common.tuple.schema.{Attribute, AttributeType, Schema}
+import edu.uci.ics.amber.engine.common.workflow.{InputPort, OutputPort}
+import edu.uci.ics.texera.workflow.operators.visualization.{
+  VisualizationConstants,
+  VisualizationOperator
+}
+import edu.uci.ics.texera.workflow.common.metadata.annotations.AutofillAttributeName
+
+class WaterfallChartOpDesc extends VisualizationOperator with PythonOperatorDescriptor {
+
+  @JsonProperty(value = "xColumn", required = true)
+  @JsonSchemaTitle("X Axis Values")
+  @JsonPropertyDescription("The column representing categories or stages")
+  @AutofillAttributeName
+  var xColumn: String = _
+
+  @JsonProperty(value = "yColumn", required = true)
+  @JsonSchemaTitle("Y Axis Values")
+  @JsonPropertyDescription("The column representing numeric values for each stage")
+  @AutofillAttributeName
+  var yColumn: String = _
+
+  override def getOutputSchema(schemas: Array[Schema]): Schema = {
+    Schema.builder().add(new Attribute("html-content", AttributeType.STRING)).build()
+  }
+
+  override def operatorInfo: OperatorInfo =
+    OperatorInfo(
+      "Waterfall Chart",
+      "Visualize data as a waterfall chart",
+      OperatorGroupConstants.VISUALIZATION_GROUP,
+      inputPorts = List(InputPort()),
+      outputPorts = List(OutputPort())
+    )
+
+  def createPlotlyFigure(): String = {
+    s"""
+       |        x_values = table['$xColumn']
+       |        y_values = table['$yColumn']
+       |
+       |        fig = go.Figure(go.Waterfall(
+       |            name="Waterfall", orientation="v",
+       |            measure=["relative"] * (len(y_values) - 1) + ["total"],
+       |            x=x_values,
+       |            y=y_values,
+       |            textposition="outside",
+       |            text=[f"{v:+}" for v in y_values],
+       |            connector={"line": {"color": "rgb(63, 63, 63)"}}
+       |        ))
+       |
+       |        fig.update_layout(showlegend=True, waterfallgap=0.3)
+       |""".stripMargin
+  }
+
+  override def generatePythonCode(): String = {
+    val finalCode =
+      s"""
+         |from pytexera import *
+         |
+         |import plotly.graph_objects as go
+         |import plotly.io
+         |
+         |class ProcessTableOperator(UDFTableOperator):
+         |
+         |    # Generate custom error message as html string
+         |    def render_error(self, error_msg) -> str:
+         |        return '''<h1>Waterfall chart is not available.</h1>
+         |                  <p>Reason is: {} </p>
+         |               '''.format(error_msg)
+         |
+         |    @overrides
+         |    def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]:
+         |        if table.empty:
+         |            yield {'html-content': self.render_error("input table is empty.")}
+         |            return
+         |        ${createPlotlyFigure()}
+         |        html = plotly.io.to_html(fig, include_plotlyjs='cdn', auto_play=False)
+         |        yield {'html-content': html}
+         |""".stripMargin
+    finalCode
+  }
+
+  // Specify the chart type as HTML visualization
+  override def chartType(): String = VisualizationConstants.HTML_VIZ
+}