backport to support geometry/geography types

ci/docker/integration/runner/Dockerfile

@@ -77,12 +77,13 @@ RUN curl -fsSL -O https://archive.apache.org/dist/spark/spark-3.5.5/spark-3.5.5-
 # if you change packages, don't forget to update them in tests/integration/helpers/cluster.py
 RUN packages="io.delta:delta-spark_2.12:3.1.0,\
 org.apache.hudi:hudi-spark3.5-bundle_2.12:1.0.1,\
-org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.3,\
+org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,\
 org.apache.hadoop:hadoop-aws:3.3.4,\
 com.amazonaws:aws-java-sdk-bundle:1.12.262,\
 org.apache.hadoop:hadoop-azure:3.3.4,\
 com.microsoft.azure:azure-storage:8.6.6,\
-org.apache.spark:spark-avro_2.12:3.5.1"\
+org.apache.spark:spark-avro_2.12:3.5.1,\
+org.apache.sedona:sedona-spark-3.5_2.12:1.6.1"\
     && /spark-3.5.5-bin-hadoop3/bin/spark-shell --packages "$packages" \
     && find /root/.ivy2/ -name '*.jar' -exec ln -sf {} /spark-3.5.5-bin-hadoop3/jars/ \;
 

ci/docker/integration/runner/requirements.txt

@@ -107,3 +107,5 @@ websocket-client==1.9.0
 wheel==0.46.3
 filelock==3.25.0
 kazoo @ git+https://github.com/ClickHouse/kazoo.git@879e0b0e4ae367a289847b43a531938f68770125
+apache-sedona==1.6.1
+shapely==2.1.0

ci/jobs/scripts/check_style/experimental_settings_ignore.txt

@@ -19,6 +19,7 @@ allow_experimental_database_materialized_postgresql
 allow_experimental_database_replicated
 allow_experimental_dynamic_type
 allow_experimental_full_text_index
+allow_experimental_geo_types_in_iceberg
 allow_experimental_insert_into_iceberg
 allow_experimental_inverted_index
 allow_experimental_join_condition

src/Core/Settings.cpp

