ggjh-159 commented on code in PR #13409:
URL: https://github.com/apache/gluten/pull/13409#discussion_r4466233681
##########
gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecExchange.java:
##########
@@ -153,8 +153,12 @@ protected Transformation<RowData> translateToPlanInternal(
LogicalTypeConverter.toVLType(getOutputType());
String id = PlanNodeIdGenerator.newId();
List<Integer> keyIndexes =
Arrays.stream(keys).boxed().collect(Collectors.toList());
+ // Keyed shuffle routes by key group (the same value the state
+ // backend derives key groups with), so each subtask receives
+ // exactly the key groups its state owns.
PartitionFunctionSpec partitionFunctionSpec =
- new HashPartitionFunctionSpec(glutenInputType, keyIndexes);
+ new KeyGroupPartitionFunctionSpec(
+ glutenInputType, keyIndexes,
DEFAULT_LOWER_BOUND_MAX_PARALLELISM);
Review Comment:
The exchange now resolves the actual max parallelism at translation time —
the job-wide value from the ExecutionConfig if set, else the same default the
scheduler derives from the parallelism (computeDefaultMaxParallelism) — and
feeds it to both the native KeyGroupPartitionFunctionSpec and the Flink
partitioner. Added a test that translates the keyed first-bid query and asserts
the partitioner carries 128 by default and 256 when explicitly configured.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
