PhysicalRecursiveCte.java
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package org.apache.doris.nereids.trees.plans.physical;
import org.apache.doris.nereids.memo.GroupExpression;
import org.apache.doris.nereids.properties.DataTrait;
import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.algebra.RecursiveCte;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.nereids.util.Utils;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.statistics.Statistics;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
/**
* PhysicalRecursiveCte is basically like PhysicalUnion
*/
public class PhysicalRecursiveCte extends AbstractPhysicalPlan implements RecursiveCte {
private final String cteName;
private final List<NamedExpression> outputs;
private final List<List<SlotReference>> regularChildrenOutputs;
private final boolean isUnionAll;
/** PhysicalRecursiveCte */
public PhysicalRecursiveCte(String cteName, boolean isUnionAll,
List<NamedExpression> outputs,
List<List<SlotReference>> childrenOutputs,
LogicalProperties logicalProperties,
List<Plan> children) {
this(cteName, isUnionAll, outputs, childrenOutputs, Optional.empty(), logicalProperties, children);
}
/** PhysicalRecursiveCte */
public PhysicalRecursiveCte(String cteName, boolean isUnionAll,
List<NamedExpression> outputs,
List<List<SlotReference>> childrenOutputs,
Optional<GroupExpression> groupExpression,
LogicalProperties logicalProperties,
List<Plan> children) {
this(cteName, isUnionAll, outputs, childrenOutputs, groupExpression, logicalProperties,
PhysicalProperties.ANY, null, children);
}
/** PhysicalRecursiveCte */
public PhysicalRecursiveCte(String cteName, boolean isUnionAll, List<NamedExpression> outputs,
List<List<SlotReference>> childrenOutputs,
Optional<GroupExpression> groupExpression, LogicalProperties logicalProperties,
PhysicalProperties physicalProperties, Statistics statistics, List<Plan> children) {
super(PlanType.PHYSICAL_RECURSIVE_CTE, groupExpression, logicalProperties, physicalProperties,
statistics, children.toArray(new Plan[0]));
this.cteName = cteName;
this.isUnionAll = isUnionAll;
this.outputs = ImmutableList.copyOf(outputs);
this.regularChildrenOutputs = ImmutableList.copyOf(childrenOutputs);
}
@Override
public boolean isUnionAll() {
return isUnionAll;
}
public String getCteName() {
return cteName;
}
@Override
public List<SlotReference> getRegularChildOutput(int i) {
return regularChildrenOutputs.get(i);
}
@Override
public List<NamedExpression> getOutputs() {
return outputs;
}
@Override
public List<Slot> computeOutput() {
return outputs.stream()
.map(NamedExpression::toSlot)
.collect(ImmutableList.toImmutableList());
}
@Override
public List<List<SlotReference>> getRegularChildrenOutputs() {
return regularChildrenOutputs;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
PhysicalRecursiveCte that = (PhysicalRecursiveCte) o;
return cteName.equals(that.cteName) && isUnionAll == that.isUnionAll && Objects.equals(outputs, that.outputs)
&& Objects.equals(regularChildrenOutputs, that.regularChildrenOutputs);
}
@Override
public int hashCode() {
return Objects.hash(cteName, isUnionAll, outputs, regularChildrenOutputs);
}
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitPhysicalRecursiveCte(this, context);
}
@Override
public List<? extends Expression> getExpressions() {
return regularChildrenOutputs.stream().flatMap(List::stream).collect(ImmutableList.toImmutableList());
}
@Override
public String toString() {
return Utils.toSqlString("PhysicalRecursiveCte" + "[" + id.asInt() + "]" + getGroupIdWithPrefix(),
"stats", statistics,
"cteName", cteName,
"isUnionAll", isUnionAll,
"outputs", outputs,
"regularChildrenOutputs", regularChildrenOutputs);
}
@Override
public String shapeInfo() {
ConnectContext context = ConnectContext.get();
if (context != null
&& context.getSessionVariable().getDetailShapePlanNodesSet().contains(getClass().getSimpleName())) {
StringBuilder builder = new StringBuilder();
builder.append(getClass().getSimpleName());
builder.append(")");
return builder.toString();
} else {
return super.shapeInfo();
}
}
@Override
public PhysicalRecursiveCte withChildren(List<Plan> children) {
return new PhysicalRecursiveCte(cteName, isUnionAll, outputs, regularChildrenOutputs, groupExpression,
getLogicalProperties(), children);
}
@Override
public PhysicalRecursiveCte withGroupExpression(Optional<GroupExpression> groupExpression) {
return new PhysicalRecursiveCte(cteName, isUnionAll, outputs, regularChildrenOutputs,
groupExpression, getLogicalProperties(), children);
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan> children) {
return new PhysicalRecursiveCte(cteName, isUnionAll, outputs, regularChildrenOutputs,
groupExpression, logicalProperties.get(), children);
}
@Override
public PhysicalRecursiveCte withPhysicalPropertiesAndStats(
PhysicalProperties physicalProperties, Statistics statistics) {
return new PhysicalRecursiveCte(cteName, isUnionAll, outputs, regularChildrenOutputs,
groupExpression, getLogicalProperties(), physicalProperties, statistics, children);
}
@Override
public PhysicalRecursiveCte resetLogicalProperties() {
return new PhysicalRecursiveCte(cteName, isUnionAll, outputs, regularChildrenOutputs,
Optional.empty(), null, physicalProperties, statistics, children);
}
@Override
public void computeUnique(DataTrait.Builder builder) {
if (!isUnionAll) {
builder.addUniqueSlot(ImmutableSet.copyOf(getOutput()));
}
}
@Override
public void computeUniform(DataTrait.Builder builder) {
// don't propagate uniform slots
}
private List<BitSet> mapSlotToIndex(Plan plan, List<Set<Slot>> equalSlotsList) {
Map<Slot, Integer> slotToIndex = new HashMap<>();
for (int i = 0; i < plan.getOutput().size(); i++) {
slotToIndex.put(plan.getOutput().get(i), i);
}
List<BitSet> equalSlotIndicesList = new ArrayList<>();
for (Set<Slot> equalSlots : equalSlotsList) {
BitSet equalSlotIndices = new BitSet();
for (Slot slot : equalSlots) {
if (slotToIndex.containsKey(slot)) {
equalSlotIndices.set(slotToIndex.get(slot));
}
}
if (equalSlotIndices.cardinality() > 1) {
equalSlotIndicesList.add(equalSlotIndices);
}
}
return equalSlotIndicesList;
}
@Override
public void computeEqualSet(DataTrait.Builder builder) {
if (children.isEmpty()) {
return;
}
// Get the list of equal slot sets and their corresponding index mappings for the first child
List<Set<Slot>> childEqualSlotsList = child(0).getLogicalProperties()
.getTrait().calAllEqualSet();
List<BitSet> childEqualSlotsIndicesList = mapSlotToIndex(child(0), childEqualSlotsList);
List<BitSet> unionEqualSlotIndicesList = new ArrayList<>(childEqualSlotsIndicesList);
// Traverse all children and find the equal sets that exist in all children
for (int i = 1; i < children.size(); i++) {
Plan child = children.get(i);
// Get the equal slot sets for the current child
childEqualSlotsList = child.getLogicalProperties().getTrait().calAllEqualSet();
// Map slots to indices for the current child
childEqualSlotsIndicesList = mapSlotToIndex(child, childEqualSlotsList);
// Only keep the equal pairs that exist in all children of the union
// This is done by calculating the intersection of all children's equal slot indices
for (BitSet unionEqualSlotIndices : unionEqualSlotIndicesList) {
BitSet intersect = new BitSet();
for (BitSet childEqualSlotIndices : childEqualSlotsIndicesList) {
if (unionEqualSlotIndices.intersects(childEqualSlotIndices)) {
intersect = childEqualSlotIndices;
break;
}
}
unionEqualSlotIndices.and(intersect);
}
}
// Build the functional dependencies for the output slots
List<Slot> outputList = getOutput();
for (BitSet equalSlotIndices : unionEqualSlotIndicesList) {
if (equalSlotIndices.cardinality() <= 1) {
continue;
}
int first = equalSlotIndices.nextSetBit(0);
int next = equalSlotIndices.nextSetBit(first + 1);
while (next > 0) {
builder.addEqualPair(outputList.get(first), outputList.get(next));
next = equalSlotIndices.nextSetBit(next + 1);
}
}
}
@Override
public void computeFd(DataTrait.Builder builder) {
// don't generate
}
}