github-actions[bot] commented on code in PR #69032: URL: https://github.com/apache/doris/pull/69032#discussion_r4292798176
########## be/src/core/data_type/data_type_spatial.h: ########## @@ -0,0 +1,70 @@ +// 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. + +#pragma once + +#include <string> + +#include "core/column/column_spatial.h" +#include "core/data_type/data_type.h" + +namespace doris { + +// The physical payload is WKB, but this DataType deliberately remains distinct +// from VARBINARY so that the external-format reader can enforce spatial semantics. +class DataTypeSpatial final : public IDataType { +public: + DataTypeSpatial(PrimitiveType primitive_type, std::string crs = "OGC:CRS84", + std::string algorithm = ""); + + const std::string get_family_name() const override { + return _primitive_type == TYPE_GEOMETRY ? "Geometry" : "Geography"; + } + PrimitiveType get_primitive_type() const override { return _primitive_type; } Review Comment: [P2] Complete spatial dispatch for admitted aggregates. FE accepts the new spatial types in ANY_VALUE/MIN/MAX, but create_aggregate_function_single_value in aggregate_function_min_max_impl.h has no GEOMETRY/GEOGRAPHY cases. Consequently SELECT ST_AsText(ANY_VALUE(geom)) FROM spatial_table fails during VAggFn initialization with "Agg Function ... is not implemented"; Geography and MIN/MAX fail likewise. The parallel NDV and HISTOGRAM creator lists also omit these types, while COLLECT_SET falls into its INVALID_TYPE branch and throws "unexpected type for collect" despite FE accepting the input. Add matching spatial implementations (SingleValueDataColumn already provides owning state for single-value aggregates), or reject deliberately unsupported operations during analysis, and cover nullable inputs/state merging. The repaired GROUP BY key dispatch does not cover these aggregate value-state dispatches. ########## be/src/exprs/function/geo/functions_geo.cpp: ########## @@ -32,18 +44,299 @@ #include "core/column/column_array.h" #include "core/column/column_execute_util.h" #include "core/column/column_nullable.h" +#include "core/column/column_varbinary.h" #include "core/data_type/data_type_array.h" #include "core/data_type/data_type_nullable.h" #include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_spatial.h" #include "core/data_type/data_type_string.h" +#include "core/data_type/data_type_varbinary.h" #include "core/data_type/define_primitive_type.h" #include "core/string_ref.h" #include "exprs/function/geo/geo_common.h" #include "exprs/function/geo/geo_types.h" +#include "exprs/function/geo/wkb_parse.h" #include "exprs/function/simple_function_factory.h" +#include "exprs/function/string_hex_util.h" namespace doris { +static bool is_spatial_type(const DataTypePtr& type) { + const auto primitive_type = remove_nullable(type)->get_primitive_type(); + return primitive_type == TYPE_GEOMETRY || primitive_type == TYPE_GEOGRAPHY; +} + +static Status validate_geography_semantics(const DataTypePtr& type, const char* function_name) { + const auto& nested_type = remove_nullable(type); + if (!is_spatial_type(nested_type)) { + return Status::OK(); + } + + const auto* spatial_type = dynamic_cast<const DataTypeSpatial*>(nested_type.get()); + DCHECK(spatial_type != nullptr); + if (spatial_type->get_primitive_type() == TYPE_GEOGRAPHY && + spatial_type->crs() == "OGC:CRS84" && spatial_type->algorithm() == "spherical") { Review Comment: [P2] Recognize case-equivalent default Geography CRS. The pinned [Iceberg 1.12 type implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/api/src/main/java/org/apache/iceberg/types/Types.java#L628-L660) preserves supplied CRS spelling and compares it case-insensitively. A valid external field created with Types.GeographyType.of("ogc:crs84", EdgeAlgorithm.SPHERICAL) therefore has the supported default semantics, but IcebergTypeMapping, FE type transport and DataTypeSpatial keep the lower-case string, and this exact comparison rejects ST_Length(geog) or ST_Distance(geog, geog). The parallel ConnectorSpatialWriteValidator also rejects default ST_GeogFromWKB output for that equivalent target. Normalize recognized CRS identifiers or apply consistent CRS equivalence at the read/write/calculation boundaries, and cover a lower/mixed-case default CRS imported from an external schema. ########## be/src/core/column/column_spatial.cpp: ########## @@ -0,0 +1,309 @@ +// 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/column/column_spatial.h" + +#include <crc32c/crc32c.h> + +#include <algorithm> +#include <cstring> + +#include "common/exception.h" +#include "core/column/columns_common.h" +#include "exec/common/sip_hash.h" +#include "exec/sort/sort_block.h" +#include "util/hash_util.hpp" + +namespace doris { + +namespace { + +template <typename HashValue> +void for_each_spatial_value(const ColumnSpatial& column, size_t start, size_t end, + const uint8_t* null_map, HashValue&& hash_value) { + for (size_t row = start; row < end; ++row) { + if (null_map == nullptr || null_map[row] == 0) { + hash_value(row, column.get_data_at(row)); + } + } +} + +} // namespace + +void ColumnSpatial::update_hash_with_value(size_t n, SipHash& hash) const { + const auto value = get_data_at(n); + hash.update(reinterpret_cast<const char*>(&value.size), sizeof(value.size)); + hash.update(value.data, value.size); +} + +void ColumnSpatial::update_hashes_with_value(uint64_t* __restrict hashes, + const uint8_t* __restrict null_data) const { + for_each_spatial_value(*this, 0, size(), null_data, [&](size_t row, StringRef value) { + hashes[row] = HashUtil::xxHash64WithSeed(value.data, value.size, hashes[row]); + }); +} + +void ColumnSpatial::update_xxHash_with_value(size_t start, size_t end, uint64_t& hash, + const uint8_t* __restrict null_data) const { + for_each_spatial_value(*this, start, end, null_data, [&](size_t, StringRef value) { + hash = HashUtil::xxHash64WithSeed(value.data, value.size, hash); + }); +} + +void ColumnSpatial::update_crcs_with_value(uint32_t* __restrict hashes, PrimitiveType /* type */, + uint32_t rows, uint32_t /* offset */, + const uint8_t* __restrict null_data) const { + DCHECK_EQ(rows, size()); + for_each_spatial_value(*this, 0, rows, null_data, [&](size_t row, StringRef value) { + hashes[row] = + HashUtil::zlib_crc_hash(value.data, cast_set<uint32_t>(value.size), hashes[row]); + }); +} + +void ColumnSpatial::update_crc_with_value(size_t start, size_t end, uint32_t& hash, + const uint8_t* __restrict null_data) const { + for_each_spatial_value(*this, start, end, null_data, [&](size_t, StringRef value) { + hash = HashUtil::zlib_crc_hash(value.data, cast_set<uint32_t>(value.size), hash); + }); +} + +void ColumnSpatial::update_crc32c_batch(uint32_t* __restrict hashes, + const uint8_t* __restrict null_map) const { + for_each_spatial_value(*this, 0, size(), null_map, [&](size_t row, StringRef value) { + hashes[row] = crc32c_extend(hashes[row], reinterpret_cast<const uint8_t*>(value.data), + value.size); + }); +} + +void ColumnSpatial::update_crc32c_single(size_t start, size_t end, uint32_t& hash, + const uint8_t* __restrict null_map) const { + for_each_spatial_value(*this, start, end, null_map, [&](size_t, StringRef value) { + hash = crc32c_extend(hash, reinterpret_cast<const uint8_t*>(value.data), value.size); + }); +} + +void ColumnSpatial::insert_data(const char* pos, size_t length) { + if (length <= StringView::kInlineSize) { + _data.emplace_back(pos, cast_set<uint32_t>(length)); + } else { + _data.emplace_back(_arena.insert(pos, length), cast_set<uint32_t>(length)); + } +} + +int ColumnSpatial::compare_at(size_t n, size_t m, const IColumn& rhs, + int /* nan_direction_hint */) const { + const auto& spatial = assert_cast<const ColumnSpatial&>(rhs); + DCHECK_EQ(_primitive_type, spatial._primitive_type); + return _data[n].compare(spatial._data[m]); +} + +MutableColumnPtr ColumnSpatial::clone_resized(size_t size) const { + auto result = create(_primitive_type); + const size_t copied = std::min(this->size(), size); + for (size_t i = 0; i < copied; ++i) { + const auto value = get_data_at(i); + result->insert_data(value.data, value.size); + } + result->insert_many_defaults(size - copied); + return result; +} + +void ColumnSpatial::insert_range_from(const IColumn& src, size_t start, size_t length) { + const auto& spatial = assert_cast<const ColumnSpatial&>(src); + DCHECK_EQ(_primitive_type, spatial._primitive_type); + if (start + length > spatial.size()) { + throw Exception(ErrorCode::INTERNAL_ERROR, + "Spatial column range start = {}, length = {} is out of bounds for {} rows", + start, length, spatial.size()); + } + for (size_t i = 0; i < length; ++i) { + const auto value = spatial.get_data_at(start + i); + insert_data(value.data, value.size); + } +} + +void ColumnSpatial::insert_indices_from(const IColumn& src, const uint32_t* begin, + const uint32_t* end) { + const auto& spatial = assert_cast<const ColumnSpatial&>(src); + DCHECK_EQ(_primitive_type, spatial._primitive_type); + for (const auto* it = begin; it != end; ++it) { + const auto value = spatial.get_data_at(*it); + insert_data(value.data, value.size); + } +} + +bool ColumnSpatial::has_enough_capacity(const IColumn& src) const { + const auto& spatial = assert_cast<const ColumnSpatial&>(src); + return _data.capacity() - _data.size() > spatial.size(); +} + +ColumnPtr ColumnSpatial::filter(const IColumn::Filter& filter, ssize_t result_size_hint) const { + column_match_filter_size(size(), filter.size()); + auto result = create(_primitive_type); + if (result_size_hint > 0) { + result->_data.reserve(result_size_hint); + } + for (size_t i = 0; i < size(); ++i) { + if (filter[i]) { + const auto value = get_data_at(i); + result->insert_data(value.data, value.size); + } + } + return result; +} + +size_t ColumnSpatial::filter(const IColumn::Filter& filter) { + column_match_filter_size(size(), filter.size()); + // Moving views within this column keeps their arena alive and avoids copying retained WKB. + size_t retained = 0; + for (size_t i = 0; i < size(); ++i) { + if (filter[i]) { + _data[retained++] = _data[i]; + } + } + _data.resize(retained); + return retained; +} + +void ColumnSpatial::erase(size_t start, size_t length) { + // Analytic trimming removes row ranges; move only views so retained WKB keeps its arena. + // Match ColumnVector: empty/out-of-range requests are no-ops and long ranges are clipped. + if (start >= size() || length == 0) { + return; + } + length = std::min(length, size() - start); + std::move(_data.begin() + start + length, _data.end(), _data.begin() + start); Review Comment: [P2] Reclaim erased WKB during analytic trimming. FIRST_VALUE(shape) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) repeatedly reaches this method through AnalyticSinkLocalState::_remove_unused_rows, but the aggregate input column lives until close and each subsequent batch appends WKB to the same _arena. Compacting only the views leaves every erased payload allocated (even 21-byte Points exceed the 12-byte inline limit), so memory grows with the entire input despite successful row trimming; large polygons can exhaust the query memory limit for a bounded frame. Periodically rebuild/reclaim the payload arena while preserving retained values, and test retained bytes across repeated append/trim cycles. The previous missing-erase exception is repaired; this is the new implementation's retained-memory issue. -- 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]
