hudi-agent commented on code in PR #19579:
URL: https://github.com/apache/hudi/pull/19579#discussion_r3756689229
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -501,86 +501,153 @@ public static DataStream<RowData>
hoodieStreamWrite(Configuration conf,
RowType rowType,
DataStream<HoodieFlinkInternalRow> dataStream) {
if (OptionsResolver.isBucketIndexType(conf)) {
- HoodieIndex.BucketIndexEngineType bucketIndexEngineType =
OptionsResolver.getBucketEngineType(conf);
- switch (bucketIndexEngineType) {
- case SIMPLE:
- // [HUDI-9036] BucketIndexPartitioner is also used in bulk insert
mode,
- // keep use of HoodieKey here in partitionCustom for now
- Partitioner<HoodieKey> partitioner =
BucketIndexPartitionerFactory.create(conf);
- SingleOutputStreamOperator<RowData> bucketWriteStream = dataStream
- .partitionCustom(
- partitioner,
- record -> new HoodieKey(record.getRecordKey(),
record.getPartitionPath()))
- .transform(
- opName("bucket_write", conf),
- TypeInformation.of(RowData.class),
- BucketStreamWriteOperator.getFactory(conf, rowType))
- .uid(opUID("bucket_write", conf))
- .setParallelism(conf.get(FlinkOptions.WRITE_TASKS));
- declareManagedMemoryIfNecessary(conf, bucketWriteStream, () ->
OptionsResolver.getWriteBufferSizeInBytes(conf));
- return bucketWriteStream;
- case CONSISTENT_HASHING:
- if (OptionsResolver.isInsertOverwrite(conf)) {
- // TODO support insert overwrite for consistent bucket index
- throw new HoodieException("Consistent hashing bucket index does
not work with insert overwrite using FLINK engine. Use simple bucket index or
Spark engine.");
- }
- SingleOutputStreamOperator<RowData> consistentBucketWriteStream =
dataStream
- .transform(
- opName("consistent_bucket_assigner", conf),
- new HoodieFlinkInternalRowTypeInfo(rowType),
- new ProcessOperator<>(new
ConsistentBucketAssignFunction(conf)))
- .uid(opUID("consistent_bucket_assigner", conf))
- .setParallelism(conf.get(FlinkOptions.BUCKET_ASSIGN_TASKS))
- .keyBy(HoodieFlinkInternalRow::getFileId)
- .transform(
- opName("consistent_bucket_write", conf),
- TypeInformation.of(RowData.class),
- BucketStreamWriteOperator.getFactory(conf, rowType))
- .uid(opUID("consistent_bucket_write", conf))
- .setParallelism(conf.get(FlinkOptions.WRITE_TASKS));
- declareManagedMemoryIfNecessary(conf, consistentBucketWriteStream,
() -> OptionsResolver.getWriteBufferSizeInBytes(conf));
- return consistentBucketWriteStream;
- default:
- throw new HoodieNotSupportedException("Unknown bucket index engine
type: " + bucketIndexEngineType);
- }
- } else {
- String writeOperatorUid = opUID("stream_write", conf);
- // uuid is used to generate operator id for the write operator, then the
bucket assign operator can send
- // operator event to the coordinator of the write operator based on the
operator id.
- // @see org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway.
- DataStream<HoodieFlinkInternalRow> bucketAssignStream =
createBucketAssignStream(dataStream, conf, rowType, writeOperatorUid);
- boolean isStreamingIndexWriteEnabled =
OptionsResolver.isStreamingIndexWriteEnabled(conf);
- SingleOutputStreamOperator<RowData> writeDatastream =
- bucketAssignStream
- // shuffle by fileId(bucket id)
- .keyBy(HoodieFlinkInternalRow::getFileId)
- .transform(
- opName("stream_write", conf),
- isStreamingIndexWriteEnabled ?
InternalTypeInfo.of(IndexRowUtils.INDEX_ROW_TYPE) :
TypeInformation.of(RowData.class),
- StreamWriteOperator.getFactory(conf, rowType))
- .uid(writeOperatorUid)
- .setParallelism(conf.get(FlinkOptions.WRITE_TASKS));
- declareManagedMemoryIfNecessary(conf, writeDatastream, () ->
OptionsResolver.getWriteBufferSizeInBytes(conf));
- if (isStreamingIndexWriteEnabled) {
- // index writing pipeline
- SingleOutputStreamOperator<RowData> indexWriteDatastream =
writeDatastream
- .partitionCustom(
- OptionsResolver.isRecordLevelIndex(conf)
- ? new RecordIndexPartitioner(conf)
- : new GlobalRecordIndexPartitioner(conf),
- IndexRowUtils::getHoodieKey)
+ return bucketStreamWrite(conf, rowType, dataStream);
+ }
+
+ String writeOperatorUid = opUID("stream_write", conf);
+ // uuid is used to generate operator id for the write operator, then the
bucket assign operator can send
+ // operator event to the coordinator of the write operator based on the
operator id.
+ // @see org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway.
+ DataStream<HoodieFlinkInternalRow> bucketAssignStream =
+ createBucketAssignStream(dataStream, conf, rowType, writeOperatorUid);
+ boolean isStreamingIndexWriteEnabled =
OptionsResolver.isStreamingIndexWriteEnabled(conf);
+ SingleOutputStreamOperator<RowData> writeDataStream = bucketAssignStream
+ // shuffle by fileId(bucket id)
+ .keyBy(HoodieFlinkInternalRow::getFileId)
+ .transform(
+ opName("stream_write", conf),
+ isStreamingIndexWriteEnabled
+ ? InternalTypeInfo.of(IndexRowUtils.INDEX_ROW_TYPE)
+ : TypeInformation.of(RowData.class),
+ StreamWriteOperator.getFactory(conf, rowType))
+ .uid(writeOperatorUid)
+ .setParallelism(conf.get(FlinkOptions.WRITE_TASKS));
+ declareManagedMemoryIfNecessary(
+ conf, writeDataStream, () ->
OptionsResolver.getWriteBufferSizeInBytes(conf));
+
+ return isStreamingIndexWriteEnabled
Review Comment:
🤖 nit: `addIndexWrite` uses a verb prefix while its sibling
`bucketStreamWrite` reads as a noun/description — could you align them?
Something like `indexWritePipeline` or `buildIndexWriteStream` would match the
style of `bucketStreamWrite` and make it clearer these are both
pipeline-builder helpers.
<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]