@@ -6992,6 +6992,9 @@ Query Iceberg table using the snapshot that was current at a specific timestamp.
 )", 0) \
     DECLARE(Int64, iceberg_snapshot_id, 0, R"(
 Query Iceberg table using the specific snapshot id.
+)", 0) \
+    DECLARE(Bool, allow_experimental_geo_types_in_iceberg, false, R"(
+Allow parsing Iceberg `geometry` and `geography` field types as ClickHouse `Geometry` (Variant) type.
 )", 0) \
     DECLARE_WITH_ALIAS(Bool, show_remote_databases_in_system_tables, false, R"(
 Enables showing remote databases (data lake catalogs, MySQL, PostgreSQL) in system tables.

src/Core/SettingsChangesHistory.cpp

@@ -41,6 +41,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         /// Note: please check if the key already exists to prevent duplicate entries.
         addSettingsChanges(settings_changes_history, "26.3.1.20001.altinityantalya",
         {
+            {"allow_experimental_geo_types_in_iceberg", false, false, "New setting to allow parsing Iceberg geometry/geography fields as Geometry type."},
             {"object_storage_cluster_join_mode", "allow", "allow", "New setting"},
             {"export_merge_tree_partition_task_timeout_seconds", "3600", "86400", "Increase default value to make it more realistic"},
             {"export_merge_tree_part_allow_lossy_cast", false, false, "New setting to gate lossy casts in EXPORT PART/PARTITION behind explicit acknowledgment"},

src/Processors/Formats/Impl/ArrowColumnToCHColumn.cpp

@@ -1097,6 +1097,10 @@ static ColumnWithTypeAndName readNonNullableColumnFromArrowColumn(
             {
                 return readColumnWithGeoData(arrow_column, column_name, *geo_metadata);
             }
+            if (type_hint && type_hint->getName() == "Geometry" && settings.allow_geoparquet_parser)
+            {
+                return readColumnWithGeoData(arrow_column, column_name, GeoColumnMetadata{GeoEncoding::WKB, GeoType::Mixed});
+            }
             return readColumnWithStringData<arrow::BinaryArray>(arrow_column, column_name);
         }
         case arrow::Type::EXTENSION:
@@ -1477,6 +1481,9 @@ static ColumnWithTypeAndName readNonNullableColumnFromArrowColumn(
         case arrow::Type::DICTIONARY:
         {
             auto & dict_info = dictionary_infos[column_name];
+            const bool is_lc_nullable = make_nullable_if_low_cardinality
+                || arrow_column->null_count() > 0
+                || (type_hint && type_hint->isLowCardinalityNullable());
 
             /// Load dictionary values only once and reuse it.
             if (!dict_info.values)
@@ -1514,11 +1521,11 @@ static ColumnWithTypeAndName readNonNullableColumnFromArrowColumn(
                     }
                 }
 
-                auto lc_type = std::make_shared<DataTypeLowCardinality>(make_nullable_if_low_cardinality ? makeNullable(dict_column.type) : dict_column.type);
+                auto lc_type = std::make_shared<DataTypeLowCardinality>(is_lc_nullable ? makeNullable(dict_column.type) : dict_column.type);
                 auto tmp_lc_column = lc_type->createColumn();
                 auto tmp_dict_column = IColumn::mutate(assert_cast<ColumnLowCardinality *>(tmp_lc_column.get())->getDictionaryPtr());
                 dynamic_cast<IColumnUnique *>(tmp_dict_column.get())->uniqueInsertRangeFrom(*dict_column.column, 0, dict_column.column->size());
-                size_t expected_dictionary_size = dict_column.column->size() + (dict_info.default_value_index == -1) + make_nullable_if_low_cardinality;
+                size_t expected_dictionary_size = dict_column.column->size() + (dict_info.default_value_index == -1) + is_lc_nullable;
                 if (tmp_dict_column->size() != expected_dictionary_size)
                 {
                     throw Exception(
@@ -1541,9 +1548,9 @@ static ColumnWithTypeAndName readNonNullableColumnFromArrowColumn(
             }
 
             auto arrow_indexes_column = std::make_shared<arrow::ChunkedArray>(indexes_array);
-            auto indexes_column = readColumnWithIndexesData(arrow_indexes_column, dict_info.default_value_index, dict_info.dictionary_size, make_nullable_if_low_cardinality);
+            auto indexes_column = readColumnWithIndexesData(arrow_indexes_column, dict_info.default_value_index, dict_info.dictionary_size, is_lc_nullable);
             auto lc_column = ColumnLowCardinality::create(dict_info.values->column, indexes_column, /*is_shared=*/true);
-            auto lc_type = std::make_shared<DataTypeLowCardinality>(make_nullable_if_low_cardinality ? makeNullable(dict_info.values->type) : dict_info.values->type);
+            auto lc_type = std::make_shared<DataTypeLowCardinality>(is_lc_nullable ? makeNullable(dict_info.values->type) : dict_info.values->type);
             return {std::move(lc_column), std::move(lc_type), column_name};
         }
 #    define DISPATCH(ARROW_NUMERIC_TYPE, CPP_NUMERIC_TYPE) \
@@ -1608,7 +1615,8 @@ static ColumnWithTypeAndName readColumnFromArrowColumn(
     const std::optional<std::unordered_map<String, String>> & parquet_columns_to_clickhouse,
     const std::optional<std::unordered_map<String, String>> & clickhouse_columns_to_parquet)
 {
-    bool read_as_nullable_column = (arrow_column->null_count() || is_nullable_column || (type_hint && (type_hint->isNullable() || type_hint->isLowCardinalityNullable()))) && !geo_metadata && settings.allow_inferring_nullable_columns;
+    bool type_hint_not_nullable_capable = type_hint && !removeNullable(type_hint)->canBeInsideNullable();
+    bool read_as_nullable_column = (arrow_column->null_count() || is_nullable_column || (type_hint && (type_hint->isNullable() || type_hint->isLowCardinalityNullable()))) && !geo_metadata && !type_hint_not_nullable_capable && settings.allow_inferring_nullable_columns;
     if (read_as_nullable_column &&
         arrow_column->type()->id() != arrow::Type::LIST &&
         arrow_column->type()->id() != arrow::Type::LARGE_LIST &&

src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp

@@ -10,6 +10,8 @@
 #include <Columns/ColumnTuple.h>
 #include <Columns/ColumnLowCardinality.h>
 #include <Columns/ColumnMap.h>
+#include <Columns/ColumnVariant.h>
+#include <Columns/ColumnsNumber.h>
 #include <Core/Block.h>
 #include <Common/Exception.h>
 #include <Common/WKB.h>
@@ -23,6 +25,7 @@
 #include <DataTypes/DataTypeFixedString.h>
 #include <DataTypes/DataTypeTime64.h>
 #include <DataTypes/DataTypeCustom.h>
+#include <DataTypes/DataTypeVariant.h>
 
 /// This file deals with schema conversion and with repetition and definition levels.
 
@@ -751,6 +754,58 @@ void prepareGeoColumn(ColumnPtr & column, DataTypePtr & type)
     if (!type->getCustomName())
         return;
 
+    if (type->getCustomName()->getName() == "Geometry")
+    {
+        const auto & col_variant = assert_cast<const ColumnVariant &>(*column);
+        const auto & variant_type = assert_cast<const DataTypeVariant &>(*type);
+        const auto & variants = variant_type.getVariants();
+
+        std::vector<std::shared_ptr<IWKBTransform>> transforms(variants.size());
+        for (size_t i = 0; i < variants.size(); ++i)
+        {
+            const auto & variant_name = variants[i]->getCustomName() ? variants[i]->getCustomName()->getName() : variants[i]->getName();
+            if (variant_name == WKBPointTransform::name)
+                transforms[i] = std::make_shared<WKBPointTransform>();
+            else if (variant_name == WKBLineStringTransform::name || variant_name == "Ring")
+                transforms[i] = std::make_shared<WKBLineStringTransform>();
+            else if (variant_name == WKBPolygonTransform::name)
+                transforms[i] = std::make_shared<WKBPolygonTransform>();
+            else if (variant_name == WKBMultiLineStringTransform::name)
+                transforms[i] = std::make_shared<WKBMultiLineStringTransform>();
+            else if (variant_name == WKBMultiPolygonTransform::name)
+                transforms[i] = std::make_shared<WKBMultiPolygonTransform>();
+        }
+
+        auto result = ColumnString::create();
+        auto null_map = ColumnUInt8::create();
+        result->reserve(col_variant.size());
+        null_map->reserve(col_variant.size());
+        for (size_t i = 0; i < col_variant.size(); ++i)
+        {
+            const auto local_discriminator = col_variant.localDiscriminatorAt(i);
+            if (local_discriminator == ColumnVariant::NULL_DISCRIMINATOR)
+            {
+                result->insertDefault();
+                null_map->insertValue(1);
+                continue;
+            }
+            const auto global_discriminator = col_variant.globalDiscriminatorAt(i);
+            if (!transforms[global_discriminator])
+                throw Exception(
+                    ErrorCodes::BAD_ARGUMENTS,
+                    "Cannot encode Geometry sub-type '{}' as WKB",
+                    variants[global_discriminator]->getName());
+            const auto & sub_column = col_variant.getVariantByLocalDiscriminator(local_discriminator);
+            Field field;
+            sub_column.get(col_variant.offsetAt(i), field);
+            result->insert(transforms[global_discriminator]->dumpObject(field));
+            null_map->insertValue(0);
+        }
+        column = ColumnNullable::create(std::move(result), std::move(null_map));
+        type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeString>());
+        return;
+    }
+
     std::shared_ptr<IWKBTransform> transform;
     if (type->getCustomName()->getName() == WKBPointTransform::name)
         transform = std::make_shared<WKBPointTransform>();

src/Processors/Formats/Impl/Parquet/SchemaConverter.cpp

@@ -886,6 +886,14 @@ void SchemaConverter::processPrimitiveColumn(
         return;
     }
 
+    if (type_hint && type_hint->getName() == "Geometry" && type == parq::Type::BYTE_ARRAY)
+    {
+        GeoColumnMetadata iceberg_geo{GeoEncoding::WKB, GeoType::Mixed};
+        out_inferred_type = getGeoDataType(GeoType::Mixed);
+        out_decoder.string_converter = std::make_shared<GeoConverter>(iceberg_geo);
+        return;
+    }
+
     if (logical.__isset.STRING || logical.__isset.JSON || logical.__isset.BSON ||
         logical.__isset.ENUM || converted == CONV::UTF8 || converted == CONV::JSON ||
         converted == CONV::BSON || converted == CONV::ENUM)

src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h

@@ -19,6 +19,8 @@ DEFINE_ICEBERG_FIELD(data_file);
 DEFINE_ICEBERG_FIELD(element);
 DEFINE_ICEBERG_FIELD(fields);
 DEFINE_ICEBERG_FIELD(float);
+DEFINE_ICEBERG_FIELD(geometry);
+DEFINE_ICEBERG_FIELD(geography);
 DEFINE_ICEBERG_FIELD(key);
 DEFINE_ICEBERG_FIELD(list)
 DEFINE_ICEBERG_FIELD(location);

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp

@@ -163,6 +163,9 @@ std::optional<Field> deserializeFieldFromBinaryRepr(const std::string & str, Dat
         return std::nullopt;
     }
 
+    if (non_nullable_type->getTypeId() == DB::TypeIndex::Variant)
+        return std::nullopt;
+
     /// For all other types except decimal binary representation
     /// matches our internal representation
     column->insertData(str.data(), str.length());

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp

@@ -401,7 +401,8 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                     continue;
 
                 if (const auto type_id = name_and_type.type->getTypeId();
-                    type_id == DB::TypeIndex::Tuple || type_id == DB::TypeIndex::Map || type_id == DB::TypeIndex::Array)
+                    type_id == DB::TypeIndex::Tuple || type_id == DB::TypeIndex::Map || type_id == DB::TypeIndex::Array
+                    || type_id == DB::TypeIndex::Variant)
                     continue;
 
                 auto left = deserializeFieldFromBinaryRepr(left_str, name_and_type.type, true);

src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp

@@ -52,6 +52,7 @@ extern const int BAD_ARGUMENTS;
 namespace Setting
 {
 extern const SettingsTimezone iceberg_timezone_for_timestamptz;
+extern const SettingsBool allow_experimental_geo_types_in_iceberg;
 }
 
 namespace
@@ -119,16 +120,73 @@ bool equals(const T & first, const T & second)
     return first_string_stream.str() == second_string_stream.str();
 }
 
-bool operator==(const Poco::JSON::Array & first, const Poco::JSON::Array & second)
+bool schemasAreIdentical(const Poco::JSON::Object & first, const Poco::JSON::Object & second, const std::unordered_map<String, String> & type_mapping);
+
+bool schemaFieldsAreStructurallyIdentical(const Poco::JSON::Object & first, const Poco::JSON::Object & second, const std::unordered_map<String, String> & type_mapping)
 {
-    return equals(first, second);
+    static constexpr const char * structural_keys[] = {f_id, f_name, f_required, f_type};
+    for (const char * key : structural_keys)
+    {
+        const bool first_has = first.has(key);
+        const bool second_has = second.has(key);
+        if (first_has != second_has)
+            return false;
+        if (!first_has)
+            continue;
+
+        if (key == f_type && first.isObject(key) && second.isObject(key))
+        {
+            const auto first_type = first.getObject(key);
+            const auto second_type = second.getObject(key);
+            if (first_type->isArray(f_fields) || second_type->isArray(f_fields))
+            {
+                if (!schemasAreIdentical(*first_type, *second_type, type_mapping))
+                    return false;
+                continue;
+            }
+        }
+
+        auto key_first = first.get(key);
+        auto key_second = second.get(key);
+        if (key == f_type && key_first.isString())
+        {
+            for (const auto & [prefix, mapped] : type_mapping)
+                if (key_first.toString().starts_with(prefix))
+                    key_first = mapped;
+        }
+        if (key == f_type && key_second.isString())
+        {
+            for (const auto & [prefix, mapped] : type_mapping)
+                if (key_second.toString().starts_with(prefix))
+                    key_second = mapped;
+        }
+
+        Poco::JSON::Object wrapper_first;
+        wrapper_first.set(key, key_first);
+        Poco::JSON::Object wrapper_second;
+        wrapper_second.set(key, key_second);
+        if (!equals(wrapper_first, wrapper_second))
+            return false;
+    }
+    return true;
 }
 
-bool schemasAreIdentical(const Poco::JSON::Object & first, const Poco::JSON::Object & second)
+bool schemasAreIdentical(const Poco::JSON::Object & first, const Poco::JSON::Object & second, const std::unordered_map<String, String> & type_mapping)
 {
     if (!first.isArray(f_fields) || !second.isArray(f_fields))
         return false;
-    return *(first.getArray(f_fields)) == *(second.getArray(f_fields));
+    const auto first_fields = first.getArray(f_fields);
+    const auto second_fields = second.getArray(f_fields);
+    if (first_fields->size() != second_fields->size())
+        return false;
+    for (UInt32 i = 0; i != first_fields->size(); ++i)
+    {
+        const auto first_field = first_fields->getObject(i);
+        const auto second_field = second_fields->getObject(i);
+        if (!first_field || !second_field || !schemaFieldsAreStructurallyIdentical(*first_field, *second_field, type_mapping))
+            return false;
+    }
+    return true;
 }
 
 std::pair<size_t, size_t> parseDecimal(const String & type_name)
@@ -160,7 +218,13 @@ void IcebergSchemaProcessor::addIcebergTableSchema(Poco::JSON::Object::Ptr schem
     if (iceberg_table_schemas_by_ids.contains(schema_id))
     {
         chassert(clickhouse_table_schemas_by_ids.contains(schema_id));
-        chassert(schemasAreIdentical(*iceberg_table_schemas_by_ids.at(schema_id), *schema_ptr));
+        std::unordered_map<String, String> type_mapping;
+        if (context_->getSettingsRef()[Setting::allow_experimental_geo_types_in_iceberg])
+        {
+            type_mapping[f_geography] = f_binary;
+            type_mapping[f_geometry] = f_binary;
+        }
+        chassert(schemasAreIdentical(*iceberg_table_schemas_by_ids.at(schema_id), *schema_ptr, type_mapping));
     }
     else
     {
@@ -253,6 +317,15 @@ DataTypePtr IcebergSchemaProcessor::getSimpleType(const String & type_name, Cont
     }
     if (type_name == f_string || type_name == f_binary)
         return std::make_shared<DataTypeString>();
+
+    if (type_name.starts_with(f_geometry) || type_name.starts_with(f_geography))
+    {
+        if (context_->getSettingsRef()[Setting::allow_experimental_geo_types_in_iceberg])
+            return DataTypeFactory::instance().get("Geometry");
+        throw Exception(
+            ErrorCodes::BAD_ARGUMENTS,
+            "Using geometry/geography types is not allowed without enabled allow_experimental_geo_types_in_iceberg flag");
+    }
     if (type_name == f_uuid)
         return std::make_shared<DataTypeUUID>();
 
@@ -352,7 +425,7 @@ DataTypePtr IcebergSchemaProcessor::getFieldType(
     {
         const String & type_name = type.extract<String>();
         auto data_type = getSimpleType(type_name, context_);
-        return required ? data_type : makeNullable(data_type);
+        return required || !data_type->canBeInsideNullable() ? data_type : makeNullable(data_type);
     }
 
     throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unexpected 'type' field: {}", type.toString());

src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp

@@ -8,6 +8,7 @@
 #include <Core/Settings.h>
 #include <Core/TypeId.h>
 #include <DataTypes/DataTypeArray.h>
+#include <DataTypes/DataTypeCustom.h>
 #include <DataTypes/DataTypesDecimal.h>
 #include <DataTypes/DataTypeMap.h>
 #include <DataTypes/DataTypeNullable.h>
@@ -564,6 +565,12 @@ std::pair<Poco::Dynamic::Var, bool> getIcebergType(DataTypePtr type, Int32 & ite
             auto type_nullable = std::static_pointer_cast<const DataTypeNullable>(type);
             return {getIcebergType(type_nullable->getNestedType(), iter).first, false};
         }
+        case TypeIndex::Variant:
+        {
+            if (type->getCustomName() && type->getCustomName()->getName() == "Geometry")
+                return {Iceberg::f_geometry, false};
+            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName());
+        }
         default:
             throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName());
     }

tests/integration/helpers/iceberg_utils.py

@@ -448,6 +448,7 @@ def create_iceberg_table(
     format="Parquet",
     order_by="",
     object_storage_cluster=False,
+    settings=None,
     **kwargs,
 ):
     if 'output_format_parquet_use_custom_encoder' in kwargs:
@@ -466,7 +467,7 @@ def create_iceberg_table(
                 run_on_cluster=run_on_cluster,
                 object_storage_cluster=object_storage_cluster,
                 **kwargs),
-            settings={"output_format_parquet_use_custom_encoder" : 0, "output_format_parquet_parallel_encoding" : 0}
+            settings={"output_format_parquet_use_custom_encoder" : 0, "output_format_parquet_parallel_encoding" : 0, **(settings or {})}
         )
     else:
         node.query(
@@ -484,6 +485,7 @@ def create_iceberg_table(
                 run_on_cluster=run_on_cluster,
                 object_storage_cluster=object_storage_cluster,
                 **kwargs),
+            settings=settings,
         )