JingsongLi commented on code in PR #6426:
URL: https://github.com/apache/paimon/pull/6426#discussion_r3523333943
##########
paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java:
##########
@@ -150,6 +196,97 @@ public static PartitionIndex loadIndex(
throw new UncheckedIOException(e);
}
}
- return new PartitionIndex(mapBuilder.build(), buckets,
targetBucketRowNumber);
+ return new PartitionIndex(
+ mapBuilder.build(), buckets, targetBucketRowNumber,
indexFileHandler, partition);
+ }
+
+ private boolean shouldRefreshWhenBucketNearFull(
+ long currentBucketRowCount,
+ long targetBucketRowNumber,
+ int minEmptyBucketsBeforeAsyncCheck) {
+ if (minEmptyBucketsBeforeAsyncCheck == -1) {
+ return false;
+ }
+
+ long threshold = targetBucketRowNumber -
minEmptyBucketsBeforeAsyncCheck;
+ return currentBucketRowCount >= threshold;
+ }
+
+ private boolean isReachedTheMinRefreshInterval(final Duration duration) {
+ return lastRefreshTime == null ||
Instant.now().isAfter(lastRefreshTime.plus(duration));
+ }
+
+ /** Called when this PartitionIndex is dropped (e.g. cleanup in
prepareCommit). */
+ public void cancelOngoingRefresh() {
+ if (refreshFuture != null && !refreshFuture.isDone()) {
+ refreshFuture.cancel(false);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cancelled ongoing refresh for partition {}",
partition);
+ }
+ }
+ }
+
+ // Merges disk row count into nonFullBucketInformation, keeping
uncommitted increments.
+ private void reconcileBucketWithDisk(int bucket, long diskNow) {
+ if (!totalBucketSet.contains(bucket)) {
+ return;
+ }
+ nonFullBucketInformation.compute(
+ bucket,
+ (b, inMemory) ->
+ inMemory == null
+ ? diskNow
+ : diskNow
+ + (inMemory
+ -
diskRowCountAtLastRefresh.getOrDefault(
+ b, inMemory)));
+ diskRowCountAtLastRefresh.put(bucket, diskNow);
+ }
+
+ private void refreshBucketsFromDisk(IntPredicate bucketFilter) {
+ if (refreshFuture == null || refreshFuture.isDone()) {
+ // Reuse the shared FileOperationThreadPool (I/O-bound scan)
rather than a dedicated
+ // executor, to avoid extra thread-pool resources and fan-out.
+ refreshFuture =
+ CompletableFuture.runAsync(
+ () -> scanAndReconcile(bucketFilter),
+ FileOperationThreadPool.getExecutorService(
Review Comment:
This still does not put a hard bound on queued refresh work.
`FileOperationThreadPool.getExecutorService(...)` uses the shared cached pool
backed by `ThreadPoolUtils.createCachedThreadPool`, whose default queue is an
unbounded `LinkedBlockingQueue`. Since every active `PartitionIndex` can
enqueue one refresh, a job with many near-full partitions can build an
unbounded backlog of disk-scan tasks. Please add a global
bound/backoff/coalescing policy, or otherwise make the refresh queue size
controlled.
##########
paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java:
##########
@@ -150,6 +196,97 @@ public static PartitionIndex loadIndex(
throw new UncheckedIOException(e);
}
}
- return new PartitionIndex(mapBuilder.build(), buckets,
targetBucketRowNumber);
+ return new PartitionIndex(
+ mapBuilder.build(), buckets, targetBucketRowNumber,
indexFileHandler, partition);
+ }
+
+ private boolean shouldRefreshWhenBucketNearFull(
+ long currentBucketRowCount,
+ long targetBucketRowNumber,
+ int minEmptyBucketsBeforeAsyncCheck) {
+ if (minEmptyBucketsBeforeAsyncCheck == -1) {
+ return false;
+ }
+
+ long threshold = targetBucketRowNumber -
minEmptyBucketsBeforeAsyncCheck;
+ return currentBucketRowCount >= threshold;
+ }
+
+ private boolean isReachedTheMinRefreshInterval(final Duration duration) {
+ return lastRefreshTime == null ||
Instant.now().isAfter(lastRefreshTime.plus(duration));
+ }
+
+ /** Called when this PartitionIndex is dropped (e.g. cleanup in
prepareCommit). */
+ public void cancelOngoingRefresh() {
+ if (refreshFuture != null && !refreshFuture.isDone()) {
+ refreshFuture.cancel(false);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cancelled ongoing refresh for partition {}",
partition);
+ }
+ }
+ }
+
+ // Merges disk row count into nonFullBucketInformation, keeping
uncommitted increments.
+ private void reconcileBucketWithDisk(int bucket, long diskNow) {
+ if (!totalBucketSet.contains(bucket)) {
Review Comment:
This is still a data race on a non-thread-safe collection. `assign` mutates
`totalBucketSet` when it creates a new bucket, while the background refresh can
read `totalBucketSet.contains(bucket)` here at the same time. The map update is
now atomic, but this `LinkedHashSet` access is not. Please either capture an
immutable owned-bucket snapshot when scheduling the refresh, or make the
total-bucket structures explicitly thread-safe/synchronized.
##########
paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java:
##########
@@ -54,20 +66,41 @@ public class PartitionIndex {
public long lastAccessedCommitIdentifier;
+ private final IndexFileHandler indexFileHandler;
+
+ private final BinaryRow partition;
+
+ private volatile CompletableFuture<Void> refreshFuture;
+
+ // null = never refreshed; the first refresh is not throttled by
min-refresh-interval.
+ private Instant lastRefreshTime;
Review Comment:
This timestamp is still written by the async refresh thread and read by the
assign thread without any visibility guarantee. The latest `computeIfPresent`
change fixes the bucket-count update race, but `lastRefreshTime` can still be
stale on the assign side, so `dynamic-bucket.min-refresh-interval` may be
ignored or applied late. Please make this field `volatile` or guard the refresh
state and timestamp with the same synchronization/atomic path.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
