englefly commented on code in PR #69032: URL: https://github.com/apache/doris/pull/69032#discussion_r4465254321
########## fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/ConnectorSpatialWriteNormalizer.java: ########## @@ -0,0 +1,540 @@ +// 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.connector.converter; + +import org.apache.doris.catalog.Column; +import org.apache.doris.nereids.CTEContext; +import org.apache.doris.nereids.analyzer.UnboundFunction; +import org.apache.doris.nereids.rules.analysis.ExpressionAnalyzer; +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.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +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.literal.ArrayLiteral; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.MapLiteral; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral; +import org.apache.doris.nereids.trees.expressions.literal.StructLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTE; +import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor; +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.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalLimit; +import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalSort; +import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.types.GeographyType; +import org.apache.doris.nereids.types.GeometryType; +import org.apache.doris.nereids.types.MapType; +import org.apache.doris.nereids.types.StructType; +import org.apache.doris.nereids.util.TypeCoercionUtils; + +import com.google.common.collect.ImmutableList; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.IntStream; + +/** Restores proven NULL producer values before spatial sink coercion. */ +public final class ConnectorSpatialWriteNormalizer { + private ConnectorSpatialWriteNormalizer() { + } + + public static LogicalPlan normalize(List<Column> targets, LogicalPlan source, CTEContext cteContext) { + return normalize(targets, source, IntStream.range(0, targets.size()).boxed() + .collect(ImmutableList.toImmutableList()), cteContext); + } + + /** Aligns data targets with source positions when a row-level sink also carries routing columns. */ + public static LogicalPlan normalize(List<Column> targets, LogicalPlan source, + List<Integer> sourceOrdinals, CTEContext cteContext) { + if (targets.stream().noneMatch(column -> column.getType().typeContainsSpatialType())) { + return source; + } + Resolver resolver = new Resolver(source, cteContext); + List<NamedExpression> outputs = new ArrayList<>(source.getOutput()); + boolean changed = false; + for (int i = 0; i < targets.size(); i++) { + Column target = targets.get(i); + if (!target.getType().typeContainsSpatialType()) { + continue; + } + int sourceOrdinal = sourceOrdinals.get(i); + Slot slot = source.getOutput().get(sourceOrdinal); + Expression value = resolver.resolve(slot, source); + DataType targetType = DataType.fromCatalogType(target.getType()); + boolean literal = isLiteralConstruction(value) || canRetargetSpatialNulls(value, targetType); + // SELECT/UPDATE reach the bound sink with CAST(ARRAY(NULL) AS ARRAY<INT>). + // Reuse MERGE's NULL-only proof before validation, without lifting arbitrary producers. + if (literal) { + value = normalizeSpatialNullsBeforeBinding(value, targetType); + } + ConnectorSpatialWriteValidator.validate(target, value); + // Never lift arbitrary producer expressions above their input scope, or + // replace outer-join NULL padding with a non-NULL container constant. + if (literal) { + Expression normalized = normalizeSpatialNulls(value, targetType); + if (!normalized.getDataType().equals(slot.getDataType())) { + // SELECT v,v can feed an ordinary BIGINT and a spatial NULL. Sharing + // v's ExprId would let project merging replace the ordinary cast's input. + outputs.set(sourceOrdinal, new Alias(normalized, slot.getName())); + changed = true; + } + } + } + return changed ? new LogicalProject<>(outputs, source) : source; + } + + private static boolean isLiteralConstruction(Expression value) { + // Only proven NULL-only casts can be retargeted; actual values and invalid source + // array-depth conversions must retain their original identity and errors. + return value instanceof Literal || (value.getClass() == Cast.class && nullArrayDepth(value) >= 0) + || ((value instanceof Array || value instanceof CreateMap + || value instanceof CreateStruct || value instanceof CreateNamedStruct) + && value.children().stream().allMatch(ConnectorSpatialWriteNormalizer::isLiteralConstruction)); + } + + private static final class Resolver { + private final CTEContext context; + private final Map<CTEId, Plan> producers = new HashMap<>(); + private final Set<CTEId> activeCtes = new HashSet<>(); + private final Map<Plan, Map<ExprId, Expression>> resolvedSlots = new IdentityHashMap<>(); + + private Resolver(Plan source, CTEContext context) { + this.context = context; + collectProducers(source); + } + + private void collectProducers(Plan source) { + if (source instanceof LogicalCTEProducer) { + LogicalCTEProducer<?> producer = (LogicalCTEProducer<?>) source; + producers.put(producer.getCteId(), producer.child()); + } + source.children().forEach(this::collectProducers); + } + + private Expression resolve(Expression value, Plan source) { + if (value instanceof Alias) { + return resolve(value.child(0), source); + } + if (value instanceof Slot) { + return resolveSlot((Slot) value, source); + } + // Only restore constant NULL construction. Expanding arbitrary IF(g, g) + // producer chains duplicates shared CTE lineage exponentially. + if (!(value instanceof Cast || value instanceof Array || value instanceof CreateMap + || value instanceof CreateStruct || value instanceof CreateNamedStruct)) { + return value; + } + List<Expression> children = value.children().stream() + .map(child -> resolve(child, source)) + .collect(ImmutableList.toImmutableList()); + // SELECT reaches sink binding before constant folding: CAST(NULL AS INT) + // and CAST(NULL AS GEOMETRY('EPSG:3857')) are still value-independent NULLs. + if (value instanceof Cast && children.get(0).isNullLiteral()) { + // A bound SELECT/UPDATE can still contain GEOGRAPHY -> INT. Retain its + // analysis error before NULL retagging and project merging erase the cast. + validateNullCastChain(value); + return new NullLiteral(value.getDataType()); + } + return children.equals(value.children()) ? value : value.withChildren(children); + } + + private Expression resolveSlot(Slot slot, Plan source) { + Map<ExprId, Expression> cached = resolvedSlots.computeIfAbsent(source, ignored -> new HashMap<>()); + Expression result = cached.get(slot.getExprId()); + if (result == null) { + result = traceSlot(slot, source); + cached.put(slot.getExprId(), result); + } + return result; + } + + private Expression traceSlot(Slot slot, Plan source) { + if (source instanceof LogicalProject || source instanceof LogicalOneRowRelation) { + List<NamedExpression> projects = source instanceof LogicalProject + ? ((LogicalProject<?>) source).getProjects() : ((LogicalOneRowRelation) source).getProjects(); + for (NamedExpression project : projects) { + if (project.getExprId().equals(slot.getExprId())) { + Plan input = source instanceof LogicalProject ? source.child(0) : source; + // A one-row correlated slot has no producer definition here. + return project instanceof Slot && input == source + ? slot : resolve(project, input); + } + } + } else if (source instanceof LogicalCTEConsumer) { + LogicalCTEConsumer consumer = (LogicalCTEConsumer) source; + Optional<Plan> producer = Optional.ofNullable(producers.get(consumer.getCteId())); + if (!producer.isPresent()) { + producer = context.findCTEContext(consumer.getName()) + .filter(cte -> cte.getCteId().equals(consumer.getCteId())) + .flatMap(cte -> cte.getAnalyzedCTEPlan(consumer.getName())).map(Plan.class::cast); + } + // Recursive CTEs do not prove constant provenance through their cycle. + if (producer.isPresent() && activeCtes.add(consumer.getCteId())) { + try { + // An analyzed outer CTE can own an inner WITH whose producers + // are absent from the sink's context and initial source tree. + collectProducers(producer.get()); + return resolve(consumer.getProducerSlot(slot), producer.get()); + } finally { + activeCtes.remove(consumer.getCteId()); + } + } + } else if (source instanceof LogicalCTEAnchor) { + return resolveSlot(slot, source.child(1)); + } else if (source instanceof LogicalSubQueryAlias || source instanceof LogicalFilter + || source instanceof LogicalLimit || source instanceof LogicalSort + || source instanceof LogicalCTE) { + return resolveSlot(slot, source.child(0)); + } + // Joins, grouping sets and set operations may add NULLs or combine values; + // their output type alone is not evidence of a constant value. + return slot; Review Comment: [P2] Preserve proven NULL provenance through UNION ALL For an Iceberg V3 Parquet table whose `shapes` column is `ARRAY<GEOMETRY>`, consider: ```sql INSERT INTO iceberg.db.t (shapes) SELECT ARRAY(CAST(NULL AS INT)) UNION ALL SELECT ARRAY(CAST(NULL AS INT)); ``` The analyzed source is `UnionAll(a: ARRAY<INT>)` over two `OneRowRelation(ARRAY(CAST(NULL AS INT)))` branches. `BindSink` runs before the later constant-folding rewrites. Here `Resolver.traceSlot()` stops at the union and returns its `ARRAY<INT>` slot, so `normalize()` cannot prove the NULL leaves and `ConnectorSpatialWriteValidator.validateType()` rejects INT -> GEOMETRY. Each branch individually is handled by the new normalizer, but combining them rejects the same NULL-only values. Please trace/normalize the corresponding branch outputs for UNION ALL and rebuild the union output types as needed, while preserving row counts and container shapes. Add an INSERT SELECT regression for both spatial kinds. This is a code-path finding; the SQL example has not been executed. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/types/DataType.java: ########## @@ -942,10 +951,15 @@ public boolean isAssignableFrom(DataType targetDataType) { } public void validateDataType() { - validateCatalogDataType(toCatalogDataType()); + validateCatalogDataType(toCatalogDataType(), false); + } + + /** Validate spatial leaves after an explicit opt-in without widening generic or Variant declarations. */ + public void validateDataTypeWithSpatialLeaves() { + validateCatalogDataType(toCatalogDataType(), true); Review Comment: [P2] Apply spatial-leaf admission to MAP expression result validation The new spatial-aware validation entry point is wired into column DDL, but `MapFromArrays.computeSignature()`, `MapFromEntries.customSignature()` and `TransformValues.customSignature()` still call `validateDataType()`, which passes `allowNestedSpatialTypes=false`. For a GEOMETRY column `g`, this means the already-supported `MAP(1, g)` construction works while the equivalent expression below is rejected during FE binding with `MAP unsupported sub-type: GEOMETRY(...)`: ```sql SELECT ST_AsBinary( element_at(map_from_arrays(array(1), array(g)), 1) ) FROM spatial_t; ``` Likewise, `transform_values((k, v) -> v, m)` rejects an imported MAP with spatial values even though it preserves the type and metadata unchanged. Please allow spatial value leaves in these expression-result validators, retaining the restrictions on spatial MAP keys and Variant fields, and add binding tests for these parallel construction/transformation paths. This is based on the validation call chain; the SQL example has not been executed. -- 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]
