Fix iceberg partition pruning by nanoseconds

docs/en/engines/table-engines/integrations/iceberg.md

@@ -98,6 +98,8 @@ The following table shows how Iceberg data types are mapped to ClickHouse data t
 | `fixed(N)` | `FixedString(N)` | |
 | `decimal(P, S)` | `Decimal(P, S)` | |
 
+Writes use the inverse mapping: `DateTime64(9)` is stored as `timestamp_ns`, and `DateTime64` with an explicit timezone and scale 9 as `timestamptz_ns`, but only when `iceberg_format_version` is 3 or higher. `DateTime` and `DateTime64` with scale 6 or less are stored as `timestamp`. Creating or altering an Iceberg table with `DateTime64` scale 9 on format version 1 or 2 raises an exception.
+
 ### Complex types {#complex-types}
 
 | Iceberg type | ClickHouse type |

src/Formats/FormatSettings.h

@@ -376,6 +376,10 @@ struct FormatSettings
         bool output_datetime_as_uint32 = false;
         bool output_date_as_uint16 = false;
         bool output_enum_as_byte_array = false;
+        /// When false (default), DateTime64 is always written as isAdjustedToUTC=true because
+        /// ClickHouse ticks are UTC. Iceberg sets this so timestamp(_ns) is false and
+        /// timestamptz(_ns) is true, matching TIMESTAMP(*, hasExplicitTimeZone()).
+        bool output_timestamp_adjusted_to_utc_from_timezone = false;
 
         bool parallel_encoding = true;
         bool write_page_index = false;

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

@@ -419,9 +419,11 @@ void preparePrimitiveColumn(ColumnPtr column, DataTypePtr type, const std::strin
             }
 
             parq::TimestampType tt;
-            /// (Shouldn't we check the DateTime64's timezone parameter here? No, the actual number
-            /// in DateTime64 column is always in UTC, regardless of the timezone parameter.)
-            tt.__set_isAdjustedToUTC(true);
+            /// ClickHouse ticks are always UTC. By default keep isAdjustedToUTC=true. Iceberg
+            /// sets output_timestamp_adjusted_to_utc_from_timezone so timestamp(_ns) is
+            /// TIMESTAMP(*, false) and timestamptz(_ns) is TIMESTAMP(*, true).
+            tt.__set_isAdjustedToUTC(
+                options.output_timestamp_adjusted_to_utc_from_timezone ? dt.hasExplicitTimeZone() : true);
             tt.__set_unit(unit);
             parq::LogicalType t;
             t.__set_TIMESTAMP(tt);

src/Processors/Formats/Impl/Parquet/Write.h

@@ -25,6 +25,7 @@ struct WriteOptions
     bool output_datetime_as_uint32 = false;
     bool output_date_as_uint16 = false;
     bool output_enum_as_byte_array = false;
+    bool output_timestamp_adjusted_to_utc_from_timezone = false;
 
     /// Note: the meaning of some compression methods here is different from
     /// wrapReadBufferWithCompressionMethod:

src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp

@@ -49,6 +49,7 @@ ParquetBlockOutputFormat::ParquetBlockOutputFormat(WriteBuffer & out_, SharedHea
     options.output_datetime_as_uint32 = format_settings.parquet.output_datetime_as_uint32;
     options.output_date_as_uint16 = format_settings.parquet.output_date_as_uint16;
     options.output_enum_as_byte_array = format_settings.parquet.output_enum_as_byte_array;
+    options.output_timestamp_adjusted_to_utc_from_timezone = format_settings.parquet.output_timestamp_adjusted_to_utc_from_timezone;
     options.data_page_size = format_settings.parquet.data_page_size;
     options.write_batch_size = format_settings.parquet.write_batch_size;
     options.write_page_index = format_settings.parquet.write_page_index;

src/Processors/tests/gtest_write_parquet_page_index.cpp

@@ -8,6 +8,8 @@
 #    include <IO/WriteBufferFromFile.h>
 #    include <Processors/Executors/CompletedPipelineExecutor.h>
 #    include <Processors/Executors/PipelineExecutor.h>
+#    include <DataTypes/DataTypeDateTime64.h>
+#    include <Processors/Formats/Impl/Parquet/Write.h>
 #    include <Processors/Formats/Impl/ParquetBlockOutputFormat.h>
 #    include <Processors/ISource.h>
 #    include <Processors/Sources/SourceFromChunks.h>
@@ -368,5 +370,26 @@ TEST(Parquet, WriteParquetPageIndexOversizedStringStats)
     ASSERT_GT(offset_index_location.value().offset, 0);
     ASSERT_GT(offset_index_location.value().length, 0);
 }
