hudi-agent commented on code in PR #19844:
URL: https://github.com/apache/hudi/pull/19844#discussion_r3954319826
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -408,36 +408,41 @@ public static DataStream<HoodieFlinkInternalRow>
bootstrap(
DataStream<RowData> dataStream,
boolean bounded,
boolean overwrite) {
- final boolean globalIndex = conf.get(FlinkOptions.INDEX_GLOBAL_ENABLED);
if (overwrite || OptionsResolver.isBucketIndexType(conf)) {
+ // overwrite and bucket index do not rely on a bootstrapped key index.
+ return rowDataToHoodieRecord(conf, rowType, dataStream);
+ }
+
+ final boolean isRLI = OptionsResolver.isGlobalRecordLevelIndex(conf) ||
OptionsResolver.isRecordLevelIndex(conf);
+ // Bounded writes require bootstrap automatically only for non-RLI indexes.
+ final boolean needsBootstrap =
conf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED) || (bounded && !isRLI);
+ if (!needsBootstrap) {
return rowDataToHoodieRecord(conf, rowType, dataStream);
}
- if (bounded && !globalIndex && OptionsResolver.isPartitionedTable(conf)) {
+
+ final boolean globalIndex = conf.get(FlinkOptions.INDEX_GLOBAL_ENABLED);
+ if (bounded && !globalIndex && !isRLI &&
OptionsResolver.isPartitionedTable(conf)) {
return boundedBootstrap(conf, rowType, dataStream);
}
- return streamBootstrap(conf, rowType, dataStream, bounded);
+
+ return streamBootstrap(conf, rowType, dataStream);
}
private static DataStream<HoodieFlinkInternalRow> streamBootstrap(
Configuration conf,
RowType rowType,
- DataStream<RowData> dataStream,
- boolean bounded) {
+ DataStream<RowData> dataStream) {
DataStream<HoodieFlinkInternalRow> dataStream1 =
rowDataToHoodieRecord(conf, rowType, dataStream);
+ AbstractBootstrapOperator bootstrapOperator =
BootstrapOperatorFactory.createInstance(conf);
Review Comment:
🤖 nit: the `.setParallelism(...)`/`.uid(...)` continuation lines picked up
an extra indent level after the de-nesting — mind aligning them back to 4
spaces under `dataStream1`?
<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/utils/Pipelines.java:
##########
@@ -408,36 +408,41 @@ public static DataStream<HoodieFlinkInternalRow>
bootstrap(
DataStream<RowData> dataStream,
boolean bounded,
boolean overwrite) {
- final boolean globalIndex = conf.get(FlinkOptions.INDEX_GLOBAL_ENABLED);
if (overwrite || OptionsResolver.isBucketIndexType(conf)) {
+ // overwrite and bucket index do not rely on a bootstrapped key index.
+ return rowDataToHoodieRecord(conf, rowType, dataStream);
+ }
+
+ final boolean isRLI = OptionsResolver.isGlobalRecordLevelIndex(conf) ||
OptionsResolver.isRecordLevelIndex(conf);
+ // Bounded writes require bootstrap automatically only for non-RLI indexes.
Review Comment:
🤖 nit: could you spell this out as `isRecordLevelIndex` (or
`usesRecordLevelIndex`)? the `isRLI` abbreviation is a bit opaque next to
`globalIndex`/`needsBootstrap`.
<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/utils/Pipelines.java:
##########
@@ -408,36 +408,41 @@ public static DataStream<HoodieFlinkInternalRow>
bootstrap(
DataStream<RowData> dataStream,
boolean bounded,
boolean overwrite) {
- final boolean globalIndex = conf.get(FlinkOptions.INDEX_GLOBAL_ENABLED);
if (overwrite || OptionsResolver.isBucketIndexType(conf)) {
+ // overwrite and bucket index do not rely on a bootstrapped key index.
+ return rowDataToHoodieRecord(conf, rowType, dataStream);
+ }
+
+ final boolean isRLI = OptionsResolver.isGlobalRecordLevelIndex(conf) ||
OptionsResolver.isRecordLevelIndex(conf);
+ // Bounded writes require bootstrap automatically only for non-RLI indexes.
+ final boolean needsBootstrap =
conf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED) || (bounded && !isRLI);
+ if (!needsBootstrap) {
return rowDataToHoodieRecord(conf, rowType, dataStream);
}
- if (bounded && !globalIndex && OptionsResolver.isPartitionedTable(conf)) {
+
+ final boolean globalIndex = conf.get(FlinkOptions.INDEX_GLOBAL_ENABLED);
+ if (bounded && !globalIndex && !isRLI &&
OptionsResolver.isPartitionedTable(conf)) {
return boundedBootstrap(conf, rowType, dataStream);
}
- return streamBootstrap(conf, rowType, dataStream, bounded);
+
+ return streamBootstrap(conf, rowType, dataStream);
}
private static DataStream<HoodieFlinkInternalRow> streamBootstrap(
Review Comment:
🤖 nit: since `streamBootstrap` now always applies the operator,
`dataStream1` could just be named `records` (or similar) — the numeric suffix
reads like a leftover.
<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]
