From 10884291c13f1d35e9dd77cf689c11b71a77b278 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Thu, 21 May 2026 11:13:59 +0200 Subject: [PATCH 1/3] Cherry-pick of https://github.com/Altinity/ClickHouse/pull/1761 with unresolved conflict markers (resolution in next commit) --- Original cherry-pick message follows: Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support Support for 'time' type in Iceberg # Conflicts: # src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp # src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp # tests/integration/test_database_iceberg/test.py --- .../Formats/Impl/AvroRowInputFormat.cpp | 14 +- .../Formats/Impl/Parquet/PrepareForWrite.cpp | 50 +++++++ src/Processors/Formats/Impl/Parquet/Write.cpp | 49 +++++- .../DataLakes/Iceberg/Constant.h | 1 + .../DataLakes/Iceberg/IcebergWrites.cpp | 91 +++++++++++- .../DataLakes/Iceberg/SchemaProcessor.cpp | 2 +- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 43 +++++- .../ObjectStorage/DataLakes/Iceberg/Utils.h | 1 + .../integration/test_database_iceberg/test.py | 140 +++++++++++++++++- .../test_write_time.py | 41 +++++ 10 files changed, 422 insertions(+), 10 deletions(-) create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_write_time.py diff --git a/src/Processors/Formats/Impl/AvroRowInputFormat.cpp b/src/Processors/Formats/Impl/AvroRowInputFormat.cpp index 2c0ddb2b8f19..55bff27fd119 100644 --- a/src/Processors/Formats/Impl/AvroRowInputFormat.cpp +++ b/src/Processors/Formats/Impl/AvroRowInputFormat.cpp @@ -173,6 +173,9 @@ static void insertNumber(IColumn & column, WhichDataType type, T value) case TypeIndex::DateTime64: convertAndInsert(assert_cast &>(column), value, type.idx); break; + case TypeIndex::Time64: + assert_cast &>(column).insertValue(static_cast(value)); + break; case TypeIndex::IPv4: convertAndInsert(assert_cast(column), value, type.idx); break; @@ -331,6 +334,8 @@ AvroDeserializer::DeserializeFn AvroDeserializer::createDeserializeFn(const avro return createDecimalDeserializeFn(root_node, target_type, false); if (target.isDateTime64()) return createDecimalDeserializeFn(root_node, target_type, false); + if (target.isTime64()) + return createDecimalDeserializeFn(root_node, target_type, false); break; case avro::AVRO_INT: if (target_type->isValueRepresentedByNumber()) @@ -1258,8 +1263,11 @@ DataTypePtr AvroSchemaReader::avroNodeToDataTypeImpl(const avro::NodePtr & node, { case avro::Type::AVRO_INT: { - if (node->logicalType().type() == avro::LogicalType::DATE) + auto logical_type = node->logicalType(); + if (logical_type.type() == avro::LogicalType::DATE) return {std::make_shared()}; + if (logical_type.type() == avro::LogicalType::TIME_MILLIS) + return {std::make_shared(3)}; return {std::make_shared()}; } @@ -1270,6 +1278,10 @@ DataTypePtr AvroSchemaReader::avroNodeToDataTypeImpl(const avro::NodePtr & node, return {std::make_shared(3)}; if (logical_type.type() == avro::LogicalType::TIMESTAMP_MICROS) return {std::make_shared(6)}; + if (logical_type.type() == avro::LogicalType::TIME_MILLIS) + return {std::make_shared(3)}; + if (logical_type.type() == avro::LogicalType::TIME_MICROS) + return {std::make_shared(6)}; return std::make_shared(); } diff --git a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp index ec85e8269cbe..66dd7e013005 100644 --- a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp +++ b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -478,6 +479,55 @@ void preparePrimitiveColumn(ColumnPtr column, DataTypePtr type, const std::strin case TypeIndex::Decimal128: decimal(16, getDecimalPrecision(*type), getDecimalScale(*type)); break; case TypeIndex::Decimal256: decimal(32, getDecimalPrecision(*type), getDecimalScale(*type)); break; + case TypeIndex::Time: + { + parq::TimeUnit unit; + unit.__set_MICROS({}); + + parq::TimeType tt; + tt.__set_isAdjustedToUTC(false); + tt.__set_unit(unit); + + parq::LogicalType t; + t.__set_TIME(tt); + types(T::INT64, parq::ConvertedType::TIME_MICROS, t); + state.datetime_multiplier = 1'000'000; + break; + } + + case TypeIndex::Time64: + { + std::optional converted; + parq::TimeUnit unit; + const auto & dt = assert_cast(*type); + UInt32 scale = dt.getScale(); + UInt32 converted_scale; + if (scale <= 6) + { + converted = parq::ConvertedType::TIME_MICROS; + unit.__set_MICROS({}); + converted_scale = 6; + } + else if (scale <= 9) + { + unit.__set_NANOS({}); + converted_scale = 9; + } + else + { + throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected Time64 scale: {}", scale); + } + + parq::TimeType tt; + tt.__set_isAdjustedToUTC(false); + tt.__set_unit(unit); + parq::LogicalType t; + t.__set_TIME(tt); + types(T::INT64, converted, t); + state.datetime_multiplier = DataTypeTime64::getScaleMultiplier(converted_scale - scale); + break; + } + default: throw Exception(ErrorCodes::UNKNOWN_TYPE, "Internal type '{}' of column '{}' is not supported for conversion into Parquet data format.", type->getFamilyName(), name); } diff --git a/src/Processors/Formats/Impl/Parquet/Write.cpp b/src/Processors/Formats/Impl/Parquet/Write.cpp index 94ee3d8de462..a808c549abf1 100644 --- a/src/Processors/Formats/Impl/Parquet/Write.cpp +++ b/src/Processors/Formats/Impl/Parquet/Write.cpp @@ -356,16 +356,17 @@ struct ConverterNumeric } }; -struct ConverterDateTime64WithMultiplier +template +struct ConverterTimeType64WithMultiplierImpl { using Statistics = StatisticsNumeric; - using Col = ColumnDecimal; + using Col = ColumnDecimal; const Col & column; Int64 multiplier; PODArray buf; - ConverterDateTime64WithMultiplier(const ColumnPtr & c, Int64 multiplier_) : column(assert_cast(*c)), multiplier(multiplier_) {} + ConverterTimeType64WithMultiplierImpl(const ColumnPtr & c, Int64 multiplier_) : column(assert_cast(*c)), multiplier(multiplier_) {} const Int64 * getBatch(size_t offset, size_t count) { @@ -383,6 +384,9 @@ struct ConverterDateTime64WithMultiplier } }; +using ConverterDateTime64WithMultiplier = ConverterTimeType64WithMultiplierImpl; +using ConverterTime64WithMultiplier = ConverterTimeType64WithMultiplierImpl; + /// Multiply DateTime by 1000 to get milliseconds (because Parquet doesn't support seconds). struct ConverterDateTime { @@ -403,6 +407,26 @@ struct ConverterDateTime } }; +struct ConverterTime +{ + using Statistics = StatisticsNumeric; + + using Col = ColumnVector; + const Col & column; + PODArray buf; + Int64 multiplier; + + ConverterTime(const ColumnPtr & c, Int64 multiplier_) : column(assert_cast(*c)), multiplier(multiplier_) {} + + const Int64 * getBatch(size_t offset, size_t count) + { + buf.resize(count); + for (size_t i = 0; i < count; ++i) + buf[i] = static_cast(column.getData()[offset + i]) * multiplier; + return buf.data(); + } +}; + struct ConverterString { using Statistics = StatisticsStringRef; @@ -1215,7 +1239,12 @@ void writeColumnChunkBody( N(Int16, Int32Type); break; } - case TypeIndex::Int32 : N(Int32, Int32Type); break; + case TypeIndex::Int32: + if (s.type->getTypeId() == TypeIndex::Time) + writeColumnImpl(s, options, out, ConverterTime(s.primitive_column, s.datetime_multiplier)); + else + N(Int32, Int32Type); + break; case TypeIndex::Int64 : N(Int64, Int64Type); break; case TypeIndex::UInt32: @@ -1228,7 +1257,6 @@ void writeColumnChunkBody( writeColumnImpl(s, options, out, ConverterDateTime(s.primitive_column)); } break; - #undef N case TypeIndex::Float32: @@ -1303,6 +1331,17 @@ void writeColumnChunkBody( case TypeIndex::Decimal256: D(Decimal256); break; #undef D + case TypeIndex::Time64: + if (s.datetime_multiplier == 1) + writeColumnImpl( + s, options, out, ConverterNumeric, Int64, Int64>( + s.primitive_column)); + else + writeColumnImpl( + s, options, out, ConverterTime64WithMultiplier( + s.primitive_column, s.datetime_multiplier)); + break; + default: throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected column type: {}", s.primitive_column->getFamilyName()); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h index 6495b22f7cf0..b87ff718e172 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h @@ -45,6 +45,7 @@ DEFINE_ICEBERG_FIELD(timestamptz); DEFINE_ICEBERG_FIELD(timestamp_ns); DEFINE_ICEBERG_FIELD(timestamptz_ns); DEFINE_ICEBERG_FIELD(type) +DEFINE_ICEBERG_FIELD(logicalType); /// this field has a camelCase name DEFINE_ICEBERG_FIELD(transform); DEFINE_ICEBERG_FIELD(direction); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index 9529764f1fdd..8b0ab5893695 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -129,6 +130,8 @@ bool canDumpIcebergStats(const Field & field, DataTypePtr type) case TypeIndex::Date32: case TypeIndex::Int64: case TypeIndex::DateTime64: + case TypeIndex::Time: + case TypeIndex::Time64: case TypeIndex::String: return true; default: @@ -144,6 +147,46 @@ std::vector dumpValue(T value) return bytes; } +DataTypePtr getTimeTypeOrNull(DataTypePtr type) +{ + if (type->isNullable()) + return getTimeTypeOrNull(assert_cast(type.get())->getNestedType()); + + const WhichDataType which(type); + if (which.isTime() || which.isTime64()) + return type; + + return nullptr; +} + +Int64 getTimeValueInMicroseconds(const Field & field, DataTypePtr type) +{ + if (type->isNullable()) + return getTimeValueInMicroseconds(field, assert_cast(type.get())->getNestedType()); + + const WhichDataType which(type); + if (which.isTime()) + { + if (field.getType() == Field::Types::Int64) + return field.safeGet() * 1'000'000; + if (field.getType() == Field::Types::UInt64) + return static_cast(field.safeGet()) * 1'000'000; + return static_cast(field.safeGet()) * 1'000'000; + } + + if (which.isTime64()) + { + const auto scale = getDecimalScale(*type); + if (scale > 6) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName()); + + const auto value = field.safeGet().getValue().value; + return value * DataTypeTime64::getScaleMultiplier(6 - scale).value; + } + + throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected Time or Time64, got {}", type->getName()); +} + std::vector dumpFieldToBytes(const Field & field, DataTypePtr type) { switch (type->getTypeId()) @@ -154,8 +197,12 @@ std::vector dumpFieldToBytes(const Field & field, DataTypePtr type) case TypeIndex::Date: case TypeIndex::Date32: return dumpValue(field.safeGet()); + case TypeIndex::Time: + return dumpValue(getTimeValueInMicroseconds(field, type)); case TypeIndex::Int64: return dumpValue(field.safeGet()); + case TypeIndex::Time64: + return dumpValue(getTimeValueInMicroseconds(field, type)); case TypeIndex::DateTime64: return dumpValue(field.safeGet().getValue().value); case TypeIndex::String: @@ -230,7 +277,16 @@ static void extendSchemaForPartitions( Poco::JSON::Object::Ptr field = new Poco::JSON::Object; field->set(Iceberg::f_field_id, 1000 + i); field->set(Iceberg::f_name, partition_columns[i]); - field->set(Iceberg::f_type, getAvroType(partition_types[i])); + auto logical_type = getAvroLogicalType(partition_types[i]); + if (!logical_type.isEmpty()) + { + Poco::JSON::Object::Ptr type_field = new Poco::JSON::Object; + type_field->set(Iceberg::f_type, getAvroType(partition_types[i])); + type_field->set(Iceberg::f_logicalType, logical_type); + field->set(Iceberg::f_type, type_field); + } + else + field->set(Iceberg::f_type, getAvroType(partition_types[i])); partition_fields->add(field); } @@ -390,9 +446,21 @@ void generateManifestFile( avro::GenericRecord & partition_record = data_file.field("partition").value(); for (size_t i = 0; i < partition_columns.size(); ++i) { +<<<<<<< HEAD /// Build the Avro datum that holds the actual partition value (without /// the surrounding union). Throws on an unsupported value type. auto make_value_datum = [&]() -> avro::GenericDatum +======= + auto partition_time_type = getTimeTypeOrNull(partition_types[i]); + if (!partition_values[i].isNull() && partition_time_type) + { + partition_record.field(partition_columns[i]) = + avro::GenericDatum(getTimeValueInMicroseconds(partition_values[i], partition_types[i])); + continue; + } + + switch (partition_values[i].getType()) +>>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) { switch (partition_values[i].getType()) { @@ -418,6 +486,7 @@ void generateManifestFile( const bool is_nullable_partition = partition_types[i]->isNullable(); const bool is_null_value = partition_values[i].getType() == Field::Types::Null; +<<<<<<< HEAD if (is_nullable_partition) { /// Nullable partition columns are encoded as Avro `["null", T]` @@ -426,6 +495,26 @@ void generateManifestFile( /// silently written as 0 because the schema was non-nullable. size_t field_index = 0; if (!partition_record.schema()->nameIndex(partition_columns[i], field_index)) +======= + case Field::Types::Float64: + partition_record.field(partition_columns[i]) = + avro::GenericDatum(partition_values[i].safeGet()); + break; + + case Field::Types::Decimal32: + partition_record.field(partition_columns[i]) = + avro::GenericDatum(partition_values[i].safeGet().getValue()); + break; + + case Field::Types::Decimal64: + partition_record.field(partition_columns[i]) = + avro::GenericDatum(partition_values[i].safeGet().getValue()); + break; + case Field::Types::Null: + break; + + default: +>>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) throw Exception( ErrorCodes::LOGICAL_ERROR, "Partition field {} not found in manifest schema", diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp index 8c6fd9ee22a2..0a88d4ce665b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp @@ -308,7 +308,7 @@ DataTypePtr IcebergSchemaProcessor::getSimpleType(const String & type_name, bool if (type_name == f_date) return std::make_shared(); if (type_name == f_time) - return std::make_shared(); + return std::make_shared(6); if (type_name == f_timestamp) return std::make_shared(6); if (type_name == f_timestamptz) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 8368128f33ac..e9f95970bc71 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -8,7 +8,11 @@ #include #include #include +<<<<<<< HEAD #include +======= +#include +>>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) #include #include #include @@ -526,6 +530,10 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite return {"timestamp", true}; case TypeIndex::Time: return {"time", true}; + case TypeIndex::Time64: + if (getDecimalScale(*type) <= 6) + return {"time", true}; + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName()); case TypeIndex::String: return {"string", true}; case TypeIndex::UUID: @@ -609,13 +617,21 @@ Poco::Dynamic::Var getAvroType(DataTypePtr type) case TypeIndex::Int32: case TypeIndex::Date: case TypeIndex::Date32: - case TypeIndex::Time: return "int"; case TypeIndex::UInt64: case TypeIndex::Int64: case TypeIndex::DateTime: case TypeIndex::DateTime64: + case TypeIndex::Time: return "long"; + case TypeIndex::Time64: + { + auto scale = getDecimalScale(*type); + if (scale <= 6) + return "long"; + else + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName()); + } case TypeIndex::Float32: return "float"; case TypeIndex::Float64: @@ -639,7 +655,32 @@ Poco::Dynamic::Var getAvroType(DataTypePtr type) } } +<<<<<<< HEAD static Poco::JSON::Object::Ptr getPartitionField( +======= +Poco::Dynamic::Var getAvroLogicalType(DataTypePtr type) +{ + if (type->isNullable()) + { + auto type_nullable = std::static_pointer_cast(type); + return getAvroLogicalType(type_nullable->getNestedType()); + } + + const WhichDataType which(type); + if (which.isTime()) + return "time-micros"; + if (which.isTime64()) + { + auto scale = getDecimalScale(*type); + if (scale <= 6) + return "time-micros"; + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported time precision for avro {}({})", type->getName(), scale); + } + return Poco::Dynamic::Var(); +} + +Poco::JSON::Object::Ptr getPartitionField( +>>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) ASTPtr partition_by_element, const std::unordered_map & column_name_to_source_id, Int32 & partition_iter) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h index 43d2c040ad59..e54a15f5e1e2 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h @@ -74,6 +74,7 @@ Poco::JSON::Object::Ptr getMetadataJSONObject( std::pair getIcebergType(DataTypePtr type, Int32 & iter); Poco::Dynamic::Var getAvroType(DataTypePtr type); +Poco::Dynamic::Var getAvroLogicalType(DataTypePtr type); /// Spec: https://iceberg.apache.org/spec/?h=metadata.json#table-metadata-fields std::pair createEmptyMetadataFile( diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 31ec8882a357..565f29b050e9 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3,8 +3,12 @@ import random import time import uuid +<<<<<<< HEAD from concurrent.futures import ThreadPoolExecutor from datetime import datetime +======= +from datetime import datetime, timedelta, time as dtime +>>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) import pyarrow as pa import pytest @@ -21,7 +25,8 @@ StringType, StructType, TimestampType, - TimestamptzType + TimestamptzType, + TimeType, ) from helpers.cluster import ClickHouseCluster @@ -1238,3 +1243,136 @@ def test_iceberg_file_progress_callback(started_cluster): f"`IcebergIterator::next` did not invoke the file-progress callback " f"(regression of PR #105413 wiring)." ) +<<<<<<< HEAD +======= + + assert res == "Jack\tBlack\nJohn\tSilver\n" + + res = node.query( + f""" + SELECT name + FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` + WHERE tag in ( + SELECT id + FROM `{table_name_local}` + ) + ORDER BY ALL + SETTINGS + object_storage_cluster='cluster_simple', + object_storage_cluster_join_mode='local' + """ + ) + + assert res == "Jack\nJohn\n" + + res = node.query( + f""" + SELECT t1.name,t2.second_name + FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` AS t1 + CROSS JOIN `{table_name_local}` AS t2 + WHERE t1.tag < 10 AND t2.id < 20 + ORDER BY ALL + SETTINGS + object_storage_cluster='cluster_simple', + object_storage_cluster_join_mode='local' + """ + ) + + assert res == "Jack\tBlack\nJack\tSilver\nJohn\tBlack\nJohn\tSilver\n" + + +def test_partitioning_by_time(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_partitioning_by_time_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + namespace = f"{root_namespace}.A" + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(namespace) + + schema = Schema( + NestedField( + field_id=1, + name="key", + field_type=TimeType(), + required=False + ), + NestedField( + field_id=2, + name="value", + field_type=StringType(), + required=False, + ), + ) + + partition_spec = PartitionSpec( + PartitionField( + source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key" + ) + ) + + table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec) + data = [{"key": dtime(12,0,0), "value": "test1"}, + {"key": dtime(13,0,0), "value": "test2"}, + {"key": dtime(14,0,0), "value": "test3"}, + ] + df = pa.Table.from_pylist(data) + table.append(df) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` ORDER BY key") == "12:00:00.000000\ttest1\n13:00:00.000000\ttest2\n14:00:00.000000\ttest3\n" + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key = '13:00:00.000000' ORDER BY key") == "13:00:00.000000\ttest2\n" + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key >= '13:00:00.000000' ORDER BY key") == "13:00:00.000000\ttest2\n14:00:00.000000\ttest3\n" + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key <= '13:00:00.000000' ORDER BY key") == "12:00:00.000000\ttest1\n13:00:00.000000\ttest2\n" + + +def test_partitioning_by_string(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_partitioning_by_string_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + namespace = f"{root_namespace}.A" + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(namespace) + + schema = Schema( + NestedField( + field_id=1, + name="key", + field_type=StringType(), + required=False + ), + NestedField( + field_id=2, + name="value", + field_type=StringType(), + required=False, + ), + NestedField( + field_id=3, + name="time_value", + field_type=TimeType(), + required=False, + ), + ) + + partition_spec = PartitionSpec( + PartitionField( + source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key" + ) + ) + + table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec) + data = [{"key": "a:b,c[d=e/f%g?h", "value": "test", "time_value": dtime(12,0,0)}] + df = pa.Table.from_pylist(data) + table.append(df) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}`") == "a:b,c[d=e/f%g?h\ttest\t12:00:00.000000\n" +>>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) diff --git a/tests/integration/test_storage_iceberg_no_spark/test_write_time.py b/tests/integration/test_storage_iceberg_no_spark/test_write_time.py new file mode 100644 index 000000000000..c4562fede71e --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_write_time.py @@ -0,0 +1,41 @@ +import pytest +from helpers.config_cluster import minio_secret_key, minio_access_key +from helpers.iceberg_utils import get_uuid_str + + +@pytest.mark.parametrize( + "time_type", + [ + "Time", + "Time64(0)", + "Time64(3)", + "Time64(6)", + "Nullable(Time)", + "Nullable(Time64(0))", + "Nullable(Time64(3))", + "Nullable(Time64(6))", + ], +) +def test_write_time(started_cluster_iceberg_no_spark, time_type): + node = started_cluster_iceberg_no_spark.instances["node1"] + + TABLE_NAME = "test_partitioning_by_time_" + get_uuid_str() + + node.query( + f"CREATE TABLE `{TABLE_NAME}` (key {time_type}, value {time_type}, comment String) ENGINE = IcebergLocal('/var/lib/clickhouse/user_files/iceberg_data/default/{TABLE_NAME}', 'Parquet') PARTITION BY key", + settings={"allow_experimental_insert_into_iceberg": 1, 'write_full_path_in_iceberg_metadata': 1} + ) + node.query( + f"INSERT INTO `{TABLE_NAME}` VALUES ('12:00:00', '12:10:00', 'test1'), ('13:00:00', '13:10:00', 'test2'), ('14:00:00', '14:10:00', 'test3');", + settings={"allow_experimental_insert_into_iceberg": 1, 'write_full_path_in_iceberg_metadata': 1} + ) + + assert node.query(f"SELECT * FROM `{TABLE_NAME}` ORDER BY key") == "12:00:00.000000\t12:10:00.000000\ttest1\n13:00:00.000000\t13:10:00.000000\ttest2\n14:00:00.000000\t14:10:00.000000\ttest3\n" + # Partition pruning + assert node.query(f"SELECT * FROM `{TABLE_NAME}` WHERE key = '13:00:00.000000' ORDER BY key") == "13:00:00.000000\t13:10:00.000000\ttest2\n" + assert node.query(f"SELECT * FROM `{TABLE_NAME}` WHERE key >= '13:00:00.000000' ORDER BY key") == "13:00:00.000000\t13:10:00.000000\ttest2\n14:00:00.000000\t14:10:00.000000\ttest3\n" + assert node.query(f"SELECT * FROM `{TABLE_NAME}` WHERE key <= '13:00:00.000000' ORDER BY key") == "12:00:00.000000\t12:10:00.000000\ttest1\n13:00:00.000000\t13:10:00.000000\ttest2\n" + # Min/Max pruning + assert node.query(f"SELECT * FROM `{TABLE_NAME}` WHERE value = '13:10:00.000000' ORDER BY key") == "13:00:00.000000\t13:10:00.000000\ttest2\n" + assert node.query(f"SELECT * FROM `{TABLE_NAME}` WHERE value >= '13:10:00.000000' ORDER BY key") == "13:00:00.000000\t13:10:00.000000\ttest2\n14:00:00.000000\t14:10:00.000000\ttest3\n" + assert node.query(f"SELECT * FROM `{TABLE_NAME}` WHERE value <= '13:10:00.000000' ORDER BY key") == "12:00:00.000000\t12:10:00.000000\ttest1\n13:00:00.000000\t13:10:00.000000\ttest2\n" From d49b6574762fae6d73d9e35b3b71667052cfa4b2 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Wed, 29 Jul 2026 20:32:05 +0200 Subject: [PATCH 2/3] Resolve conflicts in cherry-pick of #1761 Kept antalya-26.6's nullable-partition Avro union handling in generateManifestFile and routed the PR's Time/Time64 partition-value encoding through the base branch's `make_value_datum` lambda. Kept `getAvroLogicalType` from the PR and antalya-26.6's `static` linkage of `getPartitionField`; dropped test context belonging to `test_cluster_joins`, which does not exist on antalya-26.6. --- .../DataLakes/Iceberg/IcebergWrites.cpp | 66 ++++++++----------- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 9 +-- .../integration/test_database_iceberg/test.py | 43 +----------- 3 files changed, 29 insertions(+), 89 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index 8b0ab5893695..5bc1fe22fbce 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -272,21 +272,38 @@ static void extendSchemaForPartitions( const std::vector & partition_types) { Poco::JSON::Array::Ptr partition_fields = new Poco::JSON::Array; + + /// Types that need a `logicalType` annotation (Time/Time64 -> "time-micros") are + /// written as `{"type": T, "logicalType": ...}`. The annotation must decorate the + /// concrete Avro type, so for a Nullable partition column it goes inside the + /// `["null", T]` union branch, not on the union itself (an annotated union is not + /// a valid Avro schema and makes the manifest schema fail to compile). + auto make_annotated_type = [](DataTypePtr type) -> Poco::Dynamic::Var + { + auto logical_type = getAvroLogicalType(type); + if (logical_type.isEmpty()) + return getAvroType(type); + + Poco::JSON::Object::Ptr type_field = new Poco::JSON::Object; + type_field->set(Iceberg::f_type, getAvroType(type)); + type_field->set(Iceberg::f_logicalType, logical_type); + return type_field; + }; + for (size_t i = 0; i < partition_columns.size(); ++i) { Poco::JSON::Object::Ptr field = new Poco::JSON::Object; field->set(Iceberg::f_field_id, 1000 + i); field->set(Iceberg::f_name, partition_columns[i]); - auto logical_type = getAvroLogicalType(partition_types[i]); - if (!logical_type.isEmpty()) + if (partition_types[i]->isNullable()) { - Poco::JSON::Object::Ptr type_field = new Poco::JSON::Object; - type_field->set(Iceberg::f_type, getAvroType(partition_types[i])); - type_field->set(Iceberg::f_logicalType, logical_type); - field->set(Iceberg::f_type, type_field); + Poco::JSON::Array::Ptr union_array = new Poco::JSON::Array; + union_array->add("null"); + union_array->add(make_annotated_type(removeNullable(partition_types[i]))); + field->set(Iceberg::f_type, union_array); } else - field->set(Iceberg::f_type, getAvroType(partition_types[i])); + field->set(Iceberg::f_type, make_annotated_type(partition_types[i])); partition_fields->add(field); } @@ -446,22 +463,14 @@ void generateManifestFile( avro::GenericRecord & partition_record = data_file.field("partition").value(); for (size_t i = 0; i < partition_columns.size(); ++i) { -<<<<<<< HEAD /// Build the Avro datum that holds the actual partition value (without /// the surrounding union). Throws on an unsupported value type. auto make_value_datum = [&]() -> avro::GenericDatum -======= - auto partition_time_type = getTimeTypeOrNull(partition_types[i]); - if (!partition_values[i].isNull() && partition_time_type) { - partition_record.field(partition_columns[i]) = - avro::GenericDatum(getTimeValueInMicroseconds(partition_values[i], partition_types[i])); - continue; - } + auto partition_time_type = getTimeTypeOrNull(partition_types[i]); + if (partition_time_type) + return avro::GenericDatum(getTimeValueInMicroseconds(partition_values[i], partition_types[i])); - switch (partition_values[i].getType()) ->>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) - { switch (partition_values[i].getType()) { case Field::Types::Int64: @@ -486,7 +495,6 @@ void generateManifestFile( const bool is_nullable_partition = partition_types[i]->isNullable(); const bool is_null_value = partition_values[i].getType() == Field::Types::Null; -<<<<<<< HEAD if (is_nullable_partition) { /// Nullable partition columns are encoded as Avro `["null", T]` @@ -495,26 +503,6 @@ void generateManifestFile( /// silently written as 0 because the schema was non-nullable. size_t field_index = 0; if (!partition_record.schema()->nameIndex(partition_columns[i], field_index)) -======= - case Field::Types::Float64: - partition_record.field(partition_columns[i]) = - avro::GenericDatum(partition_values[i].safeGet()); - break; - - case Field::Types::Decimal32: - partition_record.field(partition_columns[i]) = - avro::GenericDatum(partition_values[i].safeGet().getValue()); - break; - - case Field::Types::Decimal64: - partition_record.field(partition_columns[i]) = - avro::GenericDatum(partition_values[i].safeGet().getValue()); - break; - case Field::Types::Null: - break; - - default: ->>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) throw Exception( ErrorCodes::LOGICAL_ERROR, "Partition field {} not found in manifest schema", diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index e9f95970bc71..180e79207c92 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -8,11 +8,8 @@ #include #include #include -<<<<<<< HEAD #include -======= #include ->>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) #include #include #include @@ -655,9 +652,6 @@ Poco::Dynamic::Var getAvroType(DataTypePtr type) } } -<<<<<<< HEAD -static Poco::JSON::Object::Ptr getPartitionField( -======= Poco::Dynamic::Var getAvroLogicalType(DataTypePtr type) { if (type->isNullable()) @@ -679,8 +673,7 @@ Poco::Dynamic::Var getAvroLogicalType(DataTypePtr type) return Poco::Dynamic::Var(); } -Poco::JSON::Object::Ptr getPartitionField( ->>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) +static Poco::JSON::Object::Ptr getPartitionField( ASTPtr partition_by_element, const std::unordered_map & column_name_to_source_id, Int32 & partition_iter) diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 565f29b050e9..5d233f89d7ca 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3,12 +3,8 @@ import random import time import uuid -<<<<<<< HEAD from concurrent.futures import ThreadPoolExecutor -from datetime import datetime -======= -from datetime import datetime, timedelta, time as dtime ->>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) +from datetime import datetime, time as dtime import pyarrow as pa import pytest @@ -1243,42 +1239,6 @@ def test_iceberg_file_progress_callback(started_cluster): f"`IcebergIterator::next` did not invoke the file-progress callback " f"(regression of PR #105413 wiring)." ) -<<<<<<< HEAD -======= - - assert res == "Jack\tBlack\nJohn\tSilver\n" - - res = node.query( - f""" - SELECT name - FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` - WHERE tag in ( - SELECT id - FROM `{table_name_local}` - ) - ORDER BY ALL - SETTINGS - object_storage_cluster='cluster_simple', - object_storage_cluster_join_mode='local' - """ - ) - - assert res == "Jack\nJohn\n" - - res = node.query( - f""" - SELECT t1.name,t2.second_name - FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` AS t1 - CROSS JOIN `{table_name_local}` AS t2 - WHERE t1.tag < 10 AND t2.id < 20 - ORDER BY ALL - SETTINGS - object_storage_cluster='cluster_simple', - object_storage_cluster_join_mode='local' - """ - ) - - assert res == "Jack\tBlack\nJack\tSilver\nJohn\tBlack\nJohn\tSilver\n" def test_partitioning_by_time(started_cluster): @@ -1375,4 +1335,3 @@ def test_partitioning_by_string(started_cluster): create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}`") == "a:b,c[d=e/f%g?h\ttest\t12:00:00.000000\n" ->>>>>>> b51229ea981 (Merge pull request #1761 from Altinity/bugfix/antalya-26.3/1535_time_type_write_support) From 24d1ace0f7a6255f70a0c218d64312026c29a7a2 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 30 Jul 2026 17:30:17 +0200 Subject: [PATCH 3/3] Fix tidy build --- src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp index 66dd7e013005..7d657c4548dd 100644 --- a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp +++ b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp @@ -501,7 +501,7 @@ void preparePrimitiveColumn(ColumnPtr column, DataTypePtr type, const std::strin parq::TimeUnit unit; const auto & dt = assert_cast(*type); UInt32 scale = dt.getScale(); - UInt32 converted_scale; + UInt32 converted_scale = 0; if (scale <= 6) { converted = parq::ConvertedType::TIME_MICROS;