github-actions[bot] commented on code in PR #69350:
URL: https://github.com/apache/doris/pull/69350#discussion_r4326770268
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java:
##########
@@ -356,103 +407,201 @@ private static Set<ExprId> traceCteOutput(
producerPlan = Optional.of(producer.child());
}
if (!producerPlan.isPresent() ||
!context.activeCteIds.add(consumer.getCteId())) {
- return Collections.emptySet();
+ return Collections.emptyMap();
}
try {
- Set<ExprId> externalExprIds =
ImmutableSet.copyOf(traceOutputLineage(
- producerPlan.get(), Collections.singleton(producerExprId),
targetColumn, context));
- context.completedCteTraces.put(traceKey, externalExprIds);
- return externalExprIds;
+ Map<ExprId, Demand> externalDemands =
ImmutableMap.copyOf(traceOutputLineage(
+ producerPlan.get(),
+ Collections.singletonMap(producerExprId, tracedDemand),
+ targetColumn,
+ context));
+ context.completedCteTraces.put(traceKey, new
CteTrace(tracedDemand, externalDemands));
+ return externalDemands;
} finally {
context.activeCteIds.remove(consumer.getCteId());
}
}
/**
- * Validates the casts on the values {@code expression} writes and returns
the ExprIds those
- * values come from: the slots they read, and the correlated slots of
their scalar subqueries,
- * which an outer scope may supply.
+ * Validates the casts on the parts of {@code expression} that {@code
demand} requires and returns the
+ * ExprIds those parts come from, with the part of each: the slots they
read, and the correlated slots
+ * of their scalar subqueries, which an outer scope may supply.
*/
- private static Set<ExprId> validateExpression(
- Expression expression, String targetColumn, TraceContext context) {
- List<Expression> valueNodes = valueNodes(expression);
+ private static Map<ExprId, Demand> validateExpression(
+ Expression expression, Demand demand, String targetColumn,
TraceContext context) {
+ List<Pair<Expression, Demand>> valueNodes = valueNodes(expression,
demand);
validateLossyCast(valueNodes, targetColumn);
- Set<ExprId> inputExprIds = new HashSet<>();
- for (Expression node : valueNodes) {
- if (node instanceof Slot) {
- inputExprIds.add(((Slot) node).getExprId());
- } else if (node instanceof ScalarSubquery) {
- inputExprIds.addAll(traceScalarSubquery((ScalarSubquery) node,
targetColumn, context));
+ Map<ExprId, Demand> inputDemands = new HashMap<>();
+ for (Pair<Expression, Demand> node : valueNodes) {
+ if (node.first instanceof Slot) {
+ inputDemands.merge(((Slot) node.first).getExprId(),
node.second, Demand::merge);
+ } else if (node.first instanceof ScalarSubquery) {
+ mergeDemands(inputDemands,
+ traceScalarSubquery((ScalarSubquery) node.first,
node.second, targetColumn, context));
}
}
- return inputExprIds;
+ return inputDemands;
}
/** Traces a scalar subquery and returns its correlated slots that the
subquery itself does not supply. */
- private static Set<ExprId> traceScalarSubquery(
- ScalarSubquery scalarSubquery, String targetColumn, TraceContext
context) {
- scalarSubquery.getTypeCoercionExpr()
- .ifPresent(coercion -> validateLossyCast(valueNodes(coercion),
targetColumn));
+ private static Map<ExprId, Demand> traceScalarSubquery(
+ ScalarSubquery scalarSubquery, Demand demand, String targetColumn,
TraceContext context) {
Plan queryPlan = scalarSubquery.getQueryPlan();
- context.collectCteProducers(queryPlan);
- Set<ExprId> unresolvedExprIds = new HashSet<>(traceOutputLineage(
- queryPlan,
-
Collections.singleton(queryPlan.getOutput().get(0).getExprId()),
+ // The type coercion casts the output of the query, so it decides
which parts of that output are written.
+ Map<ExprId, Demand> outputDemands = validateExpression(
+
scalarSubquery.getTypeCoercionExpr().orElse(queryPlan.getOutput().get(0)),
+ demand,
targetColumn,
- context));
+ context);
+ context.collectCteProducers(queryPlan);
+ Map<ExprId, Demand> unresolvedDemands = new HashMap<>(
+ traceOutputLineage(queryPlan, outputDemands, targetColumn,
context));
Set<ExprId> correlatedExprIds = new HashSet<>();
for (Slot correlateSlot : scalarSubquery.getCorrelateSlots()) {
correlatedExprIds.add(correlateSlot.getExprId());
}
- unresolvedExprIds.retainAll(correlatedExprIds);
- return unresolvedExprIds;
+ unresolvedDemands.keySet().retainAll(correlatedExprIds);
+ return unresolvedDemands;
}
/**
- * The nodes of {@code expression} whose values can reach its result, in
pre-order: every node
- * except those under the condition of an IF or of a CASE branch, which
only chooses a value, and
- * those under the partition and order keys of a window or the ORDER BY of
an aggregate function
- * such as GROUP_CONCAT, which only arrange the rows whose values are used.
+ * The nodes of {@code expression} whose values reach the parts {@code
demand} requires, in pre-order,
+ * each with the part of its own value that does. A node is left out if
none of its value does: those
+ * under the condition of an IF or of a CASE branch, under the second
argument of NULLIF or under the
+ * index or key of element_at, which only choose a value; those under the
partition and order keys of a
+ * window or the ORDER BY of an aggregate function such as GROUP_CONCAT,
which only arrange the rows whose
+ * values are used; and the arguments of a constructor or the fields of an
element_at container that are
+ * not required. The implicit cast a predicate applies to an operand is
left out as well, since only its
+ * comparison is written; the operand under it is not.
*/
- private static List<Expression> valueNodes(Expression expression) {
- List<Expression> nodes = new ArrayList<>();
- collectValueNodes(expression, nodes);
+ private static List<Pair<Expression, Demand>> valueNodes(Expression
expression, Demand demand) {
+ List<Pair<Expression, Demand>> nodes = new ArrayList<>();
+ collectValueNodes(expression, demand, nodes);
return nodes;
}
- private static void collectValueNodes(Expression expression,
List<Expression> nodes) {
- nodes.add(expression);
- if (expression instanceof If) {
- collectValueNodes(((If) expression).getTrueValue(), nodes);
- collectValueNodes(((If) expression).getFalseValue(), nodes);
- } else if (expression instanceof WhenClause) {
- collectValueNodes(((WhenClause) expression).getResult(), nodes);
+ private static void collectValueNodes(
+ Expression expression, Demand demand, List<Pair<Expression,
Demand>> nodes) {
+ if (demand.isNone()) {
+ return;
+ }
+ // Casts, set operations and constructors convert values field by
field in order, so a demand on
+ // the fields of a value is aligned to the type of each value it
passes to.
+ Demand valueDemand = demand.isAll() ? demand :
demand.alignTo(expression.getDataType());
+ nodes.add(Pair.of(expression, valueDemand));
+ Optional<List<Expression>> branchResults =
ExpressionUtils.getCaseWhenLikeBranchResults(expression);
+ if (branchResults.isPresent()) {
+ for (Expression branchResult : branchResults.get()) {
+ collectValueNodes(branchResult, valueDemand, nodes);
+ }
} else if (expression instanceof WindowExpression) {
- collectValueNodes(((WindowExpression) expression).getFunction(),
nodes);
+ collectValueNodes(((WindowExpression) expression).getFunction(),
valueDemand, nodes);
+ } else if (expression instanceof Alias || expression instanceof Cast) {
+ collectValueNodes(expression.child(0), valueDemand, nodes);
+ } else if (expression instanceof CreateNamedStruct) {
+ // The arguments alternate field names and field values.
+ for (int field = 0; field * 2 + 1 < expression.arity(); field++) {
+ collectValueNodes(expression.child(field * 2 + 1),
valueDemand.part(field), nodes);
+ }
+ } else if (expression instanceof CreateStruct) {
+ for (int field = 0; field < expression.arity(); field++) {
+ collectValueNodes(expression.child(field),
valueDemand.part(field), nodes);
+ }
+ } else if (expression instanceof Array) {
+ for (Expression element : expression.children()) {
+ collectValueNodes(element, valueDemand.part(0), nodes);
+ }
+ } else if (expression instanceof CreateMap) {
+ // The arguments alternate keys and values.
+ for (int i = 0; i < expression.arity(); i++) {
+ collectValueNodes(expression.child(i), valueDemand.part(i %
2), nodes);
+ }
+ } else if (expression instanceof ElementAt) {
+ collectValueNodes(expression.child(0), containerDemand((ElementAt)
expression, valueDemand), nodes);
+ } else if (isPredicate(expression)) {
+ for (Expression operand : expression.children()) {
+ boolean comparisonCast = operand instanceof Cast && !((Cast)
operand).isExplicitType();
Review Comment:
[P2] Apply predicate exemptions to subquery representations
For a VARIANT sink, `SELECT (SELECT payload['s'] FROM variant_source WHERE
id = 1) LIKE 'a%'` still fails the lossy-write check. Nereids represents this
as `Like(ScalarSubquery[typeCoercion=Cast(v AS VARCHAR)], 'a%')`:
`ScalarSubquery.withTypeCoercion` stores the comparison cast internally, so
this direct-`Cast` check misses it and `traceScalarSubquery` rejects it.
`payload['n'] IN (SELECT n FROM ints)` has the same problem on another
representation: `ExpressionAnalyzer.visitInSubquery` produces
`InSubquery(Cast(ElementAt(payload,'n') AS common_numeric_type), ...)`, but
`isPredicate` excludes `InSubquery`, so the fallback validates that
comparison-only cast. Both write only a Boolean, just like the newly accepted
direct LIKE/IN cases. Handle these subquery predicate representations while
retaining validation of coercions inside the producing query, and add
nonconstant subquery cases to the positive/negative tests.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java:
##########
@@ -356,103 +407,201 @@ private static Set<ExprId> traceCteOutput(
producerPlan = Optional.of(producer.child());
}
if (!producerPlan.isPresent() ||
!context.activeCteIds.add(consumer.getCteId())) {
- return Collections.emptySet();
+ return Collections.emptyMap();
}
try {
- Set<ExprId> externalExprIds =
ImmutableSet.copyOf(traceOutputLineage(
- producerPlan.get(), Collections.singleton(producerExprId),
targetColumn, context));
- context.completedCteTraces.put(traceKey, externalExprIds);
- return externalExprIds;
+ Map<ExprId, Demand> externalDemands =
ImmutableMap.copyOf(traceOutputLineage(
+ producerPlan.get(),
+ Collections.singletonMap(producerExprId, tracedDemand),
+ targetColumn,
+ context));
+ context.completedCteTraces.put(traceKey, new
CteTrace(tracedDemand, externalDemands));
+ return externalDemands;
} finally {
context.activeCteIds.remove(consumer.getCteId());
}
}
/**
- * Validates the casts on the values {@code expression} writes and returns
the ExprIds those
- * values come from: the slots they read, and the correlated slots of
their scalar subqueries,
- * which an outer scope may supply.
+ * Validates the casts on the parts of {@code expression} that {@code
demand} requires and returns the
+ * ExprIds those parts come from, with the part of each: the slots they
read, and the correlated slots
+ * of their scalar subqueries, which an outer scope may supply.
*/
- private static Set<ExprId> validateExpression(
- Expression expression, String targetColumn, TraceContext context) {
- List<Expression> valueNodes = valueNodes(expression);
+ private static Map<ExprId, Demand> validateExpression(
+ Expression expression, Demand demand, String targetColumn,
TraceContext context) {
+ List<Pair<Expression, Demand>> valueNodes = valueNodes(expression,
demand);
validateLossyCast(valueNodes, targetColumn);
- Set<ExprId> inputExprIds = new HashSet<>();
- for (Expression node : valueNodes) {
- if (node instanceof Slot) {
- inputExprIds.add(((Slot) node).getExprId());
- } else if (node instanceof ScalarSubquery) {
- inputExprIds.addAll(traceScalarSubquery((ScalarSubquery) node,
targetColumn, context));
+ Map<ExprId, Demand> inputDemands = new HashMap<>();
+ for (Pair<Expression, Demand> node : valueNodes) {
+ if (node.first instanceof Slot) {
+ inputDemands.merge(((Slot) node.first).getExprId(),
node.second, Demand::merge);
+ } else if (node.first instanceof ScalarSubquery) {
+ mergeDemands(inputDemands,
+ traceScalarSubquery((ScalarSubquery) node.first,
node.second, targetColumn, context));
}
}
- return inputExprIds;
+ return inputDemands;
}
/** Traces a scalar subquery and returns its correlated slots that the
subquery itself does not supply. */
- private static Set<ExprId> traceScalarSubquery(
- ScalarSubquery scalarSubquery, String targetColumn, TraceContext
context) {
- scalarSubquery.getTypeCoercionExpr()
- .ifPresent(coercion -> validateLossyCast(valueNodes(coercion),
targetColumn));
+ private static Map<ExprId, Demand> traceScalarSubquery(
+ ScalarSubquery scalarSubquery, Demand demand, String targetColumn,
TraceContext context) {
Plan queryPlan = scalarSubquery.getQueryPlan();
- context.collectCteProducers(queryPlan);
- Set<ExprId> unresolvedExprIds = new HashSet<>(traceOutputLineage(
- queryPlan,
-
Collections.singleton(queryPlan.getOutput().get(0).getExprId()),
+ // The type coercion casts the output of the query, so it decides
which parts of that output are written.
+ Map<ExprId, Demand> outputDemands = validateExpression(
+
scalarSubquery.getTypeCoercionExpr().orElse(queryPlan.getOutput().get(0)),
+ demand,
targetColumn,
- context));
+ context);
+ context.collectCteProducers(queryPlan);
+ Map<ExprId, Demand> unresolvedDemands = new HashMap<>(
+ traceOutputLineage(queryPlan, outputDemands, targetColumn,
context));
Set<ExprId> correlatedExprIds = new HashSet<>();
for (Slot correlateSlot : scalarSubquery.getCorrelateSlots()) {
correlatedExprIds.add(correlateSlot.getExprId());
}
- unresolvedExprIds.retainAll(correlatedExprIds);
- return unresolvedExprIds;
+ unresolvedDemands.keySet().retainAll(correlatedExprIds);
+ return unresolvedDemands;
}
/**
- * The nodes of {@code expression} whose values can reach its result, in
pre-order: every node
- * except those under the condition of an IF or of a CASE branch, which
only chooses a value, and
- * those under the partition and order keys of a window or the ORDER BY of
an aggregate function
- * such as GROUP_CONCAT, which only arrange the rows whose values are used.
+ * The nodes of {@code expression} whose values reach the parts {@code
demand} requires, in pre-order,
+ * each with the part of its own value that does. A node is left out if
none of its value does: those
+ * under the condition of an IF or of a CASE branch, under the second
argument of NULLIF or under the
+ * index or key of element_at, which only choose a value; those under the
partition and order keys of a
+ * window or the ORDER BY of an aggregate function such as GROUP_CONCAT,
which only arrange the rows whose
+ * values are used; and the arguments of a constructor or the fields of an
element_at container that are
+ * not required. The implicit cast a predicate applies to an operand is
left out as well, since only its
+ * comparison is written; the operand under it is not.
*/
- private static List<Expression> valueNodes(Expression expression) {
- List<Expression> nodes = new ArrayList<>();
- collectValueNodes(expression, nodes);
+ private static List<Pair<Expression, Demand>> valueNodes(Expression
expression, Demand demand) {
+ List<Pair<Expression, Demand>> nodes = new ArrayList<>();
+ collectValueNodes(expression, demand, nodes);
return nodes;
}
- private static void collectValueNodes(Expression expression,
List<Expression> nodes) {
- nodes.add(expression);
- if (expression instanceof If) {
- collectValueNodes(((If) expression).getTrueValue(), nodes);
- collectValueNodes(((If) expression).getFalseValue(), nodes);
- } else if (expression instanceof WhenClause) {
- collectValueNodes(((WhenClause) expression).getResult(), nodes);
+ private static void collectValueNodes(
+ Expression expression, Demand demand, List<Pair<Expression,
Demand>> nodes) {
+ if (demand.isNone()) {
+ return;
+ }
+ // Casts, set operations and constructors convert values field by
field in order, so a demand on
+ // the fields of a value is aligned to the type of each value it
passes to.
+ Demand valueDemand = demand.isAll() ? demand :
demand.alignTo(expression.getDataType());
+ nodes.add(Pair.of(expression, valueDemand));
+ Optional<List<Expression>> branchResults =
ExpressionUtils.getCaseWhenLikeBranchResults(expression);
+ if (branchResults.isPresent()) {
+ for (Expression branchResult : branchResults.get()) {
+ collectValueNodes(branchResult, valueDemand, nodes);
+ }
} else if (expression instanceof WindowExpression) {
- collectValueNodes(((WindowExpression) expression).getFunction(),
nodes);
+ collectValueNodes(((WindowExpression) expression).getFunction(),
valueDemand, nodes);
+ } else if (expression instanceof Alias || expression instanceof Cast) {
+ collectValueNodes(expression.child(0), valueDemand, nodes);
Review Comment:
[P2] Preserve field demands through recursive nullability wrappers
`AnalyzeCTE.forceOutputNullable` automatically wraps every anchor and
recursive output in `Alias(Nullable(slot))`. For a target `STRUCT<matched
BOOLEAN, payload VARIANT>`, the reduced analyzed path is `RecursiveUnion ->
Project(Nullable(s)) -> Project(named_struct(...))`. The initial `STRUCT(NONE,
ALL)` demand passes through the alias, but `Nullable` falls into the generic
branch and changes its child's demand to `ALL`. A safe scalar sibling such as
`named_struct('matched', IF(TRUE, v['n'], 1) > 1, 'payload', v)` is therefore
still rejected when forwarded through a recursive CTE, although its
direct/ordinary-CTE form is accepted. For example, with the distributed planner
enabled, use that struct as the anchor's `s` alongside `CAST(0 AS INT)` depth,
then `UNION ALL SELECT s, CAST(depth + 1 AS INT) FROM r WHERE depth < 1`, and
insert `SELECT s FROM r`. `Nullable` preserves the value and DataType, so pass
`valueDemand` through it as for Alias/Cast. Add an analyzed recursive-CTE posit
ive and a payload-loss negative; the hand-built recursive fixture omits these
wrappers.
--
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]
