irodriguezclaveria commented on code in PR #6426:
URL: https://github.com/apache/paimon/pull/6426#discussion_r3523574229
##########
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:
Hi @JingsongLi ,
Before to modify the code, what do you think about this plan to fix this:
To bound this for refreshes only, without touching the shared pool or other
tasks:
- Add a parameter for the max number of partitions that can be submitted to
the FileOperationThreadPool at a time. Default is 1 (strictly bounded, safe),
so the refresh memory peak is at most one partition's worth (scanEntries +
tempBucketInfo); the user can increase it if their setup can handle more
concurrent refreshes.
- Keep a pending structure with no duplicates — if a partition is already
pending, it won't be re-added (coalescing). It will be a FIFO queue (first in,
first out) for fairness.
- Drain it from each PartitionIndex's own future whenComplete, so only
PartitionIndex is affected. The in-flight counter and pending set are static
(global across partitions) but only touched from each partition's local
callback.
- I'll also log a warning if refreshes keep chaining continuously.
Let me know what do you think.
--
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]
