[flink] Optimize sink statistics calculation to run only when necessary

Refactor RowDataSerializationSchema to extract size estimation logic into RowDataSizeEstimator class and only enable it when PARTITION_DYNAMIC mode is used. This avoids unnecessary computation overhead for other modes. Other improvements: - Improve operator names: "Collect Statistics", "Strip Statistics" - Remove slotSharingGroup for "Strip Statistics" operator - Remove redundant test setup code in Flink 1.18/1.19 IT cases - Add comprehensive Javadoc for DistributionMode enum - Add sink name with table path for better observability

Jark Wu committed on Jan 17, 2026, 09:43 AM

Showing 22 changed files +257 additions -183 deletions

M fluss-flink/fluss-flink-1.18/src/test/java/org/apache/fluss/flink/sink/Flink118TableSinkITCase.java +1 -21

M fluss-flink/fluss-flink-1.19/src/test/java/org/apache/fluss/flink/sink/Flink119TableSinkITCase.java +1 -22

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/row/RowWithOp.java +8

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkRowDataChannelComputer.java +2 -2

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkSink.java +19 -15

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkTableSink.java +2 -2

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlussSink.java +5 -2

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlussSinkBuilder.java +8 -1

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/FlussSerializationSchema.java +3

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/RowDataSerializationSchema.java +107 -81

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/SerializerInitContextImpl.java +12 -3

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/shuffle/DataStatisticsCoordinator.java +14 -6

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/shuffle/DataStatisticsOperator.java +2 -1

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/shuffle/DistributionMode.java +43

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/shuffle/StatisticsOrRecordChannelComputer.java +1 -1

M fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java +2 -1

M fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/metrics/FlinkMetricsITCase.java -2

M fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkRowDataChannelComputerTest.java +1 -1

M fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java +19 -15

M fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlussSinkITCase.java +2 -2

M fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/shuffle/StatisticsOrRecordChannelComputerTest.java +1 -1

M website/docs/engine-flink/options.md +4 -4