This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 49c90da0a40 [improvement](fe) Reject only lossy casts that reach a
Variant leaf of the write target (#69350)
49c90da0a40 is described below
commit 49c90da0a40da3b6384569d407098ff7de601a9f
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Oct 25 08:03:36 2026 +0000
[improvement](fe) Reject only lossy casts that reach a Variant leaf of the
write target (#69350)
### What problem does this PR solve?
Issue Number: None
Related PR: #69194
Problem Summary:
**In short.** The Variant write check added in #69194 also rejects
writes that lose nothing: a scalar field of a composite target that is
computed from a Variant, and a Variant that is only compared or only
picks an element. This PR traces only the parts of a value that reach a
Variant leaf of the target column, and skips the arguments that only
compare or select. A lossy cast on a value that is written into a
Variant leaf is still rejected. This is the follow-up promised in two
review threads of #69194.
**Background.** Common-type analysis runs before the sink is bound. When
a Variant meets a number in a UNION, IF or CASE, the Variant side is
implicitly cast to DECIMAL, and an object or array Variant becomes SQL
NULL; the sink then encodes that NULL or scalar back into the Variant
column. `VariantWritePlanValidator` (#69194), which every connector's
INSERT / OVERWRITE / CTAS / row-level DML goes through, follows the
lineage of each target column that contains a Variant and rejects an
implicit Variant-to-non-Variant cast on the way. It already skips the
positions that only choose rows or values: IF and CASE conditions,
window partition and order keys, the ORDER BY of an aggregate function,
and the later inputs of EXCEPT and INTERSECT.
**The problem, and what it cost.** Two shapes of correct writes fail
with "VARIANT write cannot safely convert input column ...":
1. As soon as a target column contains one Variant leaf, the whole
column is traced, so an implicit cast in a field that is written as a
BOOLEAN is treated as a Variant loss.
2. Every argument of every expression is treated as a written value,
including arguments that are only compared or only select an element.
| Write (`v` is a VARIANT) | Target | Before | After |
|---|---|---|---|
| `named_struct('matched', COALESCE(v['n'], 0) > 1, 'payload', v)` |
`STRUCT<matched BOOLEAN, payload VARIANT>` | rejected | accepted |
| the same struct built from subquery columns, from a struct column's
fields, through `struct()`, a cast between structs, an IF or a UNION |
same | rejected | accepted |
| `map(IF(TRUE, v['k'], 1), v)` | `MAP<STRING, VARIANT>` | rejected |
accepted |
| `v['n'] > 0`, `v['n'] IN (1, 2)`, `v['s'] LIKE 'a%'`, `NOT v['flag']`
| `VARIANT` | rejected | accepted |
| `NULLIF(1, IF(TRUE, v, 1))`, `element_at(ARRAY(v), IF(TRUE, v['n'],
1))` | `VARIANT` | rejected | accepted |
| `named_struct('matched', TRUE, 'payload', IF(TRUE, v, 1))` |
`STRUCT<matched BOOLEAN, payload VARIANT>` | rejected | rejected |
| `IF(TRUE, v, 1) > 0`, `NULLIF(IF(TRUE, v, 1), 1)`,
`element_at(ARRAY(IF(TRUE, v, 1)), 1)` | `VARIANT` | rejected | rejected
|
The only workaround was an explicit CAST around the sibling field or the
comparison.
**How this PR fixes it.**
1. The validator traces a demand per ExprId instead of a set of ExprIds.
A demand says which parts of a value reach a Variant leaf: none, all of
it, or the parts of each field of a STRUCT, of the element of an ARRAY,
or of the key and the value of a MAP. The demand of a target column
comes from its type and is aligned position by position to the source
type, as the sink's struct cast is; when the shapes differ, all of the
value is traced, as before.
2. A closed set of structural expressions passes the demand on:
`named_struct` / `struct` (field i to its argument), `array` and `map`
constructors, `element_at` (a struct field chosen by its constant name
or 1-based position, an array element, a map value), casts (part by part
between values of the same shape), the results of IF / CASE / NULLIF /
NVL (`ExpressionUtils.getCaseWhenLikeBranchResults`, the model
`JoinExtractOrFromCaseWhen` already relies on), aliases, the `Nullable`
wrapper that AnalyzeCTE puts on every recursive CTE output, window
functions, set operation inputs by ordinal, recursive CTEs, CTE outputs
and scalar subqueries. Any other expression traces all of every
argument, as before, so an expression the validator does not model can
only reject more. A cast is lossy only if it loses a Variant leaf in a
demanded part.
3. Three kinds of argument are not traced, because they never become the
written value: the second argument of NULLIF, the index or key of
`element_at`, and the implicit cast that a comparison, IN (also over a
subquery), LIKE / REGEXP, IS NULL, AND / OR or NOT applies directly to
an operand, also when a compared scalar subquery keeps that cast as its
type coercion. IS TRUE and IS FALSE are analyzed into AND, NOT and IS
NULL over the operand cast to BOOLEAN, so they are covered as well. Only
that one cast is skipped; the operand under it is still traced, so
`IF(TRUE, v, 1) > 0` is still rejected. Functions whose arguments may be
returned are not listed one by one: `ELT(v['n'], 'a', 'b')`,
`SUBSTRING('abc', v['n'])` and `MAX_BY(1, IF(TRUE, v, 1))` are still
rejected and need an explicit CAST.
4. A CTE output is traced once per demand rather than once: it is traced
again only when a consumer requests a part not traced yet, at most once
per nested field of its type, so the work stays proportional to the plan
(the two existing exponential-plan tests still run in milliseconds). The
recursive CTE fixpoint works the same way per output ordinal.
5. The error message now ends with "use an explicit CAST if the
conversion is intended".
branch-4.1 and branch-4.2 scan the whole write expression for such
casts, including IF and CASE conditions, so they reject more than
master; this PR does not change them.
**Results.**
- `VariantWritePlanValidatorTest`: eight new tests (struct targets in
seven shapes, crossed struct fields by name and by position,
ARRAY<STRUCT> and MAP targets, values of another shape, compared and
selecting arguments, a CTE output traced again for more of it,
predicates over a subquery and IS TRUE analyzed up to binding as
BindSink sees them, a recursive CTE whose outputs carry Nullable). Each
accepted statement was checked to be rejected before this change, and
each rejected one to stay rejected. 19 mutations of the new rules
(dropping each structural rule, swapping map keys and values, an
off-by-one struct position, tracing the skipped arguments, reusing a CTE
trace for any part, dropping the demand on a shape mismatch, and others)
each fail at least one test. Dropping the scalar subquery coercion,
InSubquery, IsNull or Nullable handling each fails one of the last two
tests. The 21 existing tests pass unchanged.
- `test_paimon_write_variant_errors`: a Paimon table with
`STRUCT<matched: BOOLEAN, payload: VARIANT>` and a top-level VARIANT
column. The scalar field computed with `COALESCE(payload['n'], 0) > 1`,
in place and from a subquery, and the comparisons `payload['n'] > 1` and
`payload['kind'] LIKE 'object%'` are written and read back (Spark reads
the same values); a loss in `flagged.payload` or under a comparison
operand is still rejected. Before this change the first of these writes
fails with the error above.
- The other Variant write suites, all six `test_paimon_write_variant*`
and `test_iceberg_variant_read`, pass locally with this change.
### Classes and call graph
```
BindSink / BindExpression
-> VariantWritePlanValidator.validateNoLossyCoercion(targets, plan[,
sourceOrdinals][, cteContext])
Demand.of(target type).alignTo(source type)
-> traceOutputLineage(plan, ExprId -> Demand)
set operation / recursive CTE / generate / CTE consumer:
per-ordinal demands
other plans -> validateExpression(named expression, demand)
-> valueNodes(expression, demand) structural
dispatch, skipped arguments
-> validateLossyCast(nodes)
losesVariantLeaf(source, target, demand)
-> slots and scalar subqueries -> demands on the child
plans
```
---
.../datasource/VariantWritePlanValidator.java | 654 ++++++++++++++++-----
.../datasource/VariantWritePlanValidatorTest.java | 243 ++++++++
.../write/test_paimon_write_variant_errors.out | 4 +
.../write/test_paimon_write_variant_errors.groovy | 48 ++
4 files changed, 789 insertions(+), 160 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java
index 36a02ca14f1..d3d819ae444 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java
@@ -21,18 +21,32 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.common.Pair;
import org.apache.doris.nereids.CTEContext;
import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.CompoundPredicate;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.InPredicate;
+import org.apache.doris.nereids.trees.expressions.InSubquery;
+import org.apache.doris.nereids.trees.expressions.IsNull;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Not;
import org.apache.doris.nereids.trees.expressions.OrderExpression;
import org.apache.doris.nereids.trees.expressions.ScalarSubquery;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
-import org.apache.doris.nereids.trees.expressions.WhenClause;
+import org.apache.doris.nereids.trees.expressions.StringRegexPredicate;
import org.apache.doris.nereids.trees.expressions.WindowExpression;
-import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Array;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.CreateMap;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.CreateNamedStruct;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.CreateStruct;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ElementAt;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nullable;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLikeLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.algebra.RecursiveCte;
import org.apache.doris.nereids.trees.plans.algebra.SetOperation;
@@ -50,15 +64,19 @@ import org.apache.doris.nereids.types.MapType;
import org.apache.doris.nereids.types.StructField;
import org.apache.doris.nereids.types.StructType;
import org.apache.doris.nereids.types.VariantType;
+import org.apache.doris.nereids.util.ExpressionUtils;
-import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
@@ -78,7 +96,16 @@ public final class VariantWritePlanValidator {
* Variant path compared with a scalar there loses nothing that is
written. Likewise the later
* inputs of EXCEPT and INTERSECT only decide which rows of the first
input stay, and the partition
* and order keys of a window and the ORDER BY of an aggregate function
only arrange the rows whose
- * values are used.</p>
+ * values are used. For the same reason, the second argument of NULLIF,
the index or key of
+ * element_at and the cast a predicate applies to an operand, also when a
scalar subquery records it
+ * as its type coercion, are not inspected: they are only compared or
choose the element returned.</p>
+ *
+ * <p>Of a value, only the parts that reach a Variant leaf of the target
are inspected. The BOOLEAN
+ * field of a STRUCT<matched BOOLEAN, payload VARIANT> target is
written as a BOOLEAN, so a
+ * Variant cast to a scalar while computing it loses nothing in a Variant.
Struct, array and map
+ * constructors, field and element access, casts, the nullable wrapper of
recursive CTE outputs and
+ * the results of IF, CASE, NULLIF and NVL pass those parts on to their
arguments; any other
+ * expression inspects all of its arguments.</p>
*/
public static void validateNoLossyCoercion(List<Column> targetColumns,
Plan sourcePlan) {
validateNoLossyCoercion(targetColumns, sourcePlan, Optional.empty());
@@ -125,116 +152,120 @@ public final class VariantWritePlanValidator {
throw new AnalysisException("VARIANT write target and source
columns are not aligned");
}
- List<Integer> variantTargetOrdinals = new ArrayList<>();
+ Map<Integer, Demand> variantTargetDemands = new LinkedHashMap<>();
for (int i = 0; i < targetColumns.size(); i++) {
- DataType targetType =
DataType.fromCatalogType(targetColumns.get(i).getType());
- if (VariantType.containsVariant(targetType)) {
- variantTargetOrdinals.add(i);
+ Demand demand =
Demand.of(DataType.fromCatalogType(targetColumns.get(i).getType()));
+ if (!demand.isNone()) {
+ variantTargetDemands.put(i, demand);
}
}
- if (variantTargetOrdinals.isEmpty()) {
+ if (variantTargetDemands.isEmpty()) {
return;
}
TraceContext traceContext = new TraceContext(sourcePlan, cteContext);
List<Slot> sourceOutputs = sourcePlan.getOutput();
- for (int targetOrdinal : variantTargetOrdinals) {
- int sourceOrdinal = sourceOrdinals.get(targetOrdinal);
+ for (Map.Entry<Integer, Demand> targetDemand :
variantTargetDemands.entrySet()) {
+ int sourceOrdinal = sourceOrdinals.get(targetDemand.getKey());
if (sourceOrdinal < 0 || sourceOrdinal >= sourceOutputs.size()) {
throw new AnalysisException("VARIANT write target and source
columns are not aligned");
}
- Column targetColumn = targetColumns.get(targetOrdinal);
+ Slot sourceOutput = sourceOutputs.get(sourceOrdinal);
+ // The sink casts the source value to the target type, field by
field in order.
traceOutputLineage(
sourcePlan,
-
Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()),
- targetColumn.getName(),
+ Collections.singletonMap(
+ sourceOutput.getExprId(),
targetDemand.getValue().alignTo(sourceOutput.getDataType())),
+ targetColumns.get(targetDemand.getKey()).getName(),
traceContext);
}
}
/**
- * Validates the requested outputs and returns lineage ExprIds supplied by
an outer scope.
- * Scalar subqueries use the returned IDs to continue tracing correlated
columns in the
+ * Validates the required parts of the outputs and returns the parts of
the lineage ExprIds that an
+ * outer scope supplies. Scalar subqueries use them to continue tracing
correlated columns in the
* enclosing plan.
*/
- private static Set<ExprId> traceOutputLineage(
- Plan plan, Set<ExprId> requiredExprIds, String targetColumn,
TraceContext context) {
- if (requiredExprIds.isEmpty()) {
- return Collections.emptySet();
+ private static Map<ExprId, Demand> traceOutputLineage(
+ Plan plan, Map<ExprId, Demand> requiredDemands, String
targetColumn, TraceContext context) {
+ if (requiredDemands.isEmpty()) {
+ return Collections.emptyMap();
}
if (plan instanceof SetOperation) {
- return traceSetOperation(plan, (SetOperation) plan,
requiredExprIds, targetColumn, context);
+ return traceSetOperation(plan, (SetOperation) plan,
requiredDemands, targetColumn, context);
}
if (plan instanceof RecursiveCte) {
- return traceRecursiveCte(plan, (RecursiveCte) plan,
requiredExprIds, targetColumn, context);
+ return traceRecursiveCte(plan, (RecursiveCte) plan,
requiredDemands, targetColumn, context);
}
if (plan instanceof LogicalGenerate) {
- return traceGenerate((LogicalGenerate<?>) plan, requiredExprIds,
targetColumn, context);
+ return traceGenerate((LogicalGenerate<?>) plan, requiredDemands,
targetColumn, context);
}
if (plan instanceof LogicalCTEConsumer) {
- return traceCteConsumer((LogicalCTEConsumer) plan,
requiredExprIds, targetColumn, context);
+ return traceCteConsumer((LogicalCTEConsumer) plan,
requiredDemands, targetColumn, context);
}
- Set<ExprId> unresolvedExprIds = new HashSet<>(requiredExprIds);
- Set<ExprId> inputExprIds = new HashSet<>();
+ Map<ExprId, Demand> unresolvedDemands = new HashMap<>(requiredDemands);
+ Map<ExprId, Demand> inputDemands = new HashMap<>();
for (Expression expression : plan.getExpressions()) {
if (!(expression instanceof NamedExpression)) {
continue;
}
NamedExpression namedExpression = (NamedExpression) expression;
- if (!requiredExprIds.contains(namedExpression.getExprId())) {
+ Demand demand = requiredDemands.get(namedExpression.getExprId());
+ if (demand == null) {
continue;
}
- inputExprIds.addAll(validateExpression(namedExpression,
targetColumn, context));
- unresolvedExprIds.remove(namedExpression.getExprId());
+ mergeDemands(inputDemands, validateExpression(namedExpression,
demand, targetColumn, context));
+ unresolvedDemands.remove(namedExpression.getExprId());
}
- inputExprIds.addAll(unresolvedExprIds);
+ mergeDemands(inputDemands, unresolvedDemands);
- Set<ExprId> externalExprIds = new HashSet<>(inputExprIds);
+ Map<ExprId, Demand> externalDemands = new HashMap<>(inputDemands);
for (Plan child : plan.children()) {
Set<ExprId> childOutputExprIds = child.getOutputExprIdSet();
- Set<ExprId> childRequiredExprIds = new HashSet<>(inputExprIds);
- childRequiredExprIds.retainAll(childOutputExprIds);
- externalExprIds.removeAll(childOutputExprIds);
- externalExprIds.addAll(traceOutputLineage(child,
childRequiredExprIds, targetColumn, context));
+ Map<ExprId, Demand> childRequiredDemands = new
HashMap<>(inputDemands);
+ childRequiredDemands.keySet().retainAll(childOutputExprIds);
+ externalDemands.keySet().removeAll(childOutputExprIds);
+ mergeDemands(externalDemands, traceOutputLineage(child,
childRequiredDemands, targetColumn, context));
}
- return externalExprIds;
+ return externalDemands;
}
- private static Set<ExprId> traceRecursiveCte(
- Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds,
+ private static Map<ExprId, Demand> traceRecursiveCte(
+ Plan plan, RecursiveCte recursiveCte, Map<ExprId, Demand>
requiredDemands,
String targetColumn, TraceContext context) {
// The recursive term reads the previous iteration through the work
table, so a column it writes
- // can carry the value of another column. Trace those columns as well
until no new one is reached.
+ // can carry the value of another column. Trace those columns as well
until no new part of one is
+ // reached.
Map<ExprId, Integer> workTableOrdinals = workTableOrdinals(plan);
- Set<ExprId> externalExprIds = new HashSet<>();
- Set<Integer> tracedOrdinals = new HashSet<>();
- Set<Integer> pendingOrdinals = requiredOutputOrdinals(plan,
requiredExprIds);
- while (!pendingOrdinals.isEmpty()) {
- tracedOrdinals.addAll(pendingOrdinals);
- Set<ExprId> childExternalExprIds = new HashSet<>();
+ Map<ExprId, Demand> externalDemands = new HashMap<>();
+ Map<Integer, Demand> tracedDemands = new HashMap<>();
+ Map<Integer, Demand> pendingDemands = requiredOutputDemands(plan,
requiredDemands);
+ while (!pendingDemands.isEmpty()) {
+ tracedDemands.putAll(pendingDemands);
+ Map<ExprId, Demand> childExternalDemands = new HashMap<>();
for (int childIndex = 0; childIndex < plan.arity(); childIndex++) {
- List<SlotReference> childOutputs =
recursiveCte.getRegularChildOutput(childIndex);
- Set<ExprId> childRequiredExprIds = new HashSet<>();
- for (int ordinal : pendingOrdinals) {
- if (ordinal < childOutputs.size()) {
-
childRequiredExprIds.add(childOutputs.get(ordinal).getExprId());
- }
- }
- childExternalExprIds.addAll(traceOutputLineage(
- plan.child(childIndex), childRequiredExprIds,
targetColumn, context));
+ mergeDemands(childExternalDemands, traceOutputLineage(
+ plan.child(childIndex),
+
childOutputDemands(recursiveCte.getRegularChildOutput(childIndex),
pendingDemands),
+ targetColumn,
+ context));
}
- pendingOrdinals = new HashSet<>();
- for (ExprId exprId : childExternalExprIds) {
- Integer ordinal = workTableOrdinals.get(exprId);
+ pendingDemands = new HashMap<>();
+ for (Map.Entry<ExprId, Demand> childExternalDemand :
childExternalDemands.entrySet()) {
+ Integer ordinal =
workTableOrdinals.get(childExternalDemand.getKey());
if (ordinal == null) {
- externalExprIds.add(exprId);
- } else if (!tracedOrdinals.contains(ordinal)) {
- pendingOrdinals.add(ordinal);
+ externalDemands.merge(childExternalDemand.getKey(),
childExternalDemand.getValue(), Demand::merge);
+ continue;
+ }
+ Demand tracedDemand = tracedDemands.getOrDefault(ordinal,
Demand.NONE);
+ Demand reachedDemand =
tracedDemand.merge(childExternalDemand.getValue());
+ if (!reachedDemand.equals(tracedDemand)) {
+ pendingDemands.merge(ordinal, reachedDemand,
Demand::merge);
}
}
}
- return externalExprIds;
+ return externalDemands;
}
/** Maps each column of the work table the recursive term reads to the CTE
output ordinal it holds. */
@@ -253,40 +284,44 @@ public final class VariantWritePlanValidator {
return ordinals;
}
- private static Set<ExprId> traceGenerate(
- LogicalGenerate<?> generate, Set<ExprId> requiredExprIds,
+ private static Map<ExprId, Demand> traceGenerate(
+ LogicalGenerate<?> generate, Map<ExprId, Demand> requiredDemands,
String targetColumn, TraceContext context) {
- Set<ExprId> childRequiredExprIds = new HashSet<>(requiredExprIds);
+ Map<ExprId, Demand> childRequiredDemands = new
HashMap<>(requiredDemands);
List<Slot> generatorOutputs = generate.getGeneratorOutput();
List<? extends Expression> generators = generate.getGenerators();
for (int ordinal = 0; ordinal < generatorOutputs.size(); ordinal++) {
Slot generatorOutput = generatorOutputs.get(ordinal);
- if (!requiredExprIds.contains(generatorOutput.getExprId())) {
+ Demand demand = requiredDemands.get(generatorOutput.getExprId());
+ if (demand == null) {
continue;
}
- childRequiredExprIds.remove(generatorOutput.getExprId());
-
childRequiredExprIds.addAll(validateExpression(generators.get(ordinal),
targetColumn, context));
+ childRequiredDemands.remove(generatorOutput.getExprId());
+ mergeDemands(childRequiredDemands,
+ validateExpression(generators.get(ordinal), demand,
targetColumn, context));
}
Set<ExprId> childOutputExprIds = generate.child().getOutputExprIdSet();
- Set<ExprId> externalExprIds = new HashSet<>(childRequiredExprIds);
- externalExprIds.removeAll(childOutputExprIds);
- childRequiredExprIds.retainAll(childOutputExprIds);
- externalExprIds.addAll(traceOutputLineage(
- generate.child(), childRequiredExprIds, targetColumn,
context));
- return externalExprIds;
+ Map<ExprId, Demand> externalDemands = new
HashMap<>(childRequiredDemands);
+ externalDemands.keySet().removeAll(childOutputExprIds);
+ childRequiredDemands.keySet().retainAll(childOutputExprIds);
+ mergeDemands(externalDemands, traceOutputLineage(
+ generate.child(), childRequiredDemands, targetColumn,
context));
+ return externalDemands;
}
- private static Set<ExprId> traceSetOperation(
- Plan plan, SetOperation setOperation, Set<ExprId> requiredExprIds,
+ private static Map<ExprId, Demand> traceSetOperation(
+ Plan plan, SetOperation setOperation, Map<ExprId, Demand>
requiredDemands,
String targetColumn, TraceContext context) {
- Set<ExprId> externalExprIds = new HashSet<>();
- Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan,
requiredExprIds);
+ Map<ExprId, Demand> externalDemands = new HashMap<>();
+ Map<Integer, Demand> ordinalDemands = requiredOutputDemands(plan,
requiredDemands);
if (plan instanceof LogicalUnion) {
for (List<NamedExpression> constantRow : ((LogicalUnion)
plan).getConstantExprsList()) {
- for (int ordinal : requiredOrdinals) {
+ for (Map.Entry<Integer, Demand> ordinalDemand :
ordinalDemands.entrySet()) {
+ int ordinal = ordinalDemand.getKey();
if (ordinal < constantRow.size()) {
-
externalExprIds.addAll(validateExpression(constantRow.get(ordinal),
targetColumn, context));
+ mergeDemands(externalDemands, validateExpression(
+ constantRow.get(ordinal),
ordinalDemand.getValue(), targetColumn, context));
}
}
}
@@ -298,52 +333,72 @@ public final class VariantWritePlanValidator {
? 1
: setOperation.getArity();
for (int childIndex = 0; childIndex < valueInputs; childIndex++) {
- List<SlotReference> childOutputs =
setOperation.getRegularChildOutput(childIndex);
- Set<ExprId> childRequiredExprIds = new HashSet<>();
- for (int ordinal : requiredOrdinals) {
- if (ordinal < childOutputs.size()) {
-
childRequiredExprIds.add(childOutputs.get(ordinal).getExprId());
- }
- }
- externalExprIds.addAll(traceOutputLineage(
- plan.child(childIndex), childRequiredExprIds,
targetColumn, context));
+ mergeDemands(externalDemands, traceOutputLineage(
+ plan.child(childIndex),
+
childOutputDemands(setOperation.getRegularChildOutput(childIndex),
ordinalDemands),
+ targetColumn,
+ context));
}
- return externalExprIds;
+ return externalDemands;
}
- private static Set<Integer> requiredOutputOrdinals(Plan plan, Set<ExprId>
requiredExprIds) {
- Set<Integer> requiredOrdinals = new HashSet<>();
+ /** The required parts of the outputs of {@code plan}, by output ordinal.
*/
+ private static Map<Integer, Demand> requiredOutputDemands(Plan plan,
Map<ExprId, Demand> requiredDemands) {
+ Map<Integer, Demand> ordinalDemands = new HashMap<>();
List<Slot> outputs = plan.getOutput();
for (int i = 0; i < outputs.size(); i++) {
- if (requiredExprIds.contains(outputs.get(i).getExprId())) {
- requiredOrdinals.add(i);
+ Demand demand = requiredDemands.get(outputs.get(i).getExprId());
+ if (demand != null) {
+ ordinalDemands.put(i, demand);
}
}
- return requiredOrdinals;
+ return ordinalDemands;
}
- private static Set<ExprId> traceCteConsumer(
- LogicalCTEConsumer consumer, Set<ExprId> requiredExprIds,
+ /** The required parts of the outputs of one input of a set operation,
which it casts to its own outputs. */
+ private static Map<ExprId, Demand> childOutputDemands(
+ List<SlotReference> childOutputs, Map<Integer, Demand>
ordinalDemands) {
+ Map<ExprId, Demand> childDemands = new HashMap<>();
+ for (Map.Entry<Integer, Demand> ordinalDemand :
ordinalDemands.entrySet()) {
+ int ordinal = ordinalDemand.getKey();
+ if (ordinal < childOutputs.size()) {
+ SlotReference childOutput = childOutputs.get(ordinal);
+ childDemands.merge(childOutput.getExprId(),
+
ordinalDemand.getValue().alignTo(childOutput.getDataType()), Demand::merge);
+ }
+ }
+ return childDemands;
+ }
+
+ private static Map<ExprId, Demand> traceCteConsumer(
+ LogicalCTEConsumer consumer, Map<ExprId, Demand> requiredDemands,
String targetColumn, TraceContext context) {
// A producer shared by several consumers is traced once per output,
whichever sets of its outputs
- // the consumers request, so the work stays proportional to the plan
rather than to the number of
- // paths through it or of the sets of outputs requested along them.
- Set<ExprId> externalExprIds = new HashSet<>();
+ // the consumers request, and again only for a part of the output it
has not been traced for, which
+ // happens at most once per nested field of the output type. So the
work stays proportional to the
+ // plan rather than to the number of paths through it or of the sets
of outputs requested along them.
+ Map<ExprId, Demand> externalDemands = new HashMap<>();
for (Map.Entry<Slot, Slot> mapping :
consumer.getConsumerToProducerOutputMap().entrySet()) {
- if (requiredExprIds.contains(mapping.getKey().getExprId())) {
- externalExprIds.addAll(traceCteOutput(
- consumer, mapping.getValue().getExprId(),
targetColumn, context));
+ Demand demand = requiredDemands.get(mapping.getKey().getExprId());
+ if (demand != null) {
+ mergeDemands(externalDemands, traceCteOutput(
+ consumer, mapping.getValue().getExprId(), demand,
targetColumn, context));
}
}
- return externalExprIds;
+ return externalDemands;
}
- private static Set<ExprId> traceCteOutput(
- LogicalCTEConsumer consumer, ExprId producerExprId, String
targetColumn, TraceContext context) {
+ private static Map<ExprId, Demand> traceCteOutput(
+ LogicalCTEConsumer consumer, ExprId producerExprId, Demand demand,
+ String targetColumn, TraceContext context) {
Pair<CTEId, ExprId> traceKey = Pair.of(consumer.getCteId(),
producerExprId);
- Set<ExprId> completedTrace = context.completedCteTraces.get(traceKey);
+ CteTrace completedTrace = context.completedCteTraces.get(traceKey);
+ Demand tracedDemand = demand;
if (completedTrace != null) {
- return completedTrace;
+ tracedDemand = completedTrace.demand.merge(demand);
+ if (tracedDemand.equals(completedTrace.demand)) {
+ return completedTrace.externalDemands;
+ }
}
LogicalCTEProducer<?> producer =
context.cteProducers.get(consumer.getCteId());
Optional<Plan> producerPlan;
@@ -356,101 +411,223 @@ public final class VariantWritePlanValidator {
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, or the query of a scalar
subquery operand, 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
|| expression instanceof Nullable) {
+ 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()) {
+ collectValueNodes(withoutComparisonCast(operand), Demand.ALL,
nodes);
+ }
} else {
for (Expression child : expression.children()) {
if (!(child instanceof OrderExpression)) {
- collectValueNodes(child, nodes);
+ collectValueNodes(child, Demand.ALL, nodes);
}
}
}
}
- private static void validateLossyCast(List<Expression> valueNodes, String
targetColumn) {
- for (Expression node : valueNodes) {
- if (!(node instanceof Cast)) {
+ /** The demand on the container {@code elementAt} reads that puts {@code
elementDemand} on the element read. */
+ private static Demand containerDemand(ElementAt elementAt, Demand
elementDemand) {
+ DataType containerType = elementAt.child(0).getDataType();
+ if (containerType instanceof ArrayType) {
+ return Demand.composite(Demand.Kind.ARRAY,
ImmutableList.of(elementDemand));
+ }
+ if (containerType instanceof MapType) {
+ return Demand.composite(Demand.Kind.MAP,
ImmutableList.of(Demand.NONE, elementDemand));
+ }
+ if (containerType instanceof StructType) {
+ // Analysis only accepts a constant 1-based position or field name
to choose a struct field.
+ List<StructField> fields = ((StructType)
containerType).getFields();
+ Expression selector = elementAt.child(1);
+ int fieldIndex = selector instanceof IntegerLikeLiteral
+ ? ((IntegerLikeLiteral) selector).getIntValue() - 1
+ : fields.indexOf(((StructType) containerType).getField(
+ ((StringLikeLiteral) selector).getStringValue()));
+ List<Demand> fieldDemands = new
ArrayList<>(Collections.nCopies(fields.size(), Demand.NONE));
+ fieldDemands.set(fieldIndex, elementDemand);
+ return Demand.composite(Demand.Kind.STRUCT, fieldDemands);
+ }
+ return Demand.ALL;
+ }
+
+ /**
+ * Whether {@code expression} is a predicate, whose result only tells how
its operands compare or whether one
+ * is NULL. IS TRUE and IS FALSE are analyzed into AND, NOT and IS NULL
over the operand cast to BOOLEAN.
+ */
+ private static boolean isPredicate(Expression expression) {
+ return expression instanceof ComparisonPredicate
+ || expression instanceof InPredicate
+ || expression instanceof InSubquery
+ || expression instanceof StringRegexPredicate
+ || expression instanceof IsNull
+ || expression instanceof CompoundPredicate
+ || expression instanceof Not;
+ }
+
+ /**
+ * {@code operand} of a predicate without the implicit cast the predicate
applies to it. A scalar subquery
+ * records that cast as the type coercion of its output rather than under
a Cast, so the subquery is traced
+ * without it; the query that produces the value is traced either way.
+ */
+ private static Expression withoutComparisonCast(Expression operand) {
+ if (operand instanceof Cast) {
+ return ((Cast) operand).isExplicitType() ? operand :
operand.child(0);
+ }
+ if (operand instanceof ScalarSubquery && ((ScalarSubquery)
operand).getTypeCoercionExpr().isPresent()) {
+ ScalarSubquery subquery = (ScalarSubquery) operand;
+ // ScalarSubquery.withTypeCoercion casts the output as it was
coerced so far.
+ Expression coercedBefore =
subquery.getTypeCoercionExpr().get().child(0);
+ return new ScalarSubquery(subquery.getQueryPlan(),
subquery.getCorrelateSlots(),
+ coercedBefore instanceof Cast ? Optional.of(coercedBefore)
: Optional.empty(),
+ subquery.limitOneIsEliminated());
+ }
+ return operand;
+ }
+
+ private static void validateLossyCast(List<Pair<Expression, Demand>>
valueNodes, String targetColumn) {
+ for (Pair<Expression, Demand> node : valueNodes) {
+ if (!(node.first instanceof Cast)) {
continue;
}
- Cast cast = (Cast) node;
- if (!cast.isExplicitType() &&
losesVariantLeaf(cast.child().getDataType(), cast.getDataType())) {
+ Cast cast = (Cast) node.first;
+ if (!cast.isExplicitType()
+ && losesVariantLeaf(cast.child().getDataType(),
cast.getDataType(), node.second)) {
throw new AnalysisException(
"VARIANT write cannot safely convert input column '" +
targetColumn
+ "': source expression implicitly casts
VARIANT to "
- + cast.getDataType().toSql() + " before sink
analysis");
+ + cast.getDataType().toSql() + " before sink
analysis;"
+ + " use an explicit CAST if the conversion is
intended");
+ }
+ }
+ }
+
+ /** Whether a cast loses a Variant leaf in the parts of its value that
{@code demand} requires. */
+ private static boolean losesVariantLeaf(DataType sourceType, DataType
targetType, Demand demand) {
+ if (demand.isNone()) {
+ return false;
+ }
+ // The demand is aligned to the cast type. A cast from a value of
another shape does not keep the
+ // parts apart, so it is judged on the whole value.
+ if (demand.isAll() || !demand.hasShapeOf(sourceType)) {
+ return losesVariantLeaf(sourceType, targetType);
+ }
+ List<DataType> sourcePartTypes = Demand.partTypes(sourceType);
+ List<DataType> targetPartTypes = Demand.partTypes(targetType);
+ for (int i = 0; i < sourcePartTypes.size(); i++) {
+ if (losesVariantLeaf(sourcePartTypes.get(i),
targetPartTypes.get(i), demand.part(i))) {
+ return true;
}
}
+ return false;
}
private static boolean losesVariantLeaf(DataType sourceType, DataType
targetType) {
@@ -496,11 +673,168 @@ public final class VariantWritePlanValidator {
return false;
}
+ private static void mergeDemands(Map<ExprId, Demand> demands, Map<ExprId,
Demand> addedDemands) {
+ addedDemands.forEach((exprId, demand) -> demands.merge(exprId, demand,
Demand::merge));
+ }
+
+ /**
+ * The parts of a value that reach a Variant leaf of the target column:
none, all of it, or, of a STRUCT,
+ * ARRAY or MAP value, the parts of each field, of the elements, or of the
keys and of the values.
+ */
+ private static final class Demand {
+ private static final Demand NONE = new Demand(Kind.NONE,
ImmutableList.of());
+ private static final Demand ALL = new Demand(Kind.ALL,
ImmutableList.of());
+
+ private enum Kind {
+ NONE, ALL, STRUCT, ARRAY, MAP
+ }
+
+ private final Kind kind;
+ // The demands on the fields of a STRUCT, on the element of an ARRAY,
or on the key and the value of a MAP.
+ private final List<Demand> parts;
+
+ private Demand(Kind kind, List<Demand> parts) {
+ this.kind = kind;
+ this.parts = parts;
+ }
+
+ /** The Variant leaves of a value of {@code type}. */
+ private static Demand of(DataType type) {
+ if (type instanceof VariantType) {
+ return ALL;
+ }
+ List<Demand> partDemands = new ArrayList<>();
+ for (DataType partType : partTypes(type)) {
+ partDemands.add(of(partType));
+ }
+ return composite(kindOf(type), partDemands);
+ }
+
+ /** The demand of {@code kind} on these parts, or none if none of them
is required. */
+ private static Demand composite(Kind kind, List<Demand> partDemands) {
+ return partDemands.stream().allMatch(Demand::isNone)
+ ? NONE
+ : new Demand(kind, ImmutableList.copyOf(partDemands));
+ }
+
+ /** The kind of the parts of a value of {@code type}, or NONE if the
value has no parts. */
+ private static Kind kindOf(DataType type) {
+ if (type instanceof StructType) {
+ return Kind.STRUCT;
+ }
+ if (type instanceof ArrayType) {
+ return Kind.ARRAY;
+ }
+ return type instanceof MapType ? Kind.MAP : Kind.NONE;
+ }
+
+ /** The types of the fields of a STRUCT, of the element of an ARRAY,
or of the key and the value of a MAP. */
+ private static List<DataType> partTypes(DataType type) {
+ if (type instanceof StructType) {
+ List<DataType> fieldTypes = new ArrayList<>();
+ for (StructField field : ((StructType) type).getFields()) {
+ fieldTypes.add(field.getDataType());
+ }
+ return fieldTypes;
+ }
+ if (type instanceof ArrayType) {
+ return ImmutableList.of(((ArrayType) type).getItemType());
+ }
+ if (type instanceof MapType) {
+ return ImmutableList.of(((MapType) type).getKeyType(),
((MapType) type).getValueType());
+ }
+ return ImmutableList.of();
+ }
+
+ private boolean isNone() {
+ return kind == Kind.NONE;
+ }
+
+ private boolean isAll() {
+ return kind == Kind.ALL;
+ }
+
+ /** Whether a value of {@code type} has the parts this demand is on. */
+ private boolean hasShapeOf(DataType type) {
+ return kindOf(type) == kind && partTypes(type).size() ==
parts.size();
+ }
+
+ /** The demand on the field, element, key or value at {@code index} of
a STRUCT, ARRAY or MAP value. */
+ private Demand part(int index) {
+ return isAll() ? ALL : parts.get(index);
+ }
+
+ /**
+ * This demand on a value of {@code type} that is converted to the
value it is on, part by part in
+ * order. A value of another shape does not keep the parts apart, so
all of it is required.
+ */
+ private Demand alignTo(DataType type) {
+ if (isNone() || isAll()) {
+ return this;
+ }
+ if (!hasShapeOf(type)) {
+ return ALL;
+ }
+ List<DataType> partTypes = partTypes(type);
+ List<Demand> alignedParts = new ArrayList<>(parts.size());
+ for (int i = 0; i < parts.size(); i++) {
+ alignedParts.add(parts.get(i).alignTo(partTypes.get(i)));
+ }
+ return new Demand(kind, ImmutableList.copyOf(alignedParts));
+ }
+
+ /** The parts that this demand or {@code other} requires. */
+ private Demand merge(Demand other) {
+ if (isNone()) {
+ return other;
+ }
+ if (other.isNone()) {
+ return this;
+ }
+ if (isAll() || kind != other.kind || parts.size() !=
other.parts.size()) {
+ return ALL;
+ }
+ List<Demand> mergedParts = new ArrayList<>(parts.size());
+ for (int i = 0; i < parts.size(); i++) {
+ mergedParts.add(parts.get(i).merge(other.parts.get(i)));
+ }
+ return new Demand(kind, ImmutableList.copyOf(mergedParts));
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof Demand)) {
+ return false;
+ }
+ Demand that = (Demand) other;
+ return kind == that.kind && parts.equals(that.parts);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(kind, parts);
+ }
+ }
+
+ /** The part of a CTE output traced so far, with the parts of the ExprIds
that trace returned. */
+ private static final class CteTrace {
+ private final Demand demand;
+ private final Map<ExprId, Demand> externalDemands;
+
+ private CteTrace(Demand demand, Map<ExprId, Demand> externalDemands) {
+ this.demand = demand;
+ this.externalDemands = externalDemands;
+ }
+ }
+
private static final class TraceContext {
private final Map<CTEId, LogicalCTEProducer<?>> cteProducers = new
HashMap<>();
private final Set<CTEId> activeCteIds = new HashSet<>();
- // Completed traces of producer outputs, keyed by CTE and output, with
the ExprIds they return.
- private final Map<Pair<CTEId, ExprId>, Set<ExprId>> completedCteTraces
= new HashMap<>();
+ // Completed traces of producer outputs, keyed by CTE and output.
+ private final Map<Pair<CTEId, ExprId>, CteTrace> completedCteTraces =
new HashMap<>();
private final Optional<CTEContext> analyzerCteContext;
private TraceContext(Plan sourcePlan, Optional<CTEContext>
analyzerCteContext) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/VariantWritePlanValidatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/VariantWritePlanValidatorTest.java
index 863c5228258..666a4aa695c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/VariantWritePlanValidatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/VariantWritePlanValidatorTest.java
@@ -19,7 +19,13 @@ package org.apache.doris.datasource;
import org.apache.doris.catalog.Column;
import org.apache.doris.nereids.CTEContext;
+import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.jobs.executor.Analyzer;
+import org.apache.doris.nereids.rules.analysis.AnalyzeCTE;
+import org.apache.doris.nereids.rules.analysis.BindExpression;
+import org.apache.doris.nereids.rules.analysis.BindRelation;
+import org.apache.doris.nereids.rules.analysis.CheckPolicy;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.CaseWhen;
@@ -48,8 +54,14 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
import org.apache.doris.nereids.trees.plans.logical.LogicalWorkTableReference;
import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.BooleanType;
import org.apache.doris.nereids.types.ConnectorComputeVariantType;
+import org.apache.doris.nereids.types.DataType;
import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.MapType;
+import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.StructField;
+import org.apache.doris.nereids.types.StructType;
import org.apache.doris.nereids.util.MemoTestUtils;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.qe.ConnectContext;
@@ -63,6 +75,7 @@ import java.util.List;
import java.util.function.Function;
public class VariantWritePlanValidatorTest {
+ private static final String VARIANT_SOURCE = "(SELECT
parse_to_variant('{\"n\":2}') AS v) source";
@Test
public void testUnionIfAndCaseImplicitVariantLossAreRejected() {
@@ -87,6 +100,8 @@ public class VariantWritePlanValidatorTest {
"VARIANT write cannot safely convert input column
'payload'"), exception.getMessage());
Assertions.assertTrue(exception.getMessage().contains("implicitly
casts VARIANT to INT"),
exception.getMessage());
+ Assertions.assertTrue(exception.getMessage().contains("use an
explicit CAST if the conversion is intended"),
+ exception.getMessage());
}
}
@@ -518,6 +533,218 @@ public class VariantWritePlanValidatorTest {
exception.getMessage());
}
+ @Test
+ public void testScalarFieldOfStructTargetIsNotTraced() {
+ // Only the parts of a value that reach a Variant leaf of the target
are traced. The matched field of a
+ // STRUCT<matched BOOLEAN, payload VARIANT> target is written as a
BOOLEAN, so a Variant cast to a number
+ // while computing it loses nothing written, however the struct gets
its fields: built in place by
+ // named_struct or struct, from the columns of a subquery, from the
fields of a struct column, through a
+ // cast between structs, an IF or a UNION. MUTATION: tracing all of a
struct constructor, of the struct a
+ // field is read from, of the input of a cast, of an IF branch or of a
UNION input -> rejected.
+ List<String> shapes = ImmutableList.of(
+ "SELECT 1, named_struct('matched', %s, 'payload', %s) FROM " +
VARIANT_SOURCE,
+ "SELECT 1, struct(%s, %s) FROM " + VARIANT_SOURCE,
+ "SELECT 1, named_struct('matched', m, 'payload', p) FROM
(SELECT %s AS m, %s AS p FROM "
+ + VARIANT_SOURCE + ") projected",
+ "SELECT 1, named_struct('matched', s.matched, 'payload',
s.payload) FROM (SELECT named_struct("
+ + "'matched', %s, 'payload', %s) AS s FROM " +
VARIANT_SOURCE + ") nested",
+ "SELECT 1, CAST(named_struct('m', %s, 'p', %s) AS
STRUCT<matched: BOOLEAN, payload: VARIANT>) FROM "
+ + VARIANT_SOURCE,
+ "SELECT 1, IF(v['n'] > 1, named_struct('matched', %s,
'payload', %s), NULL) FROM " + VARIANT_SOURCE,
+ "SELECT 1, named_struct('matched', %s, 'payload', %s) FROM " +
VARIANT_SOURCE
+ + " UNION ALL SELECT 2, named_struct('matched', FALSE,
'payload', v) FROM " + VARIANT_SOURCE);
+ assertAccepted(flaggedTarget(), formatEach(shapes, "IF(TRUE, v['n'],
1) > 1", "v"));
+
+ // The same loss in the payload field reaches the Variant leaf and is
still rejected.
+ assertRejected(flaggedTarget(), "flagged", formatEach(shapes, "TRUE",
"IF(TRUE, v, 1)"));
+ }
+
+ @Test
+ public void testStructFieldIsTracedByTheTargetFieldItIsWrittenTo() {
+ // A field read from a struct column is traced by the target field it
is written to, not by its place in the
+ // struct it is read from: written to matched, the lossy field is not
traced; written to payload, it is
+ // rejected, whether the field is chosen by name or by position.
MUTATION: tracing all of the struct a field
+ // is read from -> the first group is rejected.
+ String nested = "(SELECT named_struct('lossy', IF(TRUE, v, 1), 'kept',
v) AS s FROM " + VARIANT_SOURCE
+ + ") nested";
+ assertAccepted(flaggedTarget(), ImmutableList.of(
+ "SELECT 1, named_struct('matched', s.lossy > 1, 'payload',
s.kept) FROM " + nested,
+ "SELECT 1, named_struct('matched', struct_element(s, 1) > 1,
'payload', struct_element(s, 2)) FROM "
+ + nested));
+ assertRejected(flaggedTarget(), "flagged", ImmutableList.of(
+ "SELECT 1, named_struct('matched', s.kept IS NULL, 'payload',
s.lossy) FROM " + nested,
+ "SELECT 1, named_struct('matched', struct_element(s, 2) IS
NULL, 'payload', struct_element(s, 1)) "
+ + "FROM " + nested));
+ }
+
+ @Test
+ public void testArrayElementsAndMapValuesFollowTheTargetType() {
+ // An ARRAY<STRUCT<matched BOOLEAN, payload VARIANT>> target traces
each element as the struct target does,
+ // and a MAP<STRING, VARIANT> target traces the values of the map but
not its keys. MUTATION: tracing all of
+ // an array or map constructor -> the accepted statements are rejected.
+ List<Column> arrayTarget = targetWithId("flagged_list",
ArrayType.of(flaggedType()));
+ assertAccepted(arrayTarget, ImmutableList.of(
+ "SELECT 1, ARRAY(named_struct('matched', IF(TRUE, v['n'], 1) >
1, 'payload', v)) FROM "
+ + VARIANT_SOURCE));
+ assertRejected(arrayTarget, "flagged_list", ImmutableList.of(
+ "SELECT 1, ARRAY(named_struct('matched', TRUE, 'payload',
IF(TRUE, v, 1))) FROM " + VARIANT_SOURCE));
+
+ List<Column> mapTarget = targetWithId(
+ "payload_map", MapType.of(StringType.INSTANCE,
ConnectorComputeVariantType.INSTANCE));
+ assertAccepted(mapTarget, ImmutableList.of("SELECT 1, map(IF(TRUE,
v['k'], 1), v) FROM " + VARIANT_SOURCE));
+ assertRejected(mapTarget, "payload_map", ImmutableList.of(
+ "SELECT 1, map('k', IF(TRUE, v, 1)) FROM " + VARIANT_SOURCE));
+ }
+
+ @Test
+ public void testValueOfAnotherShapeIsTracedWhole() {
+ // The parts of a struct target do not map onto a value of another
shape, so such a value is traced whole:
+ // a Variant written to the struct target, and an implicit cast from a
Variant to the struct. MUTATION:
+ // dropping the part of the demand that does not map onto the value ->
accepted.
+ assertRejected(flaggedTarget(), "flagged", ImmutableList.of(
+ "SELECT 1, CAST(IF(TRUE, v, 1) AS VARIANT) FROM " +
VARIANT_SOURCE));
+
+ LogicalProject<?> variantToStruct = projectExpression(variant -> new
Cast(variant, flaggedType()));
+ List<Column> structTarget = ImmutableList.of(new Column("payload",
flaggedType().toCatalogDataType()));
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () ->
VariantWritePlanValidator.validateNoLossyCoercion(structTarget,
variantToStruct));
+ Assertions.assertTrue(exception.getMessage().contains("implicitly
casts VARIANT to STRUCT<"),
+ exception.getMessage());
+ }
+
+ @Test
+ public void testSelectingAndComparingArgumentsAreNotWrittenValues() {
+ // The second argument of NULLIF is only compared with the first, the
index of an array and the key of a
+ // map only choose the element returned, and the cast a predicate
applies to an operand only serves the
+ // comparison, so a Variant lost there writes nothing. MUTATION:
tracing those arguments, or the operand
+ // cast of a predicate -> rejected.
+ assertAccepted(variantTargetWithId(), ImmutableList.of(
+ "SELECT 1, NULLIF(1, IF(TRUE, v, 1)) FROM " + VARIANT_SOURCE,
+ "SELECT 1, element_at(ARRAY(v), IF(TRUE, v['n'], 1)) FROM " +
VARIANT_SOURCE,
+ "SELECT 1, element_at(map('k', v), IF(TRUE, v['k'], 1)) FROM "
+ VARIANT_SOURCE,
+ "SELECT 1, v['n'] > 0 FROM " + VARIANT_SOURCE,
+ "SELECT 1, v['n'] IN (1, 2) FROM " + VARIANT_SOURCE,
+ "SELECT 1, v['s'] LIKE 'a%' FROM " + VARIANT_SOURCE,
+ "SELECT 1, NOT v['flag'] FROM " + VARIANT_SOURCE,
+ "SELECT 1, ARRAY_FILTER(ARRAY(1, 2), ARRAY(v['n'] > 0, TRUE))
FROM " + VARIANT_SOURCE));
+
+ // A returned argument is still traced, and only the cast a predicate
applies itself is skipped: the common
+ // type of an IF compared afterwards is still a loss. Other functions,
whose arguments may be returned, are
+ // traced whole.
+ assertRejected(variantTargetWithId(), "payload", ImmutableList.of(
+ "SELECT 1, NULLIF(IF(TRUE, v, 1), 1) FROM " + VARIANT_SOURCE,
+ "SELECT 1, element_at(ARRAY(IF(TRUE, v, 1)), 1) FROM " +
VARIANT_SOURCE,
+ "SELECT 1, IF(TRUE, v, 1) > 0 FROM " + VARIANT_SOURCE,
+ "SELECT 1, ELT(v['n'], 'a', 'b') FROM " + VARIANT_SOURCE,
+ "SELECT 1, SUBSTRING('abc', v['n']) FROM " + VARIANT_SOURCE,
+ "SELECT 1, MAX_BY(1, IF(TRUE, v, 1)) FROM " + VARIANT_SOURCE));
+ }
+
+ @Test
+ public void testSubqueryAndIsTrueOperandsOfPredicatesAreNotWrittenValues()
{
+ // When BindSink validates the source, before SubqueryToApply, a
scalar subquery keeps the cast a predicate
+ // applies to it as its type coercion, IN over a subquery is an
InSubquery, and IS TRUE / IS FALSE are AND,
+ // NOT and IS NULL over the operand cast to BOOLEAN: still only the
comparison is written. MUTATION: tracing
+ // the type coercion of a compared scalar subquery, or the operand of
an InSubquery or an IS NULL as a
+ // value -> rejected.
+ String keys = "(SELECT 1 AS k UNION ALL SELECT 2) ks";
+ assertAccepted(variantTargetWithId(), ImmutableList.of(
+ "SELECT 1, (SELECT v['s'] FROM " + VARIANT_SOURCE + ") LIKE
'a%'",
+ "SELECT 1, v['n'] IN (SELECT k FROM " + keys + ") FROM " +
VARIANT_SOURCE,
+ "SELECT 1, v['flag'] IS TRUE FROM " + VARIANT_SOURCE,
+ "SELECT 1, v['flag'] IS NOT FALSE FROM " + VARIANT_SOURCE),
+ VariantWritePlanValidatorTest::bindSource);
+
+ // The value compared is still traced: a loss in the query of the
subquery or in the operand is rejected.
+ assertRejected(variantTargetWithId(), "payload", ImmutableList.of(
+ "SELECT 1, (SELECT IF(TRUE, v, 1) FROM " + VARIANT_SOURCE + ")
LIKE 'a%'",
+ "SELECT 1, IF(TRUE, v, 1) IN (SELECT k FROM " + keys + ") FROM
" + VARIANT_SOURCE,
+ "SELECT 1, IF(TRUE, v, 1) IS TRUE FROM " + VARIANT_SOURCE),
+ VariantWritePlanValidatorTest::bindSource);
+ }
+
+ @Test
+ public void testRecursiveCteNullableOutputsKeepTheFieldsTraced() {
+ // AnalyzeCTE wraps every output of both terms of a recursive CTE as
Alias(Nullable(slot)), which keeps the
+ // value, so the matched field is still written as a BOOLEAN. Both
terms build the struct the same way: in
+ // this test the analyzer cannot bind a column of r in the recursive
term. MUTATION: tracing all of the
+ // argument of Nullable -> the first statement is rejected.
+ String recursive = "WITH RECURSIVE r AS (SELECT
named_struct('matched', %1$s, 'payload', %2$s) AS s FROM "
+ + VARIANT_SOURCE + " UNION ALL SELECT named_struct('matched',
%1$s, 'payload', %2$s) FROM r, "
+ + VARIANT_SOURCE + ") SELECT 1, s FROM r";
+ assertAccepted(flaggedTarget(),
formatEach(ImmutableList.of(recursive), "IF(TRUE, v['n'], 1) > 1", "v"),
+ VariantWritePlanValidatorTest::bindSource);
+ assertRejected(flaggedTarget(), "flagged",
formatEach(ImmutableList.of(recursive), "TRUE", "IF(TRUE, v, 1)"),
+ VariantWritePlanValidatorTest::bindSource);
+ }
+
+ @Test
+ public void testCteOutputIsTracedAgainForMoreOfIt() {
+ // flagged reads the payload field of s only, raw writes all of s into
a Variant: the producer of s is traced
+ // again for raw although it was traced for flagged. MUTATION: reusing
the trace of a CTE output whatever
+ // part of it was traced -> accepted.
+ String flags = "WITH flags AS (SELECT named_struct('matched', IF(TRUE,
v['n'], 1) > 1, 'payload', v) AS s "
+ + "FROM " + VARIANT_SOURCE + ") ";
+ assertAccepted(flaggedTarget(), ImmutableList.of(flags + "SELECT 1, s
FROM flags"));
+
+ List<Column> targets = ImmutableList.of(
+ new Column("id", org.apache.doris.catalog.Type.INT),
+ new Column("flagged", flaggedType().toCatalogDataType()),
+ new Column("raw",
ConnectorComputeVariantType.INSTANCE.toCatalogDataType()));
+ assertRejected(targets, "raw", ImmutableList.of(flags + "SELECT 1, s,
s AS raw FROM flags"));
+ }
+
+ private static List<String> formatEach(List<String> templates, Object...
arguments) {
+ return templates.stream()
+ .map(template -> String.format(template, arguments))
+ .collect(ImmutableList.toImmutableList());
+ }
+
+ private static void assertAccepted(List<Column> targets, List<String>
sqlStatements) {
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ assertAccepted(targets, sqlStatements, sql ->
PlanChecker.from(connectContext).analyze(sql).getPlan());
+ }
+
+ private static void assertAccepted(
+ List<Column> targets, List<String> sqlStatements, Function<String,
Plan> analyzer) {
+ for (String sql : sqlStatements) {
+ Plan analyzedPlan = analyzer.apply(sql);
+ Assertions.assertDoesNotThrow(
+ () ->
VariantWritePlanValidator.validateNoLossyCoercion(targets, analyzedPlan), sql);
+ }
+ }
+
+ private static void assertRejected(List<Column> targets, String
targetColumn, List<String> sqlStatements) {
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ assertRejected(targets, targetColumn, sqlStatements,
+ sql ->
PlanChecker.from(connectContext).analyze(sql).getPlan());
+ }
+
+ private static void assertRejected(List<Column> targets, String
targetColumn, List<String> sqlStatements,
+ Function<String, Plan> analyzer) {
+ for (String sql : sqlStatements) {
+ Plan analyzedPlan = analyzer.apply(sql);
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () ->
VariantWritePlanValidator.validateNoLossyCoercion(targets, analyzedPlan),
+ sql);
+ Assertions.assertTrue(exception.getMessage().contains("input
column '" + targetColumn + "'"),
+ sql + ": " + exception.getMessage());
+ }
+ }
+
+ /** The plan of {@code sql} as BindSink validates it: bound, with
subqueries still expressions. */
+ private static Plan bindSource(String sql) {
+ CascadesContext cascadesContext = MemoTestUtils.createCascadesContext(
+ MemoTestUtils.createConnectContext(), sql);
+ cascadesContext.newTableCollector(true).collect();
+ Analyzer.buildCustomAnalyzer(cascadesContext, ImmutableList.of(
+ Analyzer.topDown(new AnalyzeCTE()),
+ Analyzer.bottomUp(new BindRelation(), new CheckPolicy(), new
BindExpression()))).analyze();
+ return cascadesContext.getRewritePlan();
+ }
+
/**
* {@code WITH RECURSIVE source(a, payload) AS (SELECT IF(TRUE, CAST(v AS
INT), 1), 1 UNION ALL SELECT a, x
* FROM source) SELECT 1, payload FROM source}, shaped as AnalyzeCTE
builds it, where x is the work-table
@@ -622,4 +849,20 @@ public class VariantWritePlanValidatorTest {
new Column("id", org.apache.doris.catalog.Type.INT),
new Column("payload",
ConnectorComputeVariantType.INSTANCE.toCatalogDataType()));
}
+
+ private static StructType flaggedType() {
+ return new StructType(ImmutableList.of(
+ new StructField("matched", BooleanType.INSTANCE, true, ""),
+ new StructField("payload",
ConnectorComputeVariantType.INSTANCE, true, "")));
+ }
+
+ private static List<Column> flaggedTarget() {
+ return targetWithId("flagged", flaggedType());
+ }
+
+ private static List<Column> targetWithId(String name, DataType type) {
+ return ImmutableList.of(
+ new Column("id", org.apache.doris.catalog.Type.INT),
+ new Column(name, type.toCatalogDataType()));
+ }
}
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
index 0c20b7ea6c5..c930d72350e 100644
---
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
@@ -17,3 +17,7 @@
2 ["array",2]
3 {"kind":"object-source"}
+-- !variant_field_lineage --
+1 true {"kind":"object-n","n":2} true
+2 false {"kind":"object-n","n":2} true
+
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
index 09244b8597b..485676bb4ed 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
@@ -51,6 +51,14 @@ suite("test_paimon_write_variant_errors",
"p0,external,paimon") {
) USING paimon
TBLPROPERTIES ('file.format' = 'parquet');
+ DROP TABLE IF EXISTS paimon.${dbName}.t_variant_field_lineage;
+ CREATE TABLE paimon.${dbName}.t_variant_field_lineage (
+ id INT,
+ flagged STRUCT<matched: BOOLEAN, payload: VARIANT>,
+ compared VARIANT
+ ) USING paimon
+ TBLPROPERTIES ('file.format' = 'parquet');
+
DROP TABLE IF EXISTS paimon.${dbName}.t_variant_dml_error;
CREATE TABLE paimon.${dbName}.t_variant_dml_error (
id INT,
@@ -248,6 +256,46 @@ suite("test_paimon_write_variant_errors",
"p0,external,paimon") {
order_qt_variant_dml_after_errors """
SELECT id, CAST(payload AS STRING) FROM t_variant_dml_error ORDER
BY id
"""
+
+ // Only the parts of a value that reach a Variant leaf are checked.
flagged.matched is written as a
+ // BOOLEAN, so a Variant converted to a number to compute it loses
nothing, whether the struct is built in
+ // place or from the columns of a subquery; and the conversion a
predicate applies to compare a Variant
+ // only decides the truth value written. A loss in flagged.payload, or
under the operand of a predicate,
+ // is still rejected.
+ sql """INSERT INTO t_variant_coercion_source VALUES (3,
parse_to_variant('{"kind":"object-n","n":2}'))"""
+ sql """
+ INSERT INTO t_variant_field_lineage
+ SELECT 1, named_struct('matched', COALESCE(payload['n'], 0) > 1,
'payload', payload), payload['n'] > 1
+ FROM t_variant_coercion_source WHERE id = 3
+ """
+ sql """
+ INSERT INTO t_variant_field_lineage
+ SELECT 2, named_struct('matched', matched, 'payload', payload),
payload['kind'] LIKE 'object%'
+ FROM (
+ SELECT COALESCE(payload['n'], 0) > 2 AS matched, payload FROM
t_variant_coercion_source WHERE id = 3
+ ) source
+ """
+ test {
+ sql """
+ INSERT INTO t_variant_field_lineage
+ SELECT 3, named_struct('matched', TRUE, 'payload', IF(TRUE,
payload, 1)), NULL
+ FROM t_variant_coercion_source WHERE id = 1
+ """
+ exception "VARIANT write cannot safely convert input column
'flagged'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_field_lineage
+ SELECT 4, NULL, IF(TRUE, payload, 1) > 0
+ FROM t_variant_coercion_source WHERE id = 1
+ """
+ exception "VARIANT write cannot safely convert input column
'compared'"
+ }
+ order_qt_variant_field_lineage """
+ SELECT id, flagged.matched, flagged.payload, compared
+ FROM t_variant_field_lineage
+ ORDER BY id
+ """
} finally {
sql """SET force_jni_scanner = false"""
sql """DROP CATALOG IF EXISTS ${catalogName}"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
