github-actions[bot] commented on code in PR #69194: URL: https://github.com/apache/doris/pull/69194#discussion_r4293116118
########## fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource; + +import org.apache.doris.catalog.Column; +import org.apache.doris.nereids.CTEContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.CTEId; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +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.functions.scalar.If; +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; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer; +import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; +import org.apache.doris.nereids.trees.plans.logical.LogicalUnion; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +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 java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** Shared source-plan validation for external Variant sinks. */ +public final class VariantWritePlanValidator { + private VariantWritePlanValidator() { + } + + /** + * Rejects an implicit Variant-to-non-Variant cast in the lineage of a Variant target column. + * + * <p>Common-type analysis for UNION, IF and CASE runs before sink binding. Without this + * check, an object or array Variant can become SQL NULL while being cast to a scalar, and the + * sink will only see that scalar/NULL and encode it back as Variant. Explicit casts remain an + * intentional user conversion and are not rejected. Only the expressions whose values can reach + * the target are inspected: the condition of an IF or of a CASE branch only chooses a value, so a + * Variant path compared with a scalar there loses nothing that is written.</p> + */ + public static void validateNoLossyCoercion(List<Column> targetColumns, Plan sourcePlan) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.empty()); + } + + /** Validates a sink while its enclosing CTE producers still live in the analyzer context. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourcePlan.getOutput().size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + List<Integer> sourceOrdinals = new ArrayList<>(targetColumns.size()); + for (int i = 0; i < targetColumns.size(); i++) { + sourceOrdinals.add(i); + } + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, cteContext); + } + + /** + * Validates target columns that are embedded in a wider source output, such as row-level DML + * where the operation and row-id routing columns precede the table data columns. + */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.empty()); + } + + /** Validates mapped source ordinals with analyzer-time CTE producer visibility. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourceOrdinals.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + + List<Integer> variantTargetOrdinals = new ArrayList<>(); + for (int i = 0; i < targetColumns.size(); i++) { + DataType targetType = DataType.fromCatalogType(targetColumns.get(i).getType()); + if (VariantType.containsVariant(targetType)) { + variantTargetOrdinals.add(i); + } + } + if (variantTargetOrdinals.isEmpty()) { + return; + } + + TraceContext traceContext = new TraceContext(sourcePlan, cteContext); + List<Slot> sourceOutputs = sourcePlan.getOutput(); + for (int targetOrdinal : variantTargetOrdinals) { + int sourceOrdinal = sourceOrdinals.get(targetOrdinal); + if (sourceOrdinal < 0 || sourceOrdinal >= sourceOutputs.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + Column targetColumn = targetColumns.get(targetOrdinal); + traceOutputLineage( + sourcePlan, + Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()), + targetColumn.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 + * enclosing plan. + */ + private static Set<ExprId> traceOutputLineage( + Plan plan, Set<ExprId> requiredExprIds, String targetColumn, TraceContext context) { + if (requiredExprIds.isEmpty()) { + return Collections.emptySet(); + } + if (plan instanceof SetOperation) { + return traceSetOperation(plan, (SetOperation) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof RecursiveCte) { + return traceRecursiveCte(plan, (RecursiveCte) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalGenerate) { + return traceGenerate((LogicalGenerate<?>) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalCTEConsumer) { + return traceCteConsumer((LogicalCTEConsumer) plan, requiredExprIds, targetColumn, context); + } + + Set<ExprId> unresolvedExprIds = new HashSet<>(requiredExprIds); + Set<ExprId> inputExprIds = new HashSet<>(); + for (Expression expression : plan.getExpressions()) { + if (!(expression instanceof NamedExpression)) { + continue; + } + NamedExpression namedExpression = (NamedExpression) expression; + if (!requiredExprIds.contains(namedExpression.getExprId())) { + continue; + } + inputExprIds.addAll(validateExpression(namedExpression, targetColumn, context)); + unresolvedExprIds.remove(namedExpression.getExprId()); + } + inputExprIds.addAll(unresolvedExprIds); + + Set<ExprId> externalExprIds = new HashSet<>(inputExprIds); + 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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceRecursiveCte( + Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + for (int childIndex = 0; childIndex < plan.arity(); childIndex++) { + List<SlotReference> childOutputs = recursiveCte.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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceGenerate( + LogicalGenerate<?> generate, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> childRequiredExprIds = new HashSet<>(requiredExprIds); + 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())) { + continue; + } + childRequiredExprIds.remove(generatorOutput.getExprId()); + childRequiredExprIds.addAll(validateExpression(generators.get(ordinal), 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; + } + + private static Set<ExprId> traceSetOperation( + Plan plan, SetOperation setOperation, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + + if (plan instanceof LogicalUnion) { + for (List<NamedExpression> constantRow : ((LogicalUnion) plan).getConstantExprsList()) { + for (int ordinal : requiredOrdinals) { + if (ordinal < constantRow.size()) { + externalExprIds.addAll(validateExpression(constantRow.get(ordinal), targetColumn, context)); + } + } + } + } + + for (int childIndex = 0; childIndex < setOperation.getArity(); 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)); + } + return externalExprIds; + } + + private static Set<Integer> requiredOutputOrdinals(Plan plan, Set<ExprId> requiredExprIds) { + Set<Integer> requiredOrdinals = new HashSet<>(); + List<Slot> outputs = plan.getOutput(); + for (int i = 0; i < outputs.size(); i++) { + if (requiredExprIds.contains(outputs.get(i).getExprId())) { + requiredOrdinals.add(i); + } + } + return requiredOrdinals; + } + + private static Set<ExprId> traceCteConsumer( + LogicalCTEConsumer consumer, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + LogicalCTEProducer<?> producer = context.cteProducers.get(consumer.getCteId()); + Optional<Plan> producerPlan = producer == null + ? context.findAnalyzerCteProducer(consumer) + : Optional.of(producer.child()); + if (!producerPlan.isPresent() || !context.activeCteIds.add(consumer.getCteId())) { + return Collections.emptySet(); + } + try { + Set<ExprId> producerExprIds = new HashSet<>(); + for (Map.Entry<Slot, Slot> mapping : consumer.getConsumerToProducerOutputMap().entrySet()) { + if (requiredExprIds.contains(mapping.getKey().getExprId())) { + producerExprIds.add(mapping.getValue().getExprId()); + } + } + return traceOutputLineage(producerPlan.get(), producerExprIds, targetColumn, context); + } 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. + */ + private static Set<ExprId> validateExpression( + Expression expression, String targetColumn, TraceContext context) { + List<Expression> valueNodes = valueNodes(expression); + 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)); + } + } + return inputExprIds; + } + + /** 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)); + Plan queryPlan = scalarSubquery.getQueryPlan(); + context.collectCteProducers(queryPlan); + Set<ExprId> unresolvedExprIds = new HashSet<>(traceOutputLineage( + queryPlan, + Collections.singleton(queryPlan.getOutput().get(0).getExprId()), + targetColumn, + context)); + Set<ExprId> correlatedExprIds = new HashSet<>(); + for (Slot correlateSlot : scalarSubquery.getCorrelateSlots()) { + correlatedExprIds.add(correlateSlot.getExprId()); + } + unresolvedExprIds.retainAll(correlatedExprIds); + return unresolvedExprIds; + } + + /** + * 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. + */ + private static List<Expression> valueNodes(Expression expression) { + List<Expression> nodes = new ArrayList<>(); + collectValueNodes(expression, 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); + } else { Review Comment: [P2] Trace only the VARIANT leaves of a complex target For an externally created Iceberg target `dst(s STRUCT<matched:BOOLEAN,payload:VARIANT>)`, `INSERT INTO dst SELECT named_struct('matched', v['n'] > 1, 'payload', v) FROM src` preserves the payload but is rejected here. The reduced analyzed plan is `Sink(s) -> Project(named_struct(matched = GreaterThan(Cast(ElementAt(v,'n'), numeric), 1), payload = v)) -> Scan(v)`. `containsVariant` selects the entire record and this walk visits the BOOLEAN sibling's implicit comparison cast, so `validateLossyCast` throws even though that cast does not feed the VARIANT leaf. The same problem occurs when an outer projection selects only the record's payload. Carry the required nested value paths through constructors/extractions and aliases, so ordinary scalar siblings retain their existing coercion semantics. ########## be/src/format/table/iceberg/arrow_schema_util.cpp: ########## @@ -108,6 +110,13 @@ Status IcebergArrowSchemaConvertor::convert_to_arrow_field( arrow_type = arrow::binary(); break; + case iceberg::TypeID::VARIANT: + arrow_type = arrow::extension::variant(arrow::struct_({ Review Comment: [P2] Compact shared dictionaries before serializing VARIANT rows This newly enabled Arrow mapping reaches `DataTypeVariantV2SerDe::write_column_to_iceberg_arrow`, which calls `write_parquet_variant_arrow(..., compact_metadata=false)` and appends each row's entire batch-shared dictionary. For an unpartitioned v3 Parquet target, with `batch_size=4096` and one pipeline task, a batch of 4,096 values from `PARSE_TO_VARIANT(CONCAT('{"', REPEAT('x',240), CAST(number AS STRING), '":1}'))` has only about 1 MiB of JSON but repeats its dictionary into about 4 GB of Arrow metadata. `VParquetWriter::write` converts the whole block before row-group handling, so it hits the cumulative 2 GiB limit in [Arrow 24 BinaryBuilder](https://github.com/apache/arrow/blob/apache-arrow-24.0.0/cpp/src/arrow/array/builder_binary.h#L290-L297) and fails a valid insert (or exhausts the query memory first). Compact unused dictionary keys per row before this path writes them; the existing `compact_metadata=true` branch provides that machinery. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource; + +import org.apache.doris.catalog.Column; +import org.apache.doris.nereids.CTEContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.CTEId; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +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.functions.scalar.If; +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; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer; +import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; +import org.apache.doris.nereids.trees.plans.logical.LogicalUnion; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +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 java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** Shared source-plan validation for external Variant sinks. */ +public final class VariantWritePlanValidator { + private VariantWritePlanValidator() { + } + + /** + * Rejects an implicit Variant-to-non-Variant cast in the lineage of a Variant target column. + * + * <p>Common-type analysis for UNION, IF and CASE runs before sink binding. Without this + * check, an object or array Variant can become SQL NULL while being cast to a scalar, and the + * sink will only see that scalar/NULL and encode it back as Variant. Explicit casts remain an + * intentional user conversion and are not rejected. Only the expressions whose values can reach + * the target are inspected: the condition of an IF or of a CASE branch only chooses a value, so a + * Variant path compared with a scalar there loses nothing that is written.</p> + */ + public static void validateNoLossyCoercion(List<Column> targetColumns, Plan sourcePlan) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.empty()); + } + + /** Validates a sink while its enclosing CTE producers still live in the analyzer context. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourcePlan.getOutput().size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + List<Integer> sourceOrdinals = new ArrayList<>(targetColumns.size()); + for (int i = 0; i < targetColumns.size(); i++) { + sourceOrdinals.add(i); + } + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, cteContext); + } + + /** + * Validates target columns that are embedded in a wider source output, such as row-level DML + * where the operation and row-id routing columns precede the table data columns. + */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.empty()); + } + + /** Validates mapped source ordinals with analyzer-time CTE producer visibility. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourceOrdinals.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + + List<Integer> variantTargetOrdinals = new ArrayList<>(); + for (int i = 0; i < targetColumns.size(); i++) { + DataType targetType = DataType.fromCatalogType(targetColumns.get(i).getType()); + if (VariantType.containsVariant(targetType)) { + variantTargetOrdinals.add(i); + } + } + if (variantTargetOrdinals.isEmpty()) { + return; + } + + TraceContext traceContext = new TraceContext(sourcePlan, cteContext); + List<Slot> sourceOutputs = sourcePlan.getOutput(); + for (int targetOrdinal : variantTargetOrdinals) { + int sourceOrdinal = sourceOrdinals.get(targetOrdinal); + if (sourceOrdinal < 0 || sourceOrdinal >= sourceOutputs.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + Column targetColumn = targetColumns.get(targetOrdinal); + traceOutputLineage( + sourcePlan, + Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()), + targetColumn.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 + * enclosing plan. + */ + private static Set<ExprId> traceOutputLineage( + Plan plan, Set<ExprId> requiredExprIds, String targetColumn, TraceContext context) { + if (requiredExprIds.isEmpty()) { + return Collections.emptySet(); + } + if (plan instanceof SetOperation) { + return traceSetOperation(plan, (SetOperation) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof RecursiveCte) { + return traceRecursiveCte(plan, (RecursiveCte) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalGenerate) { + return traceGenerate((LogicalGenerate<?>) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalCTEConsumer) { + return traceCteConsumer((LogicalCTEConsumer) plan, requiredExprIds, targetColumn, context); + } + + Set<ExprId> unresolvedExprIds = new HashSet<>(requiredExprIds); + Set<ExprId> inputExprIds = new HashSet<>(); + for (Expression expression : plan.getExpressions()) { + if (!(expression instanceof NamedExpression)) { + continue; + } + NamedExpression namedExpression = (NamedExpression) expression; + if (!requiredExprIds.contains(namedExpression.getExprId())) { + continue; + } + inputExprIds.addAll(validateExpression(namedExpression, targetColumn, context)); + unresolvedExprIds.remove(namedExpression.getExprId()); + } + inputExprIds.addAll(unresolvedExprIds); + + Set<ExprId> externalExprIds = new HashSet<>(inputExprIds); + 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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceRecursiveCte( + Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + for (int childIndex = 0; childIndex < plan.arity(); childIndex++) { + List<SlotReference> childOutputs = recursiveCte.getRegularChildOutput(childIndex); + Set<ExprId> childRequiredExprIds = new HashSet<>(); + for (int ordinal : requiredOrdinals) { + if (ordinal < childOutputs.size()) { + childRequiredExprIds.add(childOutputs.get(ordinal).getExprId()); + } + } + externalExprIds.addAll(traceOutputLineage( Review Comment: [P2] Follow recursive dependencies across work-table columns Tracing only the requested ordinal misses losses that arrive through another recursive column. With `enable_nereids_distribute_planner=true` and a VARIANT sink, consider `WITH RECURSIVE r(a,b,d) AS (SELECT IF(TRUE,parse_to_variant('{"lost":true}'),1), CAST(1 AS DECIMAL(38,9)), CAST(0 AS BIGINT) UNION ALL SELECT a,a,CAST(d+1 AS BIGINT) FROM r WHERE d<1) SELECT b FROM r`. The reduced producer has anchor `a=IF(...Cast(variant AS DECIMAL)...), b=1` and recursive projection `b=worktable.a`. This method checks anchor `b` and recursive `b`, but the latter yields the work-table `a` ExprId, which generic tracing returns unresolved and the caller discards. It never checks anchor `a`, allowing the lost object's SQL NULL to be written on the second output row. Map work-table slots back to recursive ordinals and close those dependencies to a fixed point before validating both terms. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource; + +import org.apache.doris.catalog.Column; +import org.apache.doris.nereids.CTEContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.CTEId; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +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.functions.scalar.If; +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; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer; +import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; +import org.apache.doris.nereids.trees.plans.logical.LogicalUnion; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +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 java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** Shared source-plan validation for external Variant sinks. */ +public final class VariantWritePlanValidator { + private VariantWritePlanValidator() { + } + + /** + * Rejects an implicit Variant-to-non-Variant cast in the lineage of a Variant target column. + * + * <p>Common-type analysis for UNION, IF and CASE runs before sink binding. Without this + * check, an object or array Variant can become SQL NULL while being cast to a scalar, and the + * sink will only see that scalar/NULL and encode it back as Variant. Explicit casts remain an + * intentional user conversion and are not rejected. Only the expressions whose values can reach + * the target are inspected: the condition of an IF or of a CASE branch only chooses a value, so a + * Variant path compared with a scalar there loses nothing that is written.</p> + */ + public static void validateNoLossyCoercion(List<Column> targetColumns, Plan sourcePlan) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.empty()); + } + + /** Validates a sink while its enclosing CTE producers still live in the analyzer context. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourcePlan.getOutput().size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + List<Integer> sourceOrdinals = new ArrayList<>(targetColumns.size()); + for (int i = 0; i < targetColumns.size(); i++) { + sourceOrdinals.add(i); + } + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, cteContext); + } + + /** + * Validates target columns that are embedded in a wider source output, such as row-level DML + * where the operation and row-id routing columns precede the table data columns. + */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.empty()); + } + + /** Validates mapped source ordinals with analyzer-time CTE producer visibility. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourceOrdinals.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + + List<Integer> variantTargetOrdinals = new ArrayList<>(); + for (int i = 0; i < targetColumns.size(); i++) { + DataType targetType = DataType.fromCatalogType(targetColumns.get(i).getType()); + if (VariantType.containsVariant(targetType)) { + variantTargetOrdinals.add(i); + } + } + if (variantTargetOrdinals.isEmpty()) { + return; + } + + TraceContext traceContext = new TraceContext(sourcePlan, cteContext); + List<Slot> sourceOutputs = sourcePlan.getOutput(); + for (int targetOrdinal : variantTargetOrdinals) { + int sourceOrdinal = sourceOrdinals.get(targetOrdinal); + if (sourceOrdinal < 0 || sourceOrdinal >= sourceOutputs.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + Column targetColumn = targetColumns.get(targetOrdinal); + traceOutputLineage( + sourcePlan, + Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()), + targetColumn.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 + * enclosing plan. + */ + private static Set<ExprId> traceOutputLineage( + Plan plan, Set<ExprId> requiredExprIds, String targetColumn, TraceContext context) { + if (requiredExprIds.isEmpty()) { + return Collections.emptySet(); + } + if (plan instanceof SetOperation) { + return traceSetOperation(plan, (SetOperation) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof RecursiveCte) { + return traceRecursiveCte(plan, (RecursiveCte) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalGenerate) { + return traceGenerate((LogicalGenerate<?>) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalCTEConsumer) { + return traceCteConsumer((LogicalCTEConsumer) plan, requiredExprIds, targetColumn, context); + } + + Set<ExprId> unresolvedExprIds = new HashSet<>(requiredExprIds); + Set<ExprId> inputExprIds = new HashSet<>(); + for (Expression expression : plan.getExpressions()) { + if (!(expression instanceof NamedExpression)) { + continue; + } + NamedExpression namedExpression = (NamedExpression) expression; + if (!requiredExprIds.contains(namedExpression.getExprId())) { + continue; + } + inputExprIds.addAll(validateExpression(namedExpression, targetColumn, context)); + unresolvedExprIds.remove(namedExpression.getExprId()); + } + inputExprIds.addAll(unresolvedExprIds); + + Set<ExprId> externalExprIds = new HashSet<>(inputExprIds); + 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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceRecursiveCte( + Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + for (int childIndex = 0; childIndex < plan.arity(); childIndex++) { + List<SlotReference> childOutputs = recursiveCte.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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceGenerate( + LogicalGenerate<?> generate, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> childRequiredExprIds = new HashSet<>(requiredExprIds); + 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())) { + continue; + } + childRequiredExprIds.remove(generatorOutput.getExprId()); + childRequiredExprIds.addAll(validateExpression(generators.get(ordinal), 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; + } + + private static Set<ExprId> traceSetOperation( + Plan plan, SetOperation setOperation, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + + if (plan instanceof LogicalUnion) { + for (List<NamedExpression> constantRow : ((LogicalUnion) plan).getConstantExprsList()) { + for (int ordinal : requiredOrdinals) { + if (ordinal < constantRow.size()) { + externalExprIds.addAll(validateExpression(constantRow.get(ordinal), targetColumn, context)); + } + } + } + } + + for (int childIndex = 0; childIndex < setOperation.getArity(); 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)); + } + return externalExprIds; + } + + private static Set<Integer> requiredOutputOrdinals(Plan plan, Set<ExprId> requiredExprIds) { + Set<Integer> requiredOrdinals = new HashSet<>(); + List<Slot> outputs = plan.getOutput(); + for (int i = 0; i < outputs.size(); i++) { + if (requiredExprIds.contains(outputs.get(i).getExprId())) { + requiredOrdinals.add(i); + } + } + return requiredOrdinals; + } + + private static Set<ExprId> traceCteConsumer( + LogicalCTEConsumer consumer, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + LogicalCTEProducer<?> producer = context.cteProducers.get(consumer.getCteId()); + Optional<Plan> producerPlan = producer == null + ? context.findAnalyzerCteProducer(consumer) + : Optional.of(producer.child()); + if (!producerPlan.isPresent() || !context.activeCteIds.add(consumer.getCteId())) { + return Collections.emptySet(); + } + try { + Set<ExprId> producerExprIds = new HashSet<>(); + for (Map.Entry<Slot, Slot> mapping : consumer.getConsumerToProducerOutputMap().entrySet()) { + if (requiredExprIds.contains(mapping.getKey().getExprId())) { + producerExprIds.add(mapping.getValue().getExprId()); + } + } + return traceOutputLineage(producerPlan.get(), producerExprIds, targetColumn, context); + } finally { Review Comment: [P2] Memoize completed CTE lineage traversals A shared producer is revalidated for every consumer because this set remembers only the active call stack. A compact source graph `c0 := SELECT variant AS p; cN := UNION ALL(Consumer(cN-1), Consumer(cN-1)); INSERT sink SELECT p FROM cN LIMIT 0` therefore visits `c0` 2^N times; 28 levels require 268,435,456 visits plus temporary-set allocations before optimization can remove the query. This is new analyzer work: `AnalyzeCTE` stores each producer once, `BindRelation` creates leaf consumers, and the later `CTEInline` keeps twice-referenced producers shared with the default materialization settings. Cache completed producer/output validations and their external dependencies in `TraceContext`, separately from the recursion guard, so validation stays proportional to the shared plan graph. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource; + +import org.apache.doris.catalog.Column; +import org.apache.doris.nereids.CTEContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.CTEId; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +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.functions.scalar.If; +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; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer; +import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; +import org.apache.doris.nereids.trees.plans.logical.LogicalUnion; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +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 java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** Shared source-plan validation for external Variant sinks. */ +public final class VariantWritePlanValidator { + private VariantWritePlanValidator() { + } + + /** + * Rejects an implicit Variant-to-non-Variant cast in the lineage of a Variant target column. + * + * <p>Common-type analysis for UNION, IF and CASE runs before sink binding. Without this + * check, an object or array Variant can become SQL NULL while being cast to a scalar, and the + * sink will only see that scalar/NULL and encode it back as Variant. Explicit casts remain an + * intentional user conversion and are not rejected. Only the expressions whose values can reach + * the target are inspected: the condition of an IF or of a CASE branch only chooses a value, so a + * Variant path compared with a scalar there loses nothing that is written.</p> + */ + public static void validateNoLossyCoercion(List<Column> targetColumns, Plan sourcePlan) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.empty()); + } + + /** Validates a sink while its enclosing CTE producers still live in the analyzer context. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourcePlan.getOutput().size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + List<Integer> sourceOrdinals = new ArrayList<>(targetColumns.size()); + for (int i = 0; i < targetColumns.size(); i++) { + sourceOrdinals.add(i); + } + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, cteContext); + } + + /** + * Validates target columns that are embedded in a wider source output, such as row-level DML + * where the operation and row-id routing columns precede the table data columns. + */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.empty()); + } + + /** Validates mapped source ordinals with analyzer-time CTE producer visibility. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourceOrdinals.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + + List<Integer> variantTargetOrdinals = new ArrayList<>(); + for (int i = 0; i < targetColumns.size(); i++) { + DataType targetType = DataType.fromCatalogType(targetColumns.get(i).getType()); + if (VariantType.containsVariant(targetType)) { + variantTargetOrdinals.add(i); + } + } + if (variantTargetOrdinals.isEmpty()) { + return; + } + + TraceContext traceContext = new TraceContext(sourcePlan, cteContext); + List<Slot> sourceOutputs = sourcePlan.getOutput(); + for (int targetOrdinal : variantTargetOrdinals) { + int sourceOrdinal = sourceOrdinals.get(targetOrdinal); + if (sourceOrdinal < 0 || sourceOrdinal >= sourceOutputs.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + Column targetColumn = targetColumns.get(targetOrdinal); + traceOutputLineage( + sourcePlan, + Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()), + targetColumn.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 + * enclosing plan. + */ + private static Set<ExprId> traceOutputLineage( + Plan plan, Set<ExprId> requiredExprIds, String targetColumn, TraceContext context) { + if (requiredExprIds.isEmpty()) { + return Collections.emptySet(); + } + if (plan instanceof SetOperation) { + return traceSetOperation(plan, (SetOperation) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof RecursiveCte) { + return traceRecursiveCte(plan, (RecursiveCte) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalGenerate) { + return traceGenerate((LogicalGenerate<?>) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalCTEConsumer) { + return traceCteConsumer((LogicalCTEConsumer) plan, requiredExprIds, targetColumn, context); + } + + Set<ExprId> unresolvedExprIds = new HashSet<>(requiredExprIds); + Set<ExprId> inputExprIds = new HashSet<>(); + for (Expression expression : plan.getExpressions()) { + if (!(expression instanceof NamedExpression)) { + continue; + } + NamedExpression namedExpression = (NamedExpression) expression; + if (!requiredExprIds.contains(namedExpression.getExprId())) { + continue; + } + inputExprIds.addAll(validateExpression(namedExpression, targetColumn, context)); + unresolvedExprIds.remove(namedExpression.getExprId()); + } + inputExprIds.addAll(unresolvedExprIds); + + Set<ExprId> externalExprIds = new HashSet<>(inputExprIds); + 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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceRecursiveCte( + Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + for (int childIndex = 0; childIndex < plan.arity(); childIndex++) { + List<SlotReference> childOutputs = recursiveCte.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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceGenerate( + LogicalGenerate<?> generate, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> childRequiredExprIds = new HashSet<>(requiredExprIds); + 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())) { + continue; + } + childRequiredExprIds.remove(generatorOutput.getExprId()); + childRequiredExprIds.addAll(validateExpression(generators.get(ordinal), 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; + } + + private static Set<ExprId> traceSetOperation( + Plan plan, SetOperation setOperation, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + + if (plan instanceof LogicalUnion) { + for (List<NamedExpression> constantRow : ((LogicalUnion) plan).getConstantExprsList()) { + for (int ordinal : requiredOrdinals) { + if (ordinal < constantRow.size()) { + externalExprIds.addAll(validateExpression(constantRow.get(ordinal), targetColumn, context)); + } + } + } + } + + for (int childIndex = 0; childIndex < setOperation.getArity(); childIndex++) { Review Comment: [P2] Exclude EXCEPT membership-only inputs from written-value lineage For a VARIANT target, `INSERT INTO dst SELECT CAST(1 AS DECIMAL(38,9)) EXCEPT SELECT IF(TRUE,PARSE_TO_VARIANT('{"lost":true}'),1)` should write the left numeric 1: the RHS evaluates to SQL NULL and only determines membership. The reduced plan is `Sink -> Except(Left: 1, Right: IF(TRUE,Cast(variant AS DECIMAL,implicit),1))`. Iterating every set-operation child follows the RHS cast and rejects this safe write even though no RHS value is returned. This is the same value/control distinction already respected for IF/CASE conditions, but the set-operation path ignores it. Trace EXCEPT's written values through its first child; keep tracing every value-producing UNION branch. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.datasource; + +import org.apache.doris.catalog.Column; +import org.apache.doris.nereids.CTEContext; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.CTEId; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +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.functions.scalar.If; +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; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer; +import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; +import org.apache.doris.nereids.trees.plans.logical.LogicalUnion; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +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 java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** Shared source-plan validation for external Variant sinks. */ +public final class VariantWritePlanValidator { + private VariantWritePlanValidator() { + } + + /** + * Rejects an implicit Variant-to-non-Variant cast in the lineage of a Variant target column. + * + * <p>Common-type analysis for UNION, IF and CASE runs before sink binding. Without this + * check, an object or array Variant can become SQL NULL while being cast to a scalar, and the + * sink will only see that scalar/NULL and encode it back as Variant. Explicit casts remain an + * intentional user conversion and are not rejected. Only the expressions whose values can reach + * the target are inspected: the condition of an IF or of a CASE branch only chooses a value, so a + * Variant path compared with a scalar there loses nothing that is written.</p> + */ + public static void validateNoLossyCoercion(List<Column> targetColumns, Plan sourcePlan) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.empty()); + } + + /** Validates a sink while its enclosing CTE producers still live in the analyzer context. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourcePlan.getOutput().size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + List<Integer> sourceOrdinals = new ArrayList<>(targetColumns.size()); + for (int i = 0; i < targetColumns.size(); i++) { + sourceOrdinals.add(i); + } + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, cteContext); + } + + /** + * Validates target columns that are embedded in a wider source output, such as row-level DML + * where the operation and row-id routing columns precede the table data columns. + */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.empty()); + } + + /** Validates mapped source ordinals with analyzer-time CTE producer visibility. */ + public static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + CTEContext cteContext) { + validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.of(cteContext)); + } + + private static void validateNoLossyCoercion( + List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals, + Optional<CTEContext> cteContext) { + if (targetColumns.size() != sourceOrdinals.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + + List<Integer> variantTargetOrdinals = new ArrayList<>(); + for (int i = 0; i < targetColumns.size(); i++) { + DataType targetType = DataType.fromCatalogType(targetColumns.get(i).getType()); + if (VariantType.containsVariant(targetType)) { + variantTargetOrdinals.add(i); + } + } + if (variantTargetOrdinals.isEmpty()) { + return; + } + + TraceContext traceContext = new TraceContext(sourcePlan, cteContext); + List<Slot> sourceOutputs = sourcePlan.getOutput(); + for (int targetOrdinal : variantTargetOrdinals) { + int sourceOrdinal = sourceOrdinals.get(targetOrdinal); + if (sourceOrdinal < 0 || sourceOrdinal >= sourceOutputs.size()) { + throw new AnalysisException("VARIANT write target and source columns are not aligned"); + } + Column targetColumn = targetColumns.get(targetOrdinal); + traceOutputLineage( + sourcePlan, + Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()), + targetColumn.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 + * enclosing plan. + */ + private static Set<ExprId> traceOutputLineage( + Plan plan, Set<ExprId> requiredExprIds, String targetColumn, TraceContext context) { + if (requiredExprIds.isEmpty()) { + return Collections.emptySet(); + } + if (plan instanceof SetOperation) { + return traceSetOperation(plan, (SetOperation) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof RecursiveCte) { + return traceRecursiveCte(plan, (RecursiveCte) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalGenerate) { + return traceGenerate((LogicalGenerate<?>) plan, requiredExprIds, targetColumn, context); + } + if (plan instanceof LogicalCTEConsumer) { + return traceCteConsumer((LogicalCTEConsumer) plan, requiredExprIds, targetColumn, context); + } + + Set<ExprId> unresolvedExprIds = new HashSet<>(requiredExprIds); + Set<ExprId> inputExprIds = new HashSet<>(); + for (Expression expression : plan.getExpressions()) { + if (!(expression instanceof NamedExpression)) { + continue; + } + NamedExpression namedExpression = (NamedExpression) expression; + if (!requiredExprIds.contains(namedExpression.getExprId())) { + continue; + } + inputExprIds.addAll(validateExpression(namedExpression, targetColumn, context)); + unresolvedExprIds.remove(namedExpression.getExprId()); + } + inputExprIds.addAll(unresolvedExprIds); + + Set<ExprId> externalExprIds = new HashSet<>(inputExprIds); + 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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceRecursiveCte( + Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + for (int childIndex = 0; childIndex < plan.arity(); childIndex++) { + List<SlotReference> childOutputs = recursiveCte.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)); + } + return externalExprIds; + } + + private static Set<ExprId> traceGenerate( + LogicalGenerate<?> generate, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> childRequiredExprIds = new HashSet<>(requiredExprIds); + 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())) { + continue; + } + childRequiredExprIds.remove(generatorOutput.getExprId()); + childRequiredExprIds.addAll(validateExpression(generators.get(ordinal), 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; + } + + private static Set<ExprId> traceSetOperation( + Plan plan, SetOperation setOperation, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + Set<ExprId> externalExprIds = new HashSet<>(); + Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds); + + if (plan instanceof LogicalUnion) { + for (List<NamedExpression> constantRow : ((LogicalUnion) plan).getConstantExprsList()) { + for (int ordinal : requiredOrdinals) { + if (ordinal < constantRow.size()) { + externalExprIds.addAll(validateExpression(constantRow.get(ordinal), targetColumn, context)); + } + } + } + } + + for (int childIndex = 0; childIndex < setOperation.getArity(); 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)); + } + return externalExprIds; + } + + private static Set<Integer> requiredOutputOrdinals(Plan plan, Set<ExprId> requiredExprIds) { + Set<Integer> requiredOrdinals = new HashSet<>(); + List<Slot> outputs = plan.getOutput(); + for (int i = 0; i < outputs.size(); i++) { + if (requiredExprIds.contains(outputs.get(i).getExprId())) { + requiredOrdinals.add(i); + } + } + return requiredOrdinals; + } + + private static Set<ExprId> traceCteConsumer( + LogicalCTEConsumer consumer, Set<ExprId> requiredExprIds, + String targetColumn, TraceContext context) { + LogicalCTEProducer<?> producer = context.cteProducers.get(consumer.getCteId()); + Optional<Plan> producerPlan = producer == null Review Comment: [P2] Register nested producers loaded from the analyzer context A leading WITH can bypass validation when its producer contains another WITH: `WITH outer_source AS (WITH inner_source AS (SELECT IF(TRUE,parse_to_variant('{"lost":true}'),1) AS p) SELECT p FROM inner_source) INSERT INTO dst SELECT p FROM outer_source`, with a VARIANT target. `AnalyzeCTE` binds this sink before constructing the outer anchors, so the initial trace sees only `Consumer(outer_source)`. The fetched producer contains `CTEAnchor(Producer(inner_source, lossy_project), Consumer(inner_source))`, but its nested producer is never registered in `context.cteProducers`. The inner consumer's fresh ExprId causes generic traversal to skip the producer, and its local name is absent from the outer analyzer context, so the empty return silently accepts the already-coerced NULL. Collect producers from a fetched producer plan before tracing it, as `traceScalarSubquery` already does for its query plan. -- 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]
