jackwener commented on code in PR #11976:
URL: https://github.com/apache/doris/pull/11976#discussion_r962541664
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java:
##########
@@ -31,88 +39,211 @@
@Developing
public class DistributionSpecHash extends DistributionSpec {
- private final List<SlotReference> shuffledColumns;
+ private final List<ExprId> orderedShuffledColumns;
private final ShuffleType shuffleType;
- public DistributionSpecHash(List<SlotReference> shuffledColumns,
ShuffleType shuffleType) {
- // Preconditions.checkState(!shuffledColumns.isEmpty());
- this.shuffledColumns = shuffledColumns;
- this.shuffleType = shuffleType;
+ // below two attributes use for colocate join
+ private final long tableId;
+
+ private final Set<Long> partitionIds;
+
+ // use for satisfied judge
+ private final List<Set<ExprId>> equivalenceExprIds;
+
+ private final Map<ExprId, Integer> exprIdToEquivalenceSet;
+
+ /**
+ * Use for no need set table related attributes.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType) {
+ this(orderedShuffledColumns, shuffleType, -1L, Collections.emptySet());
+ }
+
+ /**
+ * Use for merge two shuffle columns.
+ */
+ public DistributionSpecHash(List<ExprId> leftColumns, List<ExprId>
rightColumns, ShuffleType shuffleType) {
+ this(leftColumns, shuffleType, -1L, Collections.emptySet());
+ Objects.requireNonNull(rightColumns);
+ Preconditions.checkArgument(leftColumns.size() == rightColumns.size());
+ for (int i = 0; i < rightColumns.size(); i++) {
+ exprIdToEquivalenceSet.put(rightColumns.get(i), i);
+ equivalenceExprIds.get(i).add(rightColumns.get(i));
+ }
+ }
+
+ /**
+ * Normal constructor.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType,
+ long tableId, Set<Long> partitionIds) {
+ this.orderedShuffledColumns =
Objects.requireNonNull(orderedShuffledColumns);
+ this.shuffleType = Objects.requireNonNull(shuffleType);
+ this.tableId = tableId;
+ this.partitionIds = Objects.requireNonNull(partitionIds);
+ this.equivalenceExprIds = Lists.newArrayList();
+ this.exprIdToEquivalenceSet = Maps.newHashMap();
+ orderedShuffledColumns.forEach(id -> {
+ exprIdToEquivalenceSet.put(id, equivalenceExprIds.size());
+ equivalenceExprIds.add(Sets.newHashSet(id));
+ });
+ }
+
+ /**
+ * Used in merge outside and put result into it.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType, long tableId,
+ Set<Long> partitionIds, List<Set<ExprId>> equivalenceExprIds,
Map<ExprId, Integer> exprIdToEquivalenceSet) {
+ this.orderedShuffledColumns =
Objects.requireNonNull(orderedShuffledColumns);
+ this.shuffleType = Objects.requireNonNull(shuffleType);
+ this.tableId = tableId;
+ this.partitionIds = Objects.requireNonNull(partitionIds);
+ this.equivalenceExprIds = Objects.requireNonNull(equivalenceExprIds);
+ this.exprIdToEquivalenceSet =
Objects.requireNonNull(exprIdToEquivalenceSet);
+ }
+
+ static DistributionSpecHash merge(DistributionSpecHash left,
DistributionSpecHash right, ShuffleType shuffleType) {
+ List<ExprId> orderedShuffledColumns = left.getOrderedShuffledColumns();
+ List<Set<ExprId>> equivalenceExprIds =
Lists.newArrayListWithCapacity(orderedShuffledColumns.size());
+ for (int i = 0; i < orderedShuffledColumns.size(); i++) {
+ Set<ExprId> equivalenceExprId = Sets.newHashSet();
+ equivalenceExprId.addAll(left.getEquivalenceExprIds().get(i));
+ equivalenceExprId.addAll(right.getEquivalenceExprIds().get(i));
+ equivalenceExprIds.add(equivalenceExprId);
+ }
+ Map<ExprId, Integer> exprIdToEquivalenceSet = Maps.newHashMap();
+ exprIdToEquivalenceSet.putAll(left.getExprIdToEquivalenceSet());
+ exprIdToEquivalenceSet.putAll(right.getExprIdToEquivalenceSet());
+ return new DistributionSpecHash(orderedShuffledColumns, shuffleType,
+ left.getTableId(), left.getPartitionIds(), equivalenceExprIds,
exprIdToEquivalenceSet);
}
- public List<SlotReference> getShuffledColumns() {
- return shuffledColumns;
+ static DistributionSpecHash merge(DistributionSpecHash left,
DistributionSpecHash right) {
+ return merge(left, right, left.getShuffleType());
+ }
+
+ public List<ExprId> getOrderedShuffledColumns() {
+ return orderedShuffledColumns;
}
public ShuffleType getShuffleType() {
return shuffleType;
}
+ public long getTableId() {
+ return tableId;
+ }
+
+ public Set<Long> getPartitionIds() {
+ return partitionIds;
+ }
+
+ public List<Set<ExprId>> getEquivalenceExprIds() {
+ return equivalenceExprIds;
+ }
+
+ public Map<ExprId, Integer> getExprIdToEquivalenceSet() {
+ return exprIdToEquivalenceSet;
+ }
+
@Override
- public boolean satisfy(DistributionSpec other) {
- if (other instanceof DistributionSpecAny) {
+ public boolean satisfy(DistributionSpec required) {
+ if (required instanceof DistributionSpecAny) {
return true;
}
- if (!(other instanceof DistributionSpecHash)) {
+ if (!(required instanceof DistributionSpecHash)) {
return false;
}
- DistributionSpecHash spec = (DistributionSpecHash) other;
+ DistributionSpecHash requiredHash = (DistributionSpecHash) required;
- if (shuffledColumns.size() > spec.shuffledColumns.size()) {
+ if (this.orderedShuffledColumns.size() >
requiredHash.orderedShuffledColumns.size()) {
return false;
}
- // TODO: need consider following logic whether is right, and maybe
need consider more.
- // TODO: consider Agg.
- // Current shuffleType is LOCAL/AGG, allow if current is contained by
other
- if (shuffleType == ShuffleType.LOCAL || spec.shuffleType ==
ShuffleType.AGG) {
- return new
HashSet<>(spec.shuffledColumns).containsAll(shuffledColumns);
+ if (requiredHash.shuffleType == ShuffleType.AGGREGATE) {
+ return containsSatisfy(requiredHash.getOrderedShuffledColumns());
}
- if (shuffleType == ShuffleType.AGG && spec.shuffleType ==
ShuffleType.JOIN) {
- return shuffledColumns.size() == spec.shuffledColumns.size()
- && shuffledColumns.equals(spec.shuffledColumns);
- } else if (shuffleType == ShuffleType.JOIN && spec.shuffleType ==
ShuffleType.AGG) {
- return new
HashSet<>(spec.shuffledColumns).containsAll(shuffledColumns);
- }
+ // if (requiredHash.shuffleType == ShuffleType.JOIN) {
+ // return equalsSatisfy(requiredHash.getOrderedShuffledColumns());
+ // }
+ //
+ // if (!this.shuffleType.equals(requiredHash.shuffleType)) {
+ // return false;
+ // }
- if (!shuffleType.equals(spec.shuffleType)) {
- return false;
- }
+ return equalsSatisfy(requiredHash.getOrderedShuffledColumns());
+ }
+
+ private boolean containsSatisfy(List<ExprId> required) {
+ BitSet containsBit = new BitSet(orderedShuffledColumns.size());
+ required.forEach(e -> {
+ if (exprIdToEquivalenceSet.containsKey(e)) {
+ containsBit.set(exprIdToEquivalenceSet.get(e));
+ }
+ });
+ return containsBit.nextClearBit(0) >= orderedShuffledColumns.size();
+ }
- return shuffledColumns.size() == spec.shuffledColumns.size()
- && shuffledColumns.equals(spec.shuffledColumns);
+ private boolean equalsSatisfy(List<ExprId> required) {
+ for (int i = 0; i < required.size(); i++) {
+ if (!equivalenceExprIds.get(i).contains(required.get(i))) {
+ return false;
+ }
+ }
+ return true;
}
@Override
public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
Review Comment:
It's include by `super.equals()`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java:
##########
@@ -31,88 +39,211 @@
@Developing
public class DistributionSpecHash extends DistributionSpec {
- private final List<SlotReference> shuffledColumns;
+ private final List<ExprId> orderedShuffledColumns;
private final ShuffleType shuffleType;
- public DistributionSpecHash(List<SlotReference> shuffledColumns,
ShuffleType shuffleType) {
- // Preconditions.checkState(!shuffledColumns.isEmpty());
- this.shuffledColumns = shuffledColumns;
- this.shuffleType = shuffleType;
+ // below two attributes use for colocate join
+ private final long tableId;
+
+ private final Set<Long> partitionIds;
+
+ // use for satisfied judge
+ private final List<Set<ExprId>> equivalenceExprIds;
+
+ private final Map<ExprId, Integer> exprIdToEquivalenceSet;
+
+ /**
+ * Use for no need set table related attributes.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType) {
+ this(orderedShuffledColumns, shuffleType, -1L, Collections.emptySet());
+ }
+
+ /**
+ * Use for merge two shuffle columns.
+ */
+ public DistributionSpecHash(List<ExprId> leftColumns, List<ExprId>
rightColumns, ShuffleType shuffleType) {
+ this(leftColumns, shuffleType, -1L, Collections.emptySet());
+ Objects.requireNonNull(rightColumns);
+ Preconditions.checkArgument(leftColumns.size() == rightColumns.size());
+ for (int i = 0; i < rightColumns.size(); i++) {
+ exprIdToEquivalenceSet.put(rightColumns.get(i), i);
+ equivalenceExprIds.get(i).add(rightColumns.get(i));
+ }
+ }
+
+ /**
+ * Normal constructor.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType,
+ long tableId, Set<Long> partitionIds) {
+ this.orderedShuffledColumns =
Objects.requireNonNull(orderedShuffledColumns);
+ this.shuffleType = Objects.requireNonNull(shuffleType);
+ this.tableId = tableId;
+ this.partitionIds = Objects.requireNonNull(partitionIds);
+ this.equivalenceExprIds = Lists.newArrayList();
+ this.exprIdToEquivalenceSet = Maps.newHashMap();
+ orderedShuffledColumns.forEach(id -> {
+ exprIdToEquivalenceSet.put(id, equivalenceExprIds.size());
+ equivalenceExprIds.add(Sets.newHashSet(id));
+ });
+ }
+
+ /**
+ * Used in merge outside and put result into it.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType, long tableId,
+ Set<Long> partitionIds, List<Set<ExprId>> equivalenceExprIds,
Map<ExprId, Integer> exprIdToEquivalenceSet) {
+ this.orderedShuffledColumns =
Objects.requireNonNull(orderedShuffledColumns);
+ this.shuffleType = Objects.requireNonNull(shuffleType);
+ this.tableId = tableId;
+ this.partitionIds = Objects.requireNonNull(partitionIds);
+ this.equivalenceExprIds = Objects.requireNonNull(equivalenceExprIds);
+ this.exprIdToEquivalenceSet =
Objects.requireNonNull(exprIdToEquivalenceSet);
+ }
+
+ static DistributionSpecHash merge(DistributionSpecHash left,
DistributionSpecHash right, ShuffleType shuffleType) {
+ List<ExprId> orderedShuffledColumns = left.getOrderedShuffledColumns();
+ List<Set<ExprId>> equivalenceExprIds =
Lists.newArrayListWithCapacity(orderedShuffledColumns.size());
+ for (int i = 0; i < orderedShuffledColumns.size(); i++) {
+ Set<ExprId> equivalenceExprId = Sets.newHashSet();
+ equivalenceExprId.addAll(left.getEquivalenceExprIds().get(i));
+ equivalenceExprId.addAll(right.getEquivalenceExprIds().get(i));
+ equivalenceExprIds.add(equivalenceExprId);
+ }
+ Map<ExprId, Integer> exprIdToEquivalenceSet = Maps.newHashMap();
+ exprIdToEquivalenceSet.putAll(left.getExprIdToEquivalenceSet());
+ exprIdToEquivalenceSet.putAll(right.getExprIdToEquivalenceSet());
+ return new DistributionSpecHash(orderedShuffledColumns, shuffleType,
+ left.getTableId(), left.getPartitionIds(), equivalenceExprIds,
exprIdToEquivalenceSet);
}
- public List<SlotReference> getShuffledColumns() {
- return shuffledColumns;
+ static DistributionSpecHash merge(DistributionSpecHash left,
DistributionSpecHash right) {
+ return merge(left, right, left.getShuffleType());
+ }
+
+ public List<ExprId> getOrderedShuffledColumns() {
+ return orderedShuffledColumns;
}
public ShuffleType getShuffleType() {
return shuffleType;
}
+ public long getTableId() {
+ return tableId;
+ }
+
+ public Set<Long> getPartitionIds() {
+ return partitionIds;
+ }
+
+ public List<Set<ExprId>> getEquivalenceExprIds() {
+ return equivalenceExprIds;
+ }
+
+ public Map<ExprId, Integer> getExprIdToEquivalenceSet() {
+ return exprIdToEquivalenceSet;
+ }
+
@Override
- public boolean satisfy(DistributionSpec other) {
- if (other instanceof DistributionSpecAny) {
+ public boolean satisfy(DistributionSpec required) {
+ if (required instanceof DistributionSpecAny) {
return true;
}
- if (!(other instanceof DistributionSpecHash)) {
+ if (!(required instanceof DistributionSpecHash)) {
return false;
}
- DistributionSpecHash spec = (DistributionSpecHash) other;
+ DistributionSpecHash requiredHash = (DistributionSpecHash) required;
- if (shuffledColumns.size() > spec.shuffledColumns.size()) {
+ if (this.orderedShuffledColumns.size() >
requiredHash.orderedShuffledColumns.size()) {
return false;
}
- // TODO: need consider following logic whether is right, and maybe
need consider more.
- // TODO: consider Agg.
- // Current shuffleType is LOCAL/AGG, allow if current is contained by
other
- if (shuffleType == ShuffleType.LOCAL || spec.shuffleType ==
ShuffleType.AGG) {
- return new
HashSet<>(spec.shuffledColumns).containsAll(shuffledColumns);
+ if (requiredHash.shuffleType == ShuffleType.AGGREGATE) {
+ return containsSatisfy(requiredHash.getOrderedShuffledColumns());
}
- if (shuffleType == ShuffleType.AGG && spec.shuffleType ==
ShuffleType.JOIN) {
- return shuffledColumns.size() == spec.shuffledColumns.size()
- && shuffledColumns.equals(spec.shuffledColumns);
- } else if (shuffleType == ShuffleType.JOIN && spec.shuffleType ==
ShuffleType.AGG) {
- return new
HashSet<>(spec.shuffledColumns).containsAll(shuffledColumns);
- }
+ // if (requiredHash.shuffleType == ShuffleType.JOIN) {
+ // return equalsSatisfy(requiredHash.getOrderedShuffledColumns());
+ // }
+ //
+ // if (!this.shuffleType.equals(requiredHash.shuffleType)) {
+ // return false;
+ // }
- if (!shuffleType.equals(spec.shuffleType)) {
- return false;
- }
+ return equalsSatisfy(requiredHash.getOrderedShuffledColumns());
+ }
+
+ private boolean containsSatisfy(List<ExprId> required) {
+ BitSet containsBit = new BitSet(orderedShuffledColumns.size());
+ required.forEach(e -> {
+ if (exprIdToEquivalenceSet.containsKey(e)) {
+ containsBit.set(exprIdToEquivalenceSet.get(e));
+ }
+ });
+ return containsBit.nextClearBit(0) >= orderedShuffledColumns.size();
+ }
- return shuffledColumns.size() == spec.shuffledColumns.size()
- && shuffledColumns.equals(spec.shuffledColumns);
+ private boolean equalsSatisfy(List<ExprId> required) {
+ for (int i = 0; i < required.size(); i++) {
+ if (!equivalenceExprIds.get(i).contains(required.get(i))) {
+ return false;
+ }
+ }
+ return true;
}
@Override
public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
if (!super.equals(o)) {
return false;
}
DistributionSpecHash that = (DistributionSpecHash) o;
- return shuffledColumns.equals(that.shuffledColumns)
- && shuffleType.equals(that.shuffleType);
- // && propertyInfo.equals(that.propertyInfo)
+ return tableId == that.tableId &&
orderedShuffledColumns.equals(that.orderedShuffledColumns)
+ && shuffleType == that.shuffleType &&
partitionIds.equals(that.partitionIds)
+ && equivalenceExprIds.equals(
+ that.equivalenceExprIds) &&
exprIdToEquivalenceSet.equals(that.exprIdToEquivalenceSet);
}
@Override
public int hashCode() {
- return Objects.hash(shuffledColumns, shuffleType);
+ return Objects.hash(orderedShuffledColumns, shuffleType, tableId,
partitionIds,
+ equivalenceExprIds, exprIdToEquivalenceSet);
+ }
+
+ @Override
+ public String toString() {
+ return "DistributionSpecHash{"
Review Comment:
use toSqlString()
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalDistribute.java:
##########
@@ -35,30 +36,40 @@
/**
* Enforcer plan.
*/
-public class PhysicalDistribution<CHILD_TYPE extends Plan> extends
PhysicalUnary<CHILD_TYPE> {
+public class PhysicalDistribute<CHILD_TYPE extends Plan> extends
PhysicalUnary<CHILD_TYPE> {
Review Comment:
why rename?
We should use noun.
Even if we need to rename, I would recommend `Distribution`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java:
##########
@@ -31,88 +39,211 @@
@Developing
public class DistributionSpecHash extends DistributionSpec {
- private final List<SlotReference> shuffledColumns;
+ private final List<ExprId> orderedShuffledColumns;
private final ShuffleType shuffleType;
- public DistributionSpecHash(List<SlotReference> shuffledColumns,
ShuffleType shuffleType) {
- // Preconditions.checkState(!shuffledColumns.isEmpty());
- this.shuffledColumns = shuffledColumns;
- this.shuffleType = shuffleType;
+ // below two attributes use for colocate join
+ private final long tableId;
+
+ private final Set<Long> partitionIds;
+
+ // use for satisfied judge
+ private final List<Set<ExprId>> equivalenceExprIds;
+
+ private final Map<ExprId, Integer> exprIdToEquivalenceSet;
+
+ /**
+ * Use for no need set table related attributes.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType) {
+ this(orderedShuffledColumns, shuffleType, -1L, Collections.emptySet());
+ }
+
+ /**
+ * Use for merge two shuffle columns.
+ */
+ public DistributionSpecHash(List<ExprId> leftColumns, List<ExprId>
rightColumns, ShuffleType shuffleType) {
+ this(leftColumns, shuffleType, -1L, Collections.emptySet());
+ Objects.requireNonNull(rightColumns);
+ Preconditions.checkArgument(leftColumns.size() == rightColumns.size());
+ for (int i = 0; i < rightColumns.size(); i++) {
+ exprIdToEquivalenceSet.put(rightColumns.get(i), i);
+ equivalenceExprIds.get(i).add(rightColumns.get(i));
+ }
+ }
+
+ /**
+ * Normal constructor.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType,
+ long tableId, Set<Long> partitionIds) {
+ this.orderedShuffledColumns =
Objects.requireNonNull(orderedShuffledColumns);
+ this.shuffleType = Objects.requireNonNull(shuffleType);
+ this.tableId = tableId;
+ this.partitionIds = Objects.requireNonNull(partitionIds);
+ this.equivalenceExprIds = Lists.newArrayList();
+ this.exprIdToEquivalenceSet = Maps.newHashMap();
+ orderedShuffledColumns.forEach(id -> {
+ exprIdToEquivalenceSet.put(id, equivalenceExprIds.size());
+ equivalenceExprIds.add(Sets.newHashSet(id));
+ });
+ }
+
+ /**
+ * Used in merge outside and put result into it.
+ */
+ public DistributionSpecHash(List<ExprId> orderedShuffledColumns,
ShuffleType shuffleType, long tableId,
+ Set<Long> partitionIds, List<Set<ExprId>> equivalenceExprIds,
Map<ExprId, Integer> exprIdToEquivalenceSet) {
+ this.orderedShuffledColumns =
Objects.requireNonNull(orderedShuffledColumns);
+ this.shuffleType = Objects.requireNonNull(shuffleType);
+ this.tableId = tableId;
+ this.partitionIds = Objects.requireNonNull(partitionIds);
+ this.equivalenceExprIds = Objects.requireNonNull(equivalenceExprIds);
+ this.exprIdToEquivalenceSet =
Objects.requireNonNull(exprIdToEquivalenceSet);
+ }
+
+ static DistributionSpecHash merge(DistributionSpecHash left,
DistributionSpecHash right, ShuffleType shuffleType) {
+ List<ExprId> orderedShuffledColumns = left.getOrderedShuffledColumns();
+ List<Set<ExprId>> equivalenceExprIds =
Lists.newArrayListWithCapacity(orderedShuffledColumns.size());
+ for (int i = 0; i < orderedShuffledColumns.size(); i++) {
+ Set<ExprId> equivalenceExprId = Sets.newHashSet();
+ equivalenceExprId.addAll(left.getEquivalenceExprIds().get(i));
+ equivalenceExprId.addAll(right.getEquivalenceExprIds().get(i));
+ equivalenceExprIds.add(equivalenceExprId);
+ }
+ Map<ExprId, Integer> exprIdToEquivalenceSet = Maps.newHashMap();
+ exprIdToEquivalenceSet.putAll(left.getExprIdToEquivalenceSet());
+ exprIdToEquivalenceSet.putAll(right.getExprIdToEquivalenceSet());
+ return new DistributionSpecHash(orderedShuffledColumns, shuffleType,
+ left.getTableId(), left.getPartitionIds(), equivalenceExprIds,
exprIdToEquivalenceSet);
}
- public List<SlotReference> getShuffledColumns() {
- return shuffledColumns;
+ static DistributionSpecHash merge(DistributionSpecHash left,
DistributionSpecHash right) {
+ return merge(left, right, left.getShuffleType());
+ }
+
+ public List<ExprId> getOrderedShuffledColumns() {
+ return orderedShuffledColumns;
}
public ShuffleType getShuffleType() {
return shuffleType;
}
+ public long getTableId() {
+ return tableId;
+ }
+
+ public Set<Long> getPartitionIds() {
+ return partitionIds;
+ }
+
+ public List<Set<ExprId>> getEquivalenceExprIds() {
+ return equivalenceExprIds;
+ }
+
+ public Map<ExprId, Integer> getExprIdToEquivalenceSet() {
+ return exprIdToEquivalenceSet;
+ }
+
@Override
- public boolean satisfy(DistributionSpec other) {
- if (other instanceof DistributionSpecAny) {
+ public boolean satisfy(DistributionSpec required) {
+ if (required instanceof DistributionSpecAny) {
return true;
}
- if (!(other instanceof DistributionSpecHash)) {
+ if (!(required instanceof DistributionSpecHash)) {
return false;
}
- DistributionSpecHash spec = (DistributionSpecHash) other;
+ DistributionSpecHash requiredHash = (DistributionSpecHash) required;
- if (shuffledColumns.size() > spec.shuffledColumns.size()) {
+ if (this.orderedShuffledColumns.size() >
requiredHash.orderedShuffledColumns.size()) {
return false;
}
- // TODO: need consider following logic whether is right, and maybe
need consider more.
- // TODO: consider Agg.
- // Current shuffleType is LOCAL/AGG, allow if current is contained by
other
- if (shuffleType == ShuffleType.LOCAL || spec.shuffleType ==
ShuffleType.AGG) {
- return new
HashSet<>(spec.shuffledColumns).containsAll(shuffledColumns);
+ if (requiredHash.shuffleType == ShuffleType.AGGREGATE) {
+ return containsSatisfy(requiredHash.getOrderedShuffledColumns());
}
- if (shuffleType == ShuffleType.AGG && spec.shuffleType ==
ShuffleType.JOIN) {
- return shuffledColumns.size() == spec.shuffledColumns.size()
- && shuffledColumns.equals(spec.shuffledColumns);
- } else if (shuffleType == ShuffleType.JOIN && spec.shuffleType ==
ShuffleType.AGG) {
- return new
HashSet<>(spec.shuffledColumns).containsAll(shuffledColumns);
- }
+ // if (requiredHash.shuffleType == ShuffleType.JOIN) {
+ // return equalsSatisfy(requiredHash.getOrderedShuffledColumns());
+ // }
+ //
+ // if (!this.shuffleType.equals(requiredHash.shuffleType)) {
+ // return false;
+ // }
- if (!shuffleType.equals(spec.shuffleType)) {
- return false;
- }
+ return equalsSatisfy(requiredHash.getOrderedShuffledColumns());
+ }
+
+ private boolean containsSatisfy(List<ExprId> required) {
+ BitSet containsBit = new BitSet(orderedShuffledColumns.size());
+ required.forEach(e -> {
+ if (exprIdToEquivalenceSet.containsKey(e)) {
+ containsBit.set(exprIdToEquivalenceSet.get(e));
+ }
+ });
+ return containsBit.nextClearBit(0) >= orderedShuffledColumns.size();
+ }
- return shuffledColumns.size() == spec.shuffledColumns.size()
- && shuffledColumns.equals(spec.shuffledColumns);
+ private boolean equalsSatisfy(List<ExprId> required) {
+ for (int i = 0; i < required.size(); i++) {
+ if (!equivalenceExprIds.get(i).contains(required.get(i))) {
+ return false;
+ }
+ }
+ return true;
}
@Override
public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
if (!super.equals(o)) {
return false;
}
DistributionSpecHash that = (DistributionSpecHash) o;
- return shuffledColumns.equals(that.shuffledColumns)
- && shuffleType.equals(that.shuffleType);
- // && propertyInfo.equals(that.propertyInfo)
+ return tableId == that.tableId &&
orderedShuffledColumns.equals(that.orderedShuffledColumns)
+ && shuffleType == that.shuffleType &&
partitionIds.equals(that.partitionIds)
+ && equivalenceExprIds.equals(
+ that.equivalenceExprIds) &&
exprIdToEquivalenceSet.equals(that.exprIdToEquivalenceSet);
Review Comment:
```suggestion
&& equivalenceExprIds.equals(that.equivalenceExprIds)
&&
exprIdToEquivalenceSet.equals(that.exprIdToEquivalenceSet);
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/OrderSpec.java:
##########
@@ -59,9 +60,22 @@ public boolean satisfy(OrderSpec other) {
return true;
}
- public GroupExpression addEnforcer(Group child) {
+ /**
+ * add a local order enforcer on child group.
+ */
+ public GroupExpression addLocalEnforcer(Group child) {
+ return new GroupExpression(
+ new PhysicalLocalQuickSort<>(orderKeys,
child.getLogicalProperties(), new GroupPlan(child)),
+ Lists.newArrayList(child)
+ );
+ }
+
+ /**
+ * add a global order enforcer on child group.
+ */
+ public GroupExpression addGlobalEnforcer(Group child) {
Review Comment:
```suggestion
public GroupExpression addGlobalQuickSortEnforcer(Group child) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/OrderSpec.java:
##########
@@ -59,9 +60,22 @@ public boolean satisfy(OrderSpec other) {
return true;
}
- public GroupExpression addEnforcer(Group child) {
+ /**
+ * add a local order enforcer on child group.
+ */
+ public GroupExpression addLocalEnforcer(Group child) {
Review Comment:
```suggestion
public GroupExpression addLocalQuickSortEnforcer(Group child) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/CostAndEnforcerJob.java:
##########
@@ -203,11 +200,13 @@ public void execute() {
}
private void enforce(PhysicalProperties outputProperty,
List<PhysicalProperties> requestChildrenProperty) {
- EnforceMissingPropertiesHelper enforceMissingPropertiesHelper = new
EnforceMissingPropertiesHelper(context,
- groupExpression, curTotalCost);
+ EnforceMissingPropertiesHelper enforceMissingPropertiesHelper
+ = new EnforceMissingPropertiesHelper(context, groupExpression,
curTotalCost);
PhysicalProperties requestedProperties =
context.getRequiredProperties();
if (!outputProperty.satisfy(requestedProperties)) {
+ // TODO: some times we cannot add enforce to satisfied
requestedProperties
Review Comment:
Example?
--
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: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
