stevenzwu commented on code in PR #10457:
URL: https://github.com/apache/iceberg/pull/10457#discussion_r1635623085


##########
flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java:
##########
@@ -103,14 +105,41 @@ public void initializeState(StateInitializationContext 
context) throws Exception
     if (context.isRestored()) {
       if (globalStatisticsState.get() == null
           || !globalStatisticsState.get().iterator().hasNext()) {
-        LOG.warn(
+        LOG.info(
             "Operator {} subtask {} doesn't have global statistics state to 
restore",
             operatorName,
             subtaskIndex);
+        // If Flink deprecates union state in the future, 
RequestGlobalStatisticsEvent can be
+        // leveraged to request global statistics from coordinator if new 
subtasks (scale-up case)
+        // has nothing to restore from.
       } else {
-        LOG.info(
-            "Operator {} subtask {} restoring global statistics state", 
operatorName, subtaskIndex);
-        this.globalStatistics = globalStatisticsState.get().iterator().next();
+        AggregatedStatistics restoredStatistics = 
globalStatisticsState.get().iterator().next();
+        // Range bounds is calculated from reservoir samplings with the 
determined partition
+        // (downstream parallelism). If downstream parallelism changed due to 
rescale, the computed
+        // range bounds array is not applicable. Operators should request 
coordinator to recompute
+        // the range bounds array using the new parallelism.
+        if (restoredStatistics.type() == StatisticsType.Sketch
+            && restoredStatistics.keySamples().length + 1 != 
downstreamParallelism) {
+          LOG.info(
+              "Operator {} subtask {} ignored restored sketch statistics as 
range bound length "
+                  + "not matching downstream parallelism due to rescale. "
+                  + "Old parallelism is {}, new parallelism is {}",
+              operatorName,
+              subtaskIndex,
+              restoredStatistics.keySamples().length + 1,
+              downstreamParallelism);
+          // Asynchronously request the latest global statistics calculated 
with new downstream
+          // parallelism. It is possible events may have started flowing 
before coordinator responds
+          // with global statistics. In this case, range partitioner would 
just blindly shuffle
+          // records in round robin fashion.

Review Comment:
   added the SketchRangePartitioner and RangePartitioner to this PR. so the 
scope is bigger now



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org

Reply via email to