Copilot commented on code in PR #67907: URL: https://github.com/apache/doris/pull/67907#discussion_r3998886704
########## fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/scalar/SpatialConstructorFunctionTest.java: ########## @@ -0,0 +1,74 @@ +// 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.nereids.trees.expressions.functions.scalar; + +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.StringLiteral; +import org.apache.doris.nereids.types.GeographyType; +import org.apache.doris.nereids.types.GeometryType; +import org.apache.doris.nereids.types.VarcharType; + +import org.junit.Assert; +import org.junit.Test; Review Comment: This test is under `org.apache.doris.nereids`, where the repository's import-control rule disallows `org.junit` and requires JUnit 5 (`fe/check/checkstyle/import-control.xml:42-45`). These imports will fail checkstyle; switch to `org.junit.jupiter.api.Test` and `Assertions`, updating the assertion calls as well. ########## be/src/core/data_type/data_type_spatial.cpp: ########## @@ -0,0 +1,139 @@ +// 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. + +#include "core/data_type/data_type_spatial.h" + +#include <cstring> + +#include "agent/be_exec_version_manager.h" +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/column/column_const.h" +#include "core/data_type/data_type.h" +#include "core/field.h" +#include "core/string_view.h" + +namespace doris { + +doris::FieldType DataTypeSpatial::get_storage_field_type() const { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Spatial types are supported only by external Iceberg tables"); + return FieldType::OLAP_FIELD_TYPE_UNKNOWN; +} + +MutableColumnPtr DataTypeSpatial::create_column() const { + return ColumnSpatial::create(_primitive_type); +} + +Status DataTypeSpatial::check_column(const IColumn& column) const { + const IColumn* nested_column = &column; + if (is_column_const(column)) { + nested_column = &assert_cast<const ColumnConst&>(column).get_data_column(); + } + const auto* spatial = check_and_get_column<ColumnSpatial>(nested_column); + if (spatial == nullptr || spatial->get_primitive_type() != _primitive_type) { + return Status::InvalidArgument("Expected {} spatial column, got {}", get_name(), + column.get_name()); + } + return Status::OK(); +} + +int64_t DataTypeSpatial::get_uncompressed_serialized_bytes(const IColumn& column, + int be_exec_version) const { + DCHECK(be_exec_version >= USE_CONST_SERDE); + const IColumn* data_column = &column; + const bool is_const = is_column_const(column); + const size_t stored_rows = is_const ? 1 : column.size(); + if (is_const) { + data_column = &assert_cast<const ColumnConst&>(column).get_data_column(); + } + const auto& spatial = assert_cast<const ColumnSpatial&>(*data_column); + size_t payload_size = 0; + for (size_t i = 0; i < stored_rows; ++i) { + payload_size += spatial.get_data_at(i).size; + } + return sizeof(bool) + sizeof(size_t) * (2 + stored_rows) + payload_size; +} + +char* DataTypeSpatial::serialize(const IColumn& column, char* buf, int be_exec_version) const { + DCHECK(be_exec_version >= USE_CONST_SERDE); + const IColumn* data_column = &column; + size_t stored_rows = 0; + buf = serialize_const_flag_and_row_num(&data_column, buf, &stored_rows); + const auto& spatial = assert_cast<const ColumnSpatial&>(*data_column); + auto* sizes = reinterpret_cast<size_t*>(buf); + for (size_t i = 0; i < stored_rows; ++i) { + unaligned_store<size_t>(&sizes[i], spatial.get_data_at(i).size); + } + char* payload = buf + sizeof(size_t) * stored_rows; + for (size_t i = 0; i < stored_rows; ++i) { + const auto value = spatial.get_data_at(i); + memcpy(payload, value.data, value.size); + payload += value.size; + } + return payload; +} + +const char* DataTypeSpatial::deserialize(const char* buf, MutableColumnPtr* column, + int be_exec_version) const { + DCHECK(be_exec_version >= USE_CONST_SERDE); + auto* original = column->get(); + size_t stored_rows = 0; + buf = deserialize_const_flag_and_row_num(buf, column, &stored_rows); + auto& spatial = assert_cast<ColumnSpatial&>(*original); + const auto* sizes = reinterpret_cast<const size_t*>(buf); + const char* payload = buf + sizeof(size_t) * stored_rows; + for (size_t i = 0; i < stored_rows; ++i) { + const size_t size = unaligned_load<size_t>(&sizes[i]); + spatial.insert_data(payload, size); + payload += size; + } + return payload; +} + +Field DataTypeSpatial::get_field(const TExprNode& /* node */) const { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Spatial literals must be constructed from WKB by a spatial function"); +} + +FieldWithDataType DataTypeSpatial::get_field_with_data_type(const IColumn& column, + size_t row_num) const { + const auto value = assert_cast<const ColumnSpatial&>(column).get_data_at(row_num); + return FieldWithDataType {.field = Field::create_field<TYPE_VARBINARY>(StringView(value)), + .base_scalar_type_id = _primitive_type}; +} + +bool DataTypeSpatial::equals(const IDataType& rhs) const { + const auto* other = dynamic_cast<const DataTypeSpatial*>(&rhs); + return other != nullptr && _primitive_type == other->_primitive_type && _crs == other->_crs && + _algorithm == other->_algorithm; +} + +void DataTypeSpatial::to_protobuf(PTypeDesc* /* ptype */, PTypeNode* /* node */, + PScalarType* /* scalar_type */) const {} Review Comment: `SlotDescriptor` serializes execution types through `IDataType::to_protobuf`, but this override is a no-op, so the new `spatial_crs`/`spatial_algorithm` fields are never sent to the BE. Non-default Iceberg GEOMETRY/GEOGRAPHY metadata is therefore lost in downstream plan descriptors, which can change Geography semantic checks and spatial write behavior. Populate the CRS for both types and the algorithm for GEOGRAPHY here. ########## be/test/exprs/function/function_geo_test.cpp: ########## @@ -94,6 +97,338 @@ TEST(VGeoFunctionsTest, function_geo_st_as_text) { } } +TEST(VGeoFunctionsTest, function_geo_st_as_text_with_spatial_wkb) { + const std::string wkb( + "\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf0?\x00\x00\x00\x00\x00\x00\x00@", 21); + for (const auto primitive_type : {TYPE_GEOMETRY, TYPE_GEOGRAPHY}) { + auto spatial_type = primitive_type == TYPE_GEOMETRY + ? std::make_shared<DataTypeSpatial>(TYPE_GEOMETRY) + : std::make_shared<DataTypeSpatial>(TYPE_GEOGRAPHY, "OGC:CRS84", + "spherical"); + auto spatial_column = ColumnSpatial::create(primitive_type); + spatial_column->insert_data(wkb.data(), wkb.size()); + + ColumnsWithTypeAndName arguments {{std::move(spatial_column), spatial_type, "spatial"}}; + auto result_type = make_nullable(std::make_shared<DataTypeString>()); + auto function = + SimpleFunctionFactory::instance().get_function("st_astext", arguments, result_type); + ASSERT_NE(nullptr, function); + + Block block; + block.insert(arguments.front()); + block.insert({nullptr, result_type, "result"}); + ASSERT_TRUE(function->execute(nullptr, block, {0}, 1, 1).ok()); + + const auto value = block.get_by_position(1).column->get_data_at(0); + EXPECT_EQ("POINT (1 2)", std::string(value.data, value.size)); + } +} + +TEST(VGeoFunctionsTest, function_geo_st_geomfromwkb_returns_raw_geometry_wkb) { + const std::string wkb( + "\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf0?\x00\x00\x00\x00\x00\x00\x00@", 21); + for (const auto& function_name : {"st_geomfromwkb", "st_geometryfromwkb"}) { + auto input_column = ColumnString::create(); + input_column->insert_data("0101000000000000000000F03F0000000000000040", 42); + auto input_type = std::make_shared<DataTypeString>(); + ColumnsWithTypeAndName arguments {{std::move(input_column), input_type, "wkb"}}; + auto result_type = make_nullable(std::make_shared<DataTypeSpatial>(TYPE_GEOMETRY)); + auto function = SimpleFunctionFactory::instance().get_function(function_name, arguments, result_type); + ASSERT_NE(nullptr, function); + + Block block; + block.insert(arguments.front()); + block.insert({nullptr, result_type, "result"}); + ASSERT_TRUE(function->execute(nullptr, block, {0}, 1, 1).ok()); + + const auto value = block.get_by_position(1).column->get_data_at(0); + EXPECT_EQ(wkb, std::string(value.data, value.size)); + } +} + +TEST(VGeoFunctionsTest, function_geo_st_geogfromwkb_returns_raw_geography_wkb) { + const std::string wkb( + "\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf0?\x00\x00\x00\x00\x00\x00\x00@", 21); + auto input_column = ColumnString::create(); + input_column->insert_data("0101000000000000000000F03F0000000000000040", 42); + auto input_type = std::make_shared<DataTypeString>(); + ColumnsWithTypeAndName arguments {{std::move(input_column), input_type, "wkb"}}; + auto result_type = make_nullable(std::make_shared<DataTypeSpatial>(TYPE_GEOGRAPHY)); Review Comment: This test constructs `DataTypeSpatial(TYPE_GEOGRAPHY)` without the required edge algorithm, so it can trigger the constructor's DCHECK before the function is exercised. Construct the result type as `GEOGRAPHY(OGC:CRS84, spherical)` instead. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java: ########## @@ -795,11 +806,32 @@ public static boolean containsVariant(Type type) { return false; } + public static boolean containsSpatial(Type type) { + if (type.isScalarType()) { + return ((ScalarType) type).isSpatialType(); + } + if (type.isArrayType()) { + return containsSpatial(((ArrayType) type).getItemType()); + } + if (type.isMapType()) { + MapType map = (MapType) type; + return containsSpatial(map.getKeyType()) || containsSpatial(map.getValueType()); + } + if (type.isStructType()) { + return ((StructType) type).getFields().stream() + .anyMatch(field -> containsSpatial(field.getType())); + } + return false; + } + public static void validateWriteSchema(Table table, List<Column> columns) { boolean writesVariant = columns.stream().anyMatch(column -> containsVariant(column.getType())); + boolean writesSpatial = columns.stream().anyMatch(column -> containsSpatial(column.getType())); FileFormat fileFormat = getFileFormat(table); - if (writesVariant) { + if (writesVariant || writesSpatial) { validateWriteSchema(columns, getFormatVersion(table), fileFormat); + } Review Comment: `containsSpatial` recursively classifies spatial fields inside structs, arrays, and maps as writable, but the Parquet writer rejects any nested spatial type with `NotImplemented` (`be/src/format/transformer/vparquet_transformer.cpp:156-159`). As a result, an INSERT into a nested GEOMETRY/GEOGRAPHY schema passes this validation and fails only during file writing; reject nested spatial schemas during analysis or add nested annotation support. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSpatialWriteAnalyzer.java: ########## @@ -0,0 +1,115 @@ +// 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.iceberg; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.ScalarType; +import org.apache.doris.catalog.Type; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.Cast; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.functions.scalar.If; +import org.apache.doris.nereids.types.DataType; + +import java.util.List; +import java.util.Objects; + +/** Analysis checks that preserve Iceberg spatial type parameters during writes. */ +public final class IcebergSpatialWriteAnalyzer { + private IcebergSpatialWriteAnalyzer() { + } + + /** + * Rejects an INSERT source whose spatial kind or parameters differ from the Iceberg target + * before sink coercion can erase that distinction. + */ + public static void validate( + List<Column> targetColumns, List<? extends NamedExpression> sourceColumns) { + if (targetColumns.size() != sourceColumns.size()) { + throw new AnalysisException("Iceberg spatial write target and source columns are not aligned"); + } + for (int i = 0; i < targetColumns.size(); ++i) { + Type targetCatalogType = targetColumns.get(i).getType(); + if (!(targetCatalogType instanceof ScalarType) + || !((ScalarType) targetCatalogType).isSpatialType()) { + continue; + } + validateSpatialConversion(sourceColumns.get(i).getDataType().toCatalogDataType(), + (ScalarType) targetCatalogType, targetColumns.get(i).getName()); + } + } + + /** Validates each MERGE action before a target cast can hide its spatial source type. */ + public static void validateMergeActions( + List<Column> targetColumns, List<? extends NamedExpression> sourceColumns) { + if (targetColumns.size() != sourceColumns.size()) { + throw new AnalysisException("Iceberg spatial write target and source columns are not aligned"); + } + for (int i = 0; i < targetColumns.size(); ++i) { + Type targetCatalogType = targetColumns.get(i).getType(); + if (targetCatalogType instanceof ScalarType && ((ScalarType) targetCatalogType).isSpatialType()) { + validateMergeActionExpression(sourceColumns.get(i), (ScalarType) targetCatalogType, + targetColumns.get(i).getName()); + } + } + } + + private static void validateMergeActionExpression( + Expression expression, ScalarType targetType, String columnName) { + if (expression instanceof Alias) { + validateMergeActionExpression(expression.child(0), targetType, columnName); + return; + } + DataType targetDataType = DataType.fromCatalogType(targetType); + if (expression instanceof If && expression.getDataType().equals(targetDataType)) { + If ifExpression = (If) expression; + validateMergeActionExpression(ifExpression.getTrueValue(), targetType, columnName); + validateMergeActionExpression(ifExpression.getFalseValue(), targetType, columnName); + return; + } + if (expression instanceof Cast && expression.getDataType().equals(targetDataType)) { + validateSpatialConversion(expression.child(0).getDataType().toCatalogDataType(), targetType, columnName); + return; + } + validateSpatialConversion(expression.getDataType().toCatalogDataType(), targetType, columnName); + } + + static void validateSpatialConversion(Type sourceType, ScalarType targetType, String columnName) { + if (!(sourceType instanceof ScalarType) || !((ScalarType) sourceType).isSpatialType()) { + throw new AnalysisException("Iceberg spatial write cannot convert input column '" + columnName + + "' from " + sourceType.toSql() + " to " + targetType.toSql()); + } Review Comment: An untyped `NULL` is represented as `Type.NULL` and is normally accepted by sink coercion for nullable targets, but this guard rejects it before `BindSink` performs the target cast. Thus `INSERT INTO ... SELECT NULL` cannot populate an optional GEOMETRY/GEOGRAPHY column. Treat `Type.NULL` as compatible and let the existing coercion type it to the spatial target. ########## be/src/exprs/function/geo/functions_geo.cpp: ########## @@ -660,16 +787,38 @@ struct StGeoFromWkb { res->insert_default(); continue; } - buf.clear(); - shape->encode_to(&buf); - res->insert_data(buf.data(), buf.size()); + if (!decode_wkb_hex(value, &wkb)) { + null_map_data[row] = 1; + res->insert_default(); + continue; + } + res->insert_data(wkb.data(), wkb.size()); } block.replace_by_position(result, ColumnNullable::create(std::move(res), std::move(null_map))); return Status::OK(); } }; +template <typename Impl> +class SpatialWkbConstructorFunction : public IFunction { +public: + static constexpr auto name = Impl::NAME; + static FunctionPtr create() { return std::make_shared<SpatialWkbConstructorFunction<Impl>>(); } + String get_name() const override { return name; } + size_t get_number_of_arguments() const override { return Impl::NUM_ARGS; } + bool is_variadic() const override { return false; } + + DataTypePtr get_return_type_impl(const DataTypes&) const override { + return make_nullable(std::make_shared<DataTypeSpatial>(Impl::OUTPUT_TYPE)); + } Review Comment: `TYPE_GEOGRAPHY` reaches this constructor with an empty algorithm, but `DataTypeSpatial` requires every Geography instance to have a non-empty algorithm. Calling `ST_GEOGFROMWKB` can therefore abort in DCHECK-enabled BE builds (and otherwise returns a type that violates the invariant). Supply the default CRS and `spherical` algorithm for Geography while keeping Geometry's algorithm empty. -- 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]
