Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -1053,7 +1053,9 @@ private LogicalPlan bindUsingJoin(MatchingContext<LogicalUsingJoin<Plan, Plan>>
ImmutableList.Builder<NamedExpression> projectExprs = ImmutableList.builder();
ImmutableList.Builder<NamedExpression> asteriskOutputs = ImmutableList.builder();

// 1. Add merged key columns (without qualifier)
// 1. Add merged key columns (without qualifier). A forwarded key keeps the preserved side's
// ExprId so qualified and unqualified references denote the same column in GROUP BY/HAVING/ORDER BY.
// Only FULL OUTER JOIN computes a new key and therefore needs a new Alias.
for (int i = 0; i < leftConjunctsSlots.size(); i++) {
Slot leftSlot = leftConjunctsSlots.get(i);
Slot rightSlot = rightConjunctsSlots.get(i);
Expand All @@ -1064,10 +1066,10 @@ private LogicalPlan bindUsingJoin(MatchingContext<LogicalUsingJoin<Plan, Plan>>
mergeExpr = new Alias(new Coalesce(leftSlot, rightSlot), colName);
} else if (joinType.isRightJoin() || joinType.isRightSemiOrAntiJoin()) {
// RIGHT OUTER / RIGHT SEMI / RIGHT ANTI: use right side (preserved side)
mergeExpr = new Alias(rightSlot, colName);
mergeExpr = rightSlot.withQualifier(ImmutableList.of());
} else {
// LEFT OUTER / INNER / LEFT SEMI / LEFT ANTI / CROSS: use left side
mergeExpr = new Alias(leftSlot, colName);
mergeExpr = leftSlot.withQualifier(ImmutableList.of());

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve CTE column order when sharing the USING-key ExprId

WITH q(x,y,z) AS (SELECT a,t1.b,t1.a FROM t1 JOIN t2 USING(a)) SELECT * FROM q now exposes its columns in the wrong order. The producer becomes:

Project(a#L, t1.b#B, t1.a#L)
  Join(t1.a = t2.a)

AnalyzeCTE wraps this with column names [x#L,y#B,z#L], retaining the ExprIds. LogicalCTEConsumer then keys producerToConsumerOutputMap by producer Slot, whose equality compares only ExprId, and computeOutput() returns that multimap's grouped values(): [x,z,y]. Star expansion binds this order before CTE inlining, which does not repair it. With a matching t1(a,b)=(1,10) row, the result is (1,1,10) instead of (1,10,1); positional consumers such as UNION or INSERT SELECT can consequently receive wrong values. Previously the merged Alias had a separate ExprId, so this query preserved its column order. The right-preserved branch has the same issue.

Preserve producer column positions in the CTE consumer alongside this identity change, and add a result-checking regression with merged and qualified keys separated by a non-key column.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed locally: the reported SQL fails on the original PR head with expected column order [x,y,z] but actual [x,z,y]. Fixed in 363d34b.

LogicalCTEConsumer now takes its output from the insertion-ordered consumer-to-producer map, whose distinct consumer keys preserve each producer position. Nullability adjustment, producer ExprId replacement and variant subpath pruning also retain that order when rebuilding the map. The USING-key identity fix remains intact.

Added inner/left/right USING CTE order checks with full planning, an actual FE-computed repeated-column result check for (1,10,1), and consumer copy/pruning/nullability/ExprId replacement tests. All 73 focused FE tests passed through run-fe-ut.sh; the full FE build and Checkstyle also passed. The result check uses the FE result-computation path; no BE execution is claimed.

}
projectExprs.add(mergeExpr);
asteriskOutputs.add(mergeExpr);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -421,15 +421,13 @@ public Plan visitLogicalPartitionTopN(LogicalPartitionTopN<? extends Plan> parti
public Plan visitLogicalCTEConsumer(LogicalCTEConsumer cteConsumer, Map<ExprId, Slot> replaceMap) {
Map<Slot, Slot> consumerToProducerOutputMap = new LinkedHashMap<>();
Multimap<Slot, Slot> producerToConsumerOutputMap = LinkedHashMultimap.create();
for (Slot producerOutputSlot : cteConsumer.getConsumerToProducerOutputMap().values()) {
Optional<Slot> newProducerOutputSlot = updateExpression(producerOutputSlot, replaceMap, true);
for (Slot consumerOutputSlot : cteConsumer.getProducerToConsumerOutputMap().get(producerOutputSlot)) {
Slot slot = newProducerOutputSlot.orElse(producerOutputSlot);
Slot newConsumerOutputSlot = consumerOutputSlot.withNullable(slot.nullable());
producerToConsumerOutputMap.put(slot, newConsumerOutputSlot);
consumerToProducerOutputMap.put(newConsumerOutputSlot, slot);
replaceMap.put(newConsumerOutputSlot.getExprId(), newConsumerOutputSlot);
}
for (Map.Entry<Slot, Slot> entry : cteConsumer.getConsumerToProducerOutputMap().entrySet()) {
Slot producerOutputSlot = entry.getValue();
Slot slot = updateExpression(producerOutputSlot, replaceMap, true).orElse(producerOutputSlot);
Slot newConsumerOutputSlot = entry.getKey().withNullable(slot.nullable());
producerToConsumerOutputMap.put(slot, newConsumerOutputSlot);
consumerToProducerOutputMap.put(newConsumerOutputSlot, slot);
replaceMap.put(newConsumerOutputSlot.getExprId(), newConsumerOutputSlot);
}
return cteConsumer.withTwoMaps(consumerToProducerOutputMap, producerToConsumerOutputMap);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -143,15 +143,15 @@ public Plan visitLogicalCTEConsumer(LogicalCTEConsumer cteConsumer, Map<ExprId,
// still references the old ExprId, so we must rebuild both maps with the new ExprIds.
Map<Slot, Slot> newConsumerToProducer = new LinkedHashMap<>();
Multimap<Slot, Slot> newProducerToConsumer = LinkedHashMultimap.create();
for (Slot producerSlot : cteConsumer.getConsumerToProducerOutputMap().values()) {
for (Map.Entry<Slot, Slot> entry : cteConsumer.getConsumerToProducerOutputMap().entrySet()) {
Slot producerSlot = entry.getValue();
ExprId newExprId = resolveExprIdChain(producerSlot.getExprId(), replaceMap);
Slot effectiveProducerSlot = newExprId != null
? (Slot) producerSlot.withExprId(newExprId)
: producerSlot;
for (Slot consumerSlot : cteConsumer.getProducerToConsumerOutputMap().get(producerSlot)) {
newProducerToConsumer.put(effectiveProducerSlot, consumerSlot);
newConsumerToProducer.put(consumerSlot, effectiveProducerSlot);
}
Slot consumerSlot = entry.getKey();
newProducerToConsumer.put(effectiveProducerSlot, consumerSlot);
newConsumerToProducer.put(consumerSlot, effectiveProducerSlot);
}
return cteConsumer.withTwoMaps(newConsumerToProducer, newProducerToConsumer);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -422,7 +422,7 @@ public Plan visitLogicalCTEConsumer(LogicalCTEConsumer cteConsumer, Context cont
.map(ExpressionTrait::getDataType).noneMatch(VariantType.class::isInstance)) {
return cteConsumer;
}
Map<Slot, Slot> consumerToProducerOutputMap = Maps.newHashMap();
Map<Slot, Slot> consumerToProducerOutputMap = Maps.newLinkedHashMap();
Multimap<Slot, Slot> producerToConsumerOutputMap = LinkedHashMultimap.create();
Map<Slot, Map<List<String>, SlotReference>> oriSlotToSubPathToSlot = Maps.newHashMap();
for (Map.Entry<Slot, Slot> consumerToProducer : cteConsumer.getConsumerToProducerOutputMap().entrySet()) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ public class LogicalCTEConsumer extends LogicalRelation implements BlockFuncDeps

private final String name;
private final CTEId cteId;
// Insertion order is the consumer column order, including repeated producer slots.
private final Map<Slot, Slot> consumerToProducerOutputMap;
private final Multimap<Slot, Slot> producerToConsumerOutputMap;

Expand Down Expand Up @@ -159,7 +160,7 @@ public LogicalCTEConsumer withRelationId(RelationId relationId) {

@Override
public List<Slot> computeOutput() {
return ImmutableList.copyOf(producerToConsumerOutputMap.values());
return ImmutableList.copyOf(consumerToProducerOutputMap.keySet());
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,16 @@

package org.apache.doris.nereids.rules.analysis;

import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.pattern.GeneratedPlanPatterns;
import org.apache.doris.nereids.rules.RulePromise;
import org.apache.doris.nereids.trees.expressions.Alias;
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.functions.scalar.Coalesce;
import org.apache.doris.nereids.trees.plans.ComputeResultSet;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.utframe.TestWithFeService;

Expand Down Expand Up @@ -119,13 +122,12 @@ void testUsingJoinProducesProjectOnTop() {
// ---- 3. Merge key expression checks ----

@Test
void testInnerJoinUsingMergeKeyIsLeftAlias() {
String sql = "SELECT * FROM t1 JOIN t2 USING(a)";
PlanChecker.from(connectContext)
.analyze(sql)
.matches(logicalProject().when(project -> project.getProjects().stream()
.filter(p -> p instanceof Alias && p.getName().equals("a"))
.anyMatch(alias -> alias.child(0) instanceof Slot)));
void testInnerJoinUsingMergeKeyKeepsLeftExprId() {
List<Slot> output = PlanChecker.from(connectContext)
.analyze("SELECT a, t1.a FROM t1 JOIN t2 USING(a)").getPlan().getOutput();
Assertions.assertEquals(output.get(1).getExprId(), output.get(0).getExprId());
Assertions.assertTrue(output.get(0).getQualifier().isEmpty());
Assertions.assertTrue(output.get(1).getQualifier().contains("t1"));
}

@Test
Expand Down Expand Up @@ -187,19 +189,11 @@ void testRightJoinUsingAsteriskOutputHasSingleKey() {

@Test
void testRightJoinMergeKeyFromRightSide() {
String sql = "SELECT * FROM t1 RIGHT JOIN t2 USING(a)";
PlanChecker.from(connectContext)
.analyze(sql)
.matches(logicalProject().when(project -> project.getProjects().stream()
.filter(p -> p instanceof Alias && p.getName().equals("a"))
.map(alias -> alias.child(0))
.filter(Slot.class::isInstance)
.map(Slot.class::cast)
.anyMatch(slot -> {
// Merge key should come from right table (t2)
return slot.getQualifier().stream()
.anyMatch(q -> q.contains("t2"));
})));
List<Slot> output = PlanChecker.from(connectContext)
.analyze("SELECT a, t2.a FROM t1 RIGHT JOIN t2 USING(a)").getPlan().getOutput();
Assertions.assertEquals(output.get(1).getExprId(), output.get(0).getExprId());
Assertions.assertTrue(output.get(0).getQualifier().isEmpty());
Assertions.assertTrue(output.get(1).getQualifier().contains("t2"));
}

// ---- 6. RIGHT SEMI JOIN merge key is visible ----
Expand Down Expand Up @@ -404,4 +398,84 @@ void testChainedFullOuterUsingJoinAsteriskOutput() {
return outputNames.equals(List.of("ak", "pk", "lv", "rv", "nv"));
}));
}

@Test
void testUsingJoinGroupByPreservedKey() {
String[] joinTypes = {
"JOIN", "LEFT JOIN", "LEFT SEMI JOIN", "LEFT ANTI JOIN",
"RIGHT JOIN", "RIGHT SEMI JOIN", "RIGHT ANTI JOIN"
};
for (String joinType : joinTypes) {
String key = joinType.startsWith("RIGHT") ? "t2.a" : "t1.a";
String from = " FROM t1 " + joinType + " t2 USING(a)";
PlanChecker.from(connectContext).checkPlannerResult(
"SELECT " + key + ", COUNT(*)" + from + " GROUP BY a");
PlanChecker.from(connectContext).checkPlannerResult(
"SELECT a, COUNT(*)" + from + " GROUP BY " + key);
PlanChecker.from(connectContext).checkPlannerResult(
"SELECT COUNT(*)" + from + " GROUP BY a HAVING " + key + " > 0 ORDER BY " + key);
PlanChecker.from(connectContext).checkPlannerResult(
"SELECT " + key + " + 1, COUNT(*)" + from + " GROUP BY a + 1");
}
}

@Test
void testUsingJoinGroupByMultipleKeys() {
PlanChecker.from(connectContext).checkPlannerResult(
"SELECT l.a, l.b, COUNT(*) FROM t1 l LEFT JOIN t1 r USING(a,b) GROUP BY a,b");
PlanChecker.from(connectContext).checkPlannerResult(
"SELECT a, b, COUNT(*) FROM t1 l LEFT JOIN t1 r USING(a,b) GROUP BY l.a,l.b");
}

@Test
void testUsingJoinGroupByDoesNotMergeNullableSide() {
String[] queries = {
"SELECT t2.a, COUNT(*) FROM t1 LEFT JOIN t2 USING(a) GROUP BY a",
"SELECT t1.a, COUNT(*) FROM t1 RIGHT JOIN t2 USING(a) GROUP BY a",
"SELECT t1.a, COUNT(*) FROM t1 FULL JOIN t2 USING(a) GROUP BY a",
"SELECT a, COUNT(*) FROM t1 FULL JOIN t2 USING(a) GROUP BY t1.a"
};
for (String sql : queries) {
AnalysisException error = Assertions.assertThrows(AnalysisException.class,
() -> PlanChecker.from(connectContext).analyze(sql), sql);
Assertions.assertTrue(error.getMessage().contains("must appear in the GROUP BY"), error.getMessage());
}
}

@Test
void testUsingJoinGroupByDerivedInput() {
PlanChecker.from(connectContext).checkPlannerResult(
"WITH l AS (SELECT 1 AS k, 10 AS v UNION ALL SELECT 2, 20), "
+ "r AS (SELECT 1 AS k UNION ALL SELECT 3) "
+ "SELECT l.k, SUM(l.v) FROM l LEFT JOIN r USING(k) "
+ "GROUP BY k HAVING l.k > 0 ORDER BY l.k");
}

@Test
void testUsingJoinCteColumnOrder() {
String[] joins = {"JOIN", "LEFT JOIN", "RIGHT JOIN"};
for (String join : joins) {
String source = join.startsWith("RIGHT") ? "t2.a" : "t1.a";
String payload = join.startsWith("RIGHT") ? "t2.c" : "t1.b";
String sql = "WITH q(x,y,z) AS (SELECT a," + payload + "," + source
+ " FROM t1 " + join + " t2 USING(a)) SELECT * FROM q";
List<Slot> output = PlanChecker.from(connectContext).analyze(sql).getPlan().getOutput();
Assertions.assertEquals(List.of("x", "y", "z"), output.stream()
.map(Slot::getName).collect(Collectors.toList()), sql);
PlanChecker.from(connectContext).checkPlannerResult(sql);
}
}

@Test
void testCteRepeatedColumnResults() {
String sql = "WITH q(x,y,z) AS (SELECT a,b,a FROM (SELECT 1 AS a,10 AS b) t) SELECT * FROM q";
PlanChecker.from(connectContext).checkPlannerResult(sql, planner -> {
PhysicalPlan plan = planner.getPhysicalPlan();
Assertions.assertTrue(plan instanceof ComputeResultSet);
Assertions.assertEquals(List.of(List.of("1", "10", "1")),
((ComputeResultSet) plan).computeResultInFe(planner.getCascadesContext(), plan.getOutput())
.orElseThrow().getResultRows());
});
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
// 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.logical;

import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
import org.apache.doris.nereids.rules.rewrite.EliminateGroupByKey;
import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.types.IntegerType;

import com.google.common.collect.ImmutableList;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

class LogicalCTEConsumerTest {
private LogicalCTEConsumer newConsumer() {
Slot key = new SlotReference("x", IntegerType.INSTANCE, false);
Slot payload = new SlotReference("y", IntegerType.INSTANCE, false);
LogicalEmptyRelation producer = new LogicalEmptyRelation(new RelationId(0),
ImmutableList.of(key, payload, key.withName("z")));
return new LogicalCTEConsumer(new RelationId(1), new CTEId(0), "q", producer);
}

private void assertColumnOrder(LogicalCTEConsumer consumer) {
List<Slot> output = consumer.getOutput();
Assertions.assertEquals(List.of("x", "y", "z"), output.stream()
.map(Slot::getName).collect(Collectors.toList()));
Assertions.assertEquals(consumer.getProducerSlot(output.get(0)).getExprId(),
consumer.getProducerSlot(output.get(2)).getExprId());
Assertions.assertNotEquals(consumer.getProducerSlot(output.get(0)).getExprId(),
consumer.getProducerSlot(output.get(1)).getExprId());
}

@Test
void testRepeatedProducerSlotKeepsColumnOrder() {
LogicalCTEConsumer consumer = newConsumer();
assertColumnOrder(consumer);
assertColumnOrder((LogicalCTEConsumer) consumer.withGroupExpression(Optional.empty()));
assertColumnOrder((LogicalCTEConsumer) consumer.withGroupExprLogicalPropChildren(
Optional.empty(), Optional.empty(), ImmutableList.of()));
}

@Test
void testPruneOutputsKeepsColumnOrder() {
LogicalCTEConsumer consumer = newConsumer();
List<Slot> output = ImmutableList.copyOf(consumer.getConsumerToProducerOutputMap().keySet());
LogicalCTEConsumer pruned = (LogicalCTEConsumer) consumer.pruneOutputs(
ImmutableList.of(output.get(1), output.get(2)));
Assertions.assertEquals(List.of("y", "z"), pruned.getOutput().stream()
.map(Slot::getName).collect(Collectors.toList()));
}

@Test
void testAdjustNullableKeepsColumnOrder() {
LogicalCTEConsumer consumer = newConsumer();
Slot producer = consumer.getConsumerToProducerOutputMap().values().iterator().next();
Map<ExprId, Slot> replaceMap = new HashMap<>();
replaceMap.put(producer.getExprId(), producer.withNullable(true));
LogicalCTEConsumer updated = (LogicalCTEConsumer) new AdjustNullable(false)
.visitLogicalCTEConsumer(consumer, replaceMap);
assertColumnOrder(updated);
Assertions.assertTrue(updated.getOutput().get(0).nullable());
Assertions.assertTrue(updated.getOutput().get(2).nullable());
Assertions.assertFalse(updated.getOutput().get(1).nullable());
}

@Test
void testReplaceProducerIdKeepsColumnOrder() {
LogicalCTEConsumer consumer = newConsumer();
Slot producer = consumer.getConsumerToProducerOutputMap().values().iterator().next();
ExprId newId = new ExprId(999);
Map<ExprId, ExprId> replaceMap = new HashMap<>();
replaceMap.put(producer.getExprId(), newId);
LogicalCTEConsumer updated = (LogicalCTEConsumer) new EliminateGroupByKey()
.visitLogicalCTEConsumer(consumer, replaceMap);
assertColumnOrder(updated);
Assertions.assertEquals(newId, updated.getProducerSlot(updated.getOutput().get(0)).getExprId());
Assertions.assertEquals(newId, updated.getProducerSlot(updated.getOutput().get(2)).getExprId());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -136,4 +136,19 @@ suite("test_using_join_merge_key", "query_p0") {
FROM l FULL OUTER JOIN r USING(pk) FULL OUTER JOIN n USING(ak)
ORDER BY l.pk NULLS LAST, r.pk NULLS LAST, n.ak NULLS LAST;
"""

// Qualified and unqualified references to a forwarded USING key must bind to the same slot.
sql """
EXPLAIN
WITH l AS (
SELECT 1 AS k, 10 AS v UNION ALL SELECT 2, 20
), r AS (
SELECT 1 AS k UNION ALL SELECT 3
)
SELECT l.k, SUM(l.v)
FROM l LEFT JOIN r USING(k)
GROUP BY k
HAVING l.k > 0
ORDER BY l.k
"""
}
Loading