morningman commented on code in PR #69350:
URL: https://github.com/apache/doris/pull/69350#discussion_r4357866621
##########
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:
Fixed in 9a809a40b5a. A predicate operand now drops the predicate's own
implicit cast in each form the analyzer gives it: under a `Cast` (as before),
as the type coercion a compared scalar subquery records
(`ScalarSubquery#withTypeCoercion`), and as the compare expression of an
`InSubquery`, which now counts as a predicate. `IsNull` counts as a predicate
too, which covers the IS TRUE / IS FALSE expansion `AND(cast, NOT(IS
NULL(cast)))`. Only that cast is skipped: the query of the subquery and the
operand under the cast are still traced.
`VariantWritePlanValidatorTest#testSubqueryAndIsTrueOperandsOfPredicatesAreNotWrittenValues`
analyzes the statements only up to binding, the plan BindSink validates
(SubqueryToApply runs later): `(SELECT v['s'] ...) LIKE 'a%'`, `v['n'] IN
(SELECT k ...)`, `v['flag'] IS TRUE` and `v['flag'] IS NOT FALSE` are accepted;
`(SELECT IF(TRUE, v, 1) ...) LIKE 'a%'`, `IF(TRUE, v, 1) IN (SELECT k ...)` and
`IF(TRUE, v, 1) IS TRUE` are rejected. (A VARIANT sca
lar subquery cannot be compared with `=` / `>`: analysis already rejects that
with "Variant does not support ordering/comparison".)
##########
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:
Fixed in 9a809a40b5a. `Nullable` now passes the demand on to its argument,
as `Alias` and `Cast` do.
`VariantWritePlanValidatorTest#testRecursiveCteNullableOutputsKeepTheFieldsTraced`
analyzes a recursive CTE with AnalyzeCTE, so both terms carry
`Alias(Nullable(...))`: `named_struct('matched', IF(TRUE, v['n'], 1) > 1,
'payload', v)` written to a `STRUCT<matched BOOLEAN, payload VARIANT>` target
is accepted, and the same CTE with `'payload', IF(TRUE, v, 1)` is rejected. The
recursive term computes the struct the same way rather than reading it from r,
because the unit-test analyzer cannot bind columns of r in the recursive term
(the same statement shape works on a cluster).
--
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]
