diff --git a/ci/docker/integration/runner/Dockerfile b/ci/docker/integration/runner/Dockerfile index 5923adc6cda5..21eb61f3e173 100644 --- a/ci/docker/integration/runner/Dockerfile +++ b/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/ \; diff --git a/ci/docker/integration/runner/requirements.txt b/ci/docker/integration/runner/requirements.txt index 0c0ff446c616..c335241fc0e2 100644 --- a/ci/docker/integration/runner/requirements.txt +++ b/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 diff --git a/ci/jobs/scripts/check_style/experimental_settings_ignore.txt b/ci/jobs/scripts/check_style/experimental_settings_ignore.txt index fc0de187e033..e9a3f2033cc2 100644 --- a/ci/jobs/scripts/check_style/experimental_settings_ignore.txt +++ b/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 diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index a11dc2ef96ac..2ce60c9310fd 100644 --- a/src/Core/Settings.cpp +++ b/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. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index ff4dea11f189..b3ceeb2e27bb 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/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"}, diff --git a/src/Processors/Formats/Impl/ArrowColumnToCHColumn.cpp b/src/Processors/Formats/Impl/ArrowColumnToCHColumn.cpp index dfc08faeceff..65751b6e34f9 100644 --- a/src/Processors/Formats/Impl/ArrowColumnToCHColumn.cpp +++ b/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_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(make_nullable_if_low_cardinality ? makeNullable(dict_column.type) : dict_column.type); + auto lc_type = std::make_shared(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(tmp_lc_column.get())->getDictionaryPtr()); dynamic_cast(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(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(make_nullable_if_low_cardinality ? makeNullable(dict_info.values->type) : dict_info.values->type); + auto lc_type = std::make_shared(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> & parquet_columns_to_clickhouse, const std::optional> & 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 && diff --git a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp index 803c8a2fb55c..95d3229ec5b8 100644 --- a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp +++ b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp @@ -10,6 +10,8 @@ #include #include #include +#include +#include #include #include #include @@ -23,6 +25,7 @@ #include #include #include +#include /// 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(*column); + const auto & variant_type = assert_cast(*type); + const auto & variants = variant_type.getVariants(); + + std::vector> 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(); + else if (variant_name == WKBLineStringTransform::name || variant_name == "Ring") + transforms[i] = std::make_shared(); + else if (variant_name == WKBPolygonTransform::name) + transforms[i] = std::make_shared(); + else if (variant_name == WKBMultiLineStringTransform::name) + transforms[i] = std::make_shared(); + else if (variant_name == WKBMultiPolygonTransform::name) + transforms[i] = std::make_shared(); + } + + 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(std::make_shared()); + return; + } + std::shared_ptr transform; if (type->getCustomName()->getName() == WKBPointTransform::name) transform = std::make_shared(); diff --git a/src/Processors/Formats/Impl/Parquet/SchemaConverter.cpp b/src/Processors/Formats/Impl/Parquet/SchemaConverter.cpp index 910e3b06c8db..20f387aea670 100644 --- a/src/Processors/Formats/Impl/Parquet/SchemaConverter.cpp +++ b/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(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) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h index f923ff6a3a7e..a063af59ebcb 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h +++ b/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); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index 1e84302758c4..9db5e72d7ac7 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -163,6 +163,9 @@ std::optional 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()); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp index 1ebdf7a52305..75f87f4cc181 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp +++ b/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); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp index dda2050c9024..307117d792b9 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp +++ b/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 & type_mapping); + +bool schemaFieldsAreStructurallyIdentical(const Poco::JSON::Object & first, const Poco::JSON::Object & second, const std::unordered_map & 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 & 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 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 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(); + + 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(); @@ -352,7 +425,7 @@ DataTypePtr IcebergSchemaProcessor::getFieldType( { const String & type_name = type.extract(); 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()); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 8564eebda3ef..0418d41f9b9b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -564,6 +565,12 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite auto type_nullable = std::static_pointer_cast(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()); } diff --git a/tests/integration/helpers/iceberg_utils.py b/tests/integration/helpers/iceberg_utils.py index 7548f61d7a48..2fac9e1a1653 100644 --- a/tests/integration/helpers/iceberg_utils.py +++ b/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, ) diff --git a/tests/integration/test_storage_iceberg_with_spark/conftest.py b/tests/integration/test_storage_iceberg_with_spark/conftest.py index 08cdda37e98e..e0e76b9e9228 100644 --- a/tests/integration/test_storage_iceberg_with_spark/conftest.py +++ b/tests/integration/test_storage_iceberg_with_spark/conftest.py @@ -32,7 +32,8 @@ def get_spark(log_dir=None): .config("spark.sql.catalog.spark_catalog.warehouse", "/var/lib/clickhouse/user_files/iceberg_data") .config( "spark.sql.extensions", - "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions", + "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions," + "org.apache.sedona.spark.SedonaSparkSessionExtension", ) .master("local") )