github-actions[bot] commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3568199830
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -528,6 +527,121 @@ 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. The lookup is
+ * equivalence-set aware, mirroring the regulator's {@code
canMapBucketKeysToRequire} /
+ * {@code calAnotherSideRequiredShuffleIds}: a bucket key can reach the
set operation through an
+ * equivalent slot (for example a join key {@code a.id} exposed only as
the projected
+ * {@code b.id}), so the ordered shuffle ExprId itself may not appear in
the child output while
+ * an equivalent one does. Each bucket key is therefore resolved through
its hash equivalence
+ * set to the output position of a visible member of that set. Returns
null when any bucket key
+ * has no visible member in the child output (the set operation does not
carry that distribution
+ * column, so the children cannot be treated as bucket-aligned).
+ */
+ static List<Integer> mapShuffleColumnsToOutputPositions(
+ PhysicalSetOperation setOperation, DistributionSpecHash childHash,
int childIndex) {
+ List<SlotReference> childOutput =
setOperation.getRegularChildrenOutputs().get(childIndex);
+ Map<ExprId, Integer> exprIdToEquivalenceSet =
childHash.getExprIdToEquivalenceSet();
+ Map<ExprId, Integer> idToOutputIndex = new LinkedHashMap<>();
+ // equivalence set index -> first output position of a slot in that set
+ Map<Integer, Integer> equivalenceSetToOutputIndex = new
LinkedHashMap<>();
+ for (int j = 0; j < childOutput.size(); j++) {
+ ExprId outputExprId = childOutput.get(j).getExprId();
+ idToOutputIndex.putIfAbsent(outputExprId, j);
+ Integer equivalenceSet = exprIdToEquivalenceSet.get(outputExprId);
+ if (equivalenceSet != null) {
+ equivalenceSetToOutputIndex.putIfAbsent(equivalenceSet, j);
+ }
Review Comment:
This still disagrees with the regulator when one hidden storage bucket key
has more than one visible equivalent set-output column and the parent request
chooses a later one. A reduced shape is:
```text
Join on u.y = p.id
UnionAll(x, y)
Project(c.id AS x, b.id AS y)
Join chain with a.id = b.id and a.id = c.id
OlapScan a -- bucketed by hidden a.id, bucket-pruned
PhysicalDistribute[STORAGE_BUCKETED on y]
...
```
`createHashRequestAccordingToParent()` makes the union children request
output `y`. The regulator accepts the basic child because hidden `a.id` is
equivalent to that requested `b.id`/`y`, and it enforces the sibling to shuffle
by `y`. But this fallback records the first visible slot in the equivalence set
by child-output order; if `x` appears before `y`, the shared proof maps the
basic child to `x` while the enforced child maps to `y`, returns false, and the
translator never sets `BUCKET_SHUFFLE`. For bucket-pruned `UNION`/`EXCEPT`,
`shouldFillUpInstances()` then skips the missing-bucket receivers and rows from
the shuffled side can be dropped.
Please make the proof use the same resolved output position as the
regulator, or otherwise make both sides choose a deterministic equivalent
output position, and add a fill-up regression with two visible equivalent keys
where the parent hashes on the later one.
--
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]
