924060929 commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3568112634
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -528,6 +527,101 @@ public PhysicalProperties
visitPhysicalSetOperation(PhysicalSetOperation setOper
return PhysicalProperties.createHash(request, firstType);
}
+ /**
+ * The basic (bucket-shuffle anchor) child candidate of a set operation:
the first NATURAL
+ * child, or else the first STORAGE_BUCKETED child, or -1 when no child is
bucket-distributed.
+ * The largest natural / storage-bucketed child keeps its buckets and the
others are
+ * bucket-shuffled to it, so this only picks the candidate; {@link
#isSetOperationBucketAligned}
+ * decides whether the children actually agree on that anchor's buckets.
+ */
+ public static int setOperationBucketBasicIndex(List<DistributionSpec>
childrenDistribution) {
+ int firstNaturalIndex = -1;
+ int firstStorageBucketedIndex = -1;
+ for (int i = 0; i < childrenDistribution.size(); i++) {
+ if (!(childrenDistribution.get(i) instanceof
DistributionSpecHash)) {
+ continue;
+ }
+ ShuffleType childShuffleType = ((DistributionSpecHash)
childrenDistribution.get(i)).getShuffleType();
+ if (childShuffleType == ShuffleType.NATURAL && firstNaturalIndex <
0) {
+ firstNaturalIndex = i;
+ } else if (childShuffleType == ShuffleType.STORAGE_BUCKETED &&
firstStorageBucketedIndex < 0) {
+ firstStorageBucketedIndex = i;
+ }
+ }
+ return firstNaturalIndex >= 0 ? firstNaturalIndex :
firstStorageBucketedIndex;
+ }
+
+ /**
+ * Whether every child of the set operation is genuinely bucket-aligned to
a single basic
+ * child, i.e. all children are hash NATURAL / STORAGE_BUCKETED on the
same storage layout
+ * (table / index / partitions) AND map their shuffle columns to the same
set-operation output
+ * positions. This is the single proof shared by {@link
#visitPhysicalSetOperation} (to decide
+ * the output distribution) and {@code PhysicalPlanTranslator} (to set the
legacy
+ * {@code BUCKET_SHUFFLE} marker), so the derived property and the marker
can never disagree: a
+ * same-layout shape whose bucket key feeds different set-operation output
positions in
+ * different children is not aligned and must not be treated as bucketed
by either side.
+ */
+ public static boolean isSetOperationBucketAligned(
+ PhysicalSetOperation setOperation, List<DistributionSpec>
childrenDistribution) {
+ int distributeToChildIndex =
setOperationBucketBasicIndex(childrenDistribution);
+ if (distributeToChildIndex < 0) {
+ return false;
+ }
+ DistributionSpecHash basicChildHash = (DistributionSpecHash)
childrenDistribution.get(distributeToChildIndex);
+ List<Integer> basicOutputPositions
+ = mapShuffleColumnsToOutputPositions(setOperation,
basicChildHash, distributeToChildIndex);
+ if (basicOutputPositions == null || basicOutputPositions.isEmpty() ||
basicChildHash.getTableId() < 0) {
+ return false;
+ }
+ for (int i = 0; i < childrenDistribution.size(); i++) {
+ if (!(childrenDistribution.get(i) instanceof
DistributionSpecHash)) {
+ return false;
+ }
+ DistributionSpecHash childHash = (DistributionSpecHash)
childrenDistribution.get(i);
+ ShuffleType childShuffleType = childHash.getShuffleType();
+ if (childShuffleType != ShuffleType.NATURAL && childShuffleType !=
ShuffleType.STORAGE_BUCKETED) {
+ return false;
+ }
+ // same storage layout (a NATURAL basic child and its enforced
STORAGE_BUCKETED siblings
+ // all carry the basic child's layout, so they share the bucket
function)
+ if (childHash.getTableId() != basicChildHash.getTableId()
+ || childHash.getSelectedIndexId() !=
basicChildHash.getSelectedIndexId()
+ ||
!childHash.getPartitionIds().equals(basicChildHash.getPartitionIds())) {
+ return false;
+ }
+ // same set-operation output positions (the bucket key must feed
the same output columns
+ // in every child, otherwise per-bucket set operation would
compare mismatched columns)
+ List<Integer> positions =
mapShuffleColumnsToOutputPositions(setOperation, childHash, i);
+ if (positions == null || !positions.equals(basicOutputPositions)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Map a child's ordered shuffle columns to the set operation output
positions, using the
+ * child's regular output slots. Returns null when any shuffle column is
not present in the
+ * child's output (so the set operation does not carry that distribution
column).
+ */
+ static List<Integer> mapShuffleColumnsToOutputPositions(
+ PhysicalSetOperation setOperation, DistributionSpecHash childHash,
int childIndex) {
+ List<SlotReference> childOutput =
setOperation.getRegularChildrenOutputs().get(childIndex);
+ Map<ExprId, Integer> idToOutputIndex = new LinkedHashMap<>();
+ for (int j = 0; j < childOutput.size(); j++) {
+ idToOutputIndex.put(childOutput.get(j).getExprId(), j);
+ }
+ List<Integer> positions = new ArrayList<>();
+ for (ExprId columnId : childHash.getOrderedShuffledColumns()) {
+ Integer index = idToOutputIndex.get(columnId);
Review Comment:
Fixed — made the mapping equivalence-set aware like the regulator.
`mapShuffleColumnsToOutputPositions` now resolves each ordered shuffle column
through the child's hash equivalence set: it prefers the shuffle ExprId's own
output position (so the existing direct shapes are unchanged) and falls back to
the output position of an equivalent slot when the bucket key is not directly
visible. So a bucket key that reaches the set operation only as a projected
join key resolves to that visible position, matching
`canMapBucketKeysToRequire`, while still requiring every bucket key to map to
the same set-output position across children.
Added the regression `bucket_shuffle_equivalent_key_fill_up`: the union's
basic child is a bucket-shuffle join output projecting `b.id AS k` (so the
storage bucket column `fill_anchor.id` is hidden and only the equivalent `k` is
visible), the anchor is pruned to 4/10 buckets by `a.id IN (0,2,4,6)`, and the
other child is shuffled with rows in the missing buckets. On a 4-BE cluster I
confirmed the union is now `PhysicalUnion[bucketShuffle]`, the anchor scan is
`tablets=4/10` while fill-up creates receivers for all 10 buckets, and the
result (21 rows) matches the execution-shuffle path. Without the equivalence
resolution the marker would be dropped, fill-up skipped, and the shuffled rows
in the 6 missing buckets lost. The rest of the suite `.out` is unchanged.
--
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]