+
+TEST(ParquetWrite, DateTime64AdjustedToUTCFromTimezone)
+{
+    auto ts = std::make_shared<DataTypeDateTime64>(9);
+    auto ts_tz = std::make_shared<DataTypeDateTime64>(9, "UTC");
+    Block sample;
+    sample.insert({ts->createColumn(), ts, "ts"});
+    sample.insert({ts_tz->createColumn(), ts_tz, "ts_tz"});
+
+    Parquet::WriteOptions default_options;
+    auto default_schema = Parquet::convertSchema(sample, default_options, std::nullopt);
+    ASSERT_EQ(default_schema.size(), 3);
+    EXPECT_TRUE(default_schema[1].logicalType.TIMESTAMP.isAdjustedToUTC);
+    EXPECT_TRUE(default_schema[2].logicalType.TIMESTAMP.isAdjustedToUTC);
+
+    Parquet::WriteOptions iceberg_options;
+    iceberg_options.output_timestamp_adjusted_to_utc_from_timezone = true;
+    auto iceberg_schema = Parquet::convertSchema(sample, iceberg_options, std::nullopt);
+    EXPECT_FALSE(iceberg_schema[1].logicalType.TIMESTAMP.isAdjustedToUTC);
+    EXPECT_TRUE(iceberg_schema[2].logicalType.TIMESTAMP.isAdjustedToUTC);
+}
 }
 #endif

src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp

@@ -156,7 +156,7 @@ DataFileMetaInfo::DataFileMetaInfo(
                 {
                     auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true);
                     auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false);
-                    if (left && right)
+                    if (left && right && !DB::Range::less(*right, *left))
                         hyperrectangle = DB::Range(*left, true, *right, true);
                 }
             }

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

