You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "irodriguezclaveria (via GitHub)" <gi...@apache.org> on 2025/10/17 14:10:57 UTC
[PR] Update index before create bucket [paimon]
irodriguezclaveria opened a new pull request, #6426:
URL: https://github.com/apache/paimon/pull/6426
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4995626353
Hi @JingsongLi ,
You're right — real compaction doesn't shrink the HASH index today. DynamicBucketIndexMaintainer only ever calls add() on the hash set, and compact() in AbstractFileStoreWrite doesn't even have a reference to it. So there's no real path where compaction frees up a bucket in the current code, and the existing tests only prove it works if you manually commit a smaller file — not something that happens in practice.
New plan: instead of trying to make compaction shrink the HASH index, we're adding a separate, lightweight structure that tracks live primary keys per bucket. It's updated right in notifyNewRecord, using the record's RowKind: insert/update adds the key, delete removes it. This gives us an always-accurate "how many live keys does this bucket have" count without touching the HASH index or waiting for compaction at all.
This extra structure only exists when the bucket-refresh feature is enabled (minEmptyBucketsBeforeAsyncCheck != -1). If refresh is off, we skip it entirely — no extra cost for users not using this feature.
This also fixes the other issues you raised, since they're all symptoms of relying on an async disk re-scan:
- No more double-counting, since there's no more diskRowCountAtLastRefresh delta to reconcile — the in-memory counter is the single source of truth.
- No more missed refresh when all non-full buckets disappear — there's no refresh trigger needed anymore, freed buckets are just visible directly in memory.
- No more buckets silently dropped when they vanish from the HASH manifest — we don't depend on the manifest at all.
- No more async refresh/backpressure complexity — no disk scans, no thread pool involved.
Let me know if this direction works for you before we start implementing.
Thanks!!
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi closed pull request #6426: [core] Add async bucket refresh for dynamic bucket mode
URL: https://github.com/apache/paimon/pull/6426
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4778563609
Hi @irodriguezclaveria, thanks for checking this carefully.
You are right that, for the compaction-only case, if a bucket was loaded by `loadIndex` and later removed from `nonFullBucketInformation` only because it became full, then the bucket id is already in `totalBucketSet` / `totalBucketArray`, and its hashes have already been loaded into `hash2Bucket`. In that narrower model, we do not need to update the total bucket structures or reload hash mappings during refresh.
The scenario I had in mind is the broader one where `refreshBucketsFromDisk` scans the latest index manifests and sees a bucket that was created after this `PartitionIndex` was loaded. That bucket may pass `bucketFilter`, but it is not in `totalBucketSet`, and its hash mappings were not read by the original `loadIndex`. If refresh adds such a bucket directly to `nonFullBucketInformation`, the in-memory state becomes inconsistent.
Since this PR is about reusing buckets freed by compaction, not discovering newly-created buckets from another writer, I think the conservative fix is to make that invariant explicit: during refresh, only merge buckets that already exist in `totalBucketSet` (and still satisfy `bucketFilter`). Then the hash mapping argument holds as you described. If we later want refresh to import newly-created buckets too, then it should also update `totalBucketSet` / `totalBucketArray` and load the hash index contents for those buckets.
For the concurrency/conservativeness question: off by default + 24h default interval + per-partition single-flight is enough for me. I do not think we need a separate executor or an additional global cap in this PR. I would just keep the refresh work bounded by the shared file-operation pool and avoid extra fan-out if possible.
For tests, please make sure the regression really loads the partition before the later on-disk change. Then verify a known bucket that became full/was removed from `nonFullBucketInformation` can be re-added after compaction lowers its row count. A second test could add a new bucket after load and verify refresh does not import it, unless we explicitly decide to support that broader behavior.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4644908131
I took another pass specifically from the perspective of whether this change is intrusive to existing paths.
The default path looks mostly non-intrusive: `dynamic-bucket.empty-bucket-threshold` defaults to `-1`, so existing jobs that do not set the new option should not enter the async refresh branch. However, once the option is enabled, the new code becomes part of the bucket assignment state machine rather than a purely side-channel optimization. I think there are still correctness risks before this is safe to merge:
1. Refreshed buckets are added only to `nonFullBucketInformation`, but not to `totalBucketSet` / `totalBucketArray`.
In `PartitionIndex.refreshBucketsFromDisk`, newly discovered buckets are merged with `nonFullBucketInformation.putIfAbsent(...)`, but the same buckets are not added to the structures used by the new-bucket creation and fallback paths. After such a refreshed bucket is consumed and removed from `nonFullBucketInformation`, the later creation path can still see it as absent from `totalBucketSet` and may try to create/reuse the same bucket id as if it were new. That makes the in-memory bucket state inconsistent.
2. Refresh only imports bucket row counts, not the hash-to-bucket mapping.
The initial `loadIndex` path reads each hash index file and fills `hash2Bucket`, which preserves the invariant that an existing key hash is always assigned back to its original bucket. The async refresh path only uses `IndexFileMeta.rowCount()` and does not read the hash index contents for newly discovered buckets. If a refreshed bucket already contains keys written by another writer or visible in a newer snapshot, this assigner can later treat those hashes as unseen and assign them elsewhere.
3. The enabled path also shares the global `FileOperationThreadPool` for manifest/index scans. That is probably acceptable if intentionally throttled, but it means the feature can affect other file operations once enabled, so the option and interval need to be conservative.
Given the above, I would suggest fixing the refresh merge so it updates all bucket state consistently, and either loads hash mappings for refreshed buckets or clearly proves that refreshed buckets can only be empty for this assigner. It would also be good to add tests for: (a) a refreshed bucket becoming full and not being re-created as a new bucket, and (b) an existing hash in a refreshed bucket still being assigned back to that bucket.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4562074883
Hi @JingsongLi ,
Let me answer your points:
- **Thread pool design**: The executor was already global (static final), but you're right that it was a dedicated one. Switched to the shared FileOperationThreadPool and removed the dedicated REFRESH_EXECUTOR and its static initializer.
- **Race condition:** Yes, exactly that window. The ideal is to ensure the threshold is large enough that the refresh has time to complete before existing buckets saturate. I add a WARN log if the value is below the recommended 5–10% of target-row-num and in the description of the parameter let me know if you considerer that this is enough or I can do a more stricly validattion.
- **621 additions**: Most of the 621 lines are tests and auto-generated docs. The actual feature logic is in paimon-core (PartitionIndex), and both engines are thin wrappers that just forward the new options.
Only two production sites construct HashBucketAssigner — HashBucketAssignerOperator (Flink) and BucketProcessor (Spark) — and both pass the new parameters identically.
- **Timeout (5 min)**: Sorry it was a mistake in my documentation, corrected. I considered adding an application-level timeout first, but Future.cancel(true) doesn't interrupt blocked native I/O — it would mark the future as done (allowing retries) but the underlying thread would still be held by FileIO until its timeout fires. Happy to add it as an opt-in if you'd prefer; otherwise I'll just document this in the javadoc.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4661465017
Hi @JingsongLi ,
Thanks for the detailed write-up. Before making any changes, I'd like to check a few things with you, because I think these may already be prevented by existing invariants and I want to make sure I'm not missing something.
1. On adding refreshed buckets to totalBucketSet / totalBucketArray:
Every bucket id the refresh can surface was already loaded by loadIndex (same partition, same bucketFilter), and nothing in this class ever removes from totalBucketSet or totalBucketArray. So when we hit putIfAbsent, the id is already in both. Is there a path you have in mind where an id could appear in the refresh without ever having been in totalBucketSet?
2. On loading hash mappings for refreshed buckets:
Hashes on disk under one of this assigner's buckets were either ingested by loadIndex (which does read full hash content) or written by this assigner during its current lifetime (and inserted into hash2Bucket at assignment time). Under the ownership model the bucketFilter enforces, I don't see a third case. Do you have a concrete scenario in mind where a hash could exist on disk in one of our buckets without already being in hash2Bucket?
3. Does the combination of "off by default + 24h default interval + per-partition single-flight" already meet the bar you had in mind for "conservative", or were you thinking of additional safeguards (e.g. a global cap on concurrent refreshes across partitions, or moving off the shared FileOperationThreadPool)?
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-5083436160
Asynchronous formula refreshes result in duplicate calculations of memory increments that have already been committed. In regression testing, the disk count increased from 2 to 3, and after the refresh, the error count became 4. Additionally, the issues I previously pointed out—that compaction does not decrement the count, empty candidates are not refreshed, and vanished buckets are not reset to zero—remain unresolved.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
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.
- 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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4827137097
Hi @JingsongLi,
Thanks for the detailed review. I have now fixed the two points you raised.
1. Refresh now updates buckets that are already present, not only absent ones
Just to clarify the intent: the goal of the refresh has always been to refresh all the buckets, not only to add new (absent) ones. When I switched to put I accidentally lost that behavior, so present buckets with a stale count were no longer being refreshed.
To preserve the uncommitted in-memory increments you mentioned, I track per bucket the disk row count seen at the last reconciliation (diskRowCountAtLastRefresh). The session delta is inMemory - lastDiskCount. On refresh I set the new value to diskNow + sessionDelta using an atomic compute, so:
- if compaction lowered the disk count, the bucket regains capacity, and
- the uncommitted in-memory increments are preserved.
I also kept the conservative invariant: refresh only reconciles buckets already in totalBucketSet, so a bucket created after load is not imported.
2. Tests
- Fixed the regression test so the partition is really loaded before the on-disk change. Before, the lazy loadIndex ran on the first assign (after the compaction commit), so the bucket was loaded synchronously instead of being discovered by the refresh.
- Added a test where a bucket already present is reconciled down after compaction (this one fails with putIfAbsent and passes with the fix).
- Added a test that a bucket created after load is not imported by the refresh.
Thanks again!
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-5159517771
Hi @JingsongLi,
I completely agree with your assessment! The test failures and bugs you just mentioned are exactly why I realized the current PR implementation (using async refreshes and the HASH index) is fundamentally flawed and should not be merged.
That is why my previous message was proposing a completely new design to replace this PR. The new plan (tracking live keys entirely in memory via RowKind insert/deletes) completely eliminates async refreshes, disk scans, and HASH index dependencies. By removing those mechanics, it naturally prevents the double-counting and vanished-bucket issues you ran into during regression testing.
I haven't written the code for this new design yet. Could you take a look at the conceptual plan in my previous comment? I'd love to get your thoughts on that new direction before I start implementing it
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4850904892
看了当前 head `cec7398932a21b2cc5a3a4e95e4e5040d9514fc9`，我的判断是：这个方向确实可以做异步优化，默认 `dynamic-bucket.empty-bucket-threshold = -1` 也让风险面默认关闭；但当前实现还不能确认线程安全和内存/backpressure 都可控，建议先修下面几个点再合。
1. `PartitionIndex.assign` 里对 `nonFullBucketInformation` 的更新仍然是 read-modify-write：先从 iterator 读 `number`，再 `entry.setValue(number + 1)`。后台 refresh 同时会在 `reconcileBucketWithDisk` 里 `compute` 同一个 bucket。两者交错时，主线程可能覆盖已经 reconcile 后的值；更严重的是主线程在 bucket 满时 `iterator.remove()` 并删除 `diskRowCountAtLastRefresh`，慢 refresh 回来后可能用 `inMemory == null ? diskNow` 把当前 checkpoint 的 in-flight 增量丢掉，导致已满 bucket 又被当作 non-full。这里最好把 assign 侧也改成基于 `compute`/CAS 的原子更新，避免 `entry.setValue` 覆盖异步 reconcile。
2. `totalBucketSet` 仍然是 `LinkedHashSet`。主线程创建 bucket 时会写 `totalBucketSet` / `totalBucketArray`，后台 refresh 在 `reconcileBucketWithDisk` 里读 `totalBucketSet.contains(bucket)`。这是普通集合的跨线程读写 data race。可以考虑 refresh 启动时复制一个 owned-bucket snapshot，或者把这部分结构改成明确的并发容器/同步边界。
3. `lastRefreshTime` 由后台线程写、assign 线程读，但不是 `volatile`，`min-refresh-interval` 的可见性不可靠。这个至少应该改成 `volatile`，或和 `refreshFuture` 的状态更新放进同一个同步/原子流程里。
4. 内存/backpressure 目前只能说“部分可控”，不能说严格有界。每个 `PartitionIndex` 最多保留一个 `refreshFuture`，运行中的 refresh 并发度受全局 `FileOperationThreadPool` 线程数限制，这是好的一面；但 `FileOperationThreadPool` 底层默认是无界 `LinkedBlockingQueue`。如果很多活跃分区同时 near-full，排队的 refresh task 数没有硬上限。每个运行中的 refresh 还会持有 `scanEntries` 结果和 `tempBucketInfo`，运行中并发有限，但单个大分区扫描的临时内存仍可能比较大。
所以我认可异步 refresh 的优化目标，但这版还不能直接说“没有线程安全问题，且内存问题可控”。建议补一个并发交错测试/压力测试，覆盖 assign 与 reconcile 同时更新同一 bucket、bucket remove 后慢 refresh 回写、以及大量分区同时触发 refresh 的场景。
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4882758848
Hi @JingsongLi,
First of all, thanks for reviewing my PR. I would love to include this in the next release, so I will make the necessary changes to ensure everything is correct.
Thanks for pointing out the errors. I have been working on the points you mentioned 3 days ago. Let me carefully review all your comments, and I will notify you when the PR is ready to be reviewed again.
Thanks again, I'm very glad we are moving this forward.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4065706151
Pull requests has been added.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4533166564
Hi @JingsongLi , thanks I'm going to review your points and I will rebase the branch. Thanks to review it !
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #6426:
URL: https://github.com/apache/paimon/pull/6426#discussion_r3459214616
##########
paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java:
##########
@@ -150,6 +189,145 @@ 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) {
+ // Only refresh if:
+ // 1. Feature is enabled (minEmptyBucketsBeforeAsyncCheck != -1)
+ // 2. Current bucket is approaching its target capacity
+ // When bucket reaches (targetBucketRowNumber - minEmptyBucketsBeforeAsyncCheck),
+ // trigger refresh to find buckets freed by compaction
+ 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));
+ }
+
+ /**
+ * Cancel any ongoing refresh operation. Should be called when this PartitionIndex is no longer
+ * needed (e.g., during 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);
+ }
+ }
+ }
+
+ private void refreshBucketsFromDisk(IntPredicate bucketFilter) {
+ // Only start refresh if not already in progress
+ if (refreshFuture == null || refreshFuture.isDone()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Triggering async refresh of bucket information for partition {}. "
+ + "Current non-full buckets: {}",
+ partition,
+ nonFullBucketInformation.size());
+ }
+
+ // Reuse the shared FileOperationThreadPool instead of creating a dedicated
+ // executor: refresh is an I/O-bound file operation (scanning index manifest
+ // entries) and the shared pool already exists for this purpose. This avoids
+ // duplicating thread-pool resources and aligns with how other file-scan paths
+ // in Paimon (FileDeletionBase, TableCommitImpl, ListUnexistingFiles) work.
+ refreshFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ List<IndexManifestEntry> files =
+ indexFileHandler.scanEntries(
+ HASH_INDEX, partition);
+
+ // Use parallel stream to scan multiple files
+ // concurrently
+ // This reduces latency when dealing with many buckets.
+ // Apply bucketFilter so we only surface buckets owned
+ // by this assigner; otherwise multiple assigners would
+ // race for buckets they do not own.
+ Map<Integer, Long> tempBucketInfo =
+ files.parallelStream()
+ .filter(
+ file ->
+ bucketFilter.test(
+ file.bucket()))
+ .filter(
+ file ->
+ file.indexFile()
+ .rowCount()
+ < targetBucketRowNumber)
+ .collect(
+ Collectors.toMap(
+ IndexManifestEntry
+ ::bucket,
+ file ->
+ file.indexFile()
+ .rowCount(),
+ (existing,
+ replacement) ->
+ existing));
+
+ // Use putIfAbsent to avoid race conditions
+ int newBucketsFound = 0;
+ for (Map.Entry<Integer, Long> entry :
+ tempBucketInfo.entrySet()) {
+ Long previous =
+ nonFullBucketInformation.putIfAbsent(
Review Comment:
This still misses the main "freed before saturation" case. If the partition was already loaded with a bucket at, say, 4/5 rows and compaction rewrites the hash index down to 1 row before the assigner removes that bucket from `nonFullBucketInformation`, the async refresh will scan the new 1-row index but `putIfAbsent` leaves the existing 4-row value untouched. `lastRefreshTime` is then advanced, so with the default 24h interval the assigner keeps the stale near-full count, removes the bucket after one more assignment, and creates another bucket instead of using the freed capacity. The refresh merge needs to handle already-present buckets too (while preserving any uncommitted in-memory increments), not only absent ones.
##########
paimon-core/src/test/java/org/apache/paimon/index/HashBucketAssignerTest.java:
##########
@@ -363,4 +368,334 @@ public void testIndexEliminate() throws IOException {
assigner.prepareCommit(3);
assertThat(assigner.currentPartitions()).isEmpty();
}
+
+ /**
+ * Test that bucket refresh is triggered when a bucket approaches target capacity, that buckets
+ * newly added on disk become discoverable by subsequent assignments, AND that the refresh
+ * respects assigner ownership (only buckets owned by this assigner are surfaced).
+ *
+ * <p>Setup uses two assigners: assigner 0 owns even buckets (0, 2, ...), assigner 1 owns odd
+ * buckets (1, 3, ...). Without the fix:
+ *
+ * <ul>
+ * <li>The async refresh would never run, so bucket 2 (added after load) would stay invisible.
+ * <li>Even if the refresh ran, it would surface bucket 1 (owned by assigner 1) into assigner
+ * 0's in-memory map, breaking the ownership invariant.
+ * </ul>
+ */
+ @Test
+ public void testRefreshTriggeredWhenBucketNearFull() throws IOException {
+ // Initial on-disk state seen by assigner 0 at load time: only bucket 0 (its own) with
+ // 2 rows. Bucket 1 also exists on disk but is owned by assigner 1.
+ commit.commit(
+ 0,
+ Arrays.asList(
+ createCommitMessage(
+ row(1),
+ 0,
+ 2,
+ fileHandler.hashIndex(row(1), 0).write(new int[] {0, 2})),
+ createCommitMessage(
+ row(1),
+ 1,
+ 2,
+ fileHandler.hashIndex(row(1), 1).write(new int[] {1, 3}))));
+
+ // numChannels=2, numAssigners=2, assignId=0 -> assigner 0 owns even buckets only.
+ // targetBucketRowNumber=5, threshold=2 -> refresh triggers when a bucket reaches 3 rows.
+ HashBucketAssigner assigner =
+ new HashBucketAssigner(
+ table.snapshotManager(),
+ commitUser,
+ fileHandler,
+ 2,
+ 2,
+ 0,
+ 5,
+ -1,
+ 2,
+ Duration.ofMillis(1));
+
+ // After the assigner has loaded, simulate another writer (or compaction) committing a
+ // new even bucket (2) that this assigner does not yet know about. We also commit
+ // additional rows to bucket 1 (owned by assigner 1) to make sure the refresh has
+ // something to filter out.
+ commit.commit(
Review Comment:
This setup does not actually commit after the assigner has loaded this partition. `HashBucketAssigner` creates the `PartitionIndex` lazily on the first `assign`, which happens below, so bucket 2 is loaded synchronously by `loadIndex` from this commit rather than discovered by the async refresh. That leaves the regression untested (and masks stale-count cases like the one above). Please force an assignment/load before this commit, then commit the refreshed index and assert that the bucket appears only after the async refresh.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-5748067837
I reviewed the current implementation and the discussion history.
**Current implementation: UNSUPPORTED; closing this PR.** The optimization goal is potentially useful, but this branch cannot provide a correct end-to-end implementation. The author has already confirmed that the asynchronous refresh plus HASH-index design is fundamentally flawed and should not be merged.
The unresolved issues are structural rather than polish: real compaction does not shrink the HASH index, reconciliation can double-count, empty/vanished candidates are not reliably reset, and asynchronous refresh introduces backpressure and recovery hazards. Keeping this implementation open would suggest a viable merge path when the required replacement design has not been coded.
A fresh PR would be welcome for the proposed live-key/in-memory direction once it has a durable recovery story and end-to-end tests covering compaction, bucket disappearance, restart, and concurrent refresh. Closing this obsolete implementation so the new design can start from a clear baseline.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4528923003
Please rebase master.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add async bucket refresh for dynamic bucket mode [paimon]
Posted by "irodriguezclaveria (via GitHub)" <gi...@apache.org>.
irodriguezclaveria commented on PR #6426:
URL: https://github.com/apache/paimon/pull/6426#issuecomment-4642880703
Hi @JingsongLi , thanks for your accurate review, I did the requested changes. Thanks !!!
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org