Repository: apache/iceberg (language: python)

Issue:
Flink: handle rescale (up or down) better in range partitioner
### Apache Iceberg version

main (development)

### Query engine

Flink

### Please describe the bug 🐞

See more context from the PR discussion: https://github.com/apache/iceberg/pull/10331#discussion_r1618890956


Discussion / hints:

Fix patch:
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java
index 52d8a2f16f99..338523b7b074 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java
@@ -18,7 +18,6 @@
  */
 package org.apache.iceberg.flink.sink.shuffle;
 
-import java.util.Comparator;
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.Set;
@@ -29,8 +28,6 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortKey;
 import org.apache.iceberg.SortOrder;
-import org.apache.iceberg.SortOrderComparators;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
@@ -38,9 +35,8 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * AggregatedStatisticsTracker is used by {@link DataStatisticsCoordinator} to track the in progress
- * {@link AggregatedStatistics} received from {@link DataStatisticsOperator} subtasks for specific
- * checkpoint.
+ * AggregatedStatisticsTracker tracks the statistics aggregation received from {@link
+ * DataStatisticsOperator} subtasks for every checkpoint.
  */
 class AggregatedStatisticsTracker {
   private static final Logger LOG = LoggerFactory.getLogger(AggregatedStatisticsTracker.class);
@@ -51,10 +47,9 @@ class AggregatedStatisticsTracker {
   private final int downstreamParallelism;
   private final StatisticsType statisticsType;
   private final int switchToSketchThreshold;
-  private final Comparator<StructLike> comparator;
   private final NavigableMap<Long, Aggregation> aggregationsPerCheckpoint;
 
-  private AggregatedStatistics completedStatistics;
+  private CompletedStatistics completedStatistics;
 
   AggregatedStatisticsTracker(
       String operatorName,
@@ -64,7 +59,7 @@ class AggregatedStatisticsTracker {
       int downstreamParallelism,
       StatisticsType statisticsType,
       int switchToSketchThreshold,
-      @Nullable AggregatedStatistics restoredStatistics) {
+      @Nullable CompletedStatistics restoredStatistics) {
     this.operatorName = operatorName;
     this.parallelism = parallelism;
     this.statisticsSerializer =
@@ -74,11 +69,10 @@ class AggregatedStatisticsTracker {
     this.switchToSketchThreshold = switchToSketchThreshold;
     this.completedStatistics = restoredStatistics;
 
-    this.comparator = SortOrderComparators.forSchema(schema, sortOrder);
     this.aggregationsPerCheckpoint = Maps.newTreeMap();
   }
 
-  AggregatedStatistics updateAndCheckCompletion(int subtask, StatisticsEvent event) {
+  CompletedStatistics updateAndCheckCompletion(int subtask, StatisticsEvent event) {
     long checkpointId = event.checkpointId();
     LOG.debug(
         "Handling statistics event from subtask {} of operator {} for checkpoint {}",
@@ -105,7 +99,6 @@ AggregatedStatistics updateAndCheckCompletion(int subtask, StatisticsEvent event
                     parallelism,
                     downstreamParallelism,
                     switchToSketchThreshold,
-                    comparator,
                     statisticsType,
                     StatisticsUtil.collectType(statisticsType, completedStatistics)));
     DataStatistics dataStatistics =
@@ -140,7 +133,6 @@ static class Aggregation {
     private final int parallelism;
     private final int downstreamParallelism;
     private final int switchToSketchThreshold;
-    private final Comparator<StructLike> comparator;
     private final StatisticsType configuredType;
     private StatisticsType currentType;
     private Map<SortKey, Long> mapStatistics;
@@ -150,14 +142,12 @@ static class Aggregation {
         int parallelism,
         int downstreamParallelism,
         int switchToSketchThreshold,
-        Comparator<StructLike> comparator,
         StatisticsType configuredType,
         StatisticsType currentType) {
       this.subtaskSet = Sets.newHashSet();
       this.parallelism = parallelism;
       this.downstreamParallelism = downstreamParallelism;
       this.switchToSketchThreshold = switchToSketchThreshold;
-      this.comparator = comparator;
       this.configuredType = configuredType;
       this.currentType = currentType;
 
@@ -246,10 +236,10 @@ private void convertCoordinatorToSketch() {
       this.mapStatistics = null;
     }
 
-    private AggregatedStatistics completedStatistics(long checkpointId) {
+    private CompletedStatistics completedStatistics(long checkpointId) {
       if (currentType == StatisticsType.Map) {
         LOG.info("Completed map statistics aggregation with {} keys", mapStatistics.size());
-        return AggregatedStatistics.fromKeyFrequency(checkpointId, mapStatistics);
+        return CompletedStatistics.fromKeyFrequency(checkpointId, mapStatistics);
       } else {
         ReservoirItemsSketch<SortKey> sketch = sketchStatistics.getResult();
         LOG.info(
@@ -258,8 +248,7 @@ private AggregatedStatistics completedStatistics(long checkpointId) {
             sketch.getK(),
             sketch.getN(),
             sketch.getNumSamples());
-        return AggregatedStatistics.fromRangeBounds(
-            checkpointId, SketchUtil.rangeBounds(downstreamParallelism, comparator, sketch));
+        return CompletedStatistics.fromKeySamples(checkpointId, sketch.getSamples());
       }
     }
   }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatistics.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/CompletedStatistics.java
similarity index 56%
rename from flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatistics.java
rename to flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/CompletedStatistics.java
index 8a91411c1d21..c0e228965ddd 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatistics.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/CompletedStatistics.java
@@ -18,46 +18,39 @@
  */
 package org.apache.iceberg.flink.sink.shuffle;
 
-import java.io.Serializable;
 import java.util.Arrays;
 import java.util.Map;
 import org.apache.iceberg.SortKey;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Objects;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 /**
- * AggregatedStatistics is used by {@link DataStatisticsCoordinator} to collect {@link
- * DataStatistics} from {@link DataStatisticsOperator} subtasks for specific checkpoint. It stores
- * the merged {@link DataStatistics} result from all reported subtasks.
+ * This is what {@link AggregatedStatisticsTracker} returns upon a completed statistics aggregation
+ * from all subtasks. It contains the raw statistics (Map or reservoir samples).
  */
-class AggregatedStatistics implements Serializable {
+class CompletedStatistics {
   private final long checkpointId;
   private final StatisticsType type;
   private final Map<SortKey, Long> keyFrequency;
-  private final SortKey[] rangeBounds;
+  private final SortKey[] keySamples;
 
-  AggregatedStatistics(
+  static CompletedStatistics fromKeyFrequency(long checkpointId, Map<SortKey, Long> stats) {
+    return new CompletedStatistics(checkpointId, StatisticsType.Map, stats, null);
+  }
+
+  static CompletedStatistics fromKeySamples(long checkpointId, SortKey[] keySamples) {
+    return new CompletedStatistics(checkpointId, StatisticsType.Sketch, null, keySamples);
+  }
+
+  CompletedStatistics(
       long checkpointId,
       StatisticsType type,
       Map<SortKey, Long> keyFrequency,
-      SortKey[] rangeBounds) {
-    Preconditions.checkArgument(
-        (keyFrequency != null && rangeBounds == null)
-            || (keyFrequency == null && rangeBounds != null),
-        "Invalid key frequency or range bounds: both are non-null or null");
+      SortKey[] keySamples) {
     this.checkpointId = checkpointId;
     this.type = type;
     this.keyFrequency = keyFrequency;
-    this.rangeBounds = rangeBounds;
-  }
-
-  static AggregatedStatistics fromKeyFrequency(long checkpointId, Map<SortKey, Long> stats) {
-    return new AggregatedStatistics(checkpointId, StatisticsType.Map, stats, null);
-  }
-
-  static AggregatedStatistics fromRangeBounds(long checkpointId, SortKey[] stats) {
-    return new AggregatedStatistics(checkpointId, StatisticsType.Sketch, null, stats);
+    this.keySamples = keySamples;
   }
 
   @Override
@@ -66,7 +59,7 @@ public String toString() {
         .add("checkpointId", checkpointId)
         .add("type", type)
         .add("keyFrequency", keyFrequency)
-        .add("rangeBounds", rangeBounds)
+        .add("keySamples", keySamples)
         .toString();
   }
 
@@ -76,20 +69,24 @@ public boolean equals(Object o) {
       return true;
     }
 
-    if (!(o instanceof AggregatedStatistics)) {
+    if (!(o instanceof CompletedStatistics)) {
       return false;
     }
 
-    AggregatedStatistics other = (AggregatedStatistics) o;
-    return Objects.equal(checkpointId, other.checkpointId())
-        && Objects.equal(type, other.type())
+    CompletedStatistics other = (CompletedStatistics) o;
+    return Objects.equal(checkpointId, other.checkpointId)
+        && Objects.equal(type, other.type)
         && Objects.equal(keyFrequency, other.keyFrequency())
-        && Arrays.equals(rangeBounds, other.rangeBounds());
+        && Arrays.equals(keySamples, other.keySamples());
   }
 
   @Override
   public int hashCode() {
-    return Objects.hashCode(checkpointId, type, keyFrequency, rangeBounds);
+    return Objects.hashCode(checkpointId, type, keyFrequency, keySamples);
+  }
+
+  long checkpointId() {
+    return checkpointId;
   }
 
   StatisticsType type() {
@@ -100,11 +97,7 @@ Map<SortKey, Long> keyFrequency() {
     return keyFrequency;
   }
 
-  SortKey[] rangeBounds() {
-    return rangeBounds;
-  }
-
-  long checkpointId() {
-    return checkpointId;
+  SortKey[] keySamples() {
+    return keySamples;
   }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsSerializer.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/CompletedStatisticsSerializer.java
similarity index 62%
rename from flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsSerializer.java
rename to flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/CompletedStatisticsSerializer.java
index 56ba5e04f05a..1ac0e386a011 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsSerializer.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/CompletedStatisticsSerializer.java
@@ -35,17 +35,17 @@
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.iceberg.SortKey;
 
-public class AggregatedStatisticsSerializer extends TypeSerializer<AggregatedStatistics> {
+class CompletedStatisticsSerializer extends TypeSerializer<CompletedStatistics> {
   private final TypeSerializer<SortKey> sortKeySerializer;
   private final EnumSerializer<StatisticsType> statisticsTypeSerializer;
   private final MapSerializer<SortKey, Long> keyFrequencySerializer;
-  private final ListSerializer<SortKey> rangeBoundsSerializer;
+  private final ListSerializer<SortKey> keySamplesSerializer;
 
-  AggregatedStatisticsSerializer(TypeSerializer<SortKey> sortKeySerializer) {
+  CompletedStatisticsSerializer(TypeSerializer<SortKey> sortKeySerializer) {
     this.sortKeySerializer = sortKeySerializer;
     this.statisticsTypeSerializer = new EnumSerializer<>(StatisticsType.class);
     this.keyFrequencySerializer = new MapSerializer<>(sortKeySerializer, LongSerializer.INSTANCE);
-    this.rangeBoundsSerializer = new ListSerializer<>(sortKeySerializer);
+    this.keySamplesSerializer = new ListSerializer<>(sortKeySerializer);
   }
 
   @Override
@@ -54,23 +54,23 @@ public boolean isImmutableType() {
   }
 
   @Override
-  public TypeSerializer<AggregatedStatistics> duplicate() {
-    return new AggregatedStatisticsSerializer(sortKeySerializer);
+  public TypeSerializer<CompletedStatistics> duplicate() {
+    return new CompletedStatisticsSerializer(sortKeySerializer);
   }
 
   @Override
-  public AggregatedStatistics createInstance() {
-    return new AggregatedStatistics(0, StatisticsType.Map, Collections.emptyMap(), null);
+  public CompletedStatistics createInstance() {
+    return CompletedStatistics.fromKeyFrequency(0L, Collections.emptyMap());
   }
 
   @Override
-  public AggregatedStatistics copy(AggregatedStatistics from) {
-    return new AggregatedStatistics(
-        from.checkpointId(), from.type(), from.keyFrequency(), from.rangeBounds());
+  public CompletedStatistics copy(CompletedStatistics from) {
+    return new CompletedStatistics(
+        from.checkpointId(), from.type(), from.keyFrequency(), from.keySamples());
   }
 
   @Override
-  public AggregatedStatistics copy(AggregatedStatistics from, AggregatedStatistics reuse) {
+  public CompletedStatistics copy(CompletedStatistics from, CompletedStatistics reuse) {
     // no benefit of reuse
     return copy(from);
   }
@@ -81,35 +81,33 @@ public int getLength() {
   }
 
   @Override
-  public void serialize(AggregatedStatistics record, DataOutputView target) throws IOException {
+  public void serialize(CompletedStatistics record, DataOutputView target) throws IOException {
     target.writeLong(record.checkpointId());
     statisticsTypeSerializer.serialize(record.type(), target);
     if (record.type() == StatisticsType.Map) {
       keyFrequencySerializer.serialize(record.keyFrequency(), target);
     } else {
-      rangeBoundsSerializer.serialize(Arrays.asList(record.rangeBounds()), target);
+      keySamplesSerializer.serialize(Arrays.asList(record.keySamples()), target);
     }
   }
 
   @Override
-  public AggregatedStatistics deserialize(DataInputView source) throws IOException {
+  public CompletedStatistics deserialize(DataInputView source) throws IOException {
     long checkpointId = source.readLong();
     StatisticsType type = statisticsTypeSerializer.deserialize(source);
-    Map<SortKey, Long> keyFrequency = null;
-    SortKey[] rangeBounds = null;
     if (type == StatisticsType.Map) {
-      keyFrequency = keyFrequencySerializer.deserialize(source);
+      Map<SortKey, Long> keyFrequency = keyFrequencySerializer.deserialize(source);
+      return CompletedStatistics.fromKeyFrequency(checkpointId, keyFrequency);
     } else {
-      List<SortKey> sortKeys = rangeBoundsSerializer.deserialize(source);
-      rangeBounds = new SortKey[sortKeys.size()];
-      rangeBounds = sortKeys.toArray(rangeBounds);
+      List<SortKey> sortKeys = keySamplesSerializer.deserialize(source);
+      SortKey[] keySamples = new SortKey[sortKeys.size()];
+      keySamples = sortKeys.toArray(keySamples);
+      return CompletedStatistics.fromKeySamples(checkpointId, keySamples);
     }
-
-    return new AggregatedStatistics(checkpointId, type, keyFrequency, rangeBounds);
   }
 
   @Override
-  public AggregatedStatistics deserialize(AggregatedStatistics reuse, DataInputView source)
+  public CompletedStatistics deserialize(CompletedStatistics reuse, DataInputView source)
       throws IOException {
     // not much benefit to reuse
     return deserialize(source);
@@ -122,11 +120,15 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
 
   @Override
   public boolean equals(Object obj) {
-    if (!(obj instanceof AggregatedStatisticsSerializer)) {
+    if (this == obj) {
+      return true;
+    }
+
+    if (obj == null || getClass() != obj.getClass()) {
       return false;
     }
 
-    AggregatedStatisticsSerializer other = (AggregatedStatisticsSerializer) obj;
+    CompletedStatisticsSerializer other = (CompletedStatisticsSerializer) obj;
     return Objects.equals(sortKeySerializer, other.sortKeySerializer);
   }
 
@@ -136,21 +138,20 @@ public int hashCode() {
   }
 
   @Override
-  public TypeSerializerSnapshot<AggregatedStatistics> snapshotConfiguration() {
-    return new AggregatedStatisticsSerializerSnapshot(this);
+  public TypeSerializerSnapshot<CompletedStatistics> snapshotConfiguration() {
+    return new CompletedStatisticsSerializerSnapshot(this);
   }
 
-  public static class AggregatedStatisticsSerializerSnapshot
-      extends CompositeTypeSerializerSnapshot<
-          AggregatedStatistics, AggregatedStatisticsSerializer> {
+  public static class CompletedStatisticsSerializerSnapshot
+      extends CompositeTypeSerializerSnapshot<CompletedStatistics, CompletedStatisticsSerializer> {
     private static final int CURRENT_VERSION = 1;
 
     /** Constructor for read instantiation. */
     @SuppressWarnings({"unused", "checkstyle:RedundantModifier"})
-    public AggregatedStatisticsSerializerSnapshot() {}
+    public CompletedStatisticsSerializerSnapshot() {}
 
     @SuppressWarnings("checkstyle:RedundantModifier")
-    public AggregatedStatisticsSerializerSnapshot(AggregatedStatisticsSerializer serializer) {
+    public CompletedStatisticsSerializerSnapshot(CompletedStatisticsSerializer serializer) {
       super(serializer);
     }
 
@@ -161,15 +162,15 @@ protected int getCurrentOuterSnapshotVersion() {
 
     @Override
     protected TypeSerializer<?>[] getNestedSerializers(
-        AggregatedStatisticsSerializer outerSerializer) {
+        CompletedStatisticsSerializer outerSerializer) {
       return new TypeSerializer<?>[] {outerSerializer.sortKeySerializer};
     }
 
     @Override
-    protected AggregatedStatisticsSerializer createOuterSerializerWithNestedSerializers(
+    protected CompletedStatisticsSerializer createOuterSerializerWithNestedSerializers(
         TypeSerializer<?>[] nestedSerializers) {
       SortKeySerializer sortKeySerializer = (SortKeySerializer) nestedSerializers[0];
-      return new AggregatedStatisticsSerializer(sortKeySerializer);
+      return new CompletedStatisticsSerializer(sortKeySerializer);
     }
   }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java
index 6e9e4f6fa6c2..3b21fbae315a 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java
@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.flink.sink.shuffle;
 
+import java.util.Comparator;
 import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
@@ -37,6 +38,8 @@
 import org.apache.flink.util.function.ThrowingRunnable;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.SortOrderComparators;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -60,17 +63,21 @@ class DataStatisticsCoordinator implements OperatorCoordinator {
   private final OperatorCoordinator.Context context;
   private final Schema schema;
   private final SortOrder sortOrder;
+  private final Comparator<StructLike> comparator;
   private final int downstreamParallelism;
   private final StatisticsType statisticsType;
+  private final double closeFileCostWeightPercentage;
 
   private final ExecutorService coordinatorExecutor;
   private final SubtaskGateways subtaskGateways;
   private final CoordinatorExecutorThreadFactory coordinatorThreadFactory;
-  private final TypeSerializer<AggregatedStatistics> aggregatedStatisticsSerializer;
+  private final TypeSerializer<CompletedStatistics> completedStatisticsSerializer;
+  private final TypeSerializer<GlobalStatistics> globalStatisticsSerializer;
 
   private transient boolean started;
   private transient AggregatedStatisticsTracker aggregatedStatisticsTracker;
-  private transient AggregatedStatistics completedStatistics;
+  private transient CompletedStatistics completedStatistics;
+  private transient GlobalStatistics globalStatistics;
 
   DataStatisticsCoordinator(
       String operatorName,
@@ -78,13 +85,16 @@ class DataStatisticsCoordinator implements OperatorCoordinator {
       Schema schema,
       SortOrder sortOrder,
       int downstreamParallelism,
-      StatisticsType statisticsType) {
+      StatisticsType statisticsType,
+      double closeFileCostWeightPercentage) {
     this.operatorName = operatorName;
     this.context = context;
     this.schema = schema;
     this.sortOrder = sortOrder;
+    this.comparator = SortOrderComparators.forSchema(schema, sortOrder);
     this.downstreamParallelism = downstreamParallelism;
     this.statisticsType = statisticsType;
+    this.closeFileCostWeightPercentage = closeFileCostWeightPercentage;
 
     this.coordinatorThreadFactory =
         new CoordinatorExecutorThreadFactory(
@@ -92,13 +102,16 @@ class DataStatisticsCoordinator implements OperatorCoordinator {
     this.coordinatorExecutor = Executors.newSingleThreadExecutor(coordinatorThreadFactory);
     this.subtaskGateways = new SubtaskGateways(operatorName, context.currentParallelism());
     SortKeySerializer sortKeySerializer = new SortKeySerializer(schema, sortOrder);
-    this.aggregatedStatisticsSerializer = new AggregatedStatisticsSerializer(sortKeySerializer);
+    this.completedStatisticsSerializer = new CompletedStatisticsSerializer(sortKeySerializer);
+    this.globalStatisticsSerializer = new GlobalStatisticsSerializer(sortKeySerializer);
   }
 
   @Override
   public void start() throws Exception {
     LOG.info("Starting data statistics coordinator: {}.", operatorName);
     this.started = true;
+
+    // statistics are restored already in resetToCheckpoint() before start() called
     this.aggregatedStatisticsTracker =
         new AggregatedStatisticsTracker(
             operatorName,
@@ -185,32 +198,98 @@ private void ensureStarted() {
   }
 
   private void handleDataStatisticRequest(int subtask, StatisticsEvent event) {
-    AggregatedStatistics aggregatedStatistics =
+    CompletedStatistics maybeCompletedStatistics =
         aggregatedStatisticsTracker.updateAndCheckCompletion(subtask, event);
 
-    if (aggregatedStatistics != null) {
-      completedStatistics = aggregatedStatistics;
-      sendAggregatedStatisticsToSubtasks(completedStatistics.checkpointId(), completedStatistics);
+    if (maybeCompletedStatistics != null) {
+      // completedStatistics contains the complete samples, which is needed to compute
+      // the range bounds in globalStatistics if downstreamParallelism changed.
+      this.completedStatistics = maybeCompletedStatistics;
+      // globalStatistics only contains assignment calculated based on Map or Sketch statistics
+      this.globalStatistics =
+          globalStatistics(
+              maybeCompletedStatistics,
+              downstreamParallelism,
+              comparator,
+              closeFileCostWeightPercentage);
+      sendGlobalStatisticsToSubtasks(globalStatistics);
+    }
+  }
+
+  private static GlobalStatistics globalStatistics(
+      CompletedStatistics completedStatistics,
+      int downstreamParallelism,
+      Comparator<StructLike> comparator,
+      double closeFileCostWeightPercentage) {
+    if (completedStatistics.type() == StatisticsType.Sketch) {
+      // range bound is a much smaller array compared to the complete samples.
+      // It helps reduce the amount of data transfer from coordinator to operator subtasks.
+      return GlobalStatistics.fromRangeBounds(
+          completedStatistics.checkpointId(),
+          SketchUtil.rangeBounds(
+              downstreamParallelism, comparator, completedStatistics.keySamples()));
+    } else {
+      return GlobalStatistics.fromMapAssignment(
+          completedStatistics.checkpointId(),
+          MapAssignment.fromKeyFrequency(
+              downstreamParallelism,
+              completedStatistics.keyFrequency(),
+              closeFileCostWeightPercentage,
+              comparator));
     }
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
-  private void sendAggregatedStatisticsToSubtasks(
-      long checkpointId, AggregatedStatistics globalStatistics) {
-    callInCoordinatorThread(
+  private void sendGlobalStatisticsToSubtasks(GlobalStatistics statistics) {
+    runInCoordinatorThread(
         () -> {
+          LOG.info(
+              "Broadcast latest global statistics from checkpoint {} to all subtasks",
+              statistics.checkpointId());
+          // applyImmediately is set to false so that operator subtasks can
+          // apply the change at checkpoint boundary
           StatisticsEvent statisticsEvent =
-              StatisticsEvent.createAggregatedStatisticsEvent(
-                  checkpointId, globalStatistics, aggregatedStatisticsSerializer);
+              StatisticsEvent.createGlobalStatisticsEvent(
+                  statistics, globalStatisticsSerializer, false);
           for (int i = 0; i < context.currentParallelism(); ++i) {
+            // Ignore future return value for potential error (e.g. subtask down).
+            // Upon restart, subtasks send request to coordinator to refresh statistics
+            // if there is any difference
             subtaskGateways.getSubtaskGateway(i).sendEvent(statisticsEvent);
           }
-
-          return null;
         },
         String.format(
             "Failed to send operator %s coordinator global data statistics for checkpoint %d",
-            operatorName, checkpointId));
+            operatorName, statistics.checkpointId()));
+  }
+
+  @SuppressWarnings("FutureReturnValueIgnored")
+  private void handleRequestGlobalStatisticsEvent(int subtask, RequestGlobalStatisticsEvent event) {
+    if (globalStatistics != null) {
+      runInCoordinatorThread(
+          () -> {
+            if (event.signature() != null && event.signature() != globalStatistics.hashCode()) {
+              LOG.debug(
+                  "Skip responding to statistics request from subtask {}, as hashCode matches or not included in the request",
+                  subtask);
+            } else {
+              LOG.info(
+                  "Send latest global statistics from checkpoint {} to subtask {}",
+                  globalStatistics.checkpointId(),
+                  subtask);
+              StatisticsEvent statisticsEvent =
+                  StatisticsEvent.createGlobalStatisticsEvent(
+                      globalStatistics, globalStatisticsSerializer, true);
+              subtaskGateways.getSubtaskGateway(subtask).sendEvent(statisticsEvent);
+            }
+          },
+          String.format(
+              "Failed to send operator %s coordinator global data statistics to requesting subtask %d for checkpoint %d",
+              operatorName, subtask, globalStatistics.checkpointId()));
+    } else {
+      LOG.info(
+          "Ignore global statistics request from subtask {} as statistics not available", subtask);
+    }
   }
 
   @Override
@@ -223,8 +302,14 @@ public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEven
               attemptNumber,
               operatorName,
               event);
-          Preconditions.checkArgument(event instanceof StatisticsEvent);
-          handleDataStatisticRequest(subtask, ((StatisticsEvent) event));
+          if (event instanceof StatisticsEvent) {
+            handleDataStatisticRequest(subtask, ((StatisticsEvent) event));
+          } else if (event instanceof RequestGlobalStatisticsEvent) {
+            handleRequestGlobalStatisticsEvent(subtask, (RequestGlobalStatisticsEvent) event);
+          } else {
+            throw new IllegalArgumentException(
+                "Invalid operator event type: " + event.getClass().getCanonicalName());
+          }
         },
         String.format(
             "handling operator event %s from subtask %d (#%d)",
@@ -240,8 +325,8 @@ public void checkpointCoordinator(long checkpointId, CompletableFuture<byte[]> r
               operatorName,
               checkpointId);
           resultFuture.complete(
-              StatisticsUtil.serializeAggregatedStatistics(
-                  completedStatistics, aggregatedStatisticsSerializer));
+              StatisticsUtil.serializeCompletedStatistics(
+                  completedStatistics, completedStatisticsSerializer));
         },
         String.format("taking checkpoint %d", checkpointId));
   }
@@ -253,7 +338,6 @@ public void notifyCheckpointComplete(long checkpointId) {}
   public void resetToCheckpoint(long checkpointId, byte[] checkpointData) {
     Preconditions.checkState(
         !started, "The coordinator %s can only be reset if it was not yet started", operatorName);
-
     if (checkpointData == null) {
       LOG.info(
           "Data statistic coordinator {} has nothing to restore from checkpoint {}",
@@ -265,8 +349,12 @@ public void resetToCheckpoint(long checkpointId, byte[] checkpointData) {
     LOG.info(
         "Restoring data statistic coordinator {} from checkpoint {}", operatorName, checkpointId);
     this.completedStatistics =
-        StatisticsUtil.deserializeAggregatedStatistics(
-            checkpointData, aggregatedStatisticsSerializer);
+        StatisticsUtil.deserializeCompletedStatistics(
+            checkpointData, completedStatisticsSerializer);
+    // recompute global statistics in case downstream parallelism changed
+    this.globalStatistics =
+        globalStatistics(
+            completedStatistics, downstreamParallelism, comparator, closeFileCostWeightPercentage);
   }
 
   @Override
@@ -290,7 +378,7 @@ public void executionAttemptFailed(int subtask, int attemptNumber, @Nullable Thr
     runInCoordinatorThread(
         () -> {
           LOG.info(
-              "Unregistering gateway after failure for subtask {} (#{}) of data statistic {}",
+              "Unregistering gateway after failure for subtask {} (#{}) of data statistics {}",
               subtask,
               attemptNumber,
               operatorName);
@@ -316,10 +404,15 @@ public void executionAttemptReady(int subtask, int attemptNumber, SubtaskGateway
   }
 
   @VisibleForTesting
-  AggregatedStatistics completedStatistics() {
+  CompletedStatistics completedStatistics() {
     return completedStatistics;
   }
 
+  @VisibleForTesting
+  GlobalStatistics globalStatistics() {
+    return globalStatistics;
+  }
+
   private static class SubtaskGateways {
     private final String operatorName;
     private final Map<Integer, SubtaskGateway>[] gateways;
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorProvider.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorProvider.java
index ffb428283785..9d7d989c298e 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorProvider.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorProvider.java
@@ -37,6 +37,7 @@ public class DataStatisticsCoordinatorProvider extends RecreateOnResetOperatorCo
   private final SortOrder sortOrder;
   private final int downstreamParallelism;
   private final StatisticsType type;
+  private final double closeFileCostWeightPercentage;
 
   public DataStatisticsCoordinatorProvider(
       String operatorName,
@@ -44,18 +45,26 @@ public DataStatisticsCoordinatorProvider(
       Schema schema,
       SortOrder sortOrder,
       int downstreamParallelism,
-      StatisticsType type) {
+      StatisticsType type,
+      double closeFileCostWeightPercentage) {
     super(operatorID);
     this.operatorName = operatorName;
     this.schema = schema;
     this.sortOrder = sortOrder;
     this.downstreamParallelism = downstreamParallelism;
     this.type = type;
+    this.closeFileCostWeightPercentage = closeFileCostWeightPercentage;
   }
 
   @Override
   public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) {
     return new DataStatisticsCoordinator(
-        operatorName, context, schema, sortOrder, downstreamParallelism, type);
+        operatorName,
+        context,
+        schema,
+        sortOrder,
+        downstreamParallelism,
+        type,
+        closeFileCostWeightPercentage);
   }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
index 2910471762cd..7995a8a5b181 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
@@ -60,16 +60,16 @@ public class DataStatisticsOperator extends AbstractStreamOperator<StatisticsOrR
   private final int downstreamParallelism;
   private final StatisticsType statisticsType;
   private final TypeSerializer<DataStatistics> taskStatisticsSerializer;
-  private final TypeSerializer<AggregatedStatistics> aggregatedStatisticsSerializer;
+  private final TypeSerializer<GlobalStatistics> globalStatisticsSerializer;
 
   private transient int parallelism;
   private transient int subtaskIndex;
-  private transient ListState<AggregatedStatistics> globalStatisticsState;
+  private transient ListState<GlobalStatistics> globalStatisticsState;
   // current statistics type may be different from the config due to possible
   // migration from Map statistics to Sketch statistics when high cardinality detected
   private transient volatile StatisticsType taskStatisticsType;
   private transient volatile DataStatistics localStatistics;
-  private transient volatile AggregatedStatistics globalStatistics;
+  private transient volatile GlobalStatistics globalStatistics;
 
   DataStatisticsOperator(
       String operatorName,
@@ -87,31 +87,57 @@ public class DataStatisticsOperator extends AbstractStreamOperator<StatisticsOrR
 
     SortKeySerializer sortKeySerializer = new SortKeySerializer(schema, sortOrder);
     this.taskStatisticsSerializer = new DataStatisticsSerializer(sortKeySerializer);
-    this.aggregatedStatisticsSerializer = new AggregatedStatisticsSerializer(sortKeySerializer);
+    this.globalStatisticsSerializer = new GlobalStatisticsSerializer(sortKeySerializer);
   }
 
   @Override
   public void initializeState(StateInitializationContext context) throws Exception {
     this.parallelism = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks();
     this.subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
+
+    // Use union state so that new subtasks can also restore global statistics during scale-up.
     this.globalStatisticsState =
         context
             .getOperatorStateStore()
             .getUnionListState(
-                new ListStateDescriptor<>("globalStatisticsState", aggregatedStatisticsSerializer));
+                new ListStateDescriptor<>("globalStatisticsState", globalStatisticsSerializer));
 
     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 {
+        GlobalStatistics restoredStatistics = globalStatisticsState.get().iterator().next();
         LOG.info(
-            "Operator {} subtask {} restoring global statistics state", operatorName, subtaskIndex);
-        this.globalStatistics = globalStatisticsState.get().iterator().next();
+            "Operator {} subtask {} restored global statistics state", operatorName, subtaskIndex);
+        this.globalStatistics = restoredStatistics;
       }
+
+      // Always request for new statistics from coordinator upon task initialization.
+      // There are a few scenarios this is needed
+      // 1. downstream writer parallelism changed due to rescale.
+      // 2. coordinator failed to send the aggregated statistics to subtask
+      //    (e.g. due to subtask failure at the time).
+      // Records may flow before coordinator can respond. Range partitioner should be
+      // able to continue to operate with potentially suboptimal behavior (in sketch case).
+      LOG.info(
+          "Operator {} subtask {} requests new global statistics from coordinator ",
+          operatorName,
+          subtaskIndex);
+      // coordinator can use the hashCode (if available) in the request event to determine
+      // if operator already has the latest global statistics and respond can be skipped.
+      // This makes the handling cheap in most situations.
+      RequestGlobalStatisticsEvent event =
+          globalStatistics != null
+              ? new RequestGlobalStatisticsEvent(globalStatistics.hashCode())
+              : new RequestGlobalStatisticsEvent();
+      operatorEventGateway.sendEventToCoordinator(event);
     }
 
     this.taskStatisticsType = StatisticsUtil.collectType(statisticsType, globalStatistics);
@@ -139,14 +165,16 @@ public void handleOperatorEvent(OperatorEvent event) {
         operatorName,
         subtaskIndex,
         statisticsEvent.checkpointId());
-    globalStatistics =
-        StatisticsUtil.deserializeAggregatedStatistics(
-            statisticsEvent.statisticsBytes(), aggregatedStatisticsSerializer);
+    this.globalStatistics =
+        StatisticsUtil.deserializeGlobalStatistics(
+            statisticsEvent.statisticsBytes(), globalStatisticsSerializer);
     checkStatisticsTypeMigration();
-    output.collect(new StreamRecord<>(StatisticsOrRecord.fromStatistics(globalStatistics)));
+    // if applyImmediately not set, wait until the checkpoint time to switch
+    if (statisticsEvent.applyImmediately()) {
+      output.collect(new StreamRecord<>(StatisticsOrRecord.fromStatistics(globalStatistics)));
+    }
   }
 
-  @SuppressWarnings("unchecked")
   @Override
   public void processElement(StreamRecord<RowData> streamRecord) {
     // collect data statistics
@@ -204,6 +232,7 @@ && getRuntimeContext().getTaskInfo().getIndexOfThisSubtask() == 0) {
         StatisticsUtil.createTaskStatistics(taskStatisticsType, parallelism, downstreamParallelism);
   }
 
+  @SuppressWarnings("unchecked")
   private void checkStatisticsTypeMigration() {
     // only check if the statisticsType config is Auto and localStatistics is currently Map type
     if (statisticsType == StatisticsType.Auto && localStatistics.type() == StatisticsType.Map) {
@@ -231,7 +260,7 @@ DataStatistics localStatistics() {
   }
 
   @VisibleForTesting
-  AggregatedStatistics globalStatistics() {
+  GlobalStatistics globalStatistics() {
     return globalStatistics;
   }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatistics.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatistics.java
new file mode 100644
index 000000000000..50ec23e9f7a2
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatistics.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import java.util.Arrays;
+import org.apache.iceberg.SortKey;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * This is used by {@link RangePartitioner} for guiding range partitioning. This is what is sent to
+ * the operator subtasks. For sketch statistics, it only contains much smaller range bounds than the
+ * complete raw samples.
+ */
+class GlobalStatistics {
+  private final long checkpointId;
+  private final StatisticsType type;
+  private final MapAssignment mapAssignment;
+  private final SortKey[] rangeBounds;
+
+  private transient Integer hashCode;
+
+  GlobalStatistics(
+      long checkpointId, StatisticsType type, MapAssignment mapAssignment, SortKey[] rangeBounds) {
+    Preconditions.checkArgument(
+        (mapAssignment != null && rangeBounds == null)
+            || (mapAssignment == null && rangeBounds != null),
+        "Invalid key assignment or range bounds: both are non-null or null");
+    this.checkpointId = checkpointId;
+    this.type = type;
+    this.mapAssignment = mapAssignment;
+    this.rangeBounds = rangeBounds;
+  }
+
+  static GlobalStatistics fromMapAssignment(long checkpointId, MapAssignment mapAssignment) {
+    return new GlobalStatistics(checkpointId, StatisticsType.Map, mapAssignment, null);
+  }
+
+  static GlobalStatistics fromRangeBounds(long checkpointId, SortKey[] rangeBounds) {
+    return new GlobalStatistics(checkpointId, StatisticsType.Sketch, null, rangeBounds);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("checkpointId", checkpointId)
+        .add("type", type)
+        .add("mapAssignment", mapAssignment)
+        .add("rangeBounds", rangeBounds)
+        .toString();
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (!(o instanceof GlobalStatistics)) {
+      return false;
+    }
+
+    GlobalStatistics other = (GlobalStatistics) o;
+    return Objects.equal(checkpointId, other.checkpointId)
+        && Objects.equal(type, other.type)
+        && Objects.equal(mapAssignment, other.mapAssignment())
+        && Arrays.equals(rangeBounds, other.rangeBounds());
+  }
+
+  @Override
+  public int hashCode() {
+    // implemented caching because coordinator can call the hashCode many times.
+    // when subtasks request statistics refresh upon initialization for reconciliation purpose,
+    // hashCode is used to check if there is any difference btw coordinator and operator state.
+    if (hashCode == null) {
+      this.hashCode = Objects.hashCode(checkpointId, type, mapAssignment, rangeBounds);
+    }
+
+    return hashCode;
+  }
+
+  long checkpointId() {
+    return checkpointId;
+  }
+
+  StatisticsType type() {
+    return type;
+  }
+
+  MapAssignment mapAssignment() {
+    return mapAssignment;
+  }
+
+  SortKey[] rangeBounds() {
+    return rangeBounds;
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsSerializer.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsSerializer.java
new file mode 100644
index 000000000000..a7fe2b30b865
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsSerializer.java
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.EnumSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.iceberg.SortKey;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+class GlobalStatisticsSerializer extends TypeSerializer<GlobalStatistics> {
+  private final TypeSerializer<SortKey> sortKeySerializer;
+  private final EnumSerializer<StatisticsType> statisticsTypeSerializer;
+  private final ListSerializer<SortKey> rangeBoundsSerializer;
+  private final ListSerializer<Integer> intsSerializer;
+  private final ListSerializer<Long> longsSerializer;
+
+  GlobalStatisticsSerializer(TypeSerializer<SortKey> sortKeySerializer) {
+    this.sortKeySerializer = sortKeySerializer;
+    this.statisticsTypeSerializer = new EnumSerializer<>(StatisticsType.class);
+    this.rangeBoundsSerializer = new ListSerializer<>(sortKeySerializer);
+    this.intsSerializer = new ListSerializer<>(IntSerializer.INSTANCE);
+    this.longsSerializer = new ListSerializer<>(LongSerializer.INSTANCE);
+  }
+
+  @Override
+  public boolean isImmutableType() {
+    return false;
+  }
+
+  @Override
+  public TypeSerializer<GlobalStatistics> duplicate() {
+    return new GlobalStatisticsSerializer(sortKeySerializer);
+  }
+
+  @Override
+  public GlobalStatistics createInstance() {
+    return GlobalStatistics.fromRangeBounds(0L, new SortKey[0]);
+  }
+
+  @Override
+  public GlobalStatistics copy(GlobalStatistics from) {
+    return new GlobalStatistics(
+        from.checkpointId(), from.type(), from.mapAssignment(), from.rangeBounds());
+  }
+
+  @Override
+  public GlobalStatistics copy(GlobalStatistics from, GlobalStatistics reuse) {
+    // no benefit of reuse
+    return copy(from);
+  }
+
+  @Override
+  public int getLength() {
+    return -1;
+  }
+
+  @Override
+  public void serialize(GlobalStatistics record, DataOutputView target) throws IOException {
+    target.writeLong(record.checkpointId());
+    statisticsTypeSerializer.serialize(record.type(), target);
+    if (record.type() == StatisticsType.Map) {
+      MapAssignment mapAssignment = record.mapAssignment();
+      target.writeInt(mapAssignment.numPartitions());
+      target.writeInt(mapAssignment.keyAssignments().size());
+      for (Map.Entry<SortKey, KeyAssignment> entry : mapAssignment.keyAssignments().entrySet()) {
+        sortKeySerializer.serialize(entry.getKey(), target);
+        KeyAssignment keyAssignment = entry.getValue();
+        intsSerializer.serialize(keyAssignment.assignedSubtasks(), target);
+        longsSerializer.serialize(keyAssignment.subtaskWeightsWithCloseFileCost(), target);
+        target.writeLong(keyAssignment.closeFileCostWeight());
+      }
+    } else {
+      rangeBoundsSerializer.serialize(Arrays.asList(record.rangeBounds()), target);
+    }
+  }
+
+  @Override
+  public GlobalStatistics deserialize(DataInputView source) throws IOException {
+    long checkpointId = source.readLong();
+    StatisticsType type = statisticsTypeSerializer.deserialize(source);
+    if (type == StatisticsType.Map) {
+      int numPartitions = source.readInt();
+      int mapSize = source.readInt();
+      Map<SortKey, KeyAssignment> keyAssignments = Maps.newHashMapWithExpectedSize(mapSize);
+      for (int i = 0; i < mapSize; ++i) {
+        SortKey sortKey = sortKeySerializer.deserialize(source);
+        List<Integer> assignedSubtasks = intsSerializer.deserialize(source);
+        List<Long> subtaskWeightsWithCloseFileCost = longsSerializer.deserialize(source);
+        long closeFileCostWeight = source.readLong();
+        keyAssignments.put(
+            sortKey,
+            new KeyAssignment(
+                assignedSubtasks, subtaskWeightsWithCloseFileCost, closeFileCostWeight));
+      }
+
+      return GlobalStatistics.fromMapAssignment(
+          checkpointId, new MapAssignment(numPartitions, keyAssignments));
+    } else {
+      List<SortKey> sortKeys = rangeBoundsSerializer.deserialize(source);
+      SortKey[] rangeBounds = new SortKey[sortKeys.size()];
+      return GlobalStatistics.fromRangeBounds(checkpointId, sortKeys.toArray(rangeBounds));
+    }
+  }
+
+  @Override
+  public GlobalStatistics deserialize(GlobalStatistics reuse, DataInputView source)
+      throws IOException {
+    // not much benefit to reuse
+    return deserialize(source);
+  }
+
+  @Override
+  public void copy(DataInputView source, DataOutputView target) throws IOException {
+    serialize(deserialize(source), target);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) {
+      return true;
+    }
+
+    if (obj == null || getClass() != obj.getClass()) {
+      return false;
+    }
+
+    GlobalStatisticsSerializer other = (GlobalStatisticsSerializer) obj;
+    return Objects.equals(sortKeySerializer, other.sortKeySerializer);
+  }
+
+  @Override
+  public int hashCode() {
+    return sortKeySerializer.hashCode();
+  }
+
+  @Override
+  public TypeSerializerSnapshot<GlobalStatistics> snapshotConfiguration() {
+    return new GlobalStatisticsSerializerSnapshot(this);
+  }
+
+  public static class GlobalStatisticsSerializerSnapshot
+      extends CompositeTypeSerializerSnapshot<GlobalStatistics, GlobalStatisticsSerializer> {
+    private static final int CURRENT_VERSION = 1;
+
+    /** Constructor for read instantiation. */
+    @SuppressWarnings({"unused", "checkstyle:RedundantModifier"})
+    public GlobalStatisticsSerializerSnapshot() {}
+
+    @SuppressWarnings("checkstyle:RedundantModifier")
+    public GlobalStatisticsSerializerSnapshot(GlobalStatisticsSerializer serializer) {
+      super(serializer);
+    }
+
+    @Override
+    protected int getCurrentOuterSnapshotVersion() {
+      return CURRENT_VERSION;
+    }
+
+    @Override
+    protected TypeSerializer<?>[] getNestedSerializers(GlobalStatisticsSerializer outerSerializer) {
+      return new TypeSerializer<?>[] {outerSerializer.sortKeySerializer};
+    }
+
+    @Override
+    protected GlobalStatisticsSerializer createOuterSerializerWithNestedSerializers(
+        TypeSerializer<?>[] nestedSerializers) {
+      SortKeySerializer sortKeySerializer = (SortKeySerializer) nestedSerializers[0];
+      return new GlobalStatisticsSerializer(sortKeySerializer);
+    }
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/KeyAssignment.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/KeyAssignment.java
new file mode 100644
index 000000000000..a164d83ac3b0
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/KeyAssignment.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/** Subtask assignment for a key for Map statistics based */
+class KeyAssignment {
+  private final List<Integer> assignedSubtasks;
+  private final List<Long> subtaskWeightsWithCloseFileCost;
+  private final long closeFileCostWeight;
+  private final long[] subtaskWeightsExcludingCloseCost;
+  private final long keyWeight;
+  private final long[] cumulativeWeights;
+
+  /**
+   * @param assignedSubtasks assigned subtasks for this key. It could be a single subtask. It could
+   *     also be multiple subtasks if the key has heavy weight that should be handled by multiple
+   *     subtasks.
+   * @param subtaskWeightsWithCloseFileCost assigned weight for each subtask. E.g., if the keyWeight
+   *     is 27 and the key is assigned to 3 subtasks, subtaskWeights could contain values as [10,
+   *     10, 7] for target weight of 10 per subtask.
+   */
+  KeyAssignment(
+      List<Integer> assignedSubtasks,
+      List<Long> subtaskWeightsWithCloseFileCost,
+      long closeFileCostWeight) {
+    Preconditions.checkArgument(
+        assignedSubtasks != null && !assignedSubtasks.isEmpty(),
+        "Invalid assigned subtasks: null or empty");
+    Preconditions.checkArgument(
+        subtaskWeightsWithCloseFileCost != null && !subtaskWeightsWithCloseFileCost.isEmpty(),
+        "Invalid assigned subtasks weights: null or empty");
+    Preconditions.checkArgument(
+        assignedSubtasks.size() == subtaskWeightsWithCloseFileCost.size(),
+        "Invalid assignment: size mismatch (tasks length = %s, weights length = %s)",
+        assignedSubtasks.size(),
+        subtaskWeightsWithCloseFileCost.size());
+    subtaskWeightsWithCloseFileCost.forEach(
+        weight ->
+            Preconditions.checkArgument(
+                weight > closeFileCostWeight,
+                "Invalid weight: should be larger than close file cost: weight = %s, close file cost = %s",
+                weight,
+                closeFileCostWeight));
+
+    this.assignedSubtasks = assignedSubtasks;
+    this.subtaskWeightsWithCloseFileCost = subtaskWeightsWithCloseFileCost;
+    this.closeFileCostWeight = closeFileCostWeight;
+    // Exclude the close file cost for key routing
+    this.subtaskWeightsExcludingCloseCost =
+        subtaskWeightsWithCloseFileCost.stream()
+            .mapToLong(weightWithCloseFileCost -> weightWithCloseFileCost - closeFileCostWeight)
+            .toArray();
+    this.keyWeight = Arrays.stream(subtaskWeightsExcludingCloseCost).sum();
+    this.cumulativeWeights = new long[subtaskWeightsExcludingCloseCost.length];
+    long cumulativeWeight = 0;
+    for (int i = 0; i < subtaskWeightsExcludingCloseCost.length; ++i) {
+      cumulativeWeight += subtaskWeightsExcludingCloseCost[i];
+      cumulativeWeights[i] = cumulativeWeight;
+    }
+  }
+
+  List<Integer> assignedSubtasks() {
+    return assignedSubtasks;
+  }
+
+  List<Long> subtaskWeightsWithCloseFileCost() {
+    return subtaskWeightsWithCloseFileCost;
+  }
+
+  long closeFileCostWeight() {
+    return closeFileCostWeight;
+  }
+
+  long[] subtaskWeightsExcludingCloseCost() {
+    return subtaskWeightsExcludingCloseCost;
+  }
+
+  /** @return subtask id */
+  int select() {
+    if (assignedSubtasks.size() == 1) {
+      // only choice. no need to run random number generator.
+      return assignedSubtasks.get(0);
+    } else {
+      long randomNumber = ThreadLocalRandom.current().nextLong(keyWeight);
+      int index = Arrays.binarySearch(cumulativeWeights, randomNumber);
+      // choose the subtask where randomNumber < cumulativeWeights[pos].
+      // this works regardless whether index is negative or not.
+      int position = Math.abs(index + 1);
+      Preconditions.checkState(
+          position < assignedSubtasks.size(),
+          "Invalid selected position: out of range. key weight = %s, random number = %s, cumulative weights array = %s",
+          keyWeight,
+          randomNumber,
+          cumulativeWeights);
+      return assignedSubtasks.get(position);
+    }
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(assignedSubtasks, subtaskWeightsWithCloseFileCost, closeFileCostWeight);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    KeyAssignment that = (KeyAssignment) o;
+    return Objects.equals(assignedSubtasks, that.assignedSubtasks)
+        && Objects.equals(subtaskWeightsWithCloseFileCost, that.subtaskWeightsWithCloseFileCost)
+        && closeFileCostWeight == that.closeFileCostWeight;
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("assignedSubtasks", assignedSubtasks)
+        .add("subtaskWeightsWithCloseFileCost", subtaskWeightsWithCloseFileCost)
+        .add("closeFileCostWeight", closeFileCostWeight)
+        .toString();
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java
new file mode 100644
index 000000000000..0abb030c2279
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import org.apache.iceberg.SortKey;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Key assignment to subtasks for Map statistics. */
+class MapAssignment {
+  private static final Logger LOG = LoggerFactory.getLogger(MapAssignment.class);
+
+  private final int numPartitions;
+  private final Map<SortKey, KeyAssignment> keyAssignments;
+
+  MapAssignment(int numPartitions, Map<SortKey, KeyAssignment> keyAssignments) {
+    Preconditions.checkArgument(keyAssignments != null, "Invalid key assignments: null");
+    this.numPartitions = numPartitions;
+    this.keyAssignments = keyAssignments;
+  }
+
+  static MapAssignment fromKeyFrequency(
+      int numPartitions,
+      Map<SortKey, Long> mapStatistics,
+      double closeFileCostWeightPercentage,
+      Comparator<StructLike> comparator) {
+    return new MapAssignment(
+        numPartitions,
+        assignment(numPartitions, mapStatistics, closeFileCostWeightPercentage, comparator));
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(numPartitions, keyAssignments);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    MapAssignment that = (MapAssignment) o;
+    return numPartitions == that.numPartitions && keyAssignments.equals(that.keyAssignments);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("numPartitions", numPartitions)
+        .add("keyAssignments", keyAssignments)
+        .toString();
+  }
+
+  int numPartitions() {
+    return numPartitions;
+  }
+
+  Map<SortKey, KeyAssignment> keyAssignments() {
+    return keyAssignments;
+  }
+
+  /**
+   * @return assignment summary for every subtask. Key is subtaskId. Value pair is (weight assigned
+   *     to the subtask, number of keys assigned to the subtask)
+   */
+  Map<Integer, Pair<Long, Integer>> assignmentInfo() {
+    Map<Integer, Pair<Long, Integer>> assignmentInfo = Maps.newTreeMap();
+    keyAssignments.forEach(
+        (key, keyAssignment) -> {
+          for (int i = 0; i < keyAssignment.assignedSubtasks().size(); ++i) {
+            int subtaskId = keyAssignment.assignedSubtasks().get(i);
+            long subtaskWeight = keyAssignment.subtaskWeightsExcludingCloseCost()[i];
+            Pair<Long, Integer> oldValue = assignmentInfo.getOrDefault(subtaskId, Pair.of(0L, 0));
+            assignmentInfo.put(
+                subtaskId, Pair.of(oldValue.first() + subtaskWeight, oldValue.second() + 1));
+          }
+        });
+
+    return assignmentInfo;
+  }
+
+  static Map<SortKey, KeyAssignment> assignment(
+      int numPartitions,
+      Map<SortKey, Long> mapStatistics,
+      double closeFileCostWeightPercentage,
+      Comparator<StructLike> comparator) {
+    mapStatistics.forEach(
+        (key, value) ->
+            Preconditions.checkArgument(
+                value > 0, "Invalid statistics: weight is 0 for key %s", key));
+
+    long totalWeight = mapStatistics.values().stream().mapToLong(l -> l).sum();
+    double targetWeightPerSubtask = ((double) totalWeight) / numPartitions;
+    long closeFileCostWeight =
+        (long) Math.ceil(targetWeightPerSubtask * closeFileCostWeightPercentage / 100);
+
+    NavigableMap<SortKey, Long> sortedStatsWithCloseFileCost = Maps.newTreeMap(comparator);
+    mapStatistics.forEach(
+        (k, v) -> {
+          int estimatedSplits = (int) Math.ceil(v / targetWeightPerSubtask);
+          long estimatedCloseFileCost = closeFileCostWeight * estimatedSplits;
+          sortedStatsWithCloseFileCost.put(k, v + estimatedCloseFileCost);
+        });
+
+    long totalWeightWithCloseFileCost =
+        sortedStatsWithCloseFileCost.values().stream().mapToLong(l -> l).sum();
+    long targetWeightPerSubtaskWithCloseFileCost =
+        (long) Math.ceil(((double) totalWeightWithCloseFileCost) / numPartitions);
+    return buildAssignment(
+        numPartitions,
+        sortedStatsWithCloseFileCost,
+        targetWeightPerSubtaskWithCloseFileCost,
+        closeFileCostWeight);
+  }
+
+  private static Map<SortKey, KeyAssignment> buildAssignment(
+      int numPartitions,
+      NavigableMap<SortKey, Long> sortedStatistics,
+      long targetWeightPerSubtask,
+      long closeFileCostWeight) {
+    Map<SortKey, KeyAssignment> assignmentMap =
+        Maps.newHashMapWithExpectedSize(sortedStatistics.size());
+    Iterator<SortKey> mapKeyIterator = sortedStatistics.keySet().iterator();
+    int subtaskId = 0;
+    SortKey currentKey = null;
+    long keyRemainingWeight = 0L;
+    long subtaskRemainingWeight = targetWeightPerSubtask;
+    List<Integer> assignedSubtasks = Lists.newArrayList();
+    List<Long> subtaskWeights = Lists.newArrayList();
+    while (mapKeyIterator.hasNext() || currentKey != null) {
+      // This should never happen because target weight is calculated using ceil function.
+      if (subtaskId >= numPartitions) {
+        LOG.error(
+            "Internal algorithm error: exhausted subtasks with unassigned keys left. number of partitions: {}, "
+                + "target weight per subtask: {}, close file cost in weight: {}, data statistics: {}",
+            numPartitions,
+            targetWeightPerSubtask,
+            closeFileCostWeight,
+            sortedStatistics);
+        throw new IllegalStateException(
+            "Internal algorithm error: exhausted subtasks with unassigned keys left");
+      }
+
+      if (currentKey == null) {
+        currentKey = mapKeyIterator.next();
+        keyRemainingWeight = sortedStatistics.get(currentKey);
+      }
+
+      assignedSubtasks.add(subtaskId);
+      if (keyRemainingWeight < subtaskRemainingWeight) {
+        // assign the remaining weight of the key to the current subtask
+        subtaskWeights.add(keyRemainingWeight);
+        subtaskRemainingWeight -= keyRemainingWeight;
+        keyRemainingWeight = 0L;
+      } else {
+        // filled up the current subtask
+        long assignedWeight = subtaskRemainingWeight;
+        keyRemainingWeight -= subtaskRemainingWeight;
+
+        // If assigned weight is less than close file cost, pad it up with close file cost.
+        // This might cause the subtask assigned weight over the target weight.
+        // But it should be no more than one close file cost. Small skew is acceptable.
+        if (assignedWeight <= closeFileCostWeight) {
+          long paddingWeight = Math.min(keyRemainingWeight, closeFileCostWeight);
+          keyRemainingWeight -= paddingWeight;
+          assignedWeight += paddingWeight;
+        }
+
+        subtaskWeights.add(assignedWeight);
+        // move on to the next subtask
+        subtaskId += 1;
+        subtaskRemainingWeight = targetWeightPerSubtask;
+      }
+
+      Preconditions.checkState(
+          assignedSubtasks.size() == subtaskWeights.size(),
+          "List size mismatch: assigned subtasks = %s, subtask weights = %s",
+          assignedSubtasks,
+          subtaskWeights);
+
+      // If the remaining key weight is smaller than the close file cost, simply skip the residual
+      // as it doesn't make sense to assign a weight smaller than close file cost to a new subtask.
+      // this might lead to some inaccuracy in weight calculation. E.g., assuming the key weight is
+      // 2 and close file cost is 2. key weight with close cost is 4. Let's assume the previous
+      // task has a weight of 3 available. So weight of 3 for this key is assigned to the task and
+      // the residual weight of 1 is dropped. Then the routing weight for this key is 1 (minus the
+      // close file cost), which is inaccurate as the true key weight should be 2.
+      // Again, this greedy algorithm is not intended to be perfect. Some small inaccuracy is
+      // expected and acceptable. Traffic distribution should still be balanced.
+      if (keyRemainingWeight > 0 && keyRemainingWeight <= closeFileCostWeight) {
+        keyRemainingWeight = 0;
+      }
+
+      if (keyRemainingWeight == 0) {
+        // finishing up the assignment for the current key
+        KeyAssignment keyAssignment =
+            new KeyAssignment(assignedSubtasks, subtaskWeights, closeFileCostWeight);
+        assignmentMap.put(currentKey, keyAssignment);
+        assignedSubtasks = Lists.newArrayList();
+        subtaskWeights = Lists.newArrayList();
+        currentKey = null;
+      }
+    }
+
+    return assignmentMap;
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapRangePartitioner.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapRangePartitioner.java
index 298426cee872..f36a078c94e0 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapRangePartitioner.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapRangePartitioner.java
@@ -18,29 +18,14 @@
  */
 package org.apache.iceberg.flink.sink.shuffle;
 
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import org.apache.flink.api.common.functions.Partitioner;
 import org.apache.flink.table.data.RowData;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortKey;
 import org.apache.iceberg.SortOrder;
-import org.apache.iceberg.SortOrderComparators;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
 import org.apache.iceberg.flink.RowDataWrapper;
-import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
-import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.apache.iceberg.util.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -61,47 +46,28 @@ class MapRangePartitioner implements Partitioner<RowData> {
 
   private final RowDataWrapper rowDataWrapper;
   private final SortKey sortKey;
-  private final Comparator<StructLike> comparator;
-  private final Map<SortKey, Long> mapStatistics;
-  private final double closeFileCostInWeightPercentage;
+  private final MapAssignment mapAssignment;
 
   // Counter that tracks how many times a new key encountered
   // where there is no traffic statistics learned about it.
   private long newSortKeyCounter;
   private long lastNewSortKeyLogTimeMilli;
 
-  // lazily computed due to the need of numPartitions
-  private Map<SortKey, KeyAssignment> assignment;
-  private NavigableMap<SortKey, Long> sortedStatsWithCloseFileCost;
-
-  MapRangePartitioner(
-      Schema schema,
-      SortOrder sortOrder,
-      Map<SortKey, Long> mapStatistics,
-      double closeFileCostInWeightPercentage) {
-    mapStatistics.forEach(
-        (key, value) ->
-            Preconditions.checkArgument(
-                value > 0, "Invalid statistics: weight is 0 for key %s", key));
-
+  MapRangePartitioner(Schema schema, SortOrder sortOrder, MapAssignment mapAssignment) {
     this.rowDataWrapper = new RowDataWrapper(FlinkSchemaUtil.convert(schema), schema.asStruct());
     this.sortKey = new SortKey(schema, sortOrder);
-    this.comparator = SortOrderComparators.forSchema(schema, sortOrder);
-    this.mapStatistics = mapStatistics;
-    this.closeFileCostInWeightPercentage = closeFileCostInWeightPercentage;
+    this.mapAssignment = mapAssignment;
     this.newSortKeyCounter = 0;
     this.lastNewSortKeyLogTimeMilli = System.currentTimeMillis();
   }
 
   @Override
   public int partition(RowData row, int numPartitions) {
-    // assignment table can only be built lazily when first referenced here,
-    // because number of partitions (downstream subtasks) is needed.
-    // the numPartitions is not available in the constructor.
-    Map<SortKey, KeyAssignment> assignmentMap = assignment(numPartitions);
     // reuse the sortKey and rowDataWrapper
     sortKey.wrap(rowDataWrapper.wrap(row));
-    KeyAssignment keyAssignment = assignmentMap.get(sortKey);
+    KeyAssignment keyAssignment = mapAssignment.keyAssignments().get(sortKey);
+
+    int partition;
     if (keyAssignment == null) {
       LOG.trace(
           "Encountered new sort key: {}. Fall back to round robin as statistics not learned yet.",
@@ -118,268 +84,12 @@ public int partition(RowData row, int numPartitions) {
         lastNewSortKeyLogTimeMilli = now;
         newSortKeyCounter = 0;
       }
-      return (int) (newSortKeyCounter % numPartitions);
+      partition = (int) (newSortKeyCounter % numPartitions);
+    } else {
+      partition = keyAssignment.select();
     }
 
-    return keyAssignment.select();
-  }
-
-  @VisibleForTesting
-  Map<SortKey, KeyAssignment> assignment(int numPartitions) {
-    if (assignment == null) {
-      long totalWeight = mapStatistics.values().stream().mapToLong(l -> l).sum();
-      double targetWeightPerSubtask = ((double) totalWeight) / numPartitions;
-      long closeFileCostInWeight =
-          (long) Math.ceil(targetWeightPerSubtask * closeFileCostInWeightPercentage / 100);
-
-      this.sortedStatsWithCloseFileCost = Maps.newTreeMap(comparator);
-      mapStatistics.forEach(
-          (k, v) -> {
-            int estimatedSplits = (int) Math.ceil(v / targetWeightPerSubtask);
-            long estimatedCloseFileCost = closeFileCostInWeight * estimatedSplits;
-            sortedStatsWithCloseFileCost.put(k, v + estimatedCloseFileCost);
-          });
-
-      long totalWeightWithCloseFileCost =
-          sortedStatsWithCloseFileCost.values().stream().mapToLong(l -> l).sum();
-      long targetWeightPerSubtaskWithCloseFileCost =
-          (long) Math.ceil(((double) totalWeightWithCloseFileCost) / numPartitions);
-      this.assignment =
-          buildAssignment(
-              numPartitions,
-              sortedStatsWithCloseFileCost,
-              targetWeightPerSubtaskWithCloseFileCost,
-              closeFileCostInWeight);
-    }
-
-    return assignment;
-  }
-
-  @VisibleForTesting
-  Map<SortKey, Long> mapStatistics() {
-    return mapStatistics;
-  }
-
-  /**
-   * Returns assignment summary for every subtask.
-   *
-   * @return assignment summary for every subtask. Key is subtaskId. Value pair is (weight assigned
-   *     to the subtask, number of keys assigned to the subtask)
-   */
-  Map<Integer, Pair<Long, Integer>> assignmentInfo() {
-    Map<Integer, Pair<Long, Integer>> assignmentInfo = Maps.newTreeMap();
-    assignment.forEach(
-        (key, keyAssignment) -> {
-          for (int i = 0; i < keyAssignment.assignedSubtasks.length; ++i) {
-            int subtaskId = keyAssignment.assignedSubtasks[i];
-            long subtaskWeight = keyAssignment.subtaskWeightsExcludingCloseCost[i];
-            Pair<Long, Integer> oldValue = assignmentInfo.getOrDefault(subtaskId, Pair.of(0L, 0));
-            assignmentInfo.put(
-                subtaskId, Pair.of(oldValue.first() + subtaskWeight, oldValue.second() + 1));
-          }
-        });
-
-    return assignmentInfo;
-  }
-
-  private Map<SortKey, KeyAssignment> buildAssignment(
-      int numPartitions,
-      NavigableMap<SortKey, Long> sortedStatistics,
-      long targetWeightPerSubtask,
-      long closeFileCostInWeight) {
-    Map<SortKey, KeyAssignment> assignmentMap =
-        Maps.newHashMapWithExpectedSize(sortedStatistics.size());
-    Iterator<SortKey> mapKeyIterator = sortedStatistics.keySet().iterator();
-    int subtaskId = 0;
-    SortKey currentKey = null;
-    long keyRemainingWeight = 0L;
-    long subtaskRemainingWeight = targetWeightPerSubtask;
-    List<Integer> assignedSubtasks = Lists.newArrayList();
-    List<Long> subtaskWeights = Lists.newArrayList();
-    while (mapKeyIterator.hasNext() || currentKey != null) {
-      // This should never happen because target weight is calculated using ceil function.
-      if (subtaskId >= numPartitions) {
-        LOG.error(
-            "Internal algorithm error: exhausted subtasks with unassigned keys left. number of partitions: {}, "
-                + "target weight per subtask: {}, close file cost in weight: {}, data statistics: {}",
-            numPartitions,
-            targetWeightPerSubtask,
-            closeFileCostInWeight,
-            sortedStatistics);
-        throw new IllegalStateException(
-            "Internal algorithm error: exhausted subtasks with unassigned keys left");
-      }
-
-      if (currentKey == null) {
-        currentKey = mapKeyIterator.next();
-        keyRemainingWeight = sortedStatistics.get(currentKey);
-      }
-
-      assignedSubtasks.add(subtaskId);
-      if (keyRemainingWeight < subtaskRemainingWeight) {
-        // assign the remaining weight of the key to the current subtask
-        subtaskWeights.add(keyRemainingWeight);
-        subtaskRemainingWeight -= keyRemainingWeight;
-        keyRemainingWeight = 0L;
-      } else {
-        // filled up the current subtask
-        long assignedWeight = subtaskRemainingWeight;
-        keyRemainingWeight -= subtaskRemainingWeight;
-
-        // If assigned weight is less than close file cost, pad it up with close file cost.
-        // This might cause the subtask assigned weight over the target weight.
-        // But it should be no more than one close file cost. Small skew is acceptable.
-        if (assignedWeight <= closeFileCostInWeight) {
-          long paddingWeight = Math.min(keyRemainingWeight, closeFileCostInWeight);
-          keyRemainingWeight -= paddingWeight;
-          assignedWeight += paddingWeight;
-        }
-
-        subtaskWeights.add(assignedWeight);
-        // move on to the next subtask
-        subtaskId += 1;
-        subtaskRemainingWeight = targetWeightPerSubtask;
-      }
-
-      Preconditions.checkState(
-          assignedSubtasks.size() == subtaskWeights.size(),
-          "List size mismatch: assigned subtasks = %s, subtask weights = %s",
-          assignedSubtasks,
-          subtaskWeights);
-
-      // If the remaining key weight is smaller than the close file cost, simply skip the residual
-      // as it doesn't make sense to assign a weight smaller than close file cost to a new subtask.
-      // this might lead to some inaccuracy in weight calculation. E.g., assuming the key weight is
-      // 2 and close file cost is 2. key weight with close cost is 4. Let's assume the previous
-      // task has a weight of 3 available. So weight of 3 for this key is assigned to the task and
-      // the residual weight of 1 is dropped. Then the routing weight for this key is 1 (minus the
-      // close file cost), which is inaccurate as the true key weight should be 2.
-      // Again, this greedy algorithm is not intended to be perfect. Some small inaccuracy is
-      // expected and acceptable. Traffic distribution should still be balanced.
-      if (keyRemainingWeight > 0 && keyRemainingWeight <= closeFileCostInWeight) {
-        keyRemainingWeight = 0;
-      }
-
-      if (keyRemainingWeight == 0) {
-        // finishing up the assignment for the current key
-        KeyAssignment keyAssignment =
-            new KeyAssignment(assignedSubtasks, subtaskWeights, closeFileCostInWeight);
-        assignmentMap.put(currentKey, keyAssignment);
-        assignedSubtasks.clear();
-        subtaskWeights.clear();
-        currentKey = null;
-      }
-    }
-
-    return assignmentMap;
-  }
-
-  /** Subtask assignment for a key */
-  @VisibleForTesting
-  static class KeyAssignment {
-    private final int[] assignedSubtasks;
-    private final long[] subtaskWeightsExcludingCloseCost;
-    private final long keyWeight;
-    private final long[] cumulativeWeights;
-
-    /**
-     * @param assignedSubtasks assigned subtasks for this key. It could be a single subtask. It
-     *     could also be multiple subtasks if the key has heavy weight that should be handled by
-     *     multiple subtasks.
-     * @param subtaskWeightsWithCloseFileCost assigned weight for each subtask. E.g., if the
-     *     keyWeight is 27 and the key is assigned to 3 subtasks, subtaskWeights could contain
-     *     values as [10, 10, 7] for target weight of 10 per subtask.
-     */
-    KeyAssignment(
-        List<Integer> assignedSubtasks,
-        List<Long> subtaskWeightsWithCloseFileCost,
-        long closeFileCostInWeight) {
-      Preconditions.checkArgument(
-          assignedSubtasks != null && !assignedSubtasks.isEmpty(),
-          "Invalid assigned subtasks: null or empty");
-      Preconditions.checkArgument(
-          subtaskWeightsWithCloseFileCost != null && !subtaskWeightsWithCloseFileCost.isEmpty(),
-          "Invalid assigned subtasks weights: null or empty");
-      Preconditions.checkArgument(
-          assignedSubtasks.size() == subtaskWeightsWithCloseFileCost.size(),
-          "Invalid assignment: size mismatch (tasks length = %s, weights length = %s)",
-          assignedSubtasks.size(),
-          subtaskWeightsWithCloseFileCost.size());
-      subtaskWeightsWithCloseFileCost.forEach(
-          weight ->
-              Preconditions.checkArgument(
-                  weight > closeFileCostInWeight,
-                  "Invalid weight: should be larger than close file cost: weight = %s, close file cost = %s",
-                  weight,
-                  closeFileCostInWeight));
-
-      this.assignedSubtasks = assignedSubtasks.stream().mapToInt(i -> i).toArray();
-      // Exclude the close file cost for key routing
-      this.subtaskWeightsExcludingCloseCost =
-          subtaskWeightsWithCloseFileCost.stream()
-              .mapToLong(weightWithCloseFileCost -> weightWithCloseFileCost - closeFileCostInWeight)
-              .toArray();
-      this.keyWeight = Arrays.stream(subtaskWeightsExcludingCloseCost).sum();
-      this.cumulativeWeights = new long[subtaskWeightsExcludingCloseCost.length];
-      long cumulativeWeight = 0;
-      for (int i = 0; i < subtaskWeightsExcludingCloseCost.length; ++i) {
-        cumulativeWeight += subtaskWeightsExcludingCloseCost[i];
-        cumulativeWeights[i] = cumulativeWeight;
-      }
-    }
-
-    /**
-     * Select a subtask for the key.
-     *
-     * @return subtask id
-     */
-    int select() {
-      if (assignedSubtasks.length == 1) {
-        // only choice. no need to run random number generator.
-        return assignedSubtasks[0];
-      } else {
-        long randomNumber = ThreadLocalRandom.current().nextLong(keyWeight);
-        int index = Arrays.binarySearch(cumulativeWeights, randomNumber);
-        // choose the subtask where randomNumber < cumulativeWeights[pos].
-        // this works regardless whether index is negative or not.
-        int position = Math.abs(index + 1);
-        Preconditions.checkState(
-            position < assignedSubtasks.length,
-            "Invalid selected position: out of range. key weight = %s, random number = %s, cumulative weights array = %s",
-            keyWeight,
-            randomNumber,
-            cumulativeWeights);
-        return assignedSubtasks[position];
-      }
-    }
-
-    @Override
-    public int hashCode() {
-      return 31 * Arrays.hashCode(assignedSubtasks)
-          + Arrays.hashCode(subtaskWeightsExcludingCloseCost);
-    }
-
-    @Override
-    public boolean equals(Object o) {
-      if (this == o) {
-        return true;
-      }
-
-      if (o == null || getClass() != o.getClass()) {
-        return false;
-      }
-
-      KeyAssignment that = (KeyAssignment) o;
-      return Arrays.equals(assignedSubtasks, that.assignedSubtasks)
-          && Arrays.equals(subtaskWeightsExcludingCloseCost, that.subtaskWeightsExcludingCloseCost);
-    }
-
-    @Override
-    public String toString() {
-      return MoreObjects.toStringHelper(this)
-          .add("assignedSubtasks", assignedSubtasks)
-          .add("subtaskWeightsExcludingCloseCost", subtaskWeightsExcludingCloseCost)
-          .toString();
-    }
+    return RangePartitioner.adjustPartitionWithRescale(
+        partition, mapAssignment.numPartitions(), numPartitions);
   }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RangePartitioner.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RangePartitioner.java
new file mode 100644
index 000000000000..83a9461233d2
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RangePartitioner.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The wrapper class */
+@Internal
+public class RangePartitioner implements Partitioner<StatisticsOrRecord> {
+  private static final Logger LOG = LoggerFactory.getLogger(RangePartitioner.class);
+
+  private final Schema schema;
+  private final SortOrder sortOrder;
+
+  private transient AtomicLong roundRobinCounter;
+  private transient Partitioner<RowData> delegatePartitioner;
+
+  public RangePartitioner(Schema schema, SortOrder sortOrder) {
+    this.schema = schema;
+    this.sortOrder = sortOrder;
+  }
+
+  @Override
+  public int partition(StatisticsOrRecord wrapper, int numPartitions) {
+    if (wrapper.hasStatistics()) {
+      this.delegatePartitioner = delegatePartitioner(wrapper.statistics());
+      return (int) (roundRobinCounter(numPartitions).getAndIncrement() % numPartitions);
+    } else {
+      if (delegatePartitioner != null) {
+        return delegatePartitioner.partition(wrapper.record(), numPartitions);
+      } else {
+        int partition = (int) (roundRobinCounter(numPartitions).getAndIncrement() % numPartitions);
+        LOG.trace("Statistics not available. Round robin to partition {}", partition);
+        return partition;
+      }
+    }
+  }
+
+  private AtomicLong roundRobinCounter(int numPartitions) {
+    if (roundRobinCounter == null) {
+      // randomize the starting point to avoid synchronization across subtasks
+      this.roundRobinCounter = new AtomicLong(new Random().nextInt(numPartitions));
+    }
+
+    return roundRobinCounter;
+  }
+
+  private Partitioner<RowData> delegatePartitioner(GlobalStatistics statistics) {
+    if (statistics.type() == StatisticsType.Map) {
+      return new MapRangePartitioner(schema, sortOrder, statistics.mapAssignment());
+    } else if (statistics.type() == StatisticsType.Sketch) {
+      return new SketchRangePartitioner(schema, sortOrder, statistics.rangeBounds());
+    } else {
+      throw new IllegalArgumentException(
+          String.format("Invalid statistics type: %s. Should be Map or Sketch", statistics.type()));
+    }
+  }
+
+  /**
+   * Util method that handles rescale (write parallelism / numPartitions change).
+   *
+   * @param partition partition caculated based on the existing statistics
+   * @param numPartitionsStatsCalculation number of partitions when the assignment was calculated
+   *     based on
+   * @param numPartitions current number of partitions
+   * @return adjusted partition if necessary.
+   */
+  static int adjustPartitionWithRescale(
+      int partition, int numPartitionsStatsCalculation, int numPartitions) {
+    if (numPartitionsStatsCalculation <= numPartitions) {
+      // no rescale or scale-up case.
+      // new subtasks are ignored and not assigned any keys, which is sub-optimal and only
+      // transient.
+      // when rescale is detected, operator requests new statistics from coordinator upon
+      // initialization.
+      return partition;
+    } else {
+      // scale-down case.
+      // Use mod % operation to distribution the over-range partitions.
+      // It can cause skew among subtasks. but the behavior is still better than
+      // discarding the statistics and falling back to round-robin (no clustering).
+      // Again, this is transient and stats refresh is requested when rescale is detected.
+      return partition % numPartitions;
+    }
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RequestGlobalStatisticsEvent.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RequestGlobalStatisticsEvent.java
new file mode 100644
index 000000000000..ce17e1964392
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/RequestGlobalStatisticsEvent.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+
+class RequestGlobalStatisticsEvent implements OperatorEvent {
+  private final Integer signature;
+
+  RequestGlobalStatisticsEvent() {
+    this.signature = null;
+  }
+
+  /** @param signature hashCode of the subtask's existing global statistics */
+  RequestGlobalStatisticsEvent(int signature) {
+    this.signature = signature;
+  }
+
+  Integer signature() {
+    return signature;
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchRangePartitioner.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchRangePartitioner.java
new file mode 100644
index 000000000000..af78271ea5dc
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchRangePartitioner.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortKey;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.SortOrderComparators;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.RowDataWrapper;
+
+class SketchRangePartitioner implements Partitioner<RowData> {
+  private final SortKey sortKey;
+  private final Comparator<StructLike> comparator;
+  private final SortKey[] rangeBounds;
+  private final RowDataWrapper rowDataWrapper;
+
+  SketchRangePartitioner(Schema schema, SortOrder sortOrder, SortKey[] rangeBounds) {
+    this.sortKey = new SortKey(schema, sortOrder);
+    this.comparator = SortOrderComparators.forSchema(schema, sortOrder);
+    this.rangeBounds = rangeBounds;
+    this.rowDataWrapper = new RowDataWrapper(FlinkSchemaUtil.convert(schema), schema.asStruct());
+  }
+
+  @Override
+  public int partition(RowData row, int numPartitions) {
+    // reuse the sortKey and rowDataWrapper
+    sortKey.wrap(rowDataWrapper.wrap(row));
+    int partition = Arrays.binarySearch(rangeBounds, sortKey, comparator);
+
+    // binarySearch either returns the match location or -[insertion point]-1
+    if (partition < 0) {
+      partition = -partition - 1;
+    }
+
+    if (partition > rangeBounds.length) {
+      partition = rangeBounds.length;
+    }
+
+    return RangePartitioner.adjustPartitionWithRescale(
+        partition, rangeBounds.length + 1, numPartitions);
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java
index bfd3082a5aa7..a58310611e8d 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java
@@ -22,7 +22,6 @@
 import java.util.Comparator;
 import java.util.Map;
 import java.util.function.Consumer;
-import org.apache.datasketches.sampling.ReservoirItemsSketch;
 import org.apache.iceberg.SortKey;
 import org.apache.iceberg.StructLike;
 
@@ -90,37 +89,32 @@ static int determineOperatorReservoirSize(int operatorParallelism, int numPartit
   /**
    * To understand how range bounds are used in range partitioning, here is an example for human
    * ages with 4 partitions: [15, 32, 60]. The 4 ranges would be
-   * <li>age <= 15
-   * <li>age > 15 && age <= 32
-   * <li>age >32 && age <= 60
-   * <li>age > 60
+   *
+   * <ul>
+   *   <li>age <= 15
+   *   <li>age > 15 && age <= 32
+   *   <li>age >32 && age <= 60
+   *   <li>age > 60
+   * </ul>
+   *
+   * <p>Assumption is that a single key is not dominant enough to span multiple subtasks.
    *
    * @param numPartitions number of partitions which maps to downstream operator parallelism
-   * @param sketch aggregated reservoir sampling sketch
-   * @return list of range partition bounds. It should be a sorted list (ascending). Number of items
-   *     should be {@code numPartitions - 1}. if numPartitions is 1, return an empty list
+   * @param samples sampled keys
+   * @return array of range partition bounds. It should be a sorted list (ascending). Number of
+   *     items should be {@code numPartitions - 1}. if numPartitions is 1, return an empty list
    */
   static SortKey[] rangeBounds(
-      int numPartitions, Comparator<StructLike> comparator, ReservoirItemsSketch<SortKey> sketch) {
-    SortKey[] sortKeys = sketch.getSamples();
-    return determineBounds(Math.min(numPartitions, sortKeys.length), comparator, sortKeys);
-  }
-
-  /**
-   * This assumes the sort keys have equal weight, which is usually the case for high-cardinality
-   * scenarios (like device_id, user_id, uuid etc.).
-   */
-  static SortKey[] determineBounds(
-      int numPartitions, Comparator<StructLike> comparator, SortKey[] sortKeys) {
+      int numPartitions, Comparator<StructLike> comparator, SortKey[] samples) {
     // sort the keys first
-    Arrays.sort(sortKeys, comparator);
+    Arrays.sort(samples, comparator);
     int numCandidates = numPartitions - 1;
     SortKey[] candidates = new SortKey[numCandidates];
-    int step = (int) Math.ceil((double) sortKeys.length / numPartitions);
+    int step = (int) Math.ceil((double) samples.length / numPartitions);
     int position = step - 1;
     int numChosen = 0;
-    while (position < sortKeys.length && numChosen < numCandidates) {
-      SortKey candidate = sortKeys[position];
+    while (position < samples.length && numChosen < numCandidates) {
+      SortKey candidate = samples[position];
       // skip duplicate values
       if (numChosen > 0 && candidate.equals(candidates[numChosen - 1])) {
         // linear probe for the next distinct value
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsEvent.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsEvent.java
index f3391d72297f..f6fcdb8b16ef 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsEvent.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsEvent.java
@@ -32,27 +32,34 @@ class StatisticsEvent implements OperatorEvent {
   private static final long serialVersionUID = 1L;
   private final long checkpointId;
   private final byte[] statisticsBytes;
+  private final boolean applyImmediately;
 
-  private StatisticsEvent(long checkpointId, byte[] statisticsBytes) {
+  private StatisticsEvent(long checkpointId, byte[] statisticsBytes, boolean applyImmediately) {
     this.checkpointId = checkpointId;
     this.statisticsBytes = statisticsBytes;
+    this.applyImmediately = applyImmediately;
   }
 
   static StatisticsEvent createTaskStatisticsEvent(
       long checkpointId,
       DataStatistics statistics,
       TypeSerializer<DataStatistics> statisticsSerializer) {
+    // applyImmediately is really only relevant for coordinator to operator event.
+    // task reported statistics is always merged immediately by the coordinator.
     return new StatisticsEvent(
-        checkpointId, StatisticsUtil.serializeDataStatistics(statistics, statisticsSerializer));
+        checkpointId,
+        StatisticsUtil.serializeDataStatistics(statistics, statisticsSerializer),
+        true);
   }
 
-  static StatisticsEvent createAggregatedStatisticsEvent(
-      long checkpointId,
-      AggregatedStatistics statistics,
-      TypeSerializer<AggregatedStatistics> statisticsSerializer) {
+  static StatisticsEvent createGlobalStatisticsEvent(
+      GlobalStatistics statistics,
+      TypeSerializer<GlobalStatistics> statisticsSerializer,
+      boolean applyImmediately) {
     return new StatisticsEvent(
-        checkpointId,
-        StatisticsUtil.serializeAggregatedStatistics(statistics, statisticsSerializer));
+        statistics.checkpointId(),
+        StatisticsUtil.serializeGlobalStatistics(statistics, statisticsSerializer),
+        applyImmediately);
   }
 
   long checkpointId() {
@@ -62,4 +69,8 @@ long checkpointId() {
   byte[] statisticsBytes() {
     return statisticsBytes;
   }
+
+  boolean applyImmediately() {
+    return applyImmediately;
+  }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecord.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecord.java
index c251ba1360fc..bc28df2b0e22 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecord.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecord.java
@@ -40,10 +40,10 @@ public class StatisticsOrRecord implements Serializable {
 
   private static final long serialVersionUID = 1L;
 
-  private AggregatedStatistics statistics;
+  private GlobalStatistics statistics;
   private RowData record;
 
-  private StatisticsOrRecord(AggregatedStatistics statistics, RowData record) {
+  private StatisticsOrRecord(GlobalStatistics statistics, RowData record) {
     Preconditions.checkArgument(
         record != null ^ statistics != null, "DataStatistics or record, not neither or both");
     this.statistics = statistics;
@@ -54,7 +54,7 @@ static StatisticsOrRecord fromRecord(RowData record) {
     return new StatisticsOrRecord(null, record);
   }
 
-  static StatisticsOrRecord fromStatistics(AggregatedStatistics statistics) {
+  static StatisticsOrRecord fromStatistics(GlobalStatistics statistics) {
     return new StatisticsOrRecord(statistics, null);
   }
 
@@ -69,7 +69,7 @@ static StatisticsOrRecord reuseRecord(
   }
 
   static StatisticsOrRecord reuseStatistics(
-      StatisticsOrRecord reuse, TypeSerializer<AggregatedStatistics> statisticsSerializer) {
+      StatisticsOrRecord reuse, TypeSerializer<GlobalStatistics> statisticsSerializer) {
     if (reuse.hasStatistics()) {
       return reuse;
     } else {
@@ -86,11 +86,11 @@ public boolean hasRecord() {
     return record != null;
   }
 
-  AggregatedStatistics statistics() {
+  GlobalStatistics statistics() {
     return statistics;
   }
 
-  void statistics(AggregatedStatistics newStatistics) {
+  void statistics(GlobalStatistics newStatistics) {
     this.statistics = newStatistics;
   }
 
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecordSerializer.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecordSerializer.java
index 7e690f61a58b..d4ae2b359679 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecordSerializer.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsOrRecordSerializer.java
@@ -30,11 +30,11 @@
 
 @Internal
 class StatisticsOrRecordSerializer extends TypeSerializer<StatisticsOrRecord> {
-  private final TypeSerializer<AggregatedStatistics> statisticsSerializer;
+  private final TypeSerializer<GlobalStatistics> statisticsSerializer;
   private final TypeSerializer<RowData> recordSerializer;
 
   StatisticsOrRecordSerializer(
-      TypeSerializer<AggregatedStatistics> statisticsSerializer,
+      TypeSerializer<GlobalStatistics> statisticsSerializer,
       TypeSerializer<RowData> recordSerializer) {
     this.statisticsSerializer = statisticsSerializer;
     this.recordSerializer = recordSerializer;
@@ -48,7 +48,7 @@ public boolean isImmutableType() {
   @SuppressWarnings("ReferenceEquality")
   @Override
   public TypeSerializer<StatisticsOrRecord> duplicate() {
-    TypeSerializer<AggregatedStatistics> duplicateStatisticsSerializer =
+    TypeSerializer<GlobalStatistics> duplicateStatisticsSerializer =
         statisticsSerializer.duplicate();
     TypeSerializer<RowData> duplicateRowDataSerializer = recordSerializer.duplicate();
     if ((statisticsSerializer != duplicateStatisticsSerializer)
@@ -84,8 +84,7 @@ public StatisticsOrRecord copy(StatisticsOrRecord from, StatisticsOrRecord reuse
       to.record(record);
     } else {
       to = StatisticsOrRecord.reuseStatistics(reuse, statisticsSerializer);
-      AggregatedStatistics statistics =
-          statisticsSerializer.copy(from.statistics(), to.statistics());
+      GlobalStatistics statistics = statisticsSerializer.copy(from.statistics(), to.statistics());
       to.statistics(statistics);
     }
 
@@ -130,7 +129,7 @@ public StatisticsOrRecord deserialize(StatisticsOrRecord reuse, DataInputView so
       to.record(record);
     } else {
       to = StatisticsOrRecord.reuseStatistics(reuse, statisticsSerializer);
-      AggregatedStatistics statistics = statisticsSerializer.deserialize(to.statistics(), source);
+      GlobalStatistics statistics = statisticsSerializer.deserialize(to.statistics(), source);
       to.statistics(statistics);
     }
 
@@ -200,8 +199,8 @@ protected TypeSerializer<?>[] getNestedSerializers(
     @Override
     protected StatisticsOrRecordSerializer createOuterSerializerWithNestedSerializers(
         TypeSerializer<?>[] nestedSerializers) {
-      TypeSerializer<AggregatedStatistics> statisticsSerializer =
-          (TypeSerializer<AggregatedStatistics>) nestedSerializers[0];
+      TypeSerializer<GlobalStatistics> statisticsSerializer =
+          (TypeSerializer<GlobalStatistics>) nestedSerializers[0];
       TypeSerializer<RowData> recordSerializer = (TypeSerializer<RowData>) nestedSerializers[1];
       return new StatisticsOrRecordSerializer(statisticsSerializer, recordSerializer);
     }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java
index 9d3d128535fe..5d48ec57ca49 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java
@@ -60,20 +60,41 @@ static DataStatistics deserializeDataStatistics(
     }
   }
 
-  static byte[] serializeAggregatedStatistics(
-      AggregatedStatistics aggregatedStatistics,
-      TypeSerializer<AggregatedStatistics> statisticsSerializer) {
+  static byte[] serializeCompletedStatistics(
+      CompletedStatistics completedStatistics,
+      TypeSerializer<CompletedStatistics> statisticsSerializer) {
     try {
       DataOutputSerializer out = new DataOutputSerializer(1024);
-      statisticsSerializer.serialize(aggregatedStatistics, out);
+      statisticsSerializer.serialize(completedStatistics, out);
       return out.getCopyOfBuffer();
     } catch (IOException e) {
       throw new UncheckedIOException("Fail to serialize aggregated statistics", e);
     }
   }
 
-  static AggregatedStatistics deserializeAggregatedStatistics(
-      byte[] bytes, TypeSerializer<AggregatedStatistics> statisticsSerializer) {
+  static CompletedStatistics deserializeCompletedStatistics(
+      byte[] bytes, TypeSerializer<CompletedStatistics> statisticsSerializer) {
+    try {
+      DataInputDeserializer input = new DataInputDeserializer(bytes);
+      return statisticsSerializer.deserialize(input);
+    } catch (IOException e) {
+      throw new UncheckedIOException("Fail to deserialize aggregated statistics", e);
+    }
+  }
+
+  static byte[] serializeGlobalStatistics(
+      GlobalStatistics globalStatistics, TypeSerializer<GlobalStatistics> statisticsSerializer) {
+    try {
+      DataOutputSerializer out = new DataOutputSerializer(1024);
+      statisticsSerializer.serialize(globalStatistics, out);
+      return out.getCopyOfBuffer();
+    } catch (IOException e) {
+      throw new UncheckedIOException("Fail to serialize aggregated statistics", e);
+    }
+  }
+
+  static GlobalStatistics deserializeGlobalStatistics(
+      byte[] bytes, TypeSerializer<GlobalStatistics> statisticsSerializer) {
     try {
       DataInputDeserializer input = new DataInputDeserializer(bytes);
       return statisticsSerializer.deserialize(input);
@@ -86,10 +107,18 @@ static StatisticsType collectType(StatisticsType config) {
     return config == StatisticsType.Sketch ? StatisticsType.Sketch : StatisticsType.Map;
   }
 
+  static StatisticsType collectType(StatisticsType config, @Nullable GlobalStatistics statistics) {
+    if (statistics != null) {
+      return statistics.type();
+    }
+
+    return collectType(config);
+  }
+
   static StatisticsType collectType(
-      StatisticsType config, @Nullable AggregatedStatistics restoredStatistics) {
-    if (restoredStatistics != null) {
-      return restoredStatistics.type();
+      StatisticsType config, @Nullable CompletedStatistics statistics) {
+    if (statistics != null) {
+      return statistics.type();
     }
 
     return collectType(config);


Test patch:
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/Fixtures.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/Fixtures.java
index c7e9f19abb02..5910bd685510 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/Fixtures.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/Fixtures.java
@@ -53,8 +53,10 @@ private Fixtures() {}
       new SortKeySerializer(SCHEMA, SORT_ORDER);
   public static final DataStatisticsSerializer TASK_STATISTICS_SERIALIZER =
       new DataStatisticsSerializer(SORT_KEY_SERIALIZER);
-  public static final AggregatedStatisticsSerializer AGGREGATED_STATISTICS_SERIALIZER =
-      new AggregatedStatisticsSerializer(SORT_KEY_SERIALIZER);
+  public static final GlobalStatisticsSerializer GLOBAL_STATISTICS_SERIALIZER =
+      new GlobalStatisticsSerializer(SORT_KEY_SERIALIZER);
+  public static final CompletedStatisticsSerializer COMPLETED_STATISTICS_SERIALIZER =
+      new CompletedStatisticsSerializer(SORT_KEY_SERIALIZER);
 
   public static final SortKey SORT_KEY = new SortKey(SCHEMA, SORT_ORDER);
   public static final Map<String, SortKey> CHAR_KEYS = createCharKeys();
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsTracker.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsTracker.java
index 52485a3c67d9..0dce1b286977 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsTracker.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsTracker.java
@@ -36,7 +36,7 @@ public void receiveNewerStatisticsEvent(StatisticsType type) {
 
     StatisticsEvent checkpoint1Subtask0StatisticsEvent =
         createStatisticsEvent(type, TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("a"));
-    AggregatedStatistics completedStatistics =
+    CompletedStatistics completedStatistics =
         tracker.updateAndCheckCompletion(0, checkpoint1Subtask0StatisticsEvent);
     assertThat(completedStatistics).isNull();
     assertThat(tracker.aggregationsPerCheckpoint().keySet()).containsExactlyInAnyOrder(1L);
@@ -86,7 +86,8 @@ public void receiveNewerStatisticsEvent(StatisticsType type) {
                   CHAR_KEYS.get("a"), 1L,
                   CHAR_KEYS.get("b"), 1L));
     } else {
-      assertThat(completedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("a"));
+      assertThat(completedStatistics.keySamples())
+          .containsExactly(CHAR_KEYS.get("a"), CHAR_KEYS.get("b"));
     }
 
     // checkpoint 2 remains
@@ -115,7 +116,7 @@ public void receiveOlderStatisticsEventTest(StatisticsType type) {
             CHAR_KEYS.get("a"),
             CHAR_KEYS.get("b"),
             CHAR_KEYS.get("b"));
-    AggregatedStatistics completedStatistics =
+    CompletedStatistics completedStatistics =
         tracker.updateAndCheckCompletion(0, checkpoint2Subtask0StatisticsEvent);
     assertThat(completedStatistics).isNull();
     assertThat(tracker.aggregationsPerCheckpoint().keySet()).containsExactlyInAnyOrder(2L);
@@ -189,7 +190,14 @@ public void receiveOlderStatisticsEventTest(StatisticsType type) {
                   CHAR_KEYS.get("a"), 2L,
                   CHAR_KEYS.get("b"), 4L));
     } else {
-      assertThat(completedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("b"));
+      assertThat(completedStatistics.keySamples())
+          .containsExactly(
+              CHAR_KEYS.get("a"),
+              CHAR_KEYS.get("b"),
+              CHAR_KEYS.get("b"),
+              CHAR_KEYS.get("a"),
+              CHAR_KEYS.get("b"),
+              CHAR_KEYS.get("b"));
     }
   }
 
@@ -207,7 +215,7 @@ public void receiveCompletedStatisticsEvent(StatisticsType type) {
             CHAR_KEYS.get("b"),
             CHAR_KEYS.get("b"));
 
-    AggregatedStatistics completedStatistics =
+    CompletedStatistics completedStatistics =
         tracker.updateAndCheckCompletion(0, checkpoint1Subtask0DataStatisticEvent);
     assertThat(completedStatistics).isNull();
     assertThat(tracker.aggregationsPerCheckpoint().keySet()).containsExactlyInAnyOrder(1L);
@@ -246,7 +254,14 @@ public void receiveCompletedStatisticsEvent(StatisticsType type) {
                   CHAR_KEYS.get("a"), 3L,
                   CHAR_KEYS.get("b"), 3L));
     } else {
-      assertThat(completedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("a"));
+      assertThat(completedStatistics.keySamples())
+          .containsExactly(
+              CHAR_KEYS.get("a"),
+              CHAR_KEYS.get("b"),
+              CHAR_KEYS.get("b"),
+              CHAR_KEYS.get("a"),
+              CHAR_KEYS.get("a"),
+              CHAR_KEYS.get("b"));
     }
 
     StatisticsEvent checkpoint2Subtask0DataStatisticEvent =
@@ -280,7 +295,8 @@ public void receiveCompletedStatisticsEvent(StatisticsType type) {
                   CHAR_KEYS.get("a"), 1L,
                   CHAR_KEYS.get("b"), 1L));
     } else {
-      assertThat(completedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("a"));
+      assertThat(completedStatistics.keySamples())
+          .containsExactly(CHAR_KEYS.get("a"), CHAR_KEYS.get("b"));
     }
   }
 
@@ -307,7 +323,7 @@ public void coordinatorSwitchToSketchOverThreshold() {
             1L,
             CHAR_KEYS.get("a"),
             CHAR_KEYS.get("b"));
-    AggregatedStatistics completedStatistics =
+    CompletedStatistics completedStatistics =
         tracker.updateAndCheckCompletion(0, checkpoint1Subtask0StatisticsEvent);
     assertThat(completedStatistics).isNull();
     assertThat(tracker.aggregationsPerCheckpoint().keySet()).containsExactlyInAnyOrder(1L);
@@ -349,8 +365,14 @@ public void coordinatorSwitchToSketchOverThreshold() {
     assertThat(tracker.aggregationsPerCheckpoint()).isEmpty();
     assertThat(completedStatistics).isNotNull();
     assertThat(completedStatistics.type()).isEqualTo(StatisticsType.Sketch);
-    assertThat(completedStatistics.rangeBounds())
-        .containsExactly(CHAR_KEYS.get("b"), CHAR_KEYS.get("d"));
+    assertThat(completedStatistics.keySamples())
+        .containsExactly(
+            CHAR_KEYS.get("a"),
+            CHAR_KEYS.get("b"),
+            CHAR_KEYS.get("c"),
+            CHAR_KEYS.get("d"),
+            CHAR_KEYS.get("e"),
+            CHAR_KEYS.get("f"));
   }
 
   @Test
@@ -376,7 +398,7 @@ public void coordinatorMapOperatorSketch() {
             1L,
             CHAR_KEYS.get("a"),
             CHAR_KEYS.get("b"));
-    AggregatedStatistics completedStatistics =
+    CompletedStatistics completedStatistics =
         tracker.updateAndCheckCompletion(0, checkpoint1Subtask0StatisticsEvent);
     assertThat(completedStatistics).isNull();
     assertThat(tracker.aggregationsPerCheckpoint().keySet()).containsExactlyInAnyOrder(1L);
@@ -419,8 +441,14 @@ public void coordinatorMapOperatorSketch() {
     assertThat(tracker.aggregationsPerCheckpoint()).isEmpty();
     assertThat(completedStatistics).isNotNull();
     assertThat(completedStatistics.type()).isEqualTo(StatisticsType.Sketch);
-    assertThat(completedStatistics.rangeBounds())
-        .containsExactly(CHAR_KEYS.get("b"), CHAR_KEYS.get("d"));
+    assertThat(completedStatistics.keySamples())
+        .containsExactly(
+            CHAR_KEYS.get("a"),
+            CHAR_KEYS.get("b"),
+            CHAR_KEYS.get("c"),
+            CHAR_KEYS.get("d"),
+            CHAR_KEYS.get("e"),
+            CHAR_KEYS.get("f"));
   }
 
   private AggregatedStatisticsTracker createTracker(StatisticsType type) {
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsSerializer.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestCompletedStatisticsSerializer.java
similarity index 70%
rename from flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsSerializer.java
rename to flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestCompletedStatisticsSerializer.java
index 0ce73fa4aaad..4ee9888934a8 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsSerializer.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestCompletedStatisticsSerializer.java
@@ -25,11 +25,11 @@
 import org.apache.iceberg.SortKey;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 
-public class TestAggregatedStatisticsSerializer extends SerializerTestBase<AggregatedStatistics> {
+public class TestCompletedStatisticsSerializer extends SerializerTestBase<CompletedStatistics> {
 
   @Override
-  protected TypeSerializer<AggregatedStatistics> createSerializer() {
-    return Fixtures.AGGREGATED_STATISTICS_SERIALIZER;
+  protected TypeSerializer<CompletedStatistics> createSerializer() {
+    return Fixtures.COMPLETED_STATISTICS_SERIALIZER;
   }
 
   @Override
@@ -38,17 +38,17 @@ protected int getLength() {
   }
 
   @Override
-  protected Class<AggregatedStatistics> getTypeClass() {
-    return AggregatedStatistics.class;
+  protected Class<CompletedStatistics> getTypeClass() {
+    return CompletedStatistics.class;
   }
 
   @Override
-  protected AggregatedStatistics[] getTestData() {
-    return new AggregatedStatistics[] {
-      AggregatedStatistics.fromKeyFrequency(
+  protected CompletedStatistics[] getTestData() {
+
+    return new CompletedStatistics[] {
+      CompletedStatistics.fromKeyFrequency(
           1L, ImmutableMap.of(CHAR_KEYS.get("a"), 1L, CHAR_KEYS.get("b"), 2L)),
-      AggregatedStatistics.fromRangeBounds(
-          2L, new SortKey[] {CHAR_KEYS.get("a"), CHAR_KEYS.get("b")})
+      CompletedStatistics.fromKeySamples(2L, new SortKey[] {CHAR_KEYS.get("a"), CHAR_KEYS.get("b")})
     };
   }
 }
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java
index fe1d07c3286a..a08a8a73e80c 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java
@@ -19,17 +19,24 @@
 package org.apache.iceberg.flink.sink.shuffle;
 
 import static org.apache.iceberg.flink.sink.shuffle.Fixtures.CHAR_KEYS;
+import static org.apache.iceberg.flink.sink.shuffle.Fixtures.NUM_SUBTASKS;
+import static org.apache.iceberg.flink.sink.shuffle.Fixtures.SORT_ORDER_COMPARTOR;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import java.time.Duration;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.operators.coordination.EventReceivingTasks;
 import org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
 import org.apache.flink.util.ExceptionUtils;
+import org.apache.iceberg.SortKey;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -45,7 +52,7 @@ public void before() throws Exception {
   }
 
   private void tasksReady(DataStatisticsCoordinator coordinator) {
-    setAllTasksReady(Fixtures.NUM_SUBTASKS, coordinator, receivingTasks);
+    setAllTasksReady(NUM_SUBTASKS, coordinator, receivingTasks);
   }
 
   @ParameterizedTest
@@ -106,20 +113,95 @@ public void testDataStatisticsEventHandling(StatisticsType type) throws Exceptio
 
       waitForCoordinatorToProcessActions(dataStatisticsCoordinator);
 
-      // Verify global data statistics is the aggregation of all subtasks data statistics
-      AggregatedStatistics aggregatedStatistics = dataStatisticsCoordinator.completedStatistics();
-      assertThat(aggregatedStatistics.checkpointId()).isEqualTo(1L);
-      assertThat(aggregatedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      Map<SortKey, Long> keyFrequency =
+          ImmutableMap.of(
+              CHAR_KEYS.get("a"), 2L,
+              CHAR_KEYS.get("b"), 3L,
+              CHAR_KEYS.get("c"), 5L);
+      MapAssignment mapAssignment =
+          MapAssignment.fromKeyFrequency(NUM_SUBTASKS, keyFrequency, 0.0d, SORT_ORDER_COMPARTOR);
+
+      CompletedStatistics completedStatistics = dataStatisticsCoordinator.completedStatistics();
+      assertThat(completedStatistics.checkpointId()).isEqualTo(1L);
+      assertThat(completedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
       if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
-        assertThat(aggregatedStatistics.keyFrequency())
-            .isEqualTo(
-                ImmutableMap.of(
-                    CHAR_KEYS.get("a"), 2L,
-                    CHAR_KEYS.get("b"), 3L,
-                    CHAR_KEYS.get("c"), 5L));
+        assertThat(completedStatistics.keyFrequency()).isEqualTo(keyFrequency);
       } else {
-        assertThat(aggregatedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("b"));
+        assertThat(completedStatistics.keySamples())
+            .containsExactly(
+                CHAR_KEYS.get("a"),
+                CHAR_KEYS.get("a"),
+                CHAR_KEYS.get("b"),
+                CHAR_KEYS.get("b"),
+                CHAR_KEYS.get("b"),
+                CHAR_KEYS.get("c"),
+                CHAR_KEYS.get("c"),
+                CHAR_KEYS.get("c"),
+                CHAR_KEYS.get("c"),
+                CHAR_KEYS.get("c"));
       }
+
+      GlobalStatistics globalStatistics = dataStatisticsCoordinator.globalStatistics();
+      assertThat(globalStatistics.checkpointId()).isEqualTo(1L);
+      assertThat(globalStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
+        assertThat(globalStatistics.mapAssignment()).isEqualTo(mapAssignment);
+      } else {
+        assertThat(globalStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("b"));
+      }
+    }
+  }
+
+  @Test
+  public void testRequestGlobalStatisticsEventHandling() throws Exception {
+    try (DataStatisticsCoordinator dataStatisticsCoordinator =
+        createCoordinator(StatisticsType.Sketch)) {
+      dataStatisticsCoordinator.start();
+      tasksReady(dataStatisticsCoordinator);
+
+      // receive request before global statistics is ready
+      dataStatisticsCoordinator.handleEventFromOperator(0, 0, new RequestGlobalStatisticsEvent());
+      assertThat(receivingTasks.getSentEventsForSubtask(0)).isEmpty();
+      assertThat(receivingTasks.getSentEventsForSubtask(1)).isEmpty();
+
+      StatisticsEvent checkpoint1Subtask0DataStatisticEvent =
+          Fixtures.createStatisticsEvent(
+              StatisticsType.Sketch, Fixtures.TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("a"));
+      StatisticsEvent checkpoint1Subtask1DataStatisticEvent =
+          Fixtures.createStatisticsEvent(
+              StatisticsType.Sketch, Fixtures.TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("b"));
+      // Handle events from operators for checkpoint 1
+      dataStatisticsCoordinator.handleEventFromOperator(
+          0, 0, checkpoint1Subtask0DataStatisticEvent);
+      dataStatisticsCoordinator.handleEventFromOperator(
+          1, 0, checkpoint1Subtask1DataStatisticEvent);
+
+      waitForCoordinatorToProcessActions(dataStatisticsCoordinator);
+      Awaitility.await("wait for statistics event")
+          .pollInterval(Duration.ofMillis(10))
+          .atMost(Duration.ofSeconds(10))
+          .until(() -> receivingTasks.getSentEventsForSubtask(0).size() == 1);
+      assertThat(receivingTasks.getSentEventsForSubtask(0).get(0))
+          .isInstanceOf(StatisticsEvent.class);
+
+      Awaitility.await("wait for statistics event")
+          .pollInterval(Duration.ofMillis(10))
+          .atMost(Duration.ofSeconds(10))
+          .until(() -> receivingTasks.getSentEventsForSubtask(1).size() == 1);
+      assertThat(receivingTasks.getSentEventsForSubtask(1).get(0))
+          .isInstanceOf(StatisticsEvent.class);
+
+      dataStatisticsCoordinator.handleEventFromOperator(1, 0, new RequestGlobalStatisticsEvent());
+
+      // coordinator should send a response to subtask 1
+      Awaitility.await("wait for statistics event")
+          .pollInterval(Duration.ofMillis(10))
+          .atMost(Duration.ofSeconds(10))
+          .until(() -> receivingTasks.getSentEventsForSubtask(1).size() == 2);
+      assertThat(receivingTasks.getSentEventsForSubtask(1).get(0))
+          .isInstanceOf(StatisticsEvent.class);
+      assertThat(receivingTasks.getSentEventsForSubtask(1).get(1))
+          .isInstanceOf(StatisticsEvent.class);
     }
   }
 
@@ -154,10 +236,11 @@ static void waitForCoordinatorToProcessActions(DataStatisticsCoordinator coordin
   private static DataStatisticsCoordinator createCoordinator(StatisticsType type) {
     return new DataStatisticsCoordinator(
         OPERATOR_NAME,
-        new MockOperatorCoordinatorContext(TEST_OPERATOR_ID, Fixtures.NUM_SUBTASKS),
+        new MockOperatorCoordinatorContext(TEST_OPERATOR_ID, NUM_SUBTASKS),
         Fixtures.SCHEMA,
         Fixtures.SORT_ORDER,
-        Fixtures.NUM_SUBTASKS,
-        type);
+        NUM_SUBTASKS,
+        type,
+        0.0d);
   }
 }
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java
index 966c8474b42b..6317f2bfde18 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java
@@ -19,16 +19,19 @@
 package org.apache.iceberg.flink.sink.shuffle;
 
 import static org.apache.iceberg.flink.sink.shuffle.Fixtures.CHAR_KEYS;
+import static org.apache.iceberg.flink.sink.shuffle.Fixtures.SORT_ORDER_COMPARTOR;
 import static org.apache.iceberg.flink.sink.shuffle.Fixtures.TASK_STATISTICS_SERIALIZER;
 import static org.apache.iceberg.flink.sink.shuffle.Fixtures.createStatisticsEvent;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.operators.coordination.EventReceivingTasks;
 import org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
 import org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator;
+import org.apache.iceberg.SortKey;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -47,7 +50,7 @@ public void before() {
   @ParameterizedTest
   @EnumSource(StatisticsType.class)
   public void testCheckpointAndReset(StatisticsType type) throws Exception {
-    DataStatisticsCoordinatorProvider provider = createProvider(type);
+    DataStatisticsCoordinatorProvider provider = createProvider(type, Fixtures.NUM_SUBTASKS);
     try (RecreateOnResetOperatorCoordinator coordinator =
         (RecreateOnResetOperatorCoordinator)
             provider.create(
@@ -62,42 +65,39 @@ public void testCheckpointAndReset(StatisticsType type) throws Exception {
 
       // Handle events from operators for checkpoint 1
       StatisticsEvent checkpoint1Subtask0StatisticsEvent =
-          createStatisticsEvent(
-              type,
-              TASK_STATISTICS_SERIALIZER,
-              1L,
-              CHAR_KEYS.get("a"),
-              CHAR_KEYS.get("b"),
-              CHAR_KEYS.get("c"));
+          createStatisticsEvent(type, TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("a"));
       coordinator.handleEventFromOperator(0, 0, checkpoint1Subtask0StatisticsEvent);
       TestDataStatisticsCoordinator.waitForCoordinatorToProcessActions(dataStatisticsCoordinator);
 
       StatisticsEvent checkpoint1Subtask1StatisticsEvent =
-          createStatisticsEvent(
-              type, TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("d"), CHAR_KEYS.get("e"));
+          createStatisticsEvent(type, TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("b"));
       coordinator.handleEventFromOperator(1, 0, checkpoint1Subtask1StatisticsEvent);
       TestDataStatisticsCoordinator.waitForCoordinatorToProcessActions(dataStatisticsCoordinator);
 
       // Verify checkpoint 1 global data statistics
-      assertThat(dataStatisticsCoordinator.completedStatistics()).isNotNull();
-      AggregatedStatistics aggregatedStatistics = dataStatisticsCoordinator.completedStatistics();
-      assertThat(aggregatedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      Map<SortKey, Long> checkpoint1KeyFrequency =
+          ImmutableMap.of(CHAR_KEYS.get("a"), 1L, CHAR_KEYS.get("b"), 1L);
+      MapAssignment checkpoint1MapAssignment =
+          MapAssignment.fromKeyFrequency(
+              Fixtures.NUM_SUBTASKS, checkpoint1KeyFrequency, 0.0d, SORT_ORDER_COMPARTOR);
+
+      CompletedStatistics completedStatistics = dataStatisticsCoordinator.completedStatistics();
+      assertThat(completedStatistics).isNotNull();
+      assertThat(completedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
+        assertThat(completedStatistics.keyFrequency()).isEqualTo(checkpoint1KeyFrequency);
+      } else {
+        assertThat(completedStatistics.keySamples())
+            .containsExactly(CHAR_KEYS.get("a"), CHAR_KEYS.get("b"));
+      }
+
+      GlobalStatistics globalStatistics = dataStatisticsCoordinator.globalStatistics();
+      assertThat(globalStatistics).isNotNull();
+      assertThat(globalStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
       if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
-        assertThat(aggregatedStatistics.keyFrequency())
-            .isEqualTo(
-                ImmutableMap.of(
-                    CHAR_KEYS.get("a"),
-                    1L,
-                    CHAR_KEYS.get("b"),
-                    1L,
-                    CHAR_KEYS.get("c"),
-                    1L,
-                    CHAR_KEYS.get("d"),
-                    1L,
-                    CHAR_KEYS.get("e"),
-                    1L));
+        assertThat(globalStatistics.mapAssignment()).isEqualTo(checkpoint1MapAssignment);
       } else {
-        assertThat(aggregatedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("c"));
+        assertThat(globalStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("a"));
       }
 
       byte[] checkpoint1Bytes = waitForCheckpoint(1L, dataStatisticsCoordinator);
@@ -114,42 +114,54 @@ public void testCheckpointAndReset(StatisticsType type) throws Exception {
       TestDataStatisticsCoordinator.waitForCoordinatorToProcessActions(dataStatisticsCoordinator);
 
       // Verify checkpoint 2 global data statistics
-      aggregatedStatistics = dataStatisticsCoordinator.completedStatistics();
-      assertThat(aggregatedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      Map<SortKey, Long> checkpoint2KeyFrequency =
+          ImmutableMap.of(CHAR_KEYS.get("d"), 1L, CHAR_KEYS.get("e"), 1L, CHAR_KEYS.get("f"), 1L);
+      MapAssignment checkpoint2MapAssignment =
+          MapAssignment.fromKeyFrequency(
+              Fixtures.NUM_SUBTASKS, checkpoint2KeyFrequency, 0.0d, SORT_ORDER_COMPARTOR);
+      completedStatistics = dataStatisticsCoordinator.completedStatistics();
+      assertThat(completedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
+        assertThat(completedStatistics.keyFrequency()).isEqualTo(checkpoint2KeyFrequency);
+      } else {
+        assertThat(completedStatistics.keySamples())
+            .containsExactly(CHAR_KEYS.get("d"), CHAR_KEYS.get("e"), CHAR_KEYS.get("f"));
+      }
+
+      globalStatistics = dataStatisticsCoordinator.globalStatistics();
+      assertThat(globalStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
       if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
-        assertThat(aggregatedStatistics.keyFrequency())
-            .isEqualTo(
-                ImmutableMap.of(
-                    CHAR_KEYS.get("d"), 1L, CHAR_KEYS.get("e"), 1L, CHAR_KEYS.get("f"), 1L));
+        assertThat(globalStatistics.mapAssignment()).isEqualTo(checkpoint2MapAssignment);
       } else {
-        assertThat(aggregatedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("e"));
+        assertThat(globalStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("e"));
       }
+
       waitForCheckpoint(2L, dataStatisticsCoordinator);
 
       // Reset coordinator to checkpoint 1
       coordinator.resetToCheckpoint(1L, checkpoint1Bytes);
       DataStatisticsCoordinator restoredDataStatisticsCoordinator =
           (DataStatisticsCoordinator) coordinator.getInternalCoordinator();
-      assertThat(dataStatisticsCoordinator).isNotEqualTo(restoredDataStatisticsCoordinator);
-      aggregatedStatistics = restoredDataStatisticsCoordinator.completedStatistics();
-      assertThat(aggregatedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      assertThat(dataStatisticsCoordinator).isNotSameAs(restoredDataStatisticsCoordinator);
+
+      completedStatistics = restoredDataStatisticsCoordinator.completedStatistics();
+      assertThat(completedStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
       // Verify restored data statistics
       if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
-        assertThat(aggregatedStatistics.keyFrequency())
-            .isEqualTo(
-                ImmutableMap.of(
-                    CHAR_KEYS.get("a"),
-                    1L,
-                    CHAR_KEYS.get("b"),
-                    1L,
-                    CHAR_KEYS.get("c"),
-                    1L,
-                    CHAR_KEYS.get("d"),
-                    1L,
-                    CHAR_KEYS.get("e"),
-                    1L));
+        assertThat(completedStatistics.keyFrequency())
+            .isEqualTo(ImmutableMap.of(CHAR_KEYS.get("a"), 1L, CHAR_KEYS.get("b"), 1L));
+      } else {
+        assertThat(completedStatistics.keySamples())
+            .containsExactly(CHAR_KEYS.get("a"), CHAR_KEYS.get("b"));
+      }
+
+      globalStatistics = restoredDataStatisticsCoordinator.globalStatistics();
+      assertThat(globalStatistics).isNotNull();
+      assertThat(globalStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
+      if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
+        assertThat(globalStatistics.mapAssignment()).isEqualTo(checkpoint1MapAssignment);
       } else {
-        assertThat(aggregatedStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("c"));
+        assertThat(globalStatistics.rangeBounds()).containsExactly(CHAR_KEYS.get("a"));
       }
     }
   }
@@ -161,13 +173,15 @@ private byte[] waitForCheckpoint(long checkpointId, DataStatisticsCoordinator co
     return future.get();
   }
 
-  private static DataStatisticsCoordinatorProvider createProvider(StatisticsType type) {
+  private static DataStatisticsCoordinatorProvider createProvider(
+      StatisticsType type, int downstreamParallelism) {
     return new DataStatisticsCoordinatorProvider(
         "DataStatisticsCoordinatorProvider",
         OPERATOR_ID,
         Fixtures.SCHEMA,
         Fixtures.SORT_ORDER,
-        Fixtures.NUM_SUBTASKS,
-        type);
+        downstreamParallelism,
+        type,
+        0.0);
   }
 }
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
index 6c864791a1bc..bc248b778184 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
@@ -19,12 +19,16 @@
 package org.apache.iceberg.flink.sink.shuffle;
 
 import static org.apache.iceberg.flink.sink.shuffle.Fixtures.CHAR_KEYS;
+import static org.apache.iceberg.flink.sink.shuffle.Fixtures.SORT_ORDER_COMPARTOR;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.apache.datasketches.sampling.ReservoirItemsSketch;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.state.OperatorStateStore;
@@ -56,7 +60,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mockito;
 
 public class TestDataStatisticsOperator {
 
@@ -75,15 +82,22 @@ public void before() throws Exception {
             new TestTaskStateManager());
   }
 
-  private DataStatisticsOperator createOperator(StatisticsType type) throws Exception {
+  private DataStatisticsOperator createOperator(StatisticsType type, int downstreamParallelism)
+      throws Exception {
     MockOperatorEventGateway mockGateway = new MockOperatorEventGateway();
+    return createOperator(type, downstreamParallelism, mockGateway);
+  }
+
+  private DataStatisticsOperator createOperator(
+      StatisticsType type, int downstreamParallelism, MockOperatorEventGateway mockGateway)
+      throws Exception {
     DataStatisticsOperator operator =
         new DataStatisticsOperator(
             "testOperator",
             Fixtures.SCHEMA,
             Fixtures.SORT_ORDER,
             mockGateway,
-            Fixtures.NUM_SUBTASKS,
+            downstreamParallelism,
             type);
     operator.setup(
         new OneInputStreamTask<String, String>(env),
@@ -96,7 +110,7 @@ private DataStatisticsOperator createOperator(StatisticsType type) throws Except
   @ParameterizedTest
   @EnumSource(StatisticsType.class)
   public void testProcessElement(StatisticsType type) throws Exception {
-    DataStatisticsOperator operator = createOperator(type);
+    DataStatisticsOperator operator = createOperator(type, Fixtures.NUM_SUBTASKS);
     try (OneInputStreamOperatorTestHarness<RowData, StatisticsOrRecord> testHarness =
         createHarness(operator)) {
       StateInitializationContext stateContext = getStateContext();
@@ -125,7 +139,7 @@ public void testProcessElement(StatisticsType type) throws Exception {
   @ParameterizedTest
   @EnumSource(StatisticsType.class)
   public void testOperatorOutput(StatisticsType type) throws Exception {
-    DataStatisticsOperator operator = createOperator(type);
+    DataStatisticsOperator operator = createOperator(type, Fixtures.NUM_SUBTASKS);
     try (OneInputStreamOperatorTestHarness<RowData, StatisticsOrRecord> testHarness =
         createHarness(operator)) {
       testHarness.processElement(
@@ -149,35 +163,47 @@ public void testOperatorOutput(StatisticsType type) throws Exception {
     }
   }
 
+  private static Stream<Arguments> provideRestoreStateParameters() {
+    return Stream.of(
+        Arguments.of(StatisticsType.Map, -1),
+        Arguments.of(StatisticsType.Map, 0),
+        Arguments.of(StatisticsType.Map, 1),
+        Arguments.of(StatisticsType.Sketch, -1),
+        Arguments.of(StatisticsType.Sketch, 0),
+        Arguments.of(StatisticsType.Sketch, 1));
+  }
+
   @ParameterizedTest
-  @EnumSource(StatisticsType.class)
-  public void testRestoreState(StatisticsType type) throws Exception {
+  @MethodSource("provideRestoreStateParameters")
+  public void testRestoreState(StatisticsType type, int parallelismAdjustment) throws Exception {
     Map<SortKey, Long> keyFrequency =
         ImmutableMap.of(CHAR_KEYS.get("a"), 2L, CHAR_KEYS.get("b"), 1L, CHAR_KEYS.get("c"), 1L);
     SortKey[] rangeBounds = new SortKey[] {CHAR_KEYS.get("a")};
-    DataStatisticsOperator operator = createOperator(type);
+    MapAssignment mapAssignment =
+        MapAssignment.fromKeyFrequency(2, keyFrequency, 0.0d, SORT_ORDER_COMPARTOR);
+    DataStatisticsOperator operator = createOperator(type, Fixtures.NUM_SUBTASKS);
     OperatorSubtaskState snapshot;
     try (OneInputStreamOperatorTestHarness<RowData, StatisticsOrRecord> testHarness1 =
         createHarness(operator)) {
-      AggregatedStatistics statistics;
+      GlobalStatistics statistics;
       if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
-        statistics = AggregatedStatistics.fromKeyFrequency(1L, keyFrequency);
+        statistics = GlobalStatistics.fromMapAssignment(1L, mapAssignment);
       } else {
-        statistics = AggregatedStatistics.fromRangeBounds(1L, rangeBounds);
+        statistics = GlobalStatistics.fromRangeBounds(1L, rangeBounds);
       }
 
       StatisticsEvent event =
-          StatisticsEvent.createAggregatedStatisticsEvent(
-              1L, statistics, Fixtures.AGGREGATED_STATISTICS_SERIALIZER);
+          StatisticsEvent.createGlobalStatisticsEvent(
+              statistics, Fixtures.GLOBAL_STATISTICS_SERIALIZER, false);
       operator.handleOperatorEvent(event);
 
-      AggregatedStatistics globalStatistics = operator.globalStatistics();
+      GlobalStatistics globalStatistics = operator.globalStatistics();
       assertThat(globalStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
       if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
-        assertThat(globalStatistics.keyFrequency()).isEqualTo(keyFrequency);
+        assertThat(globalStatistics.mapAssignment()).isEqualTo(mapAssignment);
         assertThat(globalStatistics.rangeBounds()).isNull();
       } else {
-        assertThat(globalStatistics.keyFrequency()).isNull();
+        assertThat(globalStatistics.mapAssignment()).isNull();
         assertThat(globalStatistics.rangeBounds()).isEqualTo(rangeBounds);
       }
 
@@ -186,19 +212,28 @@ public void testRestoreState(StatisticsType type) throws Exception {
 
     // Use the snapshot to initialize state for another new operator and then verify that the global
     // statistics for the new operator is same as before
-    DataStatisticsOperator restoredOperator = createOperator(type);
+    MockOperatorEventGateway spyGateway = Mockito.spy(new MockOperatorEventGateway());
+    DataStatisticsOperator restoredOperator =
+        createOperator(type, Fixtures.NUM_SUBTASKS + parallelismAdjustment, spyGateway);
     try (OneInputStreamOperatorTestHarness<RowData, StatisticsOrRecord> testHarness2 =
         new OneInputStreamOperatorTestHarness<>(restoredOperator, 2, 2, 1)) {
       testHarness2.setup();
       testHarness2.initializeState(snapshot);
 
-      AggregatedStatistics globalStatistics = restoredOperator.globalStatistics();
+      GlobalStatistics globalStatistics = restoredOperator.globalStatistics();
+      // global statistics is always restored and used initially even if
+      // downstream parallelism changed.
+      assertThat(globalStatistics).isNotNull();
+      // request is always sent to coordinator during initialization.
+      // coordinator would respond with a new global statistics that
+      // has range bound recomputed with new parallelism.
+      verify(spyGateway).sendEventToCoordinator(any(RequestGlobalStatisticsEvent.class));
       assertThat(globalStatistics.type()).isEqualTo(StatisticsUtil.collectType(type));
       if (StatisticsUtil.collectType(type) == StatisticsType.Map) {
-        assertThat(globalStatistics.keyFrequency()).isEqualTo(keyFrequency);
+        assertThat(globalStatistics.mapAssignment()).isEqualTo(mapAssignment);
         assertThat(globalStatistics.rangeBounds()).isNull();
       } else {
-        assertThat(globalStatistics.keyFrequency()).isNull();
+        assertThat(globalStatistics.mapAssignment()).isNull();
         assertThat(globalStatistics.rangeBounds()).isEqualTo(rangeBounds);
       }
     }
@@ -207,7 +242,7 @@ public void testRestoreState(StatisticsType type) throws Exception {
   @SuppressWarnings("unchecked")
   @Test
   public void testMigrationWithLocalStatsOverThreshold() throws Exception {
-    DataStatisticsOperator operator = createOperator(StatisticsType.Auto);
+    DataStatisticsOperator operator = createOperator(StatisticsType.Auto, Fixtures.NUM_SUBTASKS);
     try (OneInputStreamOperatorTestHarness<RowData, StatisticsOrRecord> testHarness =
         createHarness(operator)) {
       StateInitializationContext stateContext = getStateContext();
@@ -259,7 +294,7 @@ public void testMigrationWithLocalStatsOverThreshold() throws Exception {
   @SuppressWarnings("unchecked")
   @Test
   public void testMigrationWithGlobalSketchStatistics() throws Exception {
-    DataStatisticsOperator operator = createOperator(StatisticsType.Auto);
+    DataStatisticsOperator operator = createOperator(StatisticsType.Auto, Fixtures.NUM_SUBTASKS);
     try (OneInputStreamOperatorTestHarness<RowData, StatisticsOrRecord> testHarness =
         createHarness(operator)) {
       StateInitializationContext stateContext = getStateContext();
@@ -272,12 +307,12 @@ public void testMigrationWithGlobalSketchStatistics() throws Exception {
           .isEqualTo(ImmutableMap.of(CHAR_KEYS.get("a"), 1L));
 
       // received global statistics with sketch type
-      AggregatedStatistics globalStatistics =
-          AggregatedStatistics.fromRangeBounds(
+      GlobalStatistics globalStatistics =
+          GlobalStatistics.fromRangeBounds(
               1L, new SortKey[] {CHAR_KEYS.get("c"), CHAR_KEYS.get("f")});
       operator.handleOperatorEvent(
-          StatisticsEvent.createAggregatedStatisticsEvent(
-              1L, globalStatistics, Fixtures.AGGREGATED_STATISTICS_SERIALIZER));
+          StatisticsEvent.createGlobalStatisticsEvent(
+              globalStatistics, Fixtures.GLOBAL_STATISTICS_SERIALIZER, false));
 
       int reservoirSize =
           SketchUtil.determineOperatorReservoirSize(Fixtures.NUM_SUBTASKS, Fixtures.NUM_SUBTASKS);
@@ -310,7 +345,7 @@ private OneInputStreamOperatorTestHarness<RowData, StatisticsOrRecord> createHar
             dataStatisticsOperator, Fixtures.NUM_SUBTASKS, Fixtures.NUM_SUBTASKS, 0);
     harness.setup(
         new StatisticsOrRecordSerializer(
-            Fixtures.AGGREGATED_STATISTICS_SERIALIZER, Fixtures.ROW_SERIALIZER));
+            Fixtures.GLOBAL_STATISTICS_SERIALIZER, Fixtures.ROW_SERIALIZER));
     harness.open();
     return harness;
   }
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestGlobalStatisticsSerializer.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestGlobalStatisticsSerializer.java
new file mode 100644
index 000000000000..7afaf239c668
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestGlobalStatisticsSerializer.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.shuffle;
+
+import static org.apache.iceberg.flink.sink.shuffle.Fixtures.CHAR_KEYS;
+import static org.apache.iceberg.flink.sink.shuffle.Fixtures.SORT_ORDER_COMPARTOR;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.iceberg.SortKey;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+
+public class TestGlobalStatisticsSerializer extends SerializerTestBase<GlobalStatistics> {
+
+  @Override
+  protected TypeSerializer<GlobalStatistics> createSerializer() {
+    return Fixtures.GLOBAL_STATISTICS_SERIALIZER;
+  }
+
+  @Override
+  protected int getLength() {
+    return -1;
+  }
+
+  @Override
+  protected Class<GlobalStatistics> getTypeClass() {
+    return GlobalStatistics.class;
+  }
+
+  @Override
+  protected GlobalStatistics[] getTestData() {
+    return new GlobalStatistics[] {
+      GlobalStatistics.fromMapAssignment(
+          1L,
+          MapAssignment.fromKeyFrequency(
+              Fixtures.NUM_SUBTASKS,
+              ImmutableMap.of(CHAR_KEYS.get("a"), 1L, CHAR_KEYS.get("b"), 2L),
+              0.0d,
+              SORT_ORDER_COMPARTOR)),
+      GlobalStatistics.fromRangeBounds(2L, new SortKey[] {CHAR_KEYS.get("a"), CHAR_KEYS.get("b")})
+    };
+  }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestMapRangePartitioner.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestMapRangePartitioner.java
index d47e41768072..d5a0bebc74e7 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestMapRangePartitioner.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestMapRangePartitioner.java
@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.flink.sink.shuffle;
 
+import static org.apache.iceberg.flink.sink.shuffle.Fixtures.SORT_ORDER_COMPARTOR;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.List;
@@ -89,39 +90,35 @@ private static SortKey[] initSortKeys() {
 
   @Test
   public void testEvenlyDividableNoClosingFileCost() {
-    MapRangePartitioner partitioner =
-        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapStatistics, 0.0);
     int numPartitions = 8;
+    MapAssignment mapAssignment =
+        MapAssignment.fromKeyFrequency(numPartitions, mapStatistics, 0.0, SORT_ORDER_COMPARTOR);
 
     // each task should get targeted weight of 100 (=800/8)
-    Map<SortKey, MapRangePartitioner.KeyAssignment> expectedAssignment =
+    Map<SortKey, KeyAssignment> expectedAssignment =
         ImmutableMap.of(
             SORT_KEYS[0],
-            new MapRangePartitioner.KeyAssignment(
+            new KeyAssignment(
                 ImmutableList.of(0, 1, 2, 3), ImmutableList.of(100L, 100L, 100L, 50L), 0L),
             SORT_KEYS[1],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(3, 4, 5), ImmutableList.of(50L, 100L, 80L), 0L),
+            new KeyAssignment(ImmutableList.of(3, 4, 5), ImmutableList.of(50L, 100L, 80L), 0L),
             SORT_KEYS[2],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(5, 6), ImmutableList.of(20L, 100L), 0L),
+            new KeyAssignment(ImmutableList.of(5, 6), ImmutableList.of(20L, 100L), 0L),
             SORT_KEYS[3],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(40L), 0L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(40L), 0L),
             SORT_KEYS[4],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
             SORT_KEYS[5],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
             SORT_KEYS[6],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
             SORT_KEYS[7],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
             SORT_KEYS[8],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L),
             SORT_KEYS[9],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L));
-    Map<SortKey, MapRangePartitioner.KeyAssignment> actualAssignment =
-        partitioner.assignment(numPartitions);
-    assertThat(actualAssignment).isEqualTo(expectedAssignment);
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(10L), 0L));
+    assertThat(mapAssignment).isEqualTo(new MapAssignment(numPartitions, expectedAssignment));
 
     // key: subtask id
     // value pair: first is the assigned weight, second is the number of assigned keys
@@ -143,19 +140,20 @@ public void testEvenlyDividableNoClosingFileCost() {
             Pair.of(100L, 1),
             7,
             Pair.of(100L, 7));
-    Map<Integer, Pair<Long, Integer>> actualAssignmentInfo = partitioner.assignmentInfo();
-    assertThat(actualAssignmentInfo).isEqualTo(expectedAssignmentInfo);
+    assertThat(mapAssignment.assignmentInfo()).isEqualTo(expectedAssignmentInfo);
 
+    MapRangePartitioner partitioner =
+        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapAssignment);
     Map<Integer, Pair<AtomicLong, Set<RowData>>> partitionResults =
-        runPartitioner(partitioner, numPartitions);
+        runPartitioner(partitioner, numPartitions, mapStatistics);
     validatePartitionResults(expectedAssignmentInfo, partitionResults, 5.0);
   }
 
   @Test
   public void testEvenlyDividableWithClosingFileCost() {
-    MapRangePartitioner partitioner =
-        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapStatistics, 5.0);
     int numPartitions = 8;
+    MapAssignment mapAssignment =
+        MapAssignment.fromKeyFrequency(numPartitions, mapStatistics, 5.0, SORT_ORDER_COMPARTOR);
 
     // target subtask weight is 100 before close file cost factored in.
     // close file cost is 5 = 5% * 100.
@@ -164,35 +162,30 @@ public void testEvenlyDividableWithClosingFileCost() {
     // close-cost:  20,  15,  10,  5,  5,  5,  5,  5,  5,  5
     // after:      370, 245, 130, 45, 15, 15, 15, 15, 15, 15
     // target subtask weight with close cost per subtask is 110 (880/8)
-    Map<SortKey, MapRangePartitioner.KeyAssignment> expectedAssignment =
+    Map<SortKey, KeyAssignment> expectedAssignment =
         ImmutableMap.of(
             SORT_KEYS[0],
-            new MapRangePartitioner.KeyAssignment(
+            new KeyAssignment(
                 ImmutableList.of(0, 1, 2, 3), ImmutableList.of(110L, 110L, 110L, 40L), 5L),
             SORT_KEYS[1],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(3, 4, 5), ImmutableList.of(70L, 110L, 65L), 5L),
+            new KeyAssignment(ImmutableList.of(3, 4, 5), ImmutableList.of(70L, 110L, 65L), 5L),
             SORT_KEYS[2],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(5, 6), ImmutableList.of(45L, 85L), 5L),
+            new KeyAssignment(ImmutableList.of(5, 6), ImmutableList.of(45L, 85L), 5L),
             SORT_KEYS[3],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(6, 7), ImmutableList.of(25L, 20L), 5L),
+            new KeyAssignment(ImmutableList.of(6, 7), ImmutableList.of(25L, 20L), 5L),
             SORT_KEYS[4],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
             SORT_KEYS[5],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
             SORT_KEYS[6],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
             SORT_KEYS[7],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
             SORT_KEYS[8],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L),
             SORT_KEYS[9],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L));
-    Map<SortKey, MapRangePartitioner.KeyAssignment> actualAssignment =
-        partitioner.assignment(numPartitions);
-    assertThat(actualAssignment).isEqualTo(expectedAssignment);
+            new KeyAssignment(ImmutableList.of(7), ImmutableList.of(15L), 5L));
+    assertThat(mapAssignment.keyAssignments()).isEqualTo(expectedAssignment);
 
     // key: subtask id
     // value pair: first is the assigned weight (excluding close file cost) for the subtask,
@@ -215,51 +208,48 @@ public void testEvenlyDividableWithClosingFileCost() {
             Pair.of(100L, 2),
             7,
             Pair.of(75L, 7));
-    Map<Integer, Pair<Long, Integer>> actualAssignmentInfo = partitioner.assignmentInfo();
-    assertThat(actualAssignmentInfo).isEqualTo(expectedAssignmentInfo);
+    assertThat(mapAssignment.assignmentInfo()).isEqualTo(expectedAssignmentInfo);
 
+    MapRangePartitioner partitioner =
+        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapAssignment);
     Map<Integer, Pair<AtomicLong, Set<RowData>>> partitionResults =
-        runPartitioner(partitioner, numPartitions);
+        runPartitioner(partitioner, numPartitions, mapStatistics);
     validatePartitionResults(expectedAssignmentInfo, partitionResults, 5.0);
   }
 
   @Test
   public void testNonDividableNoClosingFileCost() {
-    MapRangePartitioner partitioner =
-        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapStatistics, 0.0);
     int numPartitions = 9;
+    MapAssignment mapAssignment =
+        MapAssignment.fromKeyFrequency(numPartitions, mapStatistics, 0.0, SORT_ORDER_COMPARTOR);
 
     // before:     350, 230, 120, 40, 10, 10, 10, 10, 10, 10
     // each task should get targeted weight of 89 = ceiling(800/9)
-    Map<SortKey, MapRangePartitioner.KeyAssignment> expectedAssignment =
+    Map<SortKey, KeyAssignment> expectedAssignment =
         ImmutableMap.of(
             SORT_KEYS[0],
-            new MapRangePartitioner.KeyAssignment(
+            new KeyAssignment(
                 ImmutableList.of(0, 1, 2, 3), ImmutableList.of(89L, 89L, 89L, 83L), 0L),
             SORT_KEYS[1],
-            new MapRangePartitioner.KeyAssignment(
+            new KeyAssignment(
                 ImmutableList.of(3, 4, 5, 6), ImmutableList.of(6L, 89L, 89L, 46L), 0L),
             SORT_KEYS[2],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(6, 7), ImmutableList.of(43L, 77L), 0L),
+            new KeyAssignment(ImmutableList.of(6, 7), ImmutableList.of(43L, 77L), 0L),
             SORT_KEYS[3],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(7, 8), ImmutableList.of(12L, 28L), 0L),
+            new KeyAssignment(ImmutableList.of(7, 8), ImmutableList.of(12L, 28L), 0L),
             SORT_KEYS[4],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
             SORT_KEYS[5],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
             SORT_KEYS[6],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
             SORT_KEYS[7],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
             SORT_KEYS[8],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L),
             SORT_KEYS[9],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L));
-    Map<SortKey, MapRangePartitioner.KeyAssignment> actualAssignment =
-        partitioner.assignment(numPartitions);
-    assertThat(actualAssignment).isEqualTo(expectedAssignment);
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(10L), 0L));
+    assertThat(mapAssignment.keyAssignments()).isEqualTo(expectedAssignment);
 
     // key: subtask id
     // value pair: first is the assigned weight, second is the number of assigned keys
@@ -283,19 +273,20 @@ public void testNonDividableNoClosingFileCost() {
             Pair.of(89L, 2),
             8,
             Pair.of(88L, 7));
-    Map<Integer, Pair<Long, Integer>> actualAssignmentInfo = partitioner.assignmentInfo();
-    assertThat(actualAssignmentInfo).isEqualTo(expectedAssignmentInfo);
+    assertThat(mapAssignment.assignmentInfo()).isEqualTo(expectedAssignmentInfo);
 
+    MapRangePartitioner partitioner =
+        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapAssignment);
     Map<Integer, Pair<AtomicLong, Set<RowData>>> partitionResults =
-        runPartitioner(partitioner, numPartitions);
+        runPartitioner(partitioner, numPartitions, mapStatistics);
     validatePartitionResults(expectedAssignmentInfo, partitionResults, 5.0);
   }
 
   @Test
   public void testNonDividableWithClosingFileCost() {
-    MapRangePartitioner partitioner =
-        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapStatistics, 5.0);
     int numPartitions = 9;
+    MapAssignment mapAssignment =
+        MapAssignment.fromKeyFrequency(numPartitions, mapStatistics, 5.0, SORT_ORDER_COMPARTOR);
 
     // target subtask weight is 89 before close file cost factored in.
     // close file cost is 5 (= 5% * 89) per file.
@@ -304,35 +295,31 @@ public void testNonDividableWithClosingFileCost() {
     // close-cost:  20,  15,  10,  5,  5,  5,  5,  5,  5,  5
     // after:      370, 245, 130, 45, 15, 15, 15, 15, 15, 15
     // target subtask weight per subtask is 98 ceiling(880/9)
-    Map<SortKey, MapRangePartitioner.KeyAssignment> expectedAssignment =
+    Map<SortKey, KeyAssignment> expectedAssignment =
         ImmutableMap.of(
             SORT_KEYS[0],
-            new MapRangePartitioner.KeyAssignment(
+            new KeyAssignment(
                 ImmutableList.of(0, 1, 2, 3), ImmutableList.of(98L, 98L, 98L, 76L), 5L),
             SORT_KEYS[1],
-            new MapRangePartitioner.KeyAssignment(
+            new KeyAssignment(
                 ImmutableList.of(3, 4, 5, 6), ImmutableList.of(22L, 98L, 98L, 27L), 5L),
             SORT_KEYS[2],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(6, 7), ImmutableList.of(71L, 59L), 5L),
+            new KeyAssignment(ImmutableList.of(6, 7), ImmutableList.of(71L, 59L), 5L),
             SORT_KEYS[3],
-            new MapRangePartitioner.KeyAssignment(
-                ImmutableList.of(7, 8), ImmutableList.of(39L, 6L), 5L),
+            new KeyAssignment(ImmutableList.of(7, 8), ImmutableList.of(39L, 6L), 5L),
             SORT_KEYS[4],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
             SORT_KEYS[5],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
             SORT_KEYS[6],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
             SORT_KEYS[7],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
             SORT_KEYS[8],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L),
             SORT_KEYS[9],
-            new MapRangePartitioner.KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L));
-    Map<SortKey, MapRangePartitioner.KeyAssignment> actualAssignment =
-        partitioner.assignment(numPartitions);
-    assertThat(actualAssignment).isEqualTo(expectedAssignment);
+            new KeyAssignment(ImmutableList.of(8), ImmutableList.of(15L), 5L));
+    assertThat(mapAssignment.keyAssignments()).isEqualTo(expectedAssignment);
 
     // key: subtask id
     // value pair: first is the assigned weight for the subtask, second is the number of keys
@@ -357,40 +344,39 @@ public void testNonDividableWithClosingFileCost() {
             Pair.of(88L, 2),
             8,
             Pair.of(61L, 7));
-    Map<Integer, Pair<Long, Integer>> actualAssignmentInfo = partitioner.assignmentInfo();
-    assertThat(actualAssignmentInfo).isEqualTo(expectedAssignmentInfo);
+    assertThat(mapAssignment.assignmentInfo()).isEqualTo(expectedAssignmentInfo);
 
+    MapRangePartitioner partitioner =
+        new MapRangePartitioner(TestFixtures.SCHEMA, SORT_ORDER, mapAssignment);
     Map<Integer, Pair<AtomicLong, Set<RowData>>> partitionResults =
-        runPartitioner(partitioner, numPartitions);
+        runPartitioner(partitioner, numPartitions, mapStatistics);
     // drift threshold is high for non-dividable scenario with close cost
     validatePartitionResults(expectedAssignmentInfo, partitionResults, 10.0);
   }
 
   private static Map<Integer, Pair<AtomicLong, Set<RowData>>> runPartitioner(
-      MapRangePartitioner partitioner, int numPartitions) {
+      MapRangePartitioner partitioner, int numPartitions, Map<SortKey, Long> mapStatistics) {
     // The Map key is the subtaskId.
     // For the map value pair, the first element is the count of assigned and
     // the second element of Set<String> is for the set of assigned keys.
     Map<Integer, Pair<AtomicLong, Set<RowData>>> partitionResults = Maps.newHashMap();
-    partitioner
-        .mapStatistics()
-        .forEach(
-            (sortKey, weight) -> {
-              String key = sortKey.get(0, String.class);
-              // run 100x times of the weight
-              long iterations = weight * 100;
-              for (int i = 0; i < iterations; ++i) {
-                RowData rowData =
-                    GenericRowData.of(
-                        StringData.fromString(key), 1, StringData.fromString("2023-06-20"));
-                int subtaskId = partitioner.partition(rowData, numPartitions);
-                partitionResults.computeIfAbsent(
-                    subtaskId, k -> Pair.of(new AtomicLong(0), Sets.newHashSet()));
-                Pair<AtomicLong, Set<RowData>> pair = partitionResults.get(subtaskId);
-                pair.first().incrementAndGet();
-                pair.second().add(rowData);
-              }
-            });
+    mapStatistics.forEach(
+        (sortKey, weight) -> {
+          String key = sortKey.get(0, String.class);
+          // run 100x times of the weight
+          long iterations = weight * 100;
+          for (int i = 0; i < iterations; ++i) {
+            RowData rowData =
+                GenericRowData.of(
+                    StringData.fromString(key), 1, StringData.fromString("2023-06-20"));
+            int subtaskId = partitioner.partition(rowData, numPartitions);
+            partitionResults.computeIfAbsent(
+                subtaskId, k -> Pair.of(new AtomicLong(0), Sets.newHashSet()));
+            Pair<AtomicLong, Set<RowData>> pair = partitionResults.get(subtaskId);
+            pair.first().incrementAndGet();
+            pair.second().add(rowData);
+          }
+        });
     return partitionResults;
   }
 
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSketchUtil.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSketchUtil.java
index 435748281f83..31dae5c76aeb 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSketchUtil.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSketchUtil.java
@@ -53,7 +53,7 @@ public void testOperatorReservoirSize() {
   @Test
   public void testRangeBoundsOneChannel() {
     assertThat(
-            SketchUtil.determineBounds(
+            SketchUtil.rangeBounds(
                 1,
                 Fixtures.SORT_ORDER_COMPARTOR,
                 new SortKey[] {
@@ -70,7 +70,7 @@ public void testRangeBoundsOneChannel() {
   @Test
   public void testRangeBoundsDivisible() {
     assertThat(
-            SketchUtil.determineBounds(
+            SketchUtil.rangeBounds(
                 3,
                 Fixtures.SORT_ORDER_COMPARTOR,
                 new SortKey[] {
@@ -88,7 +88,7 @@ public void testRangeBoundsDivisible() {
   public void testRangeBoundsNonDivisible() {
     // step is 3 = ceiling(11/4)
     assertThat(
-            SketchUtil.determineBounds(
+            SketchUtil.rangeBounds(
                 4,
                 Fixtures.SORT_ORDER_COMPARTOR,
                 new SortKey[] {
@@ -111,7 +111,7 @@ public void testRangeBoundsNonDivisible() {
   public void testRangeBoundsSkipDuplicates() {
     // step is 3 = ceiling(11/4)
     assertThat(
-            SketchUtil.determineBounds(
+            SketchUtil.rangeBounds(
                 4,
                 Fixtures.SORT_ORDER_COMPARTOR,
                 new SortKey[] {