924060929 commented on code in PR #69350:
URL: https://github.com/apache/doris/pull/69350#discussion_r4333822539
##########
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:
Confirmed by tracing the analyzer and validator at f0b2128d4dc2; this is
still unresolved. `ScalarSubquery.withTypeCoercion()` stores the comparison
cast in `typeCoercionExpr`, while `collectValueNodes()` only strips a direct
`Cast` operand, and `InSubquery` is not included in `isPredicate()`. Both paths
still reject Boolean-only writes. There is also a related analyzer-generated
shape worth covering: `payload['flag'] IS TRUE` becomes `And(Cast(v AS
BOOLEAN), Not(IsNull(Cast(v AS BOOLEAN))))` in `visitIsTrue()`. The first cast
is skipped, but the one under `IsNull` is still validated through the generic
branch. Please handle these predicate representations consistently, with
positive tests and negatives retaining loss detection inside the operand's
producing expression. This confirmation is based on source tracing; I have not
run these statements locally.
##########
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:
Confirmed at f0b2128d4dc2. `AnalyzeCTE.forceOutputNullable()` wraps both
recursive inputs as `Alias(Nullable(slot))`. A demand for only the payload
field follows `RecursiveUnion -> Project(Alias(Nullable(s))) ->
Project(named_struct(...))`; Alias preserves that demand, but Nullable takes
the generic branch and passes `Demand.ALL` to s. Consequently the scalar
sibling is traced again and a valid struct write is rejected. Please preserve
`valueDemand` through this value-preserving wrapper and add an analyzer-built
recursive-CTE positive case plus a negative case with actual loss in payload.
The hand-built recursive fixture does not contain the Nullable projections.
Source review only; no local execution.
--
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]
