This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 9b89eee2ff2 branch-4.2: [fix](MV) Accept constant query outputs when
matching a materialized view #69787 (#70021)
9b89eee2ff2 is described below
commit 9b89eee2ff202c9fbf8bcfff3dc38752349303b6
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Nov 3 17:07:17 2026 +0800
branch-4.2: [fix](MV) Accept constant query outputs when matching a
materialized view #69787 (#70021)
Cherry-picked from #69787
Co-authored-by: minghong <[email protected]>
---
.../org/apache/doris/nereids/CascadesContext.java | 12 +-
.../mv/AbstractMaterializedViewRule.java | 5 +-
.../exploration/mv/MaterializedViewUtils.java | 207 +++++++++++++++-
.../mv/MaterializedViewUtilsAlignTest.java | 273 +++++++++++++++++++++
4 files changed, 488 insertions(+), 9 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
index 0215f007022..95d8d60574a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
@@ -42,6 +42,7 @@ import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.rules.RuleFactory;
import org.apache.doris.nereids.rules.RuleSet;
import org.apache.doris.nereids.rules.exploration.mv.MaterializationContext;
+import org.apache.doris.nereids.rules.exploration.mv.MaterializedViewUtils;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.Expression;
@@ -276,15 +277,22 @@ public class CascadesContext implements ScheduleContext {
if (!statementContext.getRewrittenPlansByMv().isEmpty()) {
// copy tmp plan for mv rewrite firstly
for (Plan rewrittenPlan : rewrittenPlansByMv) {
+ // the rewritten plan may define a constant output with
another exprId than the origin plan, align
+ // the outputs before the comparison, only a same constant is
allowed to change the exprId
+ Plan alignedPlan =
MaterializedViewUtils.normalizeSinkExpressions(rewrittenPlan, plan);
+ if (alignedPlan != null &&
!alignedPlan.getLogicalProperties().equals(plan.getLogicalProperties())) {
+ alignedPlan =
MaterializedViewUtils.alignPlanOutputWithOrigin(rewrittenPlan, plan);
+ }
// aggregate_without_roll_up query_13_0 cause error into
targetGroup but differ in logical properties
// tmp rewritten plan output is different from final rewritten
plan output
- if
(!rewrittenPlan.getLogicalProperties().equals(plan.getLogicalProperties())) {
+ if (alignedPlan == null
+ ||
!alignedPlan.getLogicalProperties().equals(plan.getLogicalProperties())) {
LOG.error("rewritten plan in rbo logical properties are "
+ "different from original plan, query id
is {}",
getConnectContext().getQueryIdentifier());
continue;
}
- this.memo.copyIn(rewrittenPlan, this.memo.getRoot(), false);
+ this.memo.copyIn(alignedPlan, this.memo.getRoot(), false);
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
index 768801f8bf6..ba4b0a1b370 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java
@@ -595,7 +595,10 @@ public abstract class AbstractMaterializedViewRule
implements ExplorationRuleFac
List<Expression> rewrittenExpressions = new ArrayList<>();
for (int exprIndex = 0; exprIndex < sourceShuttledExpressions.size();
exprIndex++) {
Expression expressionShuttledToRewrite =
sourceShuttledExpressions.get(exprIndex);
- if (expressionShuttledToRewrite instanceof Literal) {
+ if (expressionShuttledToRewrite.isConstant()) {
+ // a constant expression needs no counterpart in the view:
e.g. the query column is pinned to a
+ // literal by a predicate, or it is materialized as a constant
above an aggregate, so the view does
+ // not have to output it
rewrittenExpressions.add(expressionShuttledToRewrite);
continue;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtils.java
index 71d36282c34..19ad9c0052b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtils.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtils.java
@@ -456,29 +456,224 @@ public class MaterializedViewUtils {
if (rewrittenPlan.getOutput().size() !=
originPlan.getOutput().size()) {
return null;
}
- if
(rewrittenPlan.getLogicalProperties().equals(originPlan.getLogicalProperties()))
{
- return rewrittenPlan;
- }
List<NamedExpression> rewrittenPlanOutputExprList =
rewrittenPlan.getOutputExprs();
List<? extends NamedExpression> originPlanOutputExprList =
originPlan.getOutput();
if (rewrittenPlanOutputExprList.size() !=
originPlanOutputExprList.size()) {
return null;
}
+ List<Slot> childOutputs = rewrittenPlan.child().getOutput();
+ Map<ExprId, Slot> childOutputByExprId = Maps.newHashMap();
+ for (Slot childOutput : childOutputs) {
+ childOutputByExprId.putIfAbsent(childOutput.getExprId(),
childOutput);
+ }
+ boolean childOutputsPositional = childOutputs.size() ==
rewrittenPlanOutputExprList.size();
+ // a rule can change the exprIds of the child outputs without
updating the sink outputs,
+ // e.g. when a group by key is eliminated by a uniform
predicate. The stale sink outputs
+ // have to be replaced by the output which the child produces
at the same position.
+ boolean outputsResolvable = true;
+ for (NamedExpression rewrittenExpression :
rewrittenPlanOutputExprList) {
+ if (!isExpressionResolvable(rewrittenExpression,
childOutputByExprId)) {
+ outputsResolvable = false;
+ break;
+ }
+ }
+ if (outputsResolvable
+ &&
rewrittenPlan.getLogicalProperties().equals(originPlan.getLogicalProperties()))
{
+ return rewrittenPlan;
+ }
+ if (!outputsResolvable && !childOutputsPositional) {
+ // the sink outputs reference slots which the child does
not produce and they can not
+ // be mapped to the child outputs by position, the
rewritten plan is not usable
+ return null;
+ }
List<NamedExpression> normalizedOutputExprList = new
ArrayList<>();
for (int i = 0; i < rewrittenPlanOutputExprList.size(); i++) {
NamedExpression rewrittenExpression =
rewrittenPlanOutputExprList.get(i);
+ if (childOutputsPositional &&
!isExpressionResolvable(rewrittenExpression, childOutputByExprId)) {
+ rewrittenExpression = childOutputs.get(i);
+ }
NamedExpression originalExpression =
originPlanOutputExprList.get(i);
normalizedOutputExprList.add(normalizeExpression(originalExpression,
rewrittenExpression, true));
}
- LogicalProject<Plan> project = new
LogicalProject<>(normalizedOutputExprList,
- rewrittenPlan.child());
- return rewrittenPlan.withChildren(project);
+ // Set the outputs on the sink. An output exprId which the
child does not produce yet is
+ // materialized by a project between the sink and the child,
see withSinkOutputs.
+ return withSinkOutputs(rewrittenPlan,
normalizedOutputExprList);
}
}, null);
}
+ /**
+ * Set the outputs on the sink and keep the output exprIds resolvable for
the physical translation.
+ *
+ * <p>A sink output whose exprId the child does not produce yet, e.g. the
origin exprId of a constant
+ * which the view rewrite materialized with a fresh exprId, is
materialized by a project between the
+ * sink and its child. The outputs which the child already produces are
passed through with the
+ * nullable property of the child, so a rule which merges the project with
the project of the child,
+ * e.g. MergeProjectsCBO, derives the same nullable property and the plan
keeps the logical properties
+ * of the group it is inserted into.
+ */
+ private static Plan withSinkOutputs(LogicalSink<? extends Plan> sink,
List<NamedExpression> sinkOutputs) {
+ Plan child = sink.child();
+ Map<ExprId, Slot> childOutputs = Maps.newHashMap();
+ for (Slot childOutput : child.getOutput()) {
+ childOutputs.putIfAbsent(childOutput.getExprId(), childOutput);
+ }
+ boolean needProject = false;
+ List<NamedExpression> projectOutputs = new
ArrayList<>(sinkOutputs.size());
+ for (NamedExpression sinkOutput : sinkOutputs) {
+ Slot childOutput = childOutputs.get(sinkOutput.getExprId());
+ if (childOutput == null) {
+ // the child does not produce the exprId, materialize it by
the output expression
+ projectOutputs.add(sinkOutput);
+ needProject = true;
+ } else {
+ // the child already produces the exprId, pass it through with
the nullable property of
+ // the child, the changed nullable property of the sink output
is only applied on the sink
+ projectOutputs.add(childOutput);
+ }
+ }
+ if (!needProject) {
+ return sink.withOutputExprs(sinkOutputs);
+ }
+ // The sink outputs reference the outputs of the project, not the
expressions below the project,
+ // otherwise a sink which validates its output expressions, e.g. a
result sink, would still see a
+ // slot which the sink child does not produce.
+ List<NamedExpression> projectSinkOutputs = sinkOutputs.stream()
+ .map(output -> (NamedExpression) output.toSlot())
+ .collect(ImmutableList.toImmutableList());
+ return sink.withOutputExprs(projectSinkOutputs)
+ .withChildren(new LogicalProject<>(projectOutputs, child));
+ }
+
+ private static boolean isExpressionResolvable(Expression expression,
Map<ExprId, Slot> childOutputs) {
+ for (Slot inputSlot : expression.getInputSlots()) {
+ if (!childOutputs.containsKey(inputSlot.getExprId())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Align the outputs of a plan rewritten by a materialized view with the
origin query plan, so that both plans
+ * can be alternatives of one memo group.
+ *
+ * <p>An output keeps its exprId. The exprId of the origin output may only
be substituted
+ * <ul>
+ * <li>when both plans define that output as the same constant: a
column which a predicate pins to a
+ * literal can be materialized with a fresh exprId, and the two plans
then still compute the same value,
+ * or</li>
+ * <li>when the rewritten output describes the same column as the
origin one: the name and the type have
+ * to match and the qualifier has to be kept by both sides and be
equal, so that a same named column of
+ * another relation (e.g. the two relations of a self join) is not
accepted.</li>
+ * </ul>
+ * Otherwise the two plans are not known to produce the same columns (e.g.
a self join can expose two same
+ * named columns of two relations) and the rewritten plan must not be used.
+ */
+ public static Plan alignPlanOutputWithOrigin(Plan rewrittenPlan, Plan
originPlan) {
+ List<Slot> originOutputs = originPlan.getOutput();
+ List<Slot> rewrittenOutputs = rewrittenPlan instanceof LogicalSink
+ ? ((LogicalSink<?>) rewrittenPlan).child().getOutput()
+ : rewrittenPlan.getOutput();
+ if (originOutputs.size() != rewrittenOutputs.size()) {
+ return null;
+ }
+ Map<ExprId, Expression> originOutputDefinitions =
collectNamedExpressionDefinitions(originPlan);
+ Map<ExprId, Expression> rewrittenOutputDefinitions =
collectNamedExpressionDefinitions(rewrittenPlan);
+ List<NamedExpression> alignedOutputs = new
ArrayList<>(originOutputs.size());
+ for (int i = 0; i < originOutputs.size(); i++) {
+ Slot originSlot = originOutputs.get(i);
+ Slot rewrittenSlot = rewrittenOutputs.get(i);
+ if (rewrittenSlot.getExprId().equals(originSlot.getExprId())) {
+ if (rewrittenSlot instanceof SlotReference &&
rewrittenSlot.nullable() != originSlot.nullable()) {
+ alignedOutputs.add(((SlotReference)
rewrittenSlot).withNullable(originSlot.nullable()));
+ } else {
+ alignedOutputs.add(rewrittenSlot);
+ }
+ continue;
+ }
+ // the slot itself is not a constant, the definition of its exprId
is
+ Expression originDefinition =
originOutputDefinitions.get(originSlot.getExprId());
+ Expression rewrittenDefinition =
rewrittenOutputDefinitions.get(rewrittenSlot.getExprId());
+ if (!isSameConstant(rewrittenDefinition, originDefinition)
+ && !isSameColumn(rewrittenSlot, originSlot)) {
+ return null;
+ }
+ Expression inner = rewrittenSlot.nullable() ==
originSlot.nullable()
+ ? rewrittenSlot
+ : (originSlot.nullable() ? new Nullable(rewrittenSlot) :
new NonNullable(rewrittenSlot));
+ alignedOutputs.add(new Alias(originSlot.getExprId(), inner,
originSlot.getName()));
+ }
+ if (rewrittenPlan instanceof LogicalSink) {
+ // Only the outputs of the sink have to be aligned. If an aligned
exprId is not produced by the
+ // child yet, withSinkOutputs materializes it, otherwise a sink
which translates its outputs by
+ // exprId, e.g. a file sink, can not resolve the slot.
+ return withSinkOutputs((LogicalSink<?>) rewrittenPlan,
alignedOutputs);
+ }
+ return new LogicalProject<>(alignedOutputs, rewrittenPlan);
+ }
+
+ /**
+ * Whether the rewritten output describes the same column as the origin
one. The rewritten plan is built by the
+ * materialized view rewrite which keeps the position of the query
outputs, so the name, the type, the qualifier
+ * and the sub column path have to match. The qualifier has to be kept by
both sides: a missing qualifier can not
+ * rule out that the slot is the same named column of another relation,
e.g. the two relations of a self join, so
+ * an output without a qualifier is rejected and the rewritten plan is not
used. This also rejects an output which
+ * is added or removed and shifts the columns.
+ */
+ private static boolean isSameColumn(Slot rewrittenSlot, Slot originSlot) {
+ if (!rewrittenSlot.getName().equalsIgnoreCase(originSlot.getName())
+ ||
!rewrittenSlot.getDataType().equals(originSlot.getDataType())) {
+ return false;
+ }
+ List<String> rewrittenQualifier = rewrittenSlot.getQualifier();
+ List<String> originQualifier = originSlot.getQualifier();
+ if (rewrittenQualifier.isEmpty() ||
!rewrittenQualifier.equals(originQualifier)) {
+ return false;
+ }
+ // the sub column path is part of the column identity, e.g.
variant['a'] and variant['b'] are
+ // different columns although they have the same name, type and
qualifier
+ List<String> rewrittenSubPath = rewrittenSlot instanceof SlotReference
+ ? ((SlotReference) rewrittenSlot).getSubPath() :
ImmutableList.of();
+ List<String> originSubPath = originSlot instanceof SlotReference
+ ? ((SlotReference) originSlot).getSubPath() :
ImmutableList.of();
+ return rewrittenSubPath.equals(originSubPath);
+ }
+
+ private static Map<ExprId, Expression>
collectNamedExpressionDefinitions(Plan plan) {
+ Map<ExprId, Expression> definitions = Maps.newHashMap();
+ plan.foreach(node -> {
+ for (Expression expression : ((Plan) node).getExpressions()) {
+ if (!(expression instanceof NamedExpression)) {
+ continue;
+ }
+ NamedExpression namedExpression = (NamedExpression) expression;
+ Expression existing =
definitions.get(namedExpression.getExprId());
+ // prefer the definition which materializes the value, a plain
slot reference only consumes it
+ if (existing == null || (!existing.isConstant() &&
namedExpression.isConstant())) {
+ definitions.put(namedExpression.getExprId(),
namedExpression);
+ }
+ }
+ return false;
+ });
+ return definitions;
+ }
+
+ private static boolean isSameConstant(Expression rewrittenDefinition,
Expression originDefinition) {
+ if (rewrittenDefinition == null || originDefinition == null
+ || !rewrittenDefinition.isConstant() ||
!originDefinition.isConstant()) {
+ return false;
+ }
+ Expression rewrittenConstant = rewrittenDefinition instanceof Alias
+ ? rewrittenDefinition.child(0) : rewrittenDefinition;
+ Expression originConstant = originDefinition instanceof Alias
+ ? originDefinition.child(0) : originDefinition;
+ return
rewrittenConstant.getDataType().equals(originConstant.getDataType())
+ && rewrittenConstant.equals(originConstant);
+ }
+
/**
* Normalize expression with query, keep the consistency of exprId and
nullable props with
* query
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtilsAlignTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtilsAlignTest.java
new file mode 100644
index 00000000000..666e38b2811
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtilsAlignTest.java
@@ -0,0 +1,273 @@
+// 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.rules.exploration.mv;
+
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ExprId;
+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.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFileSink;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSink;
+import org.apache.doris.nereids.types.IntegerType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Test for {@link MaterializedViewUtils#alignPlanOutputWithOrigin} and
+ * {@link MaterializedViewUtils#normalizeSinkExpressions}.
+ */
+public class MaterializedViewUtilsAlignTest {
+
+ @Test
+ public void testSameConstantWithDifferentExprIdIsAccepted() {
+ ExprId originExprId = new ExprId(10);
+ Plan origin = project(new Alias(originExprId, new IntegerLiteral(1),
"k1"));
+ Plan rewritten = project(new Alias(new ExprId(11), new
IntegerLiteral(1), "k1"));
+
+ Plan aligned =
MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten, origin);
+
+ Assertions.assertNotNull(aligned);
+ Assertions.assertEquals(originExprId,
aligned.getOutput().get(0).getExprId());
+ Assertions.assertEquals("k1", aligned.getOutput().get(0).getName());
+ }
+
+ @Test
+ public void testSameNameDifferentConstantIsRejected() {
+ Plan origin = project(new Alias(new ExprId(10), new IntegerLiteral(1),
"k1"));
+ Plan rewritten = project(new Alias(new ExprId(11), new
IntegerLiteral(2), "k1"));
+
+
Assertions.assertNull(MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten,
origin));
+ }
+
+ @Test
+ public void testSameColumnWithSameQualifierIsAccepted() {
+ ExprId originExprId = new ExprId(10);
+ Plan origin = project(slot(originExprId, ImmutableList.of("a")));
+ Plan rewritten = project(slot(new ExprId(11), ImmutableList.of("a")));
+
+ Plan aligned =
MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten, origin);
+
+ Assertions.assertNotNull(aligned);
+ Assertions.assertEquals(originExprId,
aligned.getOutput().get(0).getExprId());
+ Assertions.assertEquals("k1", aligned.getOutput().get(0).getName());
+ }
+
+ @Test
+ public void testSameNameColumnOfDifferentRelationIsRejected() {
+ Plan origin = project(slot(new ExprId(10), ImmutableList.of("a")));
+ Plan rewritten = project(slot(new ExprId(11), ImmutableList.of("b")));
+
+
Assertions.assertNull(MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten,
origin));
+ }
+
+ @Test
+ public void testSameNameColumnWithoutQualifierIsRejected() {
+ // The rewritten output lost the qualifier, so it can not be told
apart from the same named column
+ // of another relation, e.g. the other relation of a self join.
+ Plan origin = project(slot(new ExprId(10), ImmutableList.of("a")));
+ Plan rewritten = project(slot(new ExprId(11), ImmutableList.of()));
+
+
Assertions.assertNull(MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten,
origin));
+ }
+
+ @Test
+ public void testSameNameColumnWithoutQualifierOnBothSidesIsRejected() {
+ // Both sides lost the qualifier, so the outputs can still be the same
named columns of two
+ // different relations.
+ Plan origin = project(slot(new ExprId(10), ImmutableList.of()));
+ Plan rewritten = project(slot(new ExprId(11), ImmutableList.of()));
+
+
Assertions.assertNull(MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten,
origin));
+ }
+
+ @Test
+ public void testSameNameColumnOfDifferentSubPathIsRejected() {
+ // variant['a'] and variant['b'] have the same name, type and
qualifier but are different columns.
+ Plan origin = project(slot(new ExprId(10),
ImmutableList.of("a")).withSubPath(ImmutableList.of("a")));
+ Plan rewritten = project(slot(new ExprId(11),
ImmutableList.of("a")).withSubPath(ImmutableList.of("b")));
+
+
Assertions.assertNull(MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten,
origin));
+ }
+
+ @Test
+ public void testSameNameColumnOfSameSubPathIsAccepted() {
+ ExprId originExprId = new ExprId(10);
+ Plan origin = project(slot(originExprId,
ImmutableList.of("a")).withSubPath(ImmutableList.of("b")));
+ Plan rewritten = project(slot(new ExprId(11),
ImmutableList.of("a")).withSubPath(ImmutableList.of("b")));
+
+ Plan aligned =
MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten, origin);
+
+ Assertions.assertNotNull(aligned);
+ Assertions.assertEquals(originExprId,
aligned.getOutput().get(0).getExprId());
+ }
+
+ @Test
+ public void testFileSinkOutputWithFreshExprIdIsMaterialized() {
+ ExprId originExprId = new ExprId(10);
+ ExprId rewrittenExprId = new ExprId(11);
+ SlotReference rewrittenSlot = slot(rewrittenExprId,
ImmutableList.of("a"));
+ LogicalFileSink<Plan> rewritten =
fileSink(ImmutableList.of(rewrittenSlot), project(rewrittenSlot));
+ LogicalFileSink<Plan> origin =
fileSink(ImmutableList.of(slot(originExprId, ImmutableList.of("a"))),
+ oneRow());
+
+ Plan normalized =
MaterializedViewUtils.normalizeSinkExpressions(rewritten, origin);
+
+ Assertions.assertNotNull(normalized);
+ Assertions.assertEquals(originExprId,
normalized.getOutput().get(0).getExprId());
+ Assertions.assertTrue(normalized.child(0) instanceof LogicalProject);
+ assertSinkOutputsResolvable(normalized);
+ assertSinkOutputExpressionsResolvable(normalized);
+ }
+
+ @Test
+ public void testSinkWithOnlyNullablePropertyChangedDoesNotAddProject() {
+ ExprId exprId = new ExprId(10);
+ SlotReference childOutput = new SlotReference(exprId, "k1",
IntegerType.INSTANCE, true,
+ ImmutableList.of("a"));
+ LogicalOneRowRelation child = new LogicalOneRowRelation(new
RelationId(1), ImmutableList.of(childOutput));
+ LogicalFileSink<Plan> rewritten =
fileSink(ImmutableList.of(childOutput), child);
+ LogicalFileSink<Plan> origin = fileSink(ImmutableList.of(
+ new SlotReference(exprId, "k1", IntegerType.INSTANCE, false,
ImmutableList.of("a"))), oneRow());
+
+ Plan normalized =
MaterializedViewUtils.normalizeSinkExpressions(rewritten, origin);
+
+ Assertions.assertNotNull(normalized);
+ Assertions.assertSame(child, normalized.child(0));
+ Assertions.assertFalse(normalized.getOutput().get(0).nullable());
+ assertSinkOutputExpressionsResolvable(normalized);
+ }
+
+ @Test
+ public void
testMaterializedProjectKeepsChildNullableForPassThroughOutput() {
+ ExprId originFirst = new ExprId(10);
+ ExprId rewrittenFirst = new ExprId(11);
+ ExprId shared = new ExprId(12);
+ SlotReference childFirst = slot(rewrittenFirst, ImmutableList.of("a"));
+ SlotReference childShared = new SlotReference(shared, "k2",
IntegerType.INSTANCE, true,
+ ImmutableList.of("a"));
+ LogicalOneRowRelation child = new LogicalOneRowRelation(new
RelationId(1),
+ ImmutableList.of(childFirst, childShared));
+ LogicalFileSink<Plan> rewritten =
fileSink(ImmutableList.of(childFirst, childShared), child);
+ LogicalFileSink<Plan> origin = fileSink(ImmutableList.of(
+ slot(originFirst, ImmutableList.of("a")),
+ new SlotReference(shared, "k2", IntegerType.INSTANCE, false,
ImmutableList.of("a"))), oneRow());
+
+ Plan normalized =
MaterializedViewUtils.normalizeSinkExpressions(rewritten, origin);
+
+ Assertions.assertNotNull(normalized);
+ Assertions.assertEquals(originFirst,
normalized.getOutput().get(0).getExprId());
+ Assertions.assertFalse(normalized.getOutput().get(1).nullable());
+ Plan project = normalized.child(0);
+ Assertions.assertTrue(project instanceof LogicalProject);
+ // the materialized project passes the output which the child already
produces through with the
+ // nullable property of the child, so merging the projects does not
change the logical properties
+ Assertions.assertTrue(project.getOutput().get(1).nullable());
+ assertSinkOutputsResolvable(normalized);
+ assertSinkOutputExpressionsResolvable(normalized);
+ }
+
+ @Test
+ public void testAlignSinkMaterializesOutputOfFreshExprId() {
+ ExprId originExprId = new ExprId(10);
+ ExprId rewrittenExprId = new ExprId(11);
+ SlotReference rewrittenSlot = slot(rewrittenExprId,
ImmutableList.of("a"));
+ LogicalFileSink<Plan> rewritten =
fileSink(ImmutableList.of(rewrittenSlot),
+ new LogicalOneRowRelation(new RelationId(1),
ImmutableList.of(rewrittenSlot)));
+ LogicalFileSink<Plan> origin =
fileSink(ImmutableList.of(slot(originExprId, ImmutableList.of("a"))),
+ oneRow());
+
+ Plan aligned =
MaterializedViewUtils.alignPlanOutputWithOrigin(rewritten, origin);
+
+ Assertions.assertNotNull(aligned);
+ Assertions.assertEquals(originExprId,
aligned.getOutput().get(0).getExprId());
+ assertSinkOutputsResolvable(aligned);
+ assertSinkOutputExpressionsResolvable(aligned);
+ }
+
+ @Test
+ public void testStaleSinkOutputUsesChildOutput() {
+ ExprId originExprId = new ExprId(11);
+ ExprId staleExprId = new ExprId(2);
+ SlotReference childOutput = slot(originExprId, ImmutableList.of("a"));
+ LogicalOneRowRelation child = new LogicalOneRowRelation(new
RelationId(1), ImmutableList.of(childOutput));
+ // the sink still outputs the exprId from before a rule changed the
child output, the child does
+ // not produce this exprId any more
+ LogicalFileSink<Plan> rewritten =
fileSink(ImmutableList.of(slot(staleExprId, ImmutableList.of("a"))), child);
+ LogicalFileSink<Plan> origin =
fileSink(ImmutableList.of(slot(originExprId, ImmutableList.of("a"))), oneRow());
+
+ Plan normalized =
MaterializedViewUtils.normalizeSinkExpressions(rewritten, origin);
+
+ Assertions.assertNotNull(normalized);
+ Assertions.assertEquals(originExprId,
normalized.getOutput().get(0).getExprId());
+ Assertions.assertSame(child, normalized.child(0));
+ assertSinkOutputsResolvable(normalized);
+ assertSinkOutputExpressionsResolvable(normalized);
+ }
+
+ private static void assertSinkOutputsResolvable(Plan sinkPlan) {
+ // PhysicalFileSink resolves every output exprId in the child, see
+ // PhysicalPlanTranslator.visitPhysicalFileSink and
translateOutputExprs
+ Set<ExprId> childOutputExprIds = sinkPlan.child(0).getOutput().stream()
+ .map(Slot::getExprId).collect(Collectors.toSet());
+ for (Slot sinkOutput : sinkPlan.getOutput()) {
+
Assertions.assertTrue(childOutputExprIds.contains(sinkOutput.getExprId()),
+ "the child must produce the sink output exprId " +
sinkOutput.getExprId());
+ }
+ }
+
+ private static void assertSinkOutputExpressionsResolvable(Plan sinkPlan) {
+ Set<ExprId> childOutputExprIds = sinkPlan.child(0).getOutput().stream()
+ .map(Slot::getExprId).collect(Collectors.toSet());
+ for (NamedExpression output : ((LogicalSink<?>)
sinkPlan).getOutputExprs()) {
+ for (Slot inputSlot : output.getInputSlots()) {
+
Assertions.assertTrue(childOutputExprIds.contains(inputSlot.getExprId()),
+ "the sink output expression references a slot not
produced by the child: " + inputSlot);
+ }
+ }
+ }
+
+ private static SlotReference slot(ExprId exprId, ImmutableList<String>
qualifier) {
+ return new SlotReference(exprId, "k1", IntegerType.INSTANCE, false,
qualifier);
+ }
+
+ private static LogicalFileSink<Plan> fileSink(List<NamedExpression>
outputs, Plan child) {
+ return new
LogicalFileSink<>("file:///tmp/materialized_view_align_test", "csv",
+ ImmutableMap.of(), outputs, child);
+ }
+
+ private static LogicalOneRowRelation oneRow() {
+ return new LogicalOneRowRelation(new RelationId(1),
ImmutableList.of());
+ }
+
+ private static Plan project(NamedExpression... projects) {
+ return new LogicalProject<>(ImmutableList.copyOf(projects), oneRow());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
