Fix Iceberg partition and min/max pruning for timestamp_ns

src/Formats/FormatSettings.h

@@ -402,6 +402,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/AvroRowOutputFormat.cpp

@@ -211,11 +211,18 @@ AvroSerializer::SchemaWithSerializeFn AvroSerializer::createSchemaWithSerializeF
             auto schema = avro::LongSchema();
             const auto & provided_type = assert_cast<const DataTypeDateTime64 &>(*data_type);
 
-            if (provided_type.getScale() == 3)
+            const auto scale = provided_type.getScale();
+            /// Iceberg `timestamp_ns` and `timestamptz_ns` are Avro `long` values of nanosecond ticks.
+            /// contrib Avro has no `timestamp-nanos` logical type, so an Iceberg data file stores a
+            /// plain `long`. The table metadata still records `timestamp_ns`. Other scales stay a
+            /// decimal `bytes` value, which is what generic Avro output already writes.
+            const bool iceberg_timestamp_ns = scale == 9 && column_mapper;
+
+            if (scale == 3)
                 schema.root()->setLogicalType(avro::LogicalType(avro::LogicalType::TIMESTAMP_MILLIS));
-            else if (provided_type.getScale() == 6)
+            else if (scale == 6)
                 schema.root()->setLogicalType(avro::LogicalType(avro::LogicalType::TIMESTAMP_MICROS));
-            else
+            else if (!iceberg_timestamp_ns)
                 return createDecimalSchemaWithSerializeFn<DataTypeDateTime64>(data_type);
 
             return {schema, [](const IColumn & column, size_t row_num, avro::Encoder & encoder)

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

@@ -447,9 +447,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

@@ -27,6 +27,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

@@ -55,6 +55,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>
@@ -509,5 +511,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

@@ -157,7 +157,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

@@ -423,8 +423,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)
         {
@@ -1063,7 +1067,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

@@ -5,6 +5,7 @@
 #include <limits>
 
 #include <base/arithmeticOverflow.h>
+#include <base/unaligned.h>
 #include <Columns/IColumn.h>
 #include <Core/DecimalFunctions.h>
 #include <Common/Exception.h>
@@ -24,6 +25,67 @@ extern const int BAD_ARGUMENTS;
 namespace Iceberg
 {
 
+namespace
+{
+
+/// 1e16 µs is ~year 2286. Spark sentinels such as `9999-12-31` (`~2.53e17` µs)
+/// are spec-correct microseconds above this.
+constexpr Int64 microseconds_ambiguous_threshold = 10'000'000'000'000'000LL;
+/// 1e18 ns is ~year 2001; customer 2026 ns bounds are `~1.76e18`.
+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)`).
+/// `|ticks| <= 1e16` is kept as microseconds, as the spec requires. `|ticks| > 1e18`
+/// is converted from nanoseconds. `(1e16, 1e18]` fails open: it collides Spark
+/// `9999-12-31` sentinels with 1970–2001 nanoseconds.
+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)
@@ -200,6 +262,9 @@ std::optional<Field>
 deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound, bool compensate_rounding)
 {
     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/IcebergMetadata.cpp

@@ -398,7 +398,9 @@ Int32 IcebergMetadata::parseTableSchema(
 {
     const auto format_version = metadata_object->getValue<Int32>(f_format_version);
 
-    if (format_version == 2)
+    /// v2 and v3 (and later) store schemas in the `schemas` array. Only v1 has a top-level `schema`.
+    /// Using the v1 parser first for v3 looks like a spec violation and logs a false warning.
+    if (format_version >= 2)
     {
         auto [schema, current_schema_id] = parseTableSchemaV2Method(metadata_object);
         schema_processor.addIcebergTableSchema(schema, context_);

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>
@@ -291,6 +292,31 @@ 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());
+
+    /// Manifest rewrite reads the Avro `long` back as Int64 (contrib Avro has no `timestamp-nanos`).
+    if (field.getType() == Field::Types::Int64)
+        return field.safeGet<Int64>();
+    if (field.getType() == Field::Types::UInt64)
+        return static_cast<Int64>(field.safeGet<UInt64>());
+
+    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;
+}
+
 std::vector<uint8_t> dumpFieldToBytes(const Field & field, DataTypePtr type)
 {
     switch (type->getTypeId())
@@ -326,7 +352,7 @@ std::vector<uint8_t> dumpFieldToBytes(const Field & field, DataTypePtr type)
         case TypeIndex::Time64:
             return dumpValue(getTimeValueInMicroseconds(field, type));
         case TypeIndex::DateTime64:
-            return dumpValue(field.safeGet<Decimal64>().getValue().value);
+            return dumpValue(getDateTime64ValueForIcebergBounds(field, type));
         case TypeIndex::String:
         {
             auto value = field.safeGet<String>();
@@ -746,8 +772,9 @@ void generateManifestFile(
 
     auto adapter = std::make_unique<OutputStreamWriteBufferAdapter>(buf);
     avro::DataFileWriter<avro::GenericDatum> writer(std::move(adapter), schema);
-    /// avro-cpp's compiled schema loses the Iceberg field-id/element-id attributes; write the
-    /// original id-carrying JSON as the avro.schema header so external readers can plan a scan.
+    /// avro-cpp's compiled schema loses Iceberg field-id attributes, and `toJson` drops
+    /// unknown logical types such as `timestamp-nanos`. Write the original JSON as the
+    /// `avro.schema` header so external readers keep both.
     writer.setMetadata(Iceberg::f_avro_schema, schema_representation);
     writer.setMetadata(Iceberg::f_schema, json_representation);
     writer.setMetadata(Iceberg::f_schema_id, std::to_string(schema_object_to_write->getValue<Int32>(Iceberg::f_schema_id)));

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

@@ -485,6 +485,10 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                     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

@@ -34,6 +34,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
 {
 
@@ -377,7 +408,10 @@ PruningReturnStatus ManifestFilesPruner::canBePruned(
 
     if (partition_key_condition.has_value())
     {
-        std::vector<FieldRef> index_value(partition_value.begin(), partition_value.end());
+        Row wrapped_partition_value = partition_value;
+        wrapIntegerPartitionValuesAsDateTime64(wrapped_partition_value, partition_key->data_types);
+
+        std::vector<FieldRef> index_value(wrapped_partition_value.begin(), wrapped_partition_value.end());
         for (size_t i = 0; i < index_value.size(); ++i)
         {
             auto & field = index_value[i];