cshuo commented on code in PR #19837:
URL: https://github.com/apache/hudi/pull/19837#discussion_r3933101092
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/DynamicBucketAssignFunction.java:
##########
@@ -143,6 +153,15 @@ public void processElement(HoodieFlinkInternalRow record,
Context ctx, Collector
out.collect(record);
}
+ /**
+ * Applies a preloaded index record emitted by a bootstrap operator directly
to the partitioned
+ * index backend, since its {@code recordKey -> fileGroupId} mapping is
already known and does not
+ * need bucket assignment. The record carries no row data, so it is never
emitted downstream.
+ */
+ private void processIndexRecord(HoodieFlinkInternalRow record) {
+ indexBackend.bootstrap(record.getPartitionPath(), record.getRecordKey(),
record.getFileId());
Review Comment:
nit: `processIndexRecord` can be inlined.
##########
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:
Before the first checkpoint completes, `minRetainedCheckpointId` is also
`Long.MIN_VALUE`, so `lastUpdatedCheckpoint < minRetainedCheckpointId` is false
and preloaded caches cannot be evicted. Afterward, evicting reconstructible
caches under memory pressure is expected 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
