stevenzwu commented on code in PR #10331:
URL: https://github.com/apache/iceberg/pull/10331#discussion_r1628013861
##########
flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java:
##########
@@ -30,104 +42,225 @@
* {@link AggregatedStatistics} received from {@link DataStatisticsOperator}
subtasks for specific
* checkpoint.
*/
-class AggregatedStatisticsTracker<D extends DataStatistics<D, S>, S> {
+class AggregatedStatisticsTracker {
private static final Logger LOG =
LoggerFactory.getLogger(AggregatedStatisticsTracker.class);
- private static final double ACCEPT_PARTIAL_AGGR_THRESHOLD = 90;
+
private final String operatorName;
- private final TypeSerializer<DataStatistics<D, S>> statisticsSerializer;
private final int parallelism;
- private final Set<Integer> inProgressSubtaskSet;
- private volatile AggregatedStatistics<D, S> inProgressStatistics;
+ private final TypeSerializer<DataStatistics> statisticsSerializer;
+ private final int downstreamParallelism;
+ private final StatisticsType statisticsType;
+ private final int switchToSketchThreshold;
+ private final Comparator<StructLike> comparator;
+ private final NavigableMap<Long, Aggregation> aggregationsPerCheckpoint;
+
+ private volatile AggregatedStatistics completedStatistics;
Review Comment:
good question. we don't really need `volatile` here as coordinator event
handling is always single thread. let me remove the volatile.
```
this.coordinatorExecutor =
Executors.newSingleThreadExecutor(coordinatorThreadFactory);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