@@ -371,8 +371,12 @@ static void writeDataFiles(
 
         FormatFilterInfoPtr output_format_filter_info
             = std::make_shared<FormatFilterInfo>(nullptr, context, column_mapper, nullptr, nullptr);
+        auto write_format_settings = format_settings;
+        if (!write_format_settings)
+            write_format_settings = getFormatSettings(context);
+        write_format_settings->parquet.output_timestamp_adjusted_to_utc_from_timezone = true;
         auto output_format = FormatFactory::instance().getOutputFormat(
-            write_format, *write_buffer, *sample_block, context, format_settings, output_format_filter_info);
+            write_format, *write_buffer, *sample_block, context, write_format_settings, output_format_filter_info);
 
         while (true)
         {
@@ -987,7 +991,9 @@ static void writeMetadataFiles(
     auto log = getLogger("IcebergCompaction");
 
     ColumnsDescription columns_description = ColumnsDescription::fromNamesAndTypes(sample_block_->getNamesAndTypes());
-    auto [metadata_object, metadata_object_str] = createEmptyMetadataFile(table_path, columns_description, nullptr, nullptr, context);
+    const auto format_version = plan.initial_metadata_object->getValue<UInt64>(Iceberg::f_format_version);
+    auto [metadata_object, metadata_object_str] = createEmptyMetadataFile(
+        table_path, columns_description, nullptr, nullptr, context, format_version);
 
     auto current_schema_id = metadata_object->getValue<Int64>(Iceberg::f_current_schema_id);
     Poco::JSON::Object::Ptr current_schema;

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

@@ -4,6 +4,7 @@
 #include <cctype>
 #include <limits>
 
+#include <base/unaligned.h>
 #include <Columns/IColumn.h>
 #include <Common/Exception.h>
 #include <DataTypes/DataTypeNullable.h>
@@ -22,6 +23,68 @@ extern const int BAD_ARGUMENTS;
 namespace Iceberg
 {
 
+namespace
+{
+
+/// 1e16 µs is ~year 2286. That is not an Iceberg `timestamp` max: Spark sentinels
+/// such as `9999-12-31` (`~2.53e17` µs) and ClickHouse `DateTime64` max
+/// `2299-12-31` (`~1.04e16` µs) are spec-correct microseconds above this.
+/// 1e18 ns is ~year 2001; customer 2026 ns bounds are `~1.76e18`.
+constexpr Int64 microseconds_ambiguous_threshold = 10'000'000'000'000'000LL;
+constexpr Int64 nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL;
+constexpr Int64 nanoseconds_per_microsecond = 1000;
+
+bool magnitudeExceeds(Int64 value, Int64 threshold)
+{
+    return value > threshold || value < -threshold;
+}
+
+/// Convert ns ticks to us ticks without shrinking a [lower, upper] range.
+Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound)
+{
+    Int64 microseconds = nanoseconds / nanoseconds_per_microsecond;
+    if (nanoseconds % nanoseconds_per_microsecond == 0)
+        return microseconds;
+
+    /// C++ integer division truncates toward zero.
+    if (lower_bound)
+    {
+        if (nanoseconds < 0)
+            --microseconds;
+    }
+    else if (nanoseconds > 0)
+    {
+        ++microseconds;
+    }
+    return microseconds;
+}
+
+/// Iceberg timestamps are 8-byte little-endian (spec Appendix D).
+/// Some writers store nanosecond stats on Iceberg `timestamp` (`DateTime64(6)`).
+/// Convert only in a true nanosecond band (`|ticks| > 1e18`). Between 1e16 and
+/// 1e18 the value may be far-future microseconds (do not convert; fail open so
+/// min/max pruning is skipped). `|ticks| <= 1e16` is kept as microseconds.
+std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound)
+{
+    if (str.size() != sizeof(Int64))
+        return std::nullopt;
+
+    Int64 unscaled = unalignedLoadLittleEndian<Int64>(str.data());
+    const UInt32 scale = getDecimalScale(type);
+
+    if (scale == 6 && magnitudeExceeds(unscaled, microseconds_ambiguous_threshold))
+    {
+        if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold))
+            return std::nullopt;
+
+        unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound);
+    }
+
+    return DecimalField<DateTime64>(DateTime64(unscaled), scale);
+}
+
+}
+
 Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name)
 {
     if (value.getType() == Field::Types::Int64)
@@ -153,6 +216,9 @@ static std::optional<Field> deserializeDecimalBound(const std::string & str, UIn
 std::optional<Field> deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound)
 {
     auto non_nullable_type = removeNullable(expected_type);
+    if (WhichDataType(non_nullable_type).isDateTime64())
+        return deserializeDateTime64FromBinaryRepr(str, *non_nullable_type, lower_bound);
+
     auto column = non_nullable_type->createColumn();
     if (WhichDataType(non_nullable_type).isDecimal())
     {

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

@@ -10,6 +10,7 @@
 #include <Core/Range.h>
 #include <Core/Settings.h>
 #include <Core/TypeId.h>
+#include <DataTypes/DataTypeDateTime64.h>
 #include <DataTypes/DataTypeNullable.h>
 #include <DataTypes/DataTypeString.h>
 #include <DataTypes/DataTypeTime64.h>
@@ -210,6 +211,25 @@ Int64 getTimeValueInMicroseconds(const Field & field, DataTypePtr type)
     throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected Time or Time64, got {}", type->getName());
 }
 
+Int64 getDateTime64ValueForIcebergBounds(const Field & field, DataTypePtr type)
+{
+    if (type->isNullable())
+        return getDateTime64ValueForIcebergBounds(field, assert_cast<const DataTypeNullable *>(type.get())->getNestedType());
+
+    if (!WhichDataType(type).isDateTime64())
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected DateTime64, got {}", type->getName());
+
+    const auto scale = getDecimalScale(*type);
+    const auto value = field.safeGet<Decimal64>().getValue().value;
+    if (scale == 9)
+        return value;
+    if (scale > 6)
+        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName());
+
+    /// Iceberg `timestamp` bounds are microseconds. DateTime64(s) for s < 6 stores coarser ticks.
+    return value * DataTypeDateTime64::getScaleMultiplier(6 - scale).value;
+}
+
 template <typename DecimalType>
 std::vector<uint8_t> dumpDecimalValue(const Field & field)
 {
@@ -323,7 +343,7 @@ std::vector<uint8_t> dumpFieldToBytes(const Field & field, DataTypePtr type)
         case TypeIndex::UInt64:
             return dumpValue(applyVisitor(FieldVisitorConvertToNumber<Int64>(), field));
         case TypeIndex::DateTime64:
-            return dumpValue(field.safeGet<Decimal64>().getValue().value);
+            return dumpValue(getDateTime64ValueForIcebergBounds(field, type));
         case TypeIndex::String:
         {
             auto value = field.safeGet<String>();

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

@@ -476,6 +476,9 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                 auto right = deserializeFieldFromBinaryRepr(right_str, name_and_type.type, false);
                 if (!left || !right)
                     continue;
+                /// Mixed-scale timestamp stats can decode to lower > upper; skip min/max for this column.
+                if (DB::Range::less(*right, *left))
+                    continue;
 
                 hyperrectangles.emplace(column_id, DB::Range(*left, true, *right, true));
             }

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

@@ -32,6 +32,37 @@ namespace DB::ErrorCodes
     extern const int LOGICAL_ERROR;
 }
 
+namespace
+{
+
+/// Avro does not map timestamp-nanos to DateTime64, so identity partition values for
+/// Iceberg `timestamp_ns` / `timestamptz_ns` (and similarly `Time64`) arrive as Int64.
+/// `KeyCondition` compares DateTime64/Time64 predicates as DecimalField with a scale,
+/// so wrap the raw ticks before `mayBeTrueInRange`. Calendar transforms stay Int64.
+void wrapIntegerPartitionValuesAsDateTime64(Row & partition_values, const DataTypes & data_types)
+{
+    const size_t size = std::min(partition_values.size(), data_types.size());
+    for (size_t i = 0; i < size; ++i)
+    {
+        auto & field = partition_values[i];
+        if (field.isNull() || field.getType() != Field::Types::Int64)
+            continue;
+
+        auto type = removeNullable(data_types[i]);
+        WhichDataType which(type);
+        if (which.isDateTime64())
+        {
+            field = DecimalField<DateTime64>(DateTime64(field.safeGet<Int64>()), getDecimalScale(*type));
+        }
+        else if (which.isTime64())
+        {
+            field = DecimalField<Time64>(Time64(field.safeGet<Int64>()), getDecimalScale(*type));
+        }
+    }
+}
+
+}
+
 namespace DB::Iceberg
 {
 
@@ -246,7 +277,9 @@ PruningReturnStatus ManifestFilesPruner::canBePruned(
 {
     if (partition_key_condition.has_value())
     {
-        const auto & partition_value = entry->parsed_entry->partition_key_value;
+        Row partition_value = entry->parsed_entry->partition_key_value;
+        wrapIntegerPartitionValuesAsDateTime64(partition_value, partition_key->data_types);
+
         std::vector<FieldRef> index_value(partition_value.begin(), partition_value.end());
         for (size_t i = 0; i < index_value.size(); ++i)
         {

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

@@ -352,7 +352,8 @@ bool MetadataGenerator::isAddColumnApplied(const String & column_name, DataTypeP
         return false;
 
     Int32 unused_field_id = metadata_object->getValue<Int32>(Iceberg::f_last_column_id);
-    auto expected_type = Iceberg::getIcebergType(type, unused_field_id);
+    auto format_version = metadata_object->getValue<UInt64>(Iceberg::f_format_version);
+    auto expected_type = Iceberg::getIcebergType(type, unused_field_id, format_version);
 
     auto fields = current_schema->getArray(Iceberg::f_fields);
     for (UInt32 i = 0; i < fields->size(); ++i)
@@ -420,7 +421,8 @@ bool MetadataGenerator::isModifyColumnApplied(const String & column_name, DataTy
         if (type)
         {
             Int32 unused_field_id = metadata_object->getValue<Int32>(Iceberg::f_last_column_id);
-            auto expected_type = Iceberg::getIcebergType(type, unused_field_id);
+            auto format_version = metadata_object->getValue<UInt64>(Iceberg::f_format_version);
+            auto expected_type = Iceberg::getIcebergType(type, unused_field_id, format_version);
             if (field->getValue<bool>(Iceberg::f_required) != expected_type.second
                 || !icebergTypesEqualIgnoringIds(field->get(Iceberg::f_type), expected_type.first))
                 return false;
@@ -795,7 +797,8 @@ void MetadataGenerator::generateAddColumnMetadata(const String & column_name, Da
 
     auto last_column_id = metadata_object->getValue<Int32>(Iceberg::f_last_column_id);
 
-    auto new_type = Iceberg::getIcebergType(type, last_column_id);
+    auto format_version = metadata_object->getValue<UInt64>(Iceberg::f_format_version);
+    auto new_type = Iceberg::getIcebergType(type, last_column_id, format_version);
     Poco::JSON::Object::Ptr new_field = new Poco::JSON::Object;
     new_field->set(Iceberg::f_id, last_column_id + 1);
     new_field->set(Iceberg::f_name, column_name);
@@ -847,6 +850,7 @@ bool MetadataGenerator::generateModifyColumnMetadata(const String & column_name,
     auto current_schema = getCurrentSchema();
 
     auto last_column_id = metadata_object->getValue<Int32>(Iceberg::f_last_column_id);
+    auto format_version = metadata_object->getValue<UInt64>(Iceberg::f_format_version);
     auto schema_fields = current_schema->getArray(Iceberg::f_fields);
 
     /// `AFTER` the column itself keeps its position, as in `ColumnsDescription::modifyColumnOrder`.
@@ -855,7 +859,7 @@ bool MetadataGenerator::generateModifyColumnMetadata(const String & column_name,
 
     if (type)
     {
-        auto new_type = Iceberg::getIcebergType(type, last_column_id);
+        auto new_type = Iceberg::getIcebergType(type, last_column_id, format_version);
 
         for (UInt32 i = 0; i < schema_fields->size(); ++i)
         {
@@ -919,7 +923,7 @@ bool MetadataGenerator::generateModifyColumnMetadata(const String & column_name,
 
     if (type_changed)
     {
-        auto new_type = Iceberg::getIcebergType(type, last_column_id);
+        auto new_type = Iceberg::getIcebergType(type, last_column_id, format_version);
         target_field->set(Iceberg::f_type, new_type.first);
         target_field->set(Iceberg::f_required, new_type.second);
     }

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

@@ -240,12 +240,12 @@ void MultipleFileWriter::startNewFile()
     buffer = object_storage->writeObject(
         StoredObject(storage_path), WriteMode::Rewrite, std::nullopt, DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings());
 
-    if (format_settings)
-    {
-        format_settings->parquet.write_page_index = true;
-        format_settings->parquet.bloom_filter_push_down = true;
-        format_settings->parquet.filter_push_down = true;
-    }
+    if (!format_settings)
+        format_settings = getFormatSettings(context);
+    format_settings->parquet.write_page_index = true;
+    format_settings->parquet.bloom_filter_push_down = true;
+    format_settings->parquet.filter_push_down = true;
+    format_settings->parquet.output_timestamp_adjusted_to_utc_from_timezone = true;
     FormatFilterInfoPtr format_filter_info = std::make_shared<FormatFilterInfo>(nullptr, context, column_mapper, nullptr, nullptr);
     output_format = FormatFactory::instance().getOutputFormatParallelIfPossible(
         write_format, *buffer, *filtered_sample_block, context, format_settings, format_filter_info);