github-actions[bot] commented on code in PR #66620:
URL: https://github.com/apache/doris/pull/66620#discussion_r3772361334
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java:
##########
@@ -498,12 +500,17 @@ private static Plan normalizePlanWithoutLock(LogicalPlan
plan, TableIf table,
}
private static DataType targetTypeForInlineValue(
- Column column, NamedExpression value, boolean isPaimonSink) {
+ Column column, NamedExpression value, boolean isPaimonSink,
boolean isIcebergSink) {
DataType targetType = DataType.fromCatalogType(column.getType());
- return isPaimonSink
- ? PaimonVariantWriteAnalyzer.resolveInlineCoercionTarget(
- targetType, value).orElse(null)
- : targetType;
+ if (isPaimonSink) {
+ return PaimonVariantWriteAnalyzer.resolveInlineCoercionTarget(
+ targetType, value).orElse(null);
+ }
+ if (isIcebergSink) {
+ return IcebergVariantWriteAnalyzer.resolveInlineCoercionTarget(
Review Comment:
[P1] Validate VALUES sources before target casting
For an Iceberg VALUES expression, this selects the target Variant type and
`addColumnValue` installs that cast before resolving the source. The later sink
check therefore sees only compute-V2: with `enable_variant_v2=false`, `VALUES
(..., parse_to_variant(...))` hides a legacy Variant until BE rejects the
column, while MAP/STRUCT/TIMEV2/DECIMAL256 inputs can likewise pass generic
cast analysis even though the V2 kernel does not support them. Please analyze
and validate the pre-cast source (or retain and unwrap this synthetic cast as
MERGE does) before applying per-row target coercion, with legacy and
unsupported-source VALUES coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java:
##########
@@ -0,0 +1,441 @@
+// 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.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.</p>
+ */
+ public static void validateNoLossyCoercion(
+ String sinkName, List<Column> targetColumns, Plan sourcePlan) {
+ validateNoLossyCoercion(
+ sinkName, targetColumns, sourcePlan, Optional.empty());
+ }
+
+ /** Validates a sink while its enclosing CTE producers still live in the
analyzer context. */
+ public static void validateNoLossyCoercion(
+ String sinkName, List<Column> targetColumns, Plan sourcePlan,
+ CTEContext cteContext) {
+ validateNoLossyCoercion(
+ sinkName, targetColumns, sourcePlan, Optional.of(cteContext));
+ }
+
+ private static void validateNoLossyCoercion(
+ String sinkName, List<Column> targetColumns, Plan sourcePlan,
+ Optional<CTEContext> cteContext) {
+ if (targetColumns.size() != sourcePlan.getOutput().size()) {
+ throw new AnalysisException(
+ sinkName + " 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(
+ sinkName, targetColumns, sourcePlan, sourceOrdinals,
cteContext);
+ }
+
+ /**
+ * Validates target columns that are embedded in a wider source output,
such as Iceberg
+ * row-level DML where operation and row-id routing columns precede table
data columns.
+ */
+ public static void validateNoLossyCoercion(
+ String sinkName, List<Column> targetColumns, Plan sourcePlan,
+ List<Integer> sourceOrdinals) {
+ validateNoLossyCoercion(
+ sinkName, targetColumns, sourcePlan, sourceOrdinals,
Optional.empty());
+ }
+
+ /** Validates mapped source ordinals with analyzer-time CTE producer
visibility. */
+ public static void validateNoLossyCoercion(
+ String sinkName, List<Column> targetColumns, Plan sourcePlan,
+ List<Integer> sourceOrdinals, CTEContext cteContext) {
+ validateNoLossyCoercion(
+ sinkName, targetColumns, sourcePlan, sourceOrdinals,
Optional.of(cteContext));
+ }
+
+ private static void validateNoLossyCoercion(
+ String sinkName, List<Column> targetColumns, Plan sourcePlan,
+ List<Integer> sourceOrdinals, Optional<CTEContext> cteContext) {
+ if (targetColumns.size() != sourceOrdinals.size()) {
+ throw new AnalysisException(
+ sinkName + " 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(
+ sinkName + " Variant write target and source columns
are not aligned");
+ }
+ Column targetColumn = targetColumns.get(targetOrdinal);
+ traceOutputLineage(
+ sourcePlan,
+
Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()),
+ sinkName,
+ 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 sinkName, String
targetColumn,
+ TraceContext context) {
+ if (requiredExprIds.isEmpty()) {
+ return Collections.emptySet();
+ }
+ if (plan instanceof SetOperation) {
+ return traceSetOperation(
+ plan, (SetOperation) plan, requiredExprIds, sinkName,
targetColumn, context);
+ }
+ if (plan instanceof RecursiveCte) {
+ return traceRecursiveCte(
+ plan, (RecursiveCte) plan, requiredExprIds, sinkName,
targetColumn, context);
+ }
+ if (plan instanceof LogicalGenerate) {
+ return traceGenerate(
+ (LogicalGenerate<?>) plan, requiredExprIds, sinkName,
targetColumn, context);
+ }
+ if (plan instanceof LogicalCTEConsumer) {
+ return traceCteConsumer(
+ (LogicalCTEConsumer) plan, requiredExprIds, sinkName,
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, sinkName, targetColumn, context));
+ inputExprIds.addAll(namedExpression.getInputSlotExprIds());
+ 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, sinkName, targetColumn,
context));
+ }
+ return externalExprIds;
+ }
+
+ private static Set<ExprId> traceRecursiveCte(
+ Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds,
+ String sinkName, 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,
+ sinkName, targetColumn, context));
+ }
+ return externalExprIds;
+ }
+
+ private static Set<ExprId> traceGenerate(
+ LogicalGenerate<?> generate, Set<ExprId> requiredExprIds,
+ String sinkName, 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;
+ }
+ Expression generator = generators.get(ordinal);
+ childRequiredExprIds.addAll(validateExpression(
+ generator, sinkName, targetColumn, context));
+ childRequiredExprIds.remove(generatorOutput.getExprId());
+ childRequiredExprIds.addAll(generator.getInputSlotExprIds());
+ }
+ Set<ExprId> childOutputExprIds = generate.child().getOutputExprIdSet();
+ Set<ExprId> externalExprIds = new HashSet<>(childRequiredExprIds);
+ externalExprIds.removeAll(childOutputExprIds);
+ childRequiredExprIds.retainAll(childOutputExprIds);
+ externalExprIds.addAll(traceOutputLineage(
+ generate.child(), childRequiredExprIds, sinkName,
targetColumn, context));
+ return externalExprIds;
+ }
+
+ private static Set<ExprId> traceSetOperation(
+ Plan plan, SetOperation setOperation, Set<ExprId> requiredExprIds,
+ String sinkName, 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), sinkName,
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,
+ sinkName, 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 sinkName, 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, sinkName,
targetColumn, context);
+ } finally {
+ context.activeCteIds.remove(consumer.getCteId());
+ }
+ }
+
+ private static Set<ExprId> validateExpression(
+ Expression expression, String sinkName, String targetColumn,
TraceContext context) {
+ validateLossyCast(expression, sinkName, targetColumn);
Review Comment:
[P2] Ignore predicate-only Variant casts in sink lineage
This validates every cast anywhere below the selected expression, including
casts used only to compute a condition. Direct Variant-V2 subpath comparisons
intentionally insert an implicit scalar cast, so a write such as `SELECT
IF(v['n'] = 20, 7, 8)` is rejected even though only the supported scalar result
7 or 8 reaches the Variant target. The same false rejection affects Paimon and
Iceberg row-level DML. Please trace value-producing children (for IF/CASE, the
result arms) separately from predicate/index/control inputs, and add safe
subpath-predicate coverage for both sinks.
--
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]
