hudi-agent commented on code in PR #19837:
URL: https://github.com/apache/hudi/pull/19837#discussion_r3932406959
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestDynamicBucketAssignFunction.java:
##########
@@ -166,6 +186,94 @@ void testOpenInitializesBucketAssignerAndTaskOwnership()
throws Exception {
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
Review Comment:
🤖 nit: the RLI setup block (conf → `StreamerUtil.initTableIfNotExists` →
`ViewStorageProperties.createProperties` → `function.open/initializeState`) is
duplicated verbatim in both integration tests — could you extract it into a
private helper like `createOpenedFunction()` so a future setup change only
needs to land in one place?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordLevelIndexBackend.java:
##########
@@ -121,6 +121,12 @@ public void update(String partitionPath, String recordKey,
String fileId) {
cleanExecutor.runIfNecessary(() -> cleanIfNecessary(0L, partitionPath));
}
+ @Override
+ public void bootstrap(String partitionPath, String recordKey, String fileId)
{
+ BucketCache cache = getOrCreatePartitionCache(partitionPath);
Review Comment:
🤖 Since `bootstrap()` never bumps `lastUpdatedCheckpoint` (it stays
`Long.MIN_VALUE`), a preloaded-but-not-yet-written partition cache is always
the first to be evicted under memory pressure in `cleanIfNecessary`. In a
memory-constrained job with many preloaded partitions, could the preload be
largely evicted right after startup — before the first data record touches
those partitions — forcing exactly the full re-scan via `bootstrapPartition`
that this PR aims to avoid? Have you considered tagging bootstrapped caches
with the current checkpoint id (or otherwise protecting them for at least one
cycle)?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
