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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
