hudi-agent commented on code in PR #19920:
URL: https://github.com/apache/hudi/pull/19920#discussion_r3995151312
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -721,11 +725,29 @@ private static DataStream<HoodieFlinkInternalRow>
createBucketAssignStream(
* @return the compaction pipeline
*/
public static DataStreamSink<CompactionCommitEvent> compact(Configuration
conf, DataStream<RowData> dataStream) {
- DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream =
dataStream.transform("compact_plan_generate",
+ SingleOutputStreamOperator<CompactionPlanEvent> compactionPlanStream =
dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
.setParallelism(1) // plan generate must be singleton
- .setMaxParallelism(1)
+ .setMaxParallelism(1);
+ if
(conf.getOptional(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP).isPresent())
{
+ String slotSharingGroup =
conf.get(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP);
+ SlotSharingGroup.Builder slotSharingGroupBuilder =
SlotSharingGroup.newBuilder(slotSharingGroup);
+ // Flink requires cpu cores and task heap memory to be configured
together when specifying
+ // resources for a slot sharing group, reuse the TaskManager-level
configuration for both.
+ Optional<Double> cpuCores =
conf.getOptional(TaskManagerOptions.CPU_CORES);
+ Optional<MemorySize> taskHeapMemory =
conf.getOptional(TaskManagerOptions.TASK_HEAP_MEMORY);
+ if (!taskHeapMemory.isPresent()) {
+ taskHeapMemory = deriveTaskHeapMemory(conf);
+ }
+ if (cpuCores.isPresent() && taskHeapMemory.isPresent()) {
+
slotSharingGroupBuilder.setCpuCores(cpuCores.get()).setTaskHeapMemory(taskHeapMemory.get());
+ }
+ // the slot sharing group must be registered with the environment before
it can be referenced by name
+
dataStream.getExecutionEnvironment().registerSlotSharingGroup(slotSharingGroupBuilder.build());
+ compactionPlanStream =
compactionPlanStream.slotSharingGroup(slotSharingGroup);
Review Comment:
🤖 I think this ends up pulling `compact_task` and `compact_commit` into the
same group rather than isolating the planner: Flink's
`StreamGraphGenerator.determineSlotSharingGroup` makes an operator with no
explicit group inherit its input's group, and neither downstream operator sets
one. Could you explicitly pin them to `"default"`
(`.slotSharingGroup("default")`) and assert on the StreamGraph-level group
(`getStreamGraph().getStreamNode(..).getSlotSharingGroup()`) in the test, since
`Transformation.getSlotSharingGroup()` only reflects the explicitly-set value?
<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:
##########
@@ -721,11 +725,29 @@ private static DataStream<HoodieFlinkInternalRow>
createBucketAssignStream(
* @return the compaction pipeline
*/
public static DataStreamSink<CompactionCommitEvent> compact(Configuration
conf, DataStream<RowData> dataStream) {
- DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream =
dataStream.transform("compact_plan_generate",
+ SingleOutputStreamOperator<CompactionPlanEvent> compactionPlanStream =
dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
.setParallelism(1) // plan generate must be singleton
- .setMaxParallelism(1)
+ .setMaxParallelism(1);
+ if
(conf.getOptional(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP).isPresent())
{
+ String slotSharingGroup =
conf.get(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP);
+ SlotSharingGroup.Builder slotSharingGroupBuilder =
SlotSharingGroup.newBuilder(slotSharingGroup);
+ // Flink requires cpu cores and task heap memory to be configured
together when specifying
+ // resources for a slot sharing group, reuse the TaskManager-level
configuration for both.
+ Optional<Double> cpuCores =
conf.getOptional(TaskManagerOptions.CPU_CORES);
+ Optional<MemorySize> taskHeapMemory =
conf.getOptional(TaskManagerOptions.TASK_HEAP_MEMORY);
Review Comment:
🤖 The `conf` here is the Hudi table/compaction-job options rather than the
cluster's flink-conf, so `taskmanager.cpu.cores` / memory keys will only be
present if a user copies them into the table properties — is that the intended
way to enable the resource profile? Also, requesting the full TM cpu + task
heap means nothing else can be scheduled on that TM; if that's the intent, it
might be worth spelling out in the option description.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java:
##########
@@ -1025,6 +1025,16 @@ public class FlinkOptions extends HoodieConfig {
.noDefaultValue()
.withDescription("Parallelism of tasks that do actual compaction,
default same as the write task parallelism");
+ @AdvancedConfig
+ public static final ConfigOption<String>
COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP = ConfigOptions
Review Comment:
🤖 nit: the key mixes underscores and dots (`plan_generate`,
`slot_sharing_group`) whereas most other keys in this file use dot-separated
segments directly (e.g. `compaction.tasks`, `compaction.delta_commits` already
has underscores too, so double check what the dominant style is here) —
consider `compaction.plan-generate.slot-sharing-group` or matching the
surrounding keys' exact word separation so it doesn't read as a one-off.
<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:
##########
@@ -721,11 +725,29 @@ private static DataStream<HoodieFlinkInternalRow>
createBucketAssignStream(
* @return the compaction pipeline
*/
public static DataStreamSink<CompactionCommitEvent> compact(Configuration
conf, DataStream<RowData> dataStream) {
- DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream =
dataStream.transform("compact_plan_generate",
+ SingleOutputStreamOperator<CompactionPlanEvent> compactionPlanStream =
dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
.setParallelism(1) // plan generate must be singleton
- .setMaxParallelism(1)
+ .setMaxParallelism(1);
+ if
(conf.getOptional(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP).isPresent())
{
Review Comment:
🤖 nit: the slot-sharing-group setup (registering the group, deriving
CPU/heap memory) is about 15 lines inlined into `compact()`, which already does
a lot; consider extracting it into a small private helper like
`configureSlotSharingGroup(DataStream<RowData> dataStream,
SingleOutputStreamOperator<CompactionPlanEvent> stream, Configuration conf)` so
`compact()` stays focused on wiring the pipeline stages.
<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]
