This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 9903b6d83fbb feat(flink): support record index handling in dynamic
bucket assign function (#19837)
9903b6d83fbb is described below
commit 9903b6d83fbb59eecf15a5a76206c11797b6239b
Author: Peter Huang <[email protected]>
AuthorDate: Sun Sep 6 23:50:39 2026 -0700
feat(flink): support record index handling in dynamic bucket assign
function (#19837)
---
.../partitioner/DynamicBucketAssignFunction.java | 10 ++
.../index/DummyPartitionedIndexBackend.java | 5 +
.../partitioner/index/PartitionedIndexBackend.java | 14 +++
.../partitioner/index/RecordLevelIndexBackend.java | 24 +++++
.../TestDynamicBucketAssignFunction.java | 108 +++++++++++++++++++++
.../sink/partitioner/index/TestIndexBackends.java | 3 +
.../index/TestRecordLevelIndexBackend.java | 42 ++++++++
7 files changed, 206 insertions(+)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/DynamicBucketAssignFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/DynamicBucketAssignFunction.java
index 4ad34bdc406c..f1a2921ec0dd 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/DynamicBucketAssignFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/DynamicBucketAssignFunction.java
@@ -55,6 +55,11 @@ import org.apache.flink.util.Collector;
* {@code recordKey -> fileGroupId} mapping. Existing keys are routed as
updates to
* the recorded file group; new keys are assigned by {@link BucketAssigner}
and then
* written back to the backend so the streaming metadata writer can persist
the assignment to RLI.
+ *
+ * <p>Preloaded index records emitted by the bootstrap operators (e.g. {@code
RLIBootstrapOperator},
+ * {@code TimeBoundedRLIBootstrapOperator}) carry an already known {@code
recordKey -> fileGroupId}
+ * mapping and no row data. They are applied directly to the index backend to
warm up the cache and
+ * are never routed through {@link BucketAssigner} or emitted downstream.
*/
public class DynamicBucketAssignFunction
extends KeyedProcessFunctionAdapter<String, HoodieFlinkInternalRow,
HoodieFlinkInternalRow>
@@ -118,6 +123,11 @@ public class DynamicBucketAssignFunction
@Override
public void processElement(HoodieFlinkInternalRow record, Context ctx,
Collector<HoodieFlinkInternalRow> out) throws Exception {
+ if (record.isIndexRecord()) {
+ indexBackend.bootstrap(record.getPartitionPath(), record.getRecordKey(),
record.getFileId());
+ return;
+ }
+
String partitionPath = record.getPartitionPath();
String recordKey = record.getRecordKey();
String fileGroupId = indexBackend.get(partitionPath, recordKey);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/DummyPartitionedIndexBackend.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/DummyPartitionedIndexBackend.java
index 2fe8c4784294..b555aab806c6 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/DummyPartitionedIndexBackend.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/DummyPartitionedIndexBackend.java
@@ -39,6 +39,11 @@ public class DummyPartitionedIndexBackend implements
PartitionedIndexBackend {
// do nothing
}
+ @Override
+ public void bootstrap(String partitionPath, String recordKey, String fileId)
{
+ // do nothing
+ }
+
@Override
public void close() throws IOException {
// do nothing
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/PartitionedIndexBackend.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/PartitionedIndexBackend.java
index 3a2e583f41d9..170985767c4b 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/PartitionedIndexBackend.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/PartitionedIndexBackend.java
@@ -44,4 +44,18 @@ public interface PartitionedIndexBackend extends
IndexBackend {
* @param fileId the file group id, usually the file id prefix produced by
bucket assignment
*/
void update(String partitionPath, String recordKey, String fileId);
+
+ /**
+ * Applies a preloaded {@code recordKey -> fileId} mapping to the given
partition, e.g. one emitted
+ * by a bootstrap operator that has already scanned the persisted index.
+ *
+ * <p>Unlike {@link #update}, this does not trigger a scan of the persisted
index to warm up the
+ * partition cache when it does not exist yet, and the mapping is not marked
as a normal data
+ * update.
+ *
+ * @param partitionPath the partition path of the record
+ * @param recordKey the unique key identifying the record
+ * @param fileId the file group id already known for the record key
+ */
+ void bootstrap(String partitionPath, String recordKey, String fileId);
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordLevelIndexBackend.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordLevelIndexBackend.java
index 783a3bd83bef..3d19c2061ceb 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordLevelIndexBackend.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordLevelIndexBackend.java
@@ -121,6 +121,12 @@ public class RecordLevelIndexBackend implements
PartitionedIndexBackend {
cleanExecutor.runIfNecessary(() -> cleanIfNecessary(0L, partitionPath));
}
+ @Override
+ public void bootstrap(String partitionPath, String recordKey, String fileId)
{
+ BucketCache cache = getOrCreatePartitionCache(partitionPath);
+ cache.bootstrapRecordKey(recordKey, fileId);
+ }
+
/**
* Records the latest checkpoint id seen by this assign subtask.
*
@@ -163,6 +169,24 @@ public class RecordLevelIndexBackend implements
PartitionedIndexBackend {
return cache;
}
+ /**
+ * Returns the partition cache, creating an empty one without scanning the
persisted index if it
+ * does not exist yet.
+ *
+ * <p>Used for preloaded mappings that are already known to be complete for
the partition, e.g. one
+ * emitted by a bootstrap operator that has already scanned the persisted
index upstream.
+ */
+ private BucketCache getOrCreatePartitionCache(String partitionPath) {
+ BucketCache cache = partitionBucketCaches.get(partitionPath);
+ if (cache != null) {
+ return cache;
+ }
+
+ cache = createBucketCache(partitionPath);
+ partitionBucketCaches.put(partitionPath, cache);
+ return cache;
+ }
+
private BucketCache bootstrapPartition(String partitionPath) {
BucketCache cache = createBucketCache(partitionPath);
if
(!metaClient.getTableConfig().isMetadataPartitionAvailable(MetadataPartitionType.RECORD_INDEX))
{
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestDynamicBucketAssignFunction.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestDynamicBucketAssignFunction.java
index 277b5e61e4f9..51e9a5d176e0 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestDynamicBucketAssignFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestDynamicBucketAssignFunction.java
@@ -22,6 +22,7 @@ import org.apache.hudi.client.model.HoodieFlinkInternalRow;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.table.view.FileSystemViewStorageType;
import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.sink.event.Correspondent;
import org.apache.hudi.sink.partitioner.index.DummyPartitionedIndexBackend;
import org.apache.hudi.sink.partitioner.index.PartitionedIndexBackend;
@@ -49,6 +50,7 @@ import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -106,6 +108,24 @@ class TestDynamicBucketAssignFunction {
verify(indexBackend).update("partition", "key", "new-file");
}
+ @Test
+ void testProcessIndexRecordBootstrapsIndexBackendWithoutBucketAssignment()
throws Exception {
+ DynamicBucketAssignFunction function = new DynamicBucketAssignFunction(new
Configuration());
+ PartitionedIndexBackend indexBackend = mock(PartitionedIndexBackend.class);
+ BucketAssigner bucketAssigner = mock(BucketAssigner.class);
+ setField(function, "indexBackend", indexBackend);
+ setField(function, "bucketAssigner", bucketAssigner);
+ HoodieFlinkInternalRow indexRecord = new HoodieFlinkInternalRow("key",
"partition", "existing-file", "20260101000000");
+ List<HoodieFlinkInternalRow> output = new ArrayList<>();
+
+ function.processElement(indexRecord, null, collector(output));
+
+ assertEquals(0, output.size(), "Index records carry no row data and must
not be emitted downstream");
+ verify(indexBackend).bootstrap("partition", "key", "existing-file");
+ verify(bucketAssigner, never()).addInsert("partition");
+ verify(bucketAssigner, never()).addUpdate("partition", "existing-file");
+ }
+
@Test
void testCheckpointLifecycleDelegatesToBackends() throws Exception {
DynamicBucketAssignFunction function = new DynamicBucketAssignFunction(new
Configuration());
@@ -166,6 +186,94 @@ class TestDynamicBucketAssignFunction {
function.close();
}
+ @Test
+ void testBootstrappedIndexRecordIsReusedByLaterUpdate() throws Exception {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ StreamerUtil.initTableIfNotExists(conf);
+ ViewStorageProperties.createProperties(
+ conf.get(FlinkOptions.PATH),
+ FileSystemViewStorageConfig.newBuilder()
+ .withStorageType(FileSystemViewStorageType.SPILLABLE_DISK)
+ .build(),
+ conf);
+ DynamicBucketAssignFunction function = new
DynamicBucketAssignFunction(conf);
+ function.setRuntimeContext(new MockStreamingRuntimeContext(false, 1, 0));
+ function.open(new Configuration());
+ function.initializeState(mock(FunctionInitializationContext.class));
+
+ try {
+ // Simulates the preloaded index record emitted upstream by
RLIBootstrapOperator /
+ // TimeBoundedRLIBootstrapOperator for a record whose file group mapping
is already known.
+ HoodieFlinkInternalRow indexRecord =
+ new HoodieFlinkInternalRow("key", "partition", "bootstrapped-file",
"20260101000000");
+ List<HoodieFlinkInternalRow> indexOutput = new ArrayList<>();
+ function.processElement(indexRecord, null, collector(indexOutput));
+ assertEquals(0, indexOutput.size(), "Index record must not be forwarded
to the writer");
+
+ // A later update for the same key should be routed to the bootstrapped
file group instead of
+ // being assigned a brand-new bucket.
+ HoodieFlinkInternalRow updateRecord = record("key", "partition", "U");
+ List<HoodieFlinkInternalRow> updateOutput = new ArrayList<>();
+ function.processElement(updateRecord, null, collector(updateOutput));
+
+ assertEquals(1, updateOutput.size());
+ assertEquals("bootstrapped-file", updateRecord.getFileId());
+ assertEquals("U", updateRecord.getInstantTime());
+ } finally {
+ function.close();
+ }
+ }
+
+ @Test
+ void testBootstrappedIndexRecordsDoNotAffectUnrelatedKeysInSamePartition()
throws Exception {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ StreamerUtil.initTableIfNotExists(conf);
+ ViewStorageProperties.createProperties(
+ conf.get(FlinkOptions.PATH),
+ FileSystemViewStorageConfig.newBuilder()
+ .withStorageType(FileSystemViewStorageType.SPILLABLE_DISK)
+ .build(),
+ conf);
+ DynamicBucketAssignFunction function = new
DynamicBucketAssignFunction(conf);
+ function.setRuntimeContext(new MockStreamingRuntimeContext(false, 1, 0));
+ function.open(new Configuration());
+ function.initializeState(mock(FunctionInitializationContext.class));
+
+ try {
+ // Preload two keys in the same partition, as a bootstrap operator would
emit for all records it
+ // owns in that partition.
+ function.processElement(
+ new HoodieFlinkInternalRow("key1", "partition",
"bootstrapped-file-1", "20260101000000"),
+ null, collector(new ArrayList<>()));
+ function.processElement(
+ new HoodieFlinkInternalRow("key2", "partition",
"bootstrapped-file-2", "20260101000000"),
+ null, collector(new ArrayList<>()));
+
+ HoodieFlinkInternalRow update1 = record("key1", "partition", "U");
+ function.processElement(update1, null, collector(new ArrayList<>()));
+ assertEquals("bootstrapped-file-1", update1.getFileId());
+
+ HoodieFlinkInternalRow update2 = record("key2", "partition", "U");
+ function.processElement(update2, null, collector(new ArrayList<>()));
+ assertEquals("bootstrapped-file-2", update2.getFileId());
+
+ // A brand-new key in the same partition that was never preloaded must
still go through bucket
+ // assignment as an insert rather than reusing a bootstrapped file group.
+ HoodieFlinkInternalRow newRecord = record("key3", "partition", "I");
+ List<HoodieFlinkInternalRow> newOutput = new ArrayList<>();
+ function.processElement(newRecord, null, collector(newOutput));
+
+ assertEquals(1, newOutput.size());
+ assertEquals("I", newRecord.getInstantTime());
+ assertNotEquals("bootstrapped-file-1", newRecord.getFileId());
+ assertNotEquals("bootstrapped-file-2", newRecord.getFileId());
+ } finally {
+ function.close();
+ }
+ }
+
private static HoodieFlinkInternalRow record(String recordKey, String
partitionPath, String operationType) {
return new HoodieFlinkInternalRow(recordKey, partitionPath, operationType,
new GenericRowData(0));
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexBackends.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexBackends.java
index 85d6feed642c..d124731b22e5 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexBackends.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexBackends.java
@@ -51,6 +51,9 @@ class TestIndexBackends {
assertNull(backend.get("partition", "key"));
backend.update("partition", "key", "file-id");
+ assertNull(backend.get("partition", "key"));
+ backend.bootstrap("partition", "key", "file-id");
+ assertNull(backend.get("partition", "key"));
backend.close();
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordLevelIndexBackend.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordLevelIndexBackend.java
index c6f1a4191eed..83ee81f80ecc 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordLevelIndexBackend.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordLevelIndexBackend.java
@@ -219,6 +219,48 @@ public class TestRecordLevelIndexBackend {
}
}
+ @Test
+ public void testBootstrapCreatesCacheWithoutScanningPersistedIndex() throws
Exception {
+ try (RecordLevelIndexBackend backend = createBackend()) {
+ // Metrics are intentionally left unregistered: bootstrapPartition()
would NPE on the metrics
+ // field if it were invoked, so a clean call here proves bootstrap()
skips the persisted-index scan.
+ backend.bootstrap("new-partition", "key1", "file-group-1");
+
+ assertEquals(1, backend.getPartitionBucketCaches().size());
+ assertEquals("file-group-1", backend.get("new-partition", "key1"));
+ }
+ }
+
+ @Test
+ public void testBootstrapReusesExistingPartitionCache() throws Exception {
+ try (RecordLevelIndexBackend backend = createBackend()) {
+ backend.registerMetrics(new UnregisteredMetricsGroup());
+ backend.update("partition", "existing-key", "existing-file");
+
+ backend.bootstrap("partition", "new-key", "new-file");
+
+ assertEquals(1, backend.getPartitionBucketCaches().size());
+ assertEquals("existing-file", backend.get("partition", "existing-key"));
+ assertEquals("new-file", backend.get("partition", "new-key"));
+ }
+ }
+
+ @Test
+ public void testBootstrapDoesNotMarkCacheAsRecentlyUpdated() throws
Exception {
+ try (RecordLevelIndexBackend backend = createBackend()) {
+ backend.onCheckpoint(2L);
+ backend.bootstrap("bootstrapped", "key1", "file1");
+ backend.getPartitionBucketCaches().put("recent",
cacheWithHeapSize(backend, 2 * ONE_MB, 2L));
+ backend.onCheckpointComplete(new
TestCorrespondent(Collections.emptyMap()), 2L);
+
+ backend.cleanIfNecessary(0L, "recent");
+
+
assertFalse(backend.getPartitionBucketCaches().containsKey("bootstrapped"),
+ "A bootstrapped-only cache must not be protected as a recent update,
so it is evicted first");
+ assertTrue(backend.getPartitionBucketCaches().containsKey("recent"));
+ }
+ }
+
@Test
public void testLazyEvictReturnsWhenOnlyProtectedPartitionExceedsLimit()
throws Exception {
try (RecordLevelIndexBackend backend = createBackend()) {
