From 4a342773b13977b5ee167492ced59f68d2c623cf Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 30 Oct 2026 16:13:08 +0100
Subject: [PATCH 01/16] Copy the table alias onto `remote()` when wrapping a
 remote initiator

`JoinedTables` only reads the outer table function, so an old-analyzer global JOIN failed with "No alias for subquery or table function in JOIN". The inner alias stays for the query sent to the remote initiator.

https: //github.com/Altinity/ClickHouse/issues/2748
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Storages/IStorageCluster.cpp | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp
index 134ceaa4237f8..fc7a3eaa3eaf2 100644
--- a/src/Storages/IStorageCluster.cpp
+++ b/src/Storages/IStorageCluster.cpp
@@ -853,6 +853,12 @@ IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote(
     if (!table_expression->table_function)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table function in table expression");
 
+    /// `AS alias` is stored on the inner table function. `JoinedTables` reads only the outer
+    /// table function, so `remote()` must carry the same alias. Leave it on the inner function:
+    /// `addTableFunction` sends that node to the remote initiator and does not copy a non-empty
+    /// outer alias onto the replacement.
+    const String table_alias = table_expression->table_function->tryGetAlias();
+
     boost::intrusive_ptr<ASTFunction> remote_query;
 
     if (shard_addresses[0].user_specified)
@@ -868,6 +874,11 @@ IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote(
         remote_query = makeASTFunction(remote_function_name, make_intrusive<ASTLiteral>(host_name), table_expression->table_function);
     }
 
+    if (!table_alias.empty())
+    {
+        remote_query->setAlias(table_alias);
+    }
+
     table_expression->table_function = remote_query;
 
     auto remote_function = TableFunctionFactory::instance().get(remote_query, new_context);

From e6200be6c2714c6d99e7c02055fd48b898756fc9 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 30 Oct 2026 16:35:50 +0100
Subject: [PATCH 02/16] Keep Iceberg `timestamp` min/max bounds at or below
 `1e16` as microseconds

Scale-6 bounds in the normal range were dropped, so min/max pruning skipped every `timestamp` file. Values above `1e18` are still converted from nanoseconds, and `(1e16, 1e18]` still fails open.

https: //github.com/Altinity/ClickHouse/issues/2747
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Iceberg/IcebergFieldParseHelpers.cpp      | 11 ++++--
 .../gtest_iceberg_field_parse_helpers.cpp     | 39 +++++++++----------
 2 files changed, 25 insertions(+), 25 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index b8bb69a04a6ef..fe5d1f550464c 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -26,6 +26,9 @@ 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;
@@ -57,9 +60,9 @@ Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound)
 
 /// Iceberg timestamps are 8-byte little-endian (spec Appendix D).
 /// Some writers store nanosecond stats on Iceberg `timestamp` (`DateTime64(6)`).
-/// Convert only when `|ticks| > 1e18`. Otherwise fail open: `|ticks| <= 1e16`
-/// collides near-epoch nanoseconds (~115 days) with 1970–2286 microseconds, and
-/// `(1e16, 1e18]` collides Spark `9999-12-31` sentinels with 1970–2001 nanoseconds.
+/// `|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))
@@ -68,7 +71,7 @@ std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str
     Int64 unscaled = unalignedLoadLittleEndian<Int64>(str.data());
     const UInt32 scale = getDecimalScale(type);
 
-    if (scale == 6)
+    if (scale == 6 && magnitudeExceeds(unscaled, microseconds_ambiguous_threshold))
     {
         if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold))
             return std::nullopt;
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index 069a01bed01c1..a8ed01182fc36 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -29,32 +29,29 @@ Int64 decimal64Ticks(const Field & field)
 
 }
 
-TEST(IcebergFieldParseHelpers, NearEpochNanosecondDateTime64BoundsFailOpen)
+TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept)
 {
     auto type = std::make_shared<DataTypeDateTime64>(6);
-    /// One day after the epoch as nanoseconds: kept as µs this is ~1972, not 1970-01-02.
-    constexpr Int64 one_day_ns = 86'400'000'000'000LL;
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(one_day_ns), type, true).has_value());
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(one_day_ns), type, false).has_value());
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(-one_day_ns), type, true).has_value());
-}
-
-TEST(IcebergFieldParseHelpers, AmbiguousScaleDateTime64BoundsFailOpen)
-{
-    auto type = std::make_shared<DataTypeDateTime64>(6);
-    /// 2024-01-01 UTC µs is also ~19.7 days as ns. No integer test splits those scales.
-    constexpr Int64 us_2024_01_01 = 1'704'067'200'000'000LL;
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_01_01), type, true).has_value());
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(-us_2024_01_01), type, true).has_value());
+    /// 2024-02-20 10:00:00 UTC as microseconds — spec-correct Iceberg `timestamp`.
+    constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL;
+    auto lower = deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_02_20), type, true);
+    auto upper = deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_02_20), type, false);
+    ASSERT_TRUE(lower.has_value());
+    ASSERT_TRUE(upper.has_value());
+    EXPECT_EQ(decimal64Ticks(*lower), us_2024_02_20);
+    EXPECT_EQ(decimal64Ticks(*upper), us_2024_02_20);
 
-    /// 50 days after the epoch as ns (`4.32e15`) sits in the same band as 2024–2286 µs.
-    constexpr Int64 fifty_days_ns = 4'320'000'000'000'000LL;
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(fifty_days_ns), type, true).has_value());
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(fifty_days_ns), type, false).has_value());
+    /// 1960-01-01 00:00:00 UTC as microseconds.
+    constexpr Int64 us_1960_01_01 = -315'619'200'000'000LL;
+    auto negative = deserializeFieldFromBinaryRepr(littleEndianInt64(us_1960_01_01), type, true);
+    ASSERT_TRUE(negative.has_value());
+    EXPECT_EQ(decimal64Ticks(*negative), us_1960_01_01);
 
-    /// Band upper edge: 1e16 ns is ~1970-04-26, 1e16 µs is ~year 2286.
+    /// Upper edge of the microsecond band (~year 2286). Equal to the threshold, so it stays microseconds.
     constexpr Int64 one_e16 = 10'000'000'000'000'000LL;
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(one_e16), type, true).has_value());
+    auto edge = deserializeFieldFromBinaryRepr(littleEndianInt64(one_e16), type, false);
+    ASSERT_TRUE(edge.has_value());
+    EXPECT_EQ(decimal64Ticks(*edge), one_e16);
 }
 
 TEST(IcebergFieldParseHelpers, ProvenNanosecondDateTime64BoundsAreConverted)

From 799475477d3daa58b73f95455a319a3c10ad468e Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 30 Oct 2026 17:26:15 +0100
Subject: [PATCH 03/16] Decode Iceberg `timestamp` bounds with the type scale

Appendix D already sets the unit from the field type. `DateTime64(6)` stays microseconds and `DateTime64(9)` stays nanoseconds.

https: //github.com/Altinity/ClickHouse/issues/2747
Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Iceberg/IcebergFieldParseHelpers.cpp      | 55 ++-----------------
 .../gtest_iceberg_field_parse_helpers.cpp     | 24 +++-----
 2 files changed, 14 insertions(+), 65 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index fe5d1f550464c..3b8d01b99bd3a 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -26,59 +26,16 @@ 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)
+/// Iceberg Appendix D stores `timestamp` / `timestamptz` as microseconds and
+/// `timestamp_ns` / `timestamptz_ns` as nanoseconds. Both are an 8-byte
+/// little-endian int64. The ClickHouse type's scale is that unit.
+std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type)
 {
     if (str.size() != sizeof(Int64))
         return std::nullopt;
 
-    Int64 unscaled = unalignedLoadLittleEndian<Int64>(str.data());
+    const 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);
 }
 
@@ -216,7 +173,7 @@ std::optional<Field> deserializeFieldFromBinaryRepr(const std::string & str, Dat
 {
     auto non_nullable_type = removeNullable(expected_type);
     if (WhichDataType(non_nullable_type).isDateTime64())
-        return deserializeDateTime64FromBinaryRepr(str, *non_nullable_type, lower_bound);
+        return deserializeDateTime64FromBinaryRepr(str, *non_nullable_type);
 
     auto column = non_nullable_type->createColumn();
     if (WhichDataType(non_nullable_type).isDecimal())
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index a8ed01182fc36..ec96121c11fe7 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -47,28 +47,20 @@ TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept)
     ASSERT_TRUE(negative.has_value());
     EXPECT_EQ(decimal64Ticks(*negative), us_1960_01_01);
 
-    /// Upper edge of the microsecond band (~year 2286). Equal to the threshold, so it stays microseconds.
-    constexpr Int64 one_e16 = 10'000'000'000'000'000LL;
-    auto edge = deserializeFieldFromBinaryRepr(littleEndianInt64(one_e16), type, false);
-    ASSERT_TRUE(edge.has_value());
-    EXPECT_EQ(decimal64Ticks(*edge), one_e16);
+    /// Spark `9999-12-31` sentinel is spec-correct microseconds.
+    constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL;
+    auto sentinel = deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false);
+    ASSERT_TRUE(sentinel.has_value());
+    EXPECT_EQ(decimal64Ticks(*sentinel), spark_9999_us);
 }
 
-TEST(IcebergFieldParseHelpers, ProvenNanosecondDateTime64BoundsAreConverted)
+TEST(IcebergFieldParseHelpers, NanosecondDateTime64BoundsAreKept)
 {
-    auto type = std::make_shared<DataTypeDateTime64>(6);
+    auto type = std::make_shared<DataTypeDateTime64>(9);
     constexpr Int64 ns_2024_01_01 = 1'704'067'200'123'456'789LL;
     auto field = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2024_01_01), type, true);
     ASSERT_TRUE(field.has_value());
-    EXPECT_EQ(decimal64Ticks(*field), ns_2024_01_01 / 1000);
-}
-
-TEST(IcebergFieldParseHelpers, FarFutureMicrosecondDateTime64BoundsFailOpen)
-{
-    auto type = std::make_shared<DataTypeDateTime64>(6);
-    /// Spark 9999-12-31 sentinel sits in (1e16, 1e18].
-    constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL;
-    EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false).has_value());
+    EXPECT_EQ(decimal64Ticks(*field), ns_2024_01_01);
 }
 
 #endif

From 9a77f99edf5004defda7e442e6a3382ea3d7b35a Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Fri, 30 Oct 2026 19:28:34 +0100
Subject: [PATCH 04/16] Drop tests that store nanosecond bounds on Iceberg
 `timestamp`

The decoder reads bounds at the type scale. A `timestamp` column whose bounds are nanoseconds is corrupt data, so those tests are not required.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../helpers/iceberg_timestamp_ns.py           | 232 +-----------------
 .../test_pruning_datetime64_nanoseconds.py    |  80 ------
 .../test_pruning_datetime64_nanoseconds.py    |  52 ----
 3 files changed, 5 insertions(+), 359 deletions(-)

diff --git a/tests/integration/helpers/iceberg_timestamp_ns.py b/tests/integration/helpers/iceberg_timestamp_ns.py
index 3922f8d37d891..1c3a942c55c56 100644
--- a/tests/integration/helpers/iceberg_timestamp_ns.py
+++ b/tests/integration/helpers/iceberg_timestamp_ns.py
@@ -164,57 +164,13 @@ def check_ns_pruning_datetime64_microseconds(node, table, column, pruned_files,
     ) == "3\n4\n"
 
 
-def check_minmax_ns_bounds_on_timestamp_schema(node, table, column, pruned_files):
-    """Customer min/max over-prune: Iceberg `timestamp` with nanosecond min/max bytes.
-
-    OTEL `time_unix_nano` tables are often declared as Iceberg `timestamp`
-    (`DateTime64(6)`) while 8-byte lower/upper bounds stay in nanoseconds.
-    ClickHouse then deserializes those ns bytes as microseconds (year 2299),
-    so `WHERE col <= toDateTime64('2026-...', 6)` prunes every file while the
-    parquet scan (microseconds) with `use_iceberg_partition_pruning = 0` is correct.
-    """
-    type_name = node.query(
-        f"SELECT toTypeName({column}) FROM {table} LIMIT 1",
-        settings=PRUNE_OFF,
-    )
-    assert "DateTime64(6" in type_name, type_name
-    assert "DateTime64(9" not in type_name, type_name
-
-    ids = node.query(f"SELECT id FROM {table} ORDER BY id", settings=PRUNE_OFF)
-    assert ids == "1\n2\n3\n4\n"
-    # Parquet is timestamp[us] (correct 2024 wall times). Iceberg lower/upper bounds
-    # are still the original nanosecond int64 bytes, so DateTime64(6) min/max
-    # interprets them as year 2299 and over-prunes.
-    micros = node.query(
-        f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1",
-        settings=PRUNE_OFF,
-    ).strip()
-    assert micros == "1704067200123456", micros
-
-    # Same predicate shape as the customer query: scale-6 literal near a file's
-    # nanosecond upper bound. Rows 1 and 2 must survive; pruning must not drop them.
-    le_after_row2 = datetime64_literal(
-        "2024-03-15 12:00:00.000001000", scale=6, timezone=None
-    )
-    query = f"SELECT id FROM {table} WHERE {column} <= {le_after_row2} ORDER BY id"
-    # One data file covering all four timestamps; the predicate overlaps it, so
-    # min/max pruning must keep the file. Over-pruning currently returns 0 rows.
-    assert pruned_files(query) == 0
-    assert node.query(query, settings=PRUNE_ON) == "1\n2\n"
-
-    count_query = f"SELECT count() FROM {table} WHERE {column} <= {le_after_row2}"
-    assert pruned_files(count_query) == 0
-    assert node.query(count_query, settings=PRUNE_ON) == "2\n"
-
-
 def _utc_datetime_to_micros(dt):
     """Integer microseconds from Unix epoch. Do not use float `timestamp` * 1e6."""
     delta = dt - datetime(1970, 1, 1, tzinfo=timezone.utc)
     return delta.days * 86_400_000_000 + delta.seconds * 1_000_000 + delta.microseconds
 
 
-# Spec-correct Iceberg `timestamp` microseconds that sit in the ambiguous band
-# (1e16, 1e18]: converting them as nanoseconds over-prunes.
+# Spec-correct Iceberg `timestamp` microseconds, including the Spark `9999-12-31` sentinel.
 SPARK_TIMESTAMP_SENTINEL_US = _utc_datetime_to_micros(
     datetime(9999, 12, 31, 23, 59, 59, 999999, tzinfo=timezone.utc)
 )
@@ -223,79 +179,12 @@ def _utc_datetime_to_micros(dt):
 )
 
 
-# One day after the Unix epoch as nanoseconds. Kept as microseconds this is ~1972,
-# so a predicate on 1970-01-02 over-prunes unless the decoder fails open.
-NEAR_EPOCH_NS_ROWS = [
-    (86_400_000_000_000, 1, "1970-01-02 00:00:00.000000000"),
-]
-
-# 50 days after the epoch as nanoseconds (`4.32e15`). Same integer as ~2106 µs,
-# so a predicate on 1970-02-20 over-prunes unless the decoder fails open.
-OVERLAP_BAND_NS_ROWS = [
-    (4_320_000_000_000_000, 1, "1970-02-20 00:00:00.000000000"),
-]
-
-
-def check_minmax_near_epoch_ns_bounds_on_timestamp_schema(node, table, column, pruned_files):
-    """Iceberg `timestamp` whose min/max bytes are near-epoch nanoseconds must not over-prune.
-
-    One day after the epoch is `8.64e13` ns (`|ticks| <= 1e16`). The inverted-range
-    check does not fire when both bounds stay in that band.
-    """
-    type_name = node.query(
-        f"SELECT toTypeName({column}) FROM {table} LIMIT 1",
-        settings=PRUNE_OFF,
-    )
-    assert "DateTime64(6" in type_name, type_name
-    assert "DateTime64(9" not in type_name, type_name
-
-    ids = node.query(f"SELECT id FROM {table} ORDER BY id", settings=PRUNE_OFF)
-    assert ids == "1\n"
-    micros = node.query(
-        f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1",
-        settings=PRUNE_OFF,
-    ).strip()
-    assert micros == "86400000000", micros
-
-    eq_epoch_day = datetime64_literal("1970-01-02 00:00:00.000000000", scale=6, timezone=None)
-    query = f"SELECT id FROM {table} WHERE {column} = {eq_epoch_day} ORDER BY id"
-    assert pruned_files(query) == 0
-    assert node.query(query, settings=PRUNE_ON) == "1\n"
-
-
-def check_minmax_overlap_band_ns_bounds_on_timestamp_schema(node, table, column, pruned_files):
-    """Iceberg `timestamp` whose min/max bytes are leftover-band nanoseconds must not over-prune.
-
-    50 days after the epoch is `4.32e15` ns, which is also a 2024–2286 microsecond
-    timestamp. The decoder cannot tell the scales apart and must fail open.
-    """
-    type_name = node.query(
-        f"SELECT toTypeName({column}) FROM {table} LIMIT 1",
-        settings=PRUNE_OFF,
-    )
-    assert "DateTime64(6" in type_name, type_name
-    assert "DateTime64(9" not in type_name, type_name
-
-    ids = node.query(f"SELECT id FROM {table} ORDER BY id", settings=PRUNE_OFF)
-    assert ids == "1\n"
-    micros = node.query(
-        f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1",
-        settings=PRUNE_OFF,
-    ).strip()
-    assert micros == "4320000000000", micros
-
-    eq_day = datetime64_literal("1970-02-20 00:00:00.000000000", scale=6, timezone=None)
-    query = f"SELECT id FROM {table} WHERE {column} = {eq_day} ORDER BY id"
-    assert pruned_files(query) == 0
-    assert node.query(query, settings=PRUNE_ON) == "1\n"
-
-
 def check_minmax_far_future_us_upper_bound(node, table, column, pruned_files):
-    """Iceberg `timestamp` with a far-future microsecond *upper* bound must not over-prune.
+    """Iceberg `timestamp` with a far-future microsecond upper bound must keep matching rows.
 
-    Lower bound stays a real 2024 microsecond value. If the decoder treats the
-    upper bound as nanoseconds, max becomes ~1978, the [min, max] range inverts,
-    and `WHERE col <= toDateTime64('2024-03-15 ...', 6)` drops the file.
+    Lower bound stays a real 2024 microsecond value. The upper bound is the Spark
+    `9999-12-31` sentinel or the `DateTime64` maximum. A predicate on 2024 must
+    still return the matching rows.
     """
     type_name = node.query(
         f"SELECT toTypeName({column}) FROM {table} LIMIT 1",
@@ -320,117 +209,6 @@ def check_minmax_far_future_us_upper_bound(node, table, column, pruned_files):
     assert node.query(query, settings=PRUNE_ON) == "1\n2\n"
 
 
-def _scale_bound_bytes(raw, factor=1000):
-    if not isinstance(raw, (bytes, bytearray)) or len(raw) != 8:
-        return raw
-    value = struct.unpack("<q", raw)[0]
-    return struct.pack("<q", value * factor)
-
-
-def scale_timestamp_bounds_us_to_ns(obj, field_ids=(1, 2)):
-    """Multiply Iceberg timestamp (us) lower/upper bounds by 1000 so they look like ns."""
-    field_ids = set(field_ids)
-    scaled = 0
-    if isinstance(obj, dict):
-        for key in ("lower_bounds", "upper_bounds"):
-            bounds = obj.get(key)
-            if isinstance(bounds, dict):
-                for bound_key, bound_value in list(bounds.items()):
-                    try:
-                        column_id = int(bound_key)
-                    except (TypeError, ValueError):
-                        continue
-                    if column_id in field_ids:
-                        new_value = _scale_bound_bytes(bound_value)
-                        if new_value != bound_value:
-                            bounds[bound_key] = new_value
-                            scaled += 1
-            elif isinstance(bounds, list):
-                for item in bounds:
-                    if not isinstance(item, dict):
-                        continue
-                    bound_key = item.get("key", item.get("field_id"))
-                    try:
-                        column_id = int(bound_key)
-                    except (TypeError, ValueError):
-                        continue
-                    if column_id in field_ids and "value" in item:
-                        new_value = _scale_bound_bytes(item["value"])
-                        if new_value != item["value"]:
-                            item["value"] = new_value
-                            scaled += 1
-        for value_key, value in obj.items():
-            if value_key in ("lower_bounds", "upper_bounds"):
-                continue
-            scaled += scale_timestamp_bounds_us_to_ns(value, field_ids)
-    elif isinstance(obj, list):
-        for value in obj:
-            scaled += scale_timestamp_bounds_us_to_ns(value, field_ids)
-    return scaled
-
-
-def rewrite_avro_scale_timestamp_bounds_us_to_ns(
-    src: Path, dst: Path, old=None, new=None, field_ids=(1, 2)
-):
-    with open(src, "rb") as handle:
-        reader = avro.datafile.DataFileReader(handle, avro.io.DatumReader())
-        schema = reader.datum_reader.writers_schema
-        codec = reader.codec
-        meta = dict(reader.meta)
-        records = []
-        scaled = 0
-        for record in reader:
-            if old is not None and new is not None:
-                record = _deep_replace(record, old, new)
-            scaled += scale_timestamp_bounds_us_to_ns(record, field_ids)
-            records.append(record)
-        reader.close()
-
-    with open(dst, "wb") as handle:
-        writer = avro.datafile.DataFileWriter(handle, avro.io.DatumWriter(), schema, codec=codec)
-        for key, value in meta.items():
-            key_str = key.decode("utf-8") if isinstance(key, (bytes, bytearray)) else key
-            if key_str.startswith("avro."):
-                continue
-            text = _meta_text(value)
-            if old is not None and new is not None:
-                text = text.replace(old, new)
-            writer.set_meta(key_str, text.encode("utf-8"))
-        for record in records:
-            writer.append(record)
-        writer.close()
-    return scaled
-
-
-def patch_iceberg_timestamp_bounds_us_to_ns(
-    table_location: Path, old_prefix=None, new_prefix=None, field_ids=(1, 2)
-):
-    """Keep Iceberg `timestamp` / parquet microseconds, but rewrite min/max bounds as ns.
-
-    That is the customer layout: scan is DateTime64(6) in 2024/2026, while
-    `deserializeFieldFromBinaryRepr` treats 8-byte ns bounds as microseconds (year 2299)
-    and min/max pruning drops every file.
-    """
-    meta_dir = table_location / "metadata"
-    if old_prefix is not None and new_prefix is not None:
-        for metadata_file in meta_dir.glob("*.metadata.json"):
-            text = metadata_file.read_text().replace(old_prefix, new_prefix)
-            metadata_file.write_text(text)
-
-    scaled_total = 0
-    for avro_file in list(meta_dir.glob("*.avro")):
-        tmp = avro_file.with_suffix(".avro.tmp")
-        scaled = rewrite_avro_scale_timestamp_bounds_us_to_ns(
-            avro_file, tmp, old_prefix, new_prefix, field_ids=field_ids
-        )
-        if scaled or old_prefix is not None:
-            tmp.replace(avro_file)
-        else:
-            tmp.unlink()
-        scaled_total += scaled
-    assert scaled_total > 0, f"no timestamp min/max bounds scaled under {table_location}"
-
-
 def _set_bound_bytes(raw, value):
     if not isinstance(raw, (bytes, bytearray)) or len(raw) != 8:
         return raw
diff --git a/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py
index 5a78ef2a69005..c7a9beb9a6b5b 100644
--- a/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py
+++ b/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py
@@ -24,20 +24,14 @@
     DATETIME64_MAX_US,
     FILTER_LITERAL_IDS,
     FILTER_LITERALS,
-    NEAR_EPOCH_NS_ROWS,
-    OVERLAP_BAND_NS_ROWS,
     PRUNE_OFF,
     PRUNE_ON,
     ROWS,
     SPARK_TIMESTAMP_SENTINEL_US,
     check_minmax_far_future_us_upper_bound,
-    check_minmax_near_epoch_ns_bounds_on_timestamp_schema,
-    check_minmax_ns_bounds_on_timestamp_schema,
-    check_minmax_overlap_band_ns_bounds_on_timestamp_schema,
     check_ns_pruning,
     check_ns_pruning_datetime64_microseconds,
     patch_iceberg_table_long_to_timestamp_ns,
-    patch_iceberg_timestamp_bounds_us_to_ns,
     patch_iceberg_timestamp_upper_bounds,
 )
 from helpers.iceberg_utils import check_validity_and_get_prunned_files_general
@@ -123,13 +117,6 @@ def _create_timestamp_table(started_cluster, table_name, patch, rows=ROWS):
         shutil.rmtree(work, ignore_errors=True)
 
 
-def _create_timestamp_table_with_ns_bounds(started_cluster, table_name, rows=ROWS):
-    """Iceberg v2 `timestamp` (microseconds) whose manifest min/max bytes are nanoseconds."""
-    return _create_timestamp_table(
-        started_cluster, table_name, patch_iceberg_timestamp_bounds_us_to_ns, rows=rows
-    )
-
-
 def _create_timestamp_table_with_upper_bound_us(started_cluster, table_name, micros):
     """Iceberg v2 `timestamp` whose upper bounds are spec-correct far-future microseconds."""
     def patch(table_location, old_prefix, new_prefix):
@@ -244,73 +231,6 @@ def pruned_files(select_expression):
     check_ns_pruning_datetime64_microseconds(node, table, "value", pruned_files, timezone=timezone)
 
 
-def test_minmax_pruning_ns_bounds_on_timestamp_schema(started_cluster_iceberg_no_spark):
-    """Iceberg `timestamp` (`DateTime64(6)`) with nanosecond min/max bytes must not over-prune.
-
-    Customer: `WHERE time_unix_nano <= toDateTime64('2026-08-11 20:02:07.897356', 6)`
-    returns 0 with min/max pruning and the correct row count without it.
-    """
-    node = started_cluster_iceberg_no_spark.instances["node1"]
-    s3_filename = _create_timestamp_table_with_ns_bounds(
-        started_cluster_iceberg_no_spark, "minmax_ns_on_ts"
-    )
-    table = _iceberg_table(s3_filename)
-
-    def pruned_files(select_expression):
-        return check_validity_and_get_prunned_files_general(
-            node,
-            "minmax_ns_on_ts",
-            PRUNE_OFF,
-            PRUNE_ON,
-            "IcebergMinMaxIndexPrunedFiles",
-            select_expression,
-        )
-
-    check_minmax_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files)
-
-
-def test_minmax_pruning_near_epoch_ns_bounds_on_timestamp_schema(started_cluster_iceberg_no_spark):
-    """Near-epoch ns min/max on Iceberg `timestamp` must not be decoded as microseconds."""
-    node = started_cluster_iceberg_no_spark.instances["node1"]
-    s3_filename = _create_timestamp_table_with_ns_bounds(
-        started_cluster_iceberg_no_spark, "minmax_near_epoch_ns", rows=NEAR_EPOCH_NS_ROWS
-    )
-    table = _iceberg_table(s3_filename)
-
-    def pruned_files(select_expression):
-        return check_validity_and_get_prunned_files_general(
-            node,
-            "minmax_near_epoch_ns",
-            PRUNE_OFF,
-            PRUNE_ON,
-            "IcebergMinMaxIndexPrunedFiles",
-            select_expression,
-        )
-
-    check_minmax_near_epoch_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files)
-
-
-def test_minmax_pruning_overlap_band_ns_bounds_on_timestamp_schema(started_cluster_iceberg_no_spark):
-    """Ns min/max in `[1.704e15, 1e16]` on Iceberg `timestamp` must not be decoded as microseconds."""
-    node = started_cluster_iceberg_no_spark.instances["node1"]
-    s3_filename = _create_timestamp_table_with_ns_bounds(
-        started_cluster_iceberg_no_spark, "minmax_overlap_band_ns", rows=OVERLAP_BAND_NS_ROWS
-    )
-    table = _iceberg_table(s3_filename)
-
-    def pruned_files(select_expression):
-        return check_validity_and_get_prunned_files_general(
-            node,
-            "minmax_overlap_band_ns",
-            PRUNE_OFF,
-            PRUNE_ON,
-            "IcebergMinMaxIndexPrunedFiles",
-            select_expression,
-        )
-
-    check_minmax_overlap_band_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files)
-
-
 @pytest.mark.parametrize(
     "upper_bound_us",
     [SPARK_TIMESTAMP_SENTINEL_US, DATETIME64_MAX_US],
diff --git a/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py
index 229d0b4b239b8..387e1272831bd 100644
--- a/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py
+++ b/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py
@@ -18,11 +18,9 @@
     PRUNE_OFF,
     PRUNE_ON,
     ROWS,
-    check_minmax_ns_bounds_on_timestamp_schema,
     check_ns_pruning,
     check_ns_pruning_datetime64_microseconds,
     patch_iceberg_table_long_to_timestamp_ns,
-    patch_iceberg_timestamp_bounds_us_to_ns,
 )
 from helpers.iceberg_utils import (
     check_validity_and_get_prunned_files_general,
@@ -72,36 +70,6 @@ def _create_spark_ns_table(
     )
 
 
-def _create_spark_timestamp_table_with_ns_bounds(spark, started_cluster, table_name):
-    """Spark Iceberg `TIMESTAMP` (microseconds) whose manifest min/max bytes are nanoseconds."""
-    spark.sql(
-        f"""
-        CREATE TABLE {table_name} (
-            ts TIMESTAMP,
-            value TIMESTAMP,
-            id INT
-        )
-        USING iceberg
-        OPTIONS('format-version'='2')
-        """
-    )
-    values = ", ".join(
-        f"(TIMESTAMP '{utc.split('.')[0]}.{utc.split('.')[1][:6]}', "
-        f"TIMESTAMP '{utc.split('.')[0]}.{utc.split('.')[1][:6]}', {row_id})"
-        for _ts_ns, row_id, utc in ROWS
-    )
-    spark.sql(f"INSERT INTO {table_name} VALUES {values}")
-
-    table_location = Path(f"{WAREHOUSE_PREFIX}/{table_name}")
-    patch_iceberg_timestamp_bounds_us_to_ns(table_location)
-    default_upload_directory(
-        started_cluster,
-        "s3",
-        f"/iceberg_data/default/{table_name}/",
-        f"/iceberg_data/default/{table_name}/",
-    )
-
-
 @pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS)
 def test_spark_partition_pruning_datetime64_nanoseconds(started_cluster_iceberg_with_spark, make_literal):
     """Identity partition pruning on Spark-written Iceberg timestamp_ns must not drop matching rows."""
@@ -159,23 +127,3 @@ def pruned_files(select_expression):
         )
 
     check_ns_pruning_datetime64_microseconds(node, table, "value", pruned_files, timezone=timezone)
-
-
-def test_spark_minmax_pruning_ns_bounds_on_timestamp_schema(started_cluster_iceberg_with_spark):
-    """Nanosecond parquet/bounds with Iceberg `timestamp` (DateTime64(6)) must not over-prune."""
-    node = started_cluster_iceberg_with_spark.instances["node1"]
-    spark = started_cluster_iceberg_with_spark.spark_session
-    table_name = "test_spark_minmax_ns_on_ts_" + get_uuid_str()
-    _create_spark_timestamp_table_with_ns_bounds(
-        spark, started_cluster_iceberg_with_spark, table_name
-    )
-    table = get_creation_expression(
-        "s3", table_name, started_cluster_iceberg_with_spark, table_function=True
-    )
-
-    def pruned_files(select_expression):
-        return check_validity_and_get_prunned_files_general(
-            node, table_name, PRUNE_OFF, PRUNE_ON, "IcebergMinMaxIndexPrunedFiles", select_expression
-        )
-
-    check_minmax_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files)

From 0b54964b19e779f68945923ff2f8afd67e976a5e Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 10:06:43 +0100
Subject: [PATCH 05/16] Keep secrets when serializing a table function into a
 distributed query plan

With `serialize_query_plan=1` and `prefer_localhost_replica=0`, `ClusterProxy::SelectStreamFactory`
builds a local `QueryPlan` for each shard and sends it over the native protocol via
`ReadFromTableFunctionStep`. The table function AST was formatted with secrets hidden
(`show_secrets_=false`), so a function like `s3`/`s3Cluster` arrived on the executing
shard with its access key or secret replaced by the `[HIDDEN]` literal, which the shard
then used to sign requests, failing with `403 Forbidden`.

This AST only travels inside the cluster and is never shown to the user, so format it
with `formatWithSecretsOneLine` instead, matching the plain-SQL distributed query path
(`IStorageCluster::readFromRemoteInitiator`, `ClusterProxy::rewriteSelectQuery`).

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Planner/PlannerJoinTree.cpp | 18 ++++++++----------
 1 file changed, 8 insertions(+), 10 deletions(-)

diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp
index 931b5b08d6638..500a3516e5bf2 100644
--- a/src/Planner/PlannerJoinTree.cpp
+++ b/src/Planner/PlannerJoinTree.cpp
@@ -1491,16 +1491,14 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres
                         auto table_function_ast = table_function_node->toAST();
                         table_function_ast->setAlias({});
 
-                        WriteBufferFromOwnString out;
-                        IAST::FormatSettings format_settings(
-                            /*one_line=*/true,
-                            IdentifierQuotingRule::WhenNecessary,
-                            IdentifierQuotingStyle::Backticks,
-                            /*show_secrets_=*/false);
-
-                        table_function_ast->format(out, format_settings);
-
-                        auto table_function_serialized_ast = std::move(out.str());
+                        /// This serialized AST only travels inside the cluster, over the native
+                        /// protocol, to build the same table function on the executing shard (see
+                        /// `ReadFromTableFunctionStep` / `resolveStorages`). It is never shown to the
+                        /// user, so keep secrets, same as the plain-SQL distributed query path
+                        /// (`IStorageCluster::readFromRemoteInitiator` / `ClusterProxy::rewriteSelectQuery`
+                        /// use `formatWithSecretsOneLine` for the same reason). Hiding them here would
+                        /// send a masked literal (e.g. a redacted S3 secret key) to the remote shard.
+                        auto table_function_serialized_ast = table_function_ast->formatWithSecretsOneLine();
 
                         auto reading_from_table_function = std::make_unique<ReadFromTableFunctionStep>(
                             sample_block,

From 3b96dd1856d836efe3f71ad09d94238e9b343c05 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 14:43:02 +0100
Subject: [PATCH 06/16] Broadcast the right side of an old-analyzer global
 object-storage join

`convertToRemote` sent the local table name to one replica, so a replica without that table returned `UNKNOWN_TABLE`. `RIGHT`, `FULL`, and `PASTE` still stay on the initiator.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Storages/IStorageCluster.cpp              | 49 ++++++++++++++++++-
 .../test_remote_initiator.py                  | 25 +++++-----
 2 files changed, 62 insertions(+), 12 deletions(-)

diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp
index fc7a3eaa3eaf2..e1294e3c0c628 100644
--- a/src/Storages/IStorageCluster.cpp
+++ b/src/Storages/IStorageCluster.cpp
@@ -42,6 +42,9 @@
 #include <Analyzer/InDepthQueryTreeVisitor.h>
 #include <Analyzer/Utils.h>
 #include <Interpreters/TreeRewriter.h>
+#include <Interpreters/DatabaseCatalog.h>
+#include <Interpreters/GlobalSubqueriesVisitor.h>
+#include <Interpreters/PreparedSets.h>
 #include <Core/Joins.h>
 #include <Parsers/ASTExpressionList.h>
 #include <Parsers/ASTFunction.h>
@@ -409,6 +412,46 @@ void IStorageCluster::rewriteASTForGlobalJoin(ASTPtr & query)
     rewriteASTJoinsToGlobal(query);
 }
 
+/// The initiator interpreter usually does this already: `IStorageCluster::isRemote` is true, so
+/// `GlobalSubqueriesVisitor` replaces a `GLOBAL` right side with an external table. This pass
+/// covers a query that still names a local table when `convertToRemote` runs. A second pass
+/// is a no-op once that external table is in the context. Sets are added to the interpreter's
+/// `prepared_sets`, which `DelayedCreatingSetsStep` builds before the rows are sent.
+namespace
+{
+void broadcastOldAnalyzerGlobalJoin(ASTPtr & query_to_send, const SelectQueryInfo & query_info, const ContextPtr & context)
+{
+    IStorageCluster::rewriteASTForGlobalJoin(query_to_send);
+
+    auto prepared_sets = query_info.prepared_sets ? query_info.prepared_sets : std::make_shared<PreparedSets>();
+    TemporaryTablesMapping external_tables;
+    bool has_global_subqueries = false;
+
+    /// `table_join` stays null: the right side is read with all of its columns. Passing the
+    /// analyzed join here would rebuild its actions against an AST the first pass may already
+    /// have rewritten to an external table.
+    GlobalSubqueriesVisitor::Data data(
+        context,
+        /* subquery_depth */ 0,
+        /* is_remote */ true,
+        /* is_explain */ false,
+        external_tables,
+        prepared_sets,
+        has_global_subqueries,
+        /* table_join */ nullptr);
+    GlobalSubqueriesVisitor(data).visit(query_to_send);
+
+    /// The read context is const. External tables live on the query context, and a later
+    /// `Context::createCopy` still sees them through `getExternalTables`.
+    auto mutable_context = context->getQueryContext();
+    for (auto & [name, holder] : external_tables)
+    {
+        if (!mutable_context->tryResolveStorageID(StorageID{"", name}, Context::ResolveExternal))
+            mutable_context->addExternalTable(name, std::move(*holder));
+    }
+}
+}
+
 /*
 Try to make subquery to send on nodes
 Converts
@@ -455,14 +498,18 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded(
     }
     case ObjectStorageClusterJoinMode::GLOBAL:
     {
-        /// Old analyzer: ship the left scan only. `INNER`/`LEFT`/`CROSS` stay in `query_to_send`.
+        /// Old analyzer. `RIGHT`/`FULL`/`PASTE` stay on the initiator. `INNER`/`LEFT`/`CROSS`
+        /// stay in `query_to_send`, with the right side replaced by an external table.
         if (!query_info.query_tree)
         {
             if (queryHasRightOrFullJoin(query_to_send))
             {
                 auto info = getQueryJoinInfoFromAST(query_to_send);
                 rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context);
+                return;
             }
+
+            broadcastOldAnalyzerGlobalJoin(query_to_send, query_info, context);
             return;
         }
 
diff --git a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
index 77891e65fb449..22612fee96018 100644
--- a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
+++ b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
@@ -144,14 +144,17 @@ def execute_spark_query(query: str):
     instance.query(f"CREATE TABLE {TABLE2_NAME} (tag INT, number2 INT) ENGINE=Memory")
     instance.query(f"INSERT INTO {TABLE2_NAME} VALUES (1, 2)")
 
-    res = instance.query(f"""
-        SELECT t1.tag, t1.number, t2.number2
-        FROM {TABLE_NAME} AS t1
-        JOIN {TABLE2_NAME} AS t2 USING (tag)
-        SETTINGS
-            object_storage_remote_initiator=1,
-            object_storage_remote_initiator_cluster='cluster_simple',
-            object_storage_cluster_join_mode='global',
-            allow_experimental_analyzer=0
-        """)
-    assert res == "1\t1\t2\n"
+    # `convertToRemote` picks one replica of `cluster_simple`. The Memory table exists only on
+    # node1, so a single run can pass by landing there. Repeat until every replica is likely.
+    for _ in range(9):
+        res = instance.query(f"""
+            SELECT t1.tag, t1.number, t2.number2
+            FROM {TABLE_NAME} AS t1
+            JOIN {TABLE2_NAME} AS t2 USING (tag)
+            SETTINGS
+                object_storage_remote_initiator=1,
+                object_storage_remote_initiator_cluster='cluster_simple',
+                object_storage_cluster_join_mode='global',
+                allow_experimental_analyzer=0
+            """)
+        assert res == "1\t1\t2\n"

From 88d08827ad3ce002c97272e24847ccbe16b1b614 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 16:38:50 +0100
Subject: [PATCH 07/16] Convert Iceberg `timestamp` min/max above `1e18` from
 nanoseconds

Bounds at or below `1e18` stay microseconds, so ordinary `timestamp` pruning still runs. The previous threshold dropped those bounds and skipped the column.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Iceberg/IcebergFieldParseHelpers.cpp      | 48 ++++++++++++++++---
 .../gtest_iceberg_field_parse_helpers.cpp     | 22 ++++++++-
 2 files changed, 63 insertions(+), 7 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index 3b8d01b99bd3a..6b2566602500d 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -26,16 +26,52 @@ namespace Iceberg
 namespace
 {
 
-/// Iceberg Appendix D stores `timestamp` / `timestamptz` as microseconds and
-/// `timestamp_ns` / `timestamptz_ns` as nanoseconds. Both are an 8-byte
-/// little-endian int64. The ClickHouse type's scale is that unit.
-std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type)
+/// 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)`).
+/// Convert only when `|ticks| > 1e18`. A smaller value stays microseconds.
+std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound)
 {
     if (str.size() != sizeof(Int64))
         return std::nullopt;
 
-    const Int64 unscaled = unalignedLoadLittleEndian<Int64>(str.data());
+    Int64 unscaled = unalignedLoadLittleEndian<Int64>(str.data());
     const UInt32 scale = getDecimalScale(type);
+
+    if (scale == 6)
+    {
+        if (magnitudeExceeds(unscaled, nanoseconds_lower_threshold))
+            unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound);
+    }
+
     return DecimalField<DateTime64>(DateTime64(unscaled), scale);
 }
 
@@ -173,7 +209,7 @@ std::optional<Field> deserializeFieldFromBinaryRepr(const std::string & str, Dat
 {
     auto non_nullable_type = removeNullable(expected_type);
     if (WhichDataType(non_nullable_type).isDateTime64())
-        return deserializeDateTime64FromBinaryRepr(str, *non_nullable_type);
+        return deserializeDateTime64FromBinaryRepr(str, *non_nullable_type, lower_bound);
 
     auto column = non_nullable_type->createColumn();
     if (WhichDataType(non_nullable_type).isDecimal())
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index ec96121c11fe7..b522af192c970 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -47,11 +47,31 @@ TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept)
     ASSERT_TRUE(negative.has_value());
     EXPECT_EQ(decimal64Ticks(*negative), us_1960_01_01);
 
-    /// Spark `9999-12-31` sentinel is spec-correct microseconds.
+    /// Spark `9999-12-31` sentinel is spec-correct microseconds, below `1e18`.
     constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL;
     auto sentinel = deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false);
     ASSERT_TRUE(sentinel.has_value());
     EXPECT_EQ(decimal64Ticks(*sentinel), spark_9999_us);
+
+    /// `1e18` itself does not exceed the threshold, so it stays microseconds.
+    constexpr Int64 exactly_1e18 = 1'000'000'000'000'000'000LL;
+    auto edge = deserializeFieldFromBinaryRepr(littleEndianInt64(exactly_1e18), type, true);
+    ASSERT_TRUE(edge.has_value());
+    EXPECT_EQ(decimal64Ticks(*edge), exactly_1e18);
+}
+
+TEST(IcebergFieldParseHelpers, NanosecondBoundsAbove1e18OnTimestampAreConverted)
+{
+    auto type = std::make_shared<DataTypeDateTime64>(6);
+    /// #2240: `2026-08-11 19:02:06.576815622` stored as nanoseconds on `timestamp`.
+    constexpr Int64 ns_2026_08_11 = 1'786'474'926'576'815'622LL;
+    auto lower = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2026_08_11), type, true);
+    auto upper = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2026_08_11), type, false);
+    ASSERT_TRUE(lower.has_value());
+    ASSERT_TRUE(upper.has_value());
+    /// Division truncates toward zero; the upper bound widens by one microsecond.
+    EXPECT_EQ(decimal64Ticks(*lower), 1'786'474'926'576'815LL);
+    EXPECT_EQ(decimal64Ticks(*upper), 1'786'474'926'576'816LL);
 }
 
 TEST(IcebergFieldParseHelpers, NanosecondDateTime64BoundsAreKept)

From 66be23d429c461d4ab2f4116c237c95c2ecd9909 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 16:57:39 +0100
Subject: [PATCH 08/16] Check both units for Iceberg `timestamp` min/max
 between `1e16` and `1e18`

Bounds below `1e16` stay microseconds and bounds at or above `1e18` are nanoseconds. A file in the middle band is pruned only when the predicate misses both readings.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Iceberg/IcebergFieldParseHelpers.cpp      | 69 +++++++++++++++++--
 .../Iceberg/IcebergFieldParseHelpers.h        | 16 +++++
 .../Iceberg/ManifestFileIterator.cpp          | 29 ++++++--
 .../Iceberg/ManifestFilesPruning.cpp          | 19 ++++-
 .../DataLakes/Iceberg/ManifestFilesPruning.h  |  5 +-
 .../gtest_iceberg_field_parse_helpers.cpp     | 41 +++++++++--
 6 files changed, 160 insertions(+), 19 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index 6b2566602500d..3031c03aa3095 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -26,13 +26,14 @@ namespace Iceberg
 namespace
 {
 
-/// 1e18 ns is ~year 2001; customer 2026 ns bounds are `~1.76e18`.
+/// `1e16` us is `2286-11-20`. `1e18` ns is `2001-09-09`.
+constexpr Int64 microseconds_only_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)
+bool magnitudeBelow(Int64 value, Int64 threshold)
 {
-    return value > threshold || value < -threshold;
+    return value > -threshold && value < threshold;
 }
 
 /// Convert ns ticks to us ticks without shrinking a [lower, upper] range.
@@ -57,7 +58,8 @@ Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound)
 
 /// Iceberg timestamps are 8-byte little-endian (spec Appendix D).
 /// Some writers store nanosecond stats on Iceberg `timestamp` (`DateTime64(6)`).
-/// Convert only when `|ticks| > 1e18`. A smaller value stays microseconds.
+/// `|ticks| >= 1e18` is nanoseconds. A smaller value stays microseconds here;
+/// the min/max pruner also tries nanoseconds for `|ticks| >= 1e16`.
 std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound)
 {
     if (str.size() != sizeof(Int64))
@@ -68,13 +70,30 @@ std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str
 
     if (scale == 6)
     {
-        if (magnitudeExceeds(unscaled, nanoseconds_lower_threshold))
+        if (!magnitudeBelow(unscaled, nanoseconds_lower_threshold))
             unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound);
     }
 
     return DecimalField<DateTime64>(DateTime64(unscaled), scale);
 }
 
+std::optional<Int64> loadDateTime64Ticks(const std::string & str)
+{
+    if (str.size() != sizeof(Int64))
+        return std::nullopt;
+    return unalignedLoadLittleEndian<Int64>(str.data());
+}
+
+void appendRange(std::vector<DateTime64BoundRange> & ranges, Int64 lower_ticks, Int64 upper_ticks, UInt32 scale)
+{
+    if (lower_ticks > upper_ticks)
+        return;
+
+    ranges.push_back(DateTime64BoundRange{
+        DecimalField<DateTime64>(DateTime64(lower_ticks), scale),
+        DecimalField<DateTime64>(DateTime64(upper_ticks), scale)});
+}
+
 }
 
 Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name)
@@ -205,6 +224,46 @@ static std::optional<Field> deserializeDecimalBound(const std::string & str, UIn
     return DecimalField<DecimalType>(unscaled_value, scale);
 }
 
+std::vector<DateTime64BoundRange> deserializeDateTime64MinMaxRanges(
+    const std::string & lower_str,
+    const std::string & upper_str,
+    DataTypePtr type)
+{
+    std::vector<DateTime64BoundRange> ranges;
+    const auto lower_ticks = loadDateTime64Ticks(lower_str);
+    const auto upper_ticks = loadDateTime64Ticks(upper_str);
+    if (!lower_ticks || !upper_ticks)
+        return ranges;
+
+    const UInt32 scale = getDecimalScale(*removeNullable(type));
+    if (scale != 6)
+    {
+        appendRange(ranges, *lower_ticks, *upper_ticks, scale);
+        return ranges;
+    }
+
+    /// Both ends must accept a unit. A lower bound below `1e16` and an upper bound
+    /// of `1e18` produce no range, and min/max pruning skips the column.
+    const bool allow_microseconds = magnitudeBelow(*lower_ticks, nanoseconds_lower_threshold)
+        && magnitudeBelow(*upper_ticks, nanoseconds_lower_threshold);
+    const bool allow_nanoseconds = !magnitudeBelow(*lower_ticks, microseconds_only_threshold)
+        && !magnitudeBelow(*upper_ticks, microseconds_only_threshold);
+
+    if (allow_microseconds)
+        appendRange(ranges, *lower_ticks, *upper_ticks, scale);
+
+    if (allow_nanoseconds)
+    {
+        appendRange(
+            ranges,
+            nanosecondsToMicrosecondsForBound(*lower_ticks, true),
+            nanosecondsToMicrosecondsForBound(*upper_ticks, false),
+            scale);
+    }
+
+    return ranges;
+}
+
 std::optional<Field> deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound)
 {
     auto non_nullable_type = removeNullable(expected_type);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
index 21156647fd4b4..ad8d023b5c7af 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
@@ -32,6 +32,22 @@ std::vector<Int64> fieldToInt64Array(const Field & value, std::string_view conte
 /// See https://iceberg.apache.org/spec/#appendix-d-single-value-serialization
 std::optional<Field> deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound);
 
+/// One interpretation of an Iceberg `timestamp` lower/upper pair.
+struct DateTime64BoundRange
+{
+    Field lower;
+    Field upper;
+};
+
+/// `DateTime64(6)` min/max used for file pruning.
+/// `|ticks| < 1e16` is microseconds. `|ticks| >= 1e18` is nanoseconds.
+/// A value between them can be either, so both ranges are returned.
+/// `DateTime64(9)` returns the raw nanosecond range.
+std::vector<DateTime64BoundRange> deserializeDateTime64MinMaxRanges(
+    const std::string & lower_str,
+    const std::string & upper_str,
+    DataTypePtr type);
+
 }
 
 #endif
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
index 5ef953b9a2086..aa31da13f6734 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
@@ -87,7 +87,9 @@ namespace
         return false;
     }
 
-    void addRowLineageHyperrectangles(std::unordered_map<Int32, DB::Range> & hyperrectangles, const ProcessedManifestFileEntry & entry)
+    void addRowLineageHyperrectangles(
+        std::unordered_map<Int32, std::vector<DB::Range>> & hyperrectangles,
+        const ProcessedManifestFileEntry & entry)
     {
         const auto & parsed_entry = *entry.parsed_entry;
         if (!entry.first_row_id.has_value() || parsed_entry.record_count <= 0 || entry.sequence_number < 0)
@@ -110,17 +112,17 @@ namespace
             {
                 if (column_presence_is_known)
                 {
-                    hyperrectangles.emplace(field_id, DB::Range(inherited_lower_bound, true, inherited_upper_bound, true));
+                    hyperrectangles[field_id].emplace_back(inherited_lower_bound, true, inherited_upper_bound, true);
                     continue;
                 }
             }
             else if (auto range = getMaterializedRowLineageRange(parsed_entry, field_id))
             {
-                hyperrectangles.emplace(field_id, *range);
+                hyperrectangles[field_id].emplace_back(*range);
                 continue;
             }
 
-            hyperrectangles.emplace(field_id, DB::Range(UInt64(0), true, inherited_upper_bound, true));
+            hyperrectangles[field_id].emplace_back(UInt64(0), true, inherited_upper_bound, true);
         }
     }
 }
@@ -500,7 +502,7 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
     if (filter_dag)
     {
         /// Compute per-column hyperrectangles for DATA files
-        std::unordered_map<Int32, DB::Range> hyperrectangles;
+        std::unordered_map<Int32, std::vector<DB::Range>> hyperrectangles;
         if (parsed_entry->content_type == FileContentType::DATA)
         {
             for (const auto & [column_id, bounds] : parsed_entry->value_bounds)
@@ -525,6 +527,21 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                     || type_id == DB::TypeIndex::Variant)
                     continue;
 
+                if (WhichDataType(name_and_type.type).isDateTime64())
+                {
+                    /// `|ticks| < 1e16` is one microsecond range. `|ticks| >= 1e18` is one
+                    /// nanosecond range. Values between them produce both, and the file is
+                    /// pruned only when the predicate misses every range.
+                    const auto timestamp_ranges = deserializeDateTime64MinMaxRanges(left_str, right_str, name_and_type.type);
+                    if (timestamp_ranges.empty())
+                        continue;
+
+                    auto & ranges = hyperrectangles[column_id];
+                    for (const auto & timestamp_range : timestamp_ranges)
+                        ranges.emplace_back(timestamp_range.lower, true, timestamp_range.upper, true);
+                    continue;
+                }
+
                 auto left = deserializeFieldFromBinaryRepr(left_str, name_and_type.type, true);
                 auto right = deserializeFieldFromBinaryRepr(right_str, name_and_type.type, false);
                 if (!left || !right)
@@ -533,7 +550,7 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                 if (DB::Range::less(*right, *left))
                     continue;
 
-                hyperrectangles.emplace(column_id, DB::Range(*left, true, *right, true));
+                hyperrectangles[column_id].emplace_back(*left, true, *right, true);
             }
 
             addRowLineageHyperrectangles(hyperrectangles, *entry);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp
index c2d400056a47b..662671c352a51 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp
@@ -274,7 +274,8 @@ Field reinterpretPartitionDateTime64(const Field & field, const IDataType & type
 }
 
 PruningReturnStatus ManifestFilesPruner::canBePruned(
-    const ProcessedManifestFileEntryPtr & entry, const std::unordered_map<Int32, DB::Range> & entry_hyperrectangles) const
+    const ProcessedManifestFileEntryPtr & entry,
+    const std::unordered_map<Int32, std::vector<DB::Range>> & entry_hyperrectangles) const
 {
     if (partition_key_condition.has_value())
     {
@@ -344,7 +345,21 @@ PruningReturnStatus ManifestFilesPruner::canBePruned(
         if (rect_it == entry_hyperrectangles.end())
             continue;
 
-        if (has_no_nulls && !key_condition.mayBeTrueInRange(1, &rect_it->second.left, &rect_it->second.right, {name_and_type->type}))
+        if (!has_no_nulls)
+            continue;
+
+        /// A `timestamp` column in `[1e16, 1e18)` has two ranges. Prune only when every range misses.
+        bool may_be_true = false;
+        for (const auto & range : rect_it->second)
+        {
+            if (key_condition.mayBeTrueInRange(1, &range.left, &range.right, {name_and_type->type}))
+            {
+                may_be_true = true;
+                break;
+            }
+        }
+
+        if (!may_be_true)
         {
             return PruningReturnStatus::MIN_MAX_INDEX_PRUNED;
         }
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h
index 732edb21b58da..3a228fc6b981c 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h
@@ -3,6 +3,7 @@
 
 #include <optional>
 #include <unordered_map>
+#include <vector>
 
 #include <Core/NamesAndTypes.h>
 #include <Parsers/IAST_fwd.h>
@@ -66,7 +67,9 @@ class ManifestFilesPruner
         const ManifestFileIterator & manifest_file,
         DB::ContextPtr context);
 
-    PruningReturnStatus canBePruned(const ProcessedManifestFileEntryPtr & entry, const std::unordered_map<Int32, DB::Range> & entry_hyperrectangles) const;
+    PruningReturnStatus canBePruned(
+        const ProcessedManifestFileEntryPtr & entry,
+        const std::unordered_map<Int32, std::vector<DB::Range>> & entry_hyperrectangles) const;
 };
 
 }
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index b522af192c970..6f089e9d9450f 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -52,12 +52,33 @@ TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept)
     auto sentinel = deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false);
     ASSERT_TRUE(sentinel.has_value());
     EXPECT_EQ(decimal64Ticks(*sentinel), spark_9999_us);
+}
 
-    /// `1e18` itself does not exceed the threshold, so it stays microseconds.
-    constexpr Int64 exactly_1e18 = 1'000'000'000'000'000'000LL;
-    auto edge = deserializeFieldFromBinaryRepr(littleEndianInt64(exactly_1e18), type, true);
-    ASSERT_TRUE(edge.has_value());
-    EXPECT_EQ(decimal64Ticks(*edge), exactly_1e18);
+TEST(IcebergFieldParseHelpers, TimestampBoundsUseBothRangesBetween1e16And1e18)
+{
+    auto type = std::make_shared<DataTypeDateTime64>(6);
+    constexpr Int64 middle = 200'000'000'000'000'000LL;
+    const auto bytes = littleEndianInt64(middle);
+    auto ranges = deserializeDateTime64MinMaxRanges(bytes, bytes, type);
+    ASSERT_EQ(ranges.size(), 2);
+    EXPECT_EQ(decimal64Ticks(ranges[0].lower), middle);
+    EXPECT_EQ(decimal64Ticks(ranges[0].upper), middle);
+    EXPECT_EQ(decimal64Ticks(ranges[1].lower), middle / 1000);
+    EXPECT_EQ(decimal64Ticks(ranges[1].upper), middle / 1000);
+
+    /// Below `1e16` there is no nanosecond interpretation.
+    constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL;
+    const auto small = littleEndianInt64(us_2024_02_20);
+    auto microseconds = deserializeDateTime64MinMaxRanges(small, small, type);
+    ASSERT_EQ(microseconds.size(), 1);
+    EXPECT_EQ(decimal64Ticks(microseconds[0].lower), us_2024_02_20);
+
+    /// A small lower bound keeps the Spark sentinel as microseconds only.
+    constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL;
+    auto sentinel = deserializeDateTime64MinMaxRanges(small, littleEndianInt64(spark_9999_us), type);
+    ASSERT_EQ(sentinel.size(), 1);
+    EXPECT_EQ(decimal64Ticks(sentinel[0].lower), us_2024_02_20);
+    EXPECT_EQ(decimal64Ticks(sentinel[0].upper), spark_9999_us);
 }
 
 TEST(IcebergFieldParseHelpers, NanosecondBoundsAbove1e18OnTimestampAreConverted)
@@ -72,6 +93,16 @@ TEST(IcebergFieldParseHelpers, NanosecondBoundsAbove1e18OnTimestampAreConverted)
     /// Division truncates toward zero; the upper bound widens by one microsecond.
     EXPECT_EQ(decimal64Ticks(*lower), 1'786'474'926'576'815LL);
     EXPECT_EQ(decimal64Ticks(*upper), 1'786'474'926'576'816LL);
+
+    /// `1e18` itself is nanoseconds. The pair has only the converted range.
+    constexpr Int64 exactly_1e18 = 1'000'000'000'000'000'000LL;
+    auto edge = deserializeFieldFromBinaryRepr(littleEndianInt64(exactly_1e18), type, true);
+    ASSERT_TRUE(edge.has_value());
+    EXPECT_EQ(decimal64Ticks(*edge), exactly_1e18 / 1000);
+    const auto edge_bytes = littleEndianInt64(exactly_1e18);
+    auto edge_ranges = deserializeDateTime64MinMaxRanges(edge_bytes, edge_bytes, type);
+    ASSERT_EQ(edge_ranges.size(), 1);
+    EXPECT_EQ(decimal64Ticks(edge_ranges[0].lower), exactly_1e18 / 1000);
 }
 
 TEST(IcebergFieldParseHelpers, NanosecondDateTime64BoundsAreKept)

From 9cd745a4fb814e9ea01122abf912b97f4fa7b5b3 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 17:16:14 +0100
Subject: [PATCH 09/16] Apply Iceberg `timestamp` min/max bands to optional
 columns

Optional fields are `Nullable(DateTime64)`, so the outer type check skipped both units between `1e16` and `1e18`.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../DataLakes/Iceberg/ManifestFileIterator.cpp             | 3 ++-
 .../Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp    | 7 +++++++
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
index aa31da13f6734..4465d71ce3e80 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
@@ -527,7 +527,8 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                     || type_id == DB::TypeIndex::Variant)
                     continue;
 
-                if (WhichDataType(name_and_type.type).isDateTime64())
+                /// Optional Iceberg fields are `Nullable(DateTime64)`. The outer type is not `DateTime64`.
+                if (WhichDataType(removeNullable(name_and_type.type)).isDateTime64())
                 {
                     /// `|ticks| < 1e16` is one microsecond range. `|ticks| >= 1e18` is one
                     /// nanosecond range. Values between them produce both, and the file is
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index 6f089e9d9450f..09055d10e0bdd 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -7,6 +7,7 @@
 #include <base/unaligned.h>
 #include <Core/Field.h>
 #include <DataTypes/DataTypeDateTime64.h>
+#include <DataTypes/DataTypeNullable.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h>
 
 using namespace DB;
@@ -66,6 +67,12 @@ TEST(IcebergFieldParseHelpers, TimestampBoundsUseBothRangesBetween1e16And1e18)
     EXPECT_EQ(decimal64Ticks(ranges[1].lower), middle / 1000);
     EXPECT_EQ(decimal64Ticks(ranges[1].upper), middle / 1000);
 
+    /// Optional Iceberg `timestamp` is `Nullable(DateTime64(6))` and uses the same two ranges.
+    auto nullable_ranges = deserializeDateTime64MinMaxRanges(bytes, bytes, makeNullable(type));
+    ASSERT_EQ(nullable_ranges.size(), 2);
+    EXPECT_EQ(decimal64Ticks(nullable_ranges[0].lower), middle);
+    EXPECT_EQ(decimal64Ticks(nullable_ranges[1].lower), middle / 1000);
+
     /// Below `1e16` there is no nanosecond interpretation.
     constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL;
     const auto small = littleEndianInt64(us_2024_02_20);

From 604a615cc91b4257eb22c8804481047b8606a239 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 17:23:38 +0100
Subject: [PATCH 10/16] Skip Iceberg constant columns for `timestamp` bounds
 between `1e16` and `1e18`

That band can be either unit, so the shortcut must read the file. Pruning and the shortcut share the same two thresholds.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../DataLakes/IDataLakeMetadata.cpp           | 20 ++++++++++++++++++-
 .../Iceberg/IcebergFieldParseHelpers.cpp      | 18 +++++------------
 .../Iceberg/IcebergFieldParseHelpers.h        | 17 ++++++++++++++++
 .../gtest_iceberg_field_parse_helpers.cpp     | 10 ++++++++++
 4 files changed, 51 insertions(+), 14 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
index 562e7ec40938c..c0f6ff1101727 100644
--- a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
@@ -2,7 +2,10 @@
 #include <Storages/ObjectStorage/StorageObjectStorageSource.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h>
+#include <base/unaligned.h>
 #include <Core/TypeId.h>
+#include <DataTypes/DataTypeNullable.h>
+#include <DataTypes/DataTypesDecimal.h>
 #include <IO/ReadHelpers.h>
 #include <IO/WriteHelpers.h>
 #include <Core/Field.h>
@@ -158,7 +161,22 @@ DataFileMetaInfo::DataFileMetaInfo(
                     auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true);
                     auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false);
                     if (left && right && !DB::Range::less(*right, *left))
-                        hyperrectangle = DB::Range(*left, true, *right, true);
+                    {
+                        const auto non_nullable_type = removeNullable(type);
+                        bool ambiguous_timestamp = false;
+                        if (WhichDataType(non_nullable_type).isDateTime64() && getDecimalScale(*non_nullable_type) == 6
+                            && left_str.size() == sizeof(Int64) && right_str.size() == sizeof(Int64))
+                        {
+                            const Int64 left_ticks = unalignedLoadLittleEndian<Int64>(left_str.data());
+                            const Int64 right_ticks = unalignedLoadLittleEndian<Int64>(right_str.data());
+                            /// `[1e16, 1e18)` can be microseconds or nanoseconds, so the constant-column
+                            /// shortcut must not substitute either reading.
+                            ambiguous_timestamp = Iceberg::isAmbiguousIcebergTimestampBound(left_ticks)
+                                || Iceberg::isAmbiguousIcebergTimestampBound(right_ticks);
+                        }
+                        if (!ambiguous_timestamp)
+                            hyperrectangle = DB::Range(*left, true, *right, true);
+                    }
                 }
             }
         }
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index 3031c03aa3095..1162f9cc69bce 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -26,16 +26,8 @@ namespace Iceberg
 namespace
 {
 
-/// `1e16` us is `2286-11-20`. `1e18` ns is `2001-09-09`.
-constexpr Int64 microseconds_only_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 magnitudeBelow(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)
 {
@@ -70,7 +62,7 @@ std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str
 
     if (scale == 6)
     {
-        if (!magnitudeBelow(unscaled, nanoseconds_lower_threshold))
+        if (!icebergTimestampMagnitudeBelow(unscaled, iceberg_timestamp_nanoseconds_lower_threshold))
             unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound);
     }
 
@@ -244,10 +236,10 @@ std::vector<DateTime64BoundRange> deserializeDateTime64MinMaxRanges(
 
     /// Both ends must accept a unit. A lower bound below `1e16` and an upper bound
     /// of `1e18` produce no range, and min/max pruning skips the column.
-    const bool allow_microseconds = magnitudeBelow(*lower_ticks, nanoseconds_lower_threshold)
-        && magnitudeBelow(*upper_ticks, nanoseconds_lower_threshold);
-    const bool allow_nanoseconds = !magnitudeBelow(*lower_ticks, microseconds_only_threshold)
-        && !magnitudeBelow(*upper_ticks, microseconds_only_threshold);
+    const bool allow_microseconds = icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_nanoseconds_lower_threshold)
+        && icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_nanoseconds_lower_threshold);
+    const bool allow_nanoseconds = !icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_microseconds_only_threshold)
+        && !icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_microseconds_only_threshold);
 
     if (allow_microseconds)
         appendRange(ranges, *lower_ticks, *upper_ticks, scale);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
index ad8d023b5c7af..2e8b8076997d9 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
@@ -16,6 +16,23 @@
 namespace DB::Iceberg
 {
 
+/// `iceberg_timestamp_microseconds_only_threshold` microseconds is `2286-11-20`.
+/// `iceberg_timestamp_nanoseconds_lower_threshold` nanoseconds is `2001-09-09`.
+constexpr Int64 iceberg_timestamp_microseconds_only_threshold = 10'000'000'000'000'000LL;
+constexpr Int64 iceberg_timestamp_nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL;
+
+inline bool icebergTimestampMagnitudeBelow(Int64 value, Int64 threshold)
+{
+    return value > -threshold && value < threshold;
+}
+
+/// `[iceberg_timestamp_microseconds_only_threshold, iceberg_timestamp_nanoseconds_lower_threshold)`.
+inline bool isAmbiguousIcebergTimestampBound(Int64 ticks)
+{
+    return !icebergTimestampMagnitudeBelow(ticks, iceberg_timestamp_microseconds_only_threshold)
+        && icebergTimestampMagnitudeBelow(ticks, iceberg_timestamp_nanoseconds_lower_threshold);
+}
+
 /// Convert a Field to Int64, accepting Int64 and UInt64 (with overflow check).
 Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name);
 
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index 09055d10e0bdd..ce42947aad74e 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -55,6 +55,16 @@ TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept)
     EXPECT_EQ(decimal64Ticks(*sentinel), spark_9999_us);
 }
 
+TEST(IcebergFieldParseHelpers, AmbiguousTimestampBoundsAreTheOpenMiddleBand)
+{
+    EXPECT_FALSE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_microseconds_only_threshold - 1));
+    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_microseconds_only_threshold));
+    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(200'000'000'000'000'000LL));
+    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(-200'000'000'000'000'000LL));
+    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_nanoseconds_lower_threshold - 1));
+    EXPECT_FALSE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_nanoseconds_lower_threshold));
+}
+
 TEST(IcebergFieldParseHelpers, TimestampBoundsUseBothRangesBetween1e16And1e18)
 {
     auto type = std::make_shared<DataTypeDateTime64>(6);

From 12930fc83b12d277af31a766eb1e95a1437e5713 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 17:37:31 +0100
Subject: [PATCH 11/16] Write Iceberg `DateTime64` min/max bounds little-endian

The reader already loads these bounds as little-endian `long`s. `test_minmax_pruning` now checks Spark-written `timestamp` values with `unix_micros`.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../Iceberg/IcebergFieldParseHelpers.cpp      |  7 +++++++
 .../Iceberg/IcebergFieldParseHelpers.h        |  3 +++
 .../DataLakes/Iceberg/IcebergWrites.cpp       | 11 ++++++++++-
 .../gtest_iceberg_field_parse_helpers.cpp     | 19 +++++++++++++++++++
 .../test_minmax_pruning.py                    |  9 +++++++++
 5 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index 1162f9cc69bce..e08ec74c85861 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -88,6 +88,13 @@ void appendRange(std::vector<DateTime64BoundRange> & ranges, Int64 lower_ticks,
 
 }
 
+std::string serializeDateTime64IcebergBound(Int64 ticks)
+{
+    std::string bytes(sizeof(Int64), '\0');
+    unalignedStoreLittleEndian<Int64>(bytes.data(), ticks);
+    return bytes;
+}
+
 Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name)
 {
     if (value.getType() == Field::Types::Int64)
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
index 2e8b8076997d9..f0fd586d9d5cb 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
@@ -49,6 +49,9 @@ std::vector<Int64> fieldToInt64Array(const Field & value, std::string_view conte
 /// See https://iceberg.apache.org/spec/#appendix-d-single-value-serialization
 std::optional<Field> deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound);
 
+/// Appendix D stores `timestamp` and `timestamp_ns` as an 8-byte little-endian `long`.
+std::string serializeDateTime64IcebergBound(Int64 ticks);
+
 /// One interpretation of an Iceberg `timestamp` lower/upper pair.
 struct DateTime64BoundRange
 {
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
index e07dff8ff28f8..a6d42b1c51acf 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
@@ -37,6 +37,7 @@
 #include <Storages/ObjectStorage/DataLakes/DataLakeStorageSettings.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/AvroSchema.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/Constant.h>
+#include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h>
@@ -163,6 +164,8 @@ bool isNaNPartitionValue(const Field & field, DataTypePtr type)
     }
 }
 
+/// Integers, floats, dates and times are copied in host order. Their reader uses `insertData`, which
+/// expects that layout. `DateTime64` is the exception: it is loaded with `unalignedLoadLittleEndian`.
 template <typename T>
 std::vector<uint8_t> dumpValue(T value)
 {
@@ -171,6 +174,12 @@ std::vector<uint8_t> dumpValue(T value)
     return bytes;
 }
 
+std::vector<uint8_t> dumpDateTime64Bound(Int64 ticks)
+{
+    const auto bytes = Iceberg::serializeDateTime64IcebergBound(ticks);
+    return std::vector<uint8_t>(bytes.begin(), bytes.end());
+}
+
 DataTypePtr getTimeTypeOrNull(DataTypePtr type)
 {
     if (type->isNullable())
@@ -349,7 +358,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(getDateTime64ValueForIcebergBounds(field, type));
+            return dumpDateTime64Bound(getDateTime64ValueForIcebergBounds(field, type));
         case TypeIndex::String:
         {
             auto value = field.safeGet<String>();
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index ce42947aad74e..a8878d480e504 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -122,6 +122,25 @@ TEST(IcebergFieldParseHelpers, NanosecondBoundsAbove1e18OnTimestampAreConverted)
     EXPECT_EQ(decimal64Ticks(edge_ranges[0].lower), exactly_1e18 / 1000);
 }
 
+TEST(IcebergFieldParseHelpers, DateTime64BoundsAreStoredLittleEndian)
+{
+    auto type = std::make_shared<DataTypeDateTime64>(6);
+    /// 2024-02-20 10:00:00 UTC. The written bytes must be little-endian, then read back unchanged.
+    constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL;
+    const auto bytes = serializeDateTime64IcebergBound(us_2024_02_20);
+    EXPECT_EQ(bytes, littleEndianInt64(us_2024_02_20));
+    auto field = deserializeFieldFromBinaryRepr(bytes, type, true);
+    ASSERT_TRUE(field.has_value());
+    EXPECT_EQ(decimal64Ticks(*field), us_2024_02_20);
+
+    constexpr Int64 us_1960_01_01 = -315'619'200'000'000LL;
+    const auto negative_bytes = serializeDateTime64IcebergBound(us_1960_01_01);
+    EXPECT_EQ(negative_bytes, littleEndianInt64(us_1960_01_01));
+    auto negative = deserializeFieldFromBinaryRepr(negative_bytes, type, false);
+    ASSERT_TRUE(negative.has_value());
+    EXPECT_EQ(decimal64Ticks(*negative), us_1960_01_01);
+}
+
 TEST(IcebergFieldParseHelpers, NanosecondDateTime64BoundsAreKept)
 {
     auto type = std::make_shared<DataTypeDateTime64>(9);
diff --git a/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py b/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py
index c62e6029b8513..5265d08769e1c 100644
--- a/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py
+++ b/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py
@@ -81,6 +81,15 @@ def execute_spark_query(query: str):
         ))
         creation_expression = TABLE_NAME
 
+    spark_timestamps = spark.sql(
+        f"SELECT unix_micros(ts), unix_micros(time_struct.b) FROM {TABLE_NAME} ORDER BY tag"
+    ).collect()
+    expected_timestamps = "".join(f"{row[0]}\t{row[1]}\n" for row in spark_timestamps)
+    assert instance.query(
+        f"SELECT toUnixTimestamp64Micro(ts), toUnixTimestamp64Micro(time_struct.b) "
+        f"FROM {creation_expression} ORDER BY tag"
+    ) == expected_timestamps
+
     def check_validity_and_get_prunned_files(select_expression):
         settings1 = {
             "use_iceberg_partition_pruning": 0,

From 237fb0baaaf7fcaf80d342b4efdc16e2e0e9257d Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 18:29:07 +0100
Subject: [PATCH 12/16] Keep Iceberg `timestamp` bounds below `1e16` as
 microseconds

`86400000000000` is `1972-09-27` as microseconds. The test locks that single range so it is not read as `1970-01-02` nanoseconds.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../gtest_iceberg_field_parse_helpers.cpp     | 27 ++++++++++++++++++-
 1 file changed, 26 insertions(+), 1 deletion(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index a8878d480e504..1129a9effe102 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -83,7 +83,7 @@ TEST(IcebergFieldParseHelpers, TimestampBoundsUseBothRangesBetween1e16And1e18)
     EXPECT_EQ(decimal64Ticks(nullable_ranges[0].lower), middle);
     EXPECT_EQ(decimal64Ticks(nullable_ranges[1].lower), middle / 1000);
 
-    /// Below `1e16` there is no nanosecond interpretation.
+    /// `2024-02-20` is below `1e16`, so it has one microsecond range.
     constexpr Int64 us_2024_02_20 = 1'708'423'200'000'000LL;
     const auto small = littleEndianInt64(us_2024_02_20);
     auto microseconds = deserializeDateTime64MinMaxRanges(small, small, type);
@@ -98,6 +98,31 @@ TEST(IcebergFieldParseHelpers, TimestampBoundsUseBothRangesBetween1e16And1e18)
     EXPECT_EQ(decimal64Ticks(sentinel[0].upper), spark_9999_us);
 }
 
+TEST(IcebergFieldParseHelpers, BoundsBelow1e16StayMicroseconds)
+{
+    auto type = std::make_shared<DataTypeDateTime64>(6);
+    /// `86400000000000` is one day in nanoseconds (`1970-01-02`) and 1000 days in
+    /// microseconds (`1972-09-27`). It is below `1e16`, so Iceberg `timestamp` keeps
+    /// the integer and does not add a nanosecond range. The constant-column shortcut
+    /// may use this point: it is not the ambiguous middle band.
+    constexpr Int64 ticks = 86'400'000'000'000LL;
+    EXPECT_FALSE(isAmbiguousIcebergTimestampBound(ticks));
+    EXPECT_LT(ticks, iceberg_timestamp_microseconds_only_threshold);
+
+    const auto bytes = littleEndianInt64(ticks);
+    auto lower = deserializeFieldFromBinaryRepr(bytes, type, true);
+    auto upper = deserializeFieldFromBinaryRepr(bytes, type, false);
+    ASSERT_TRUE(lower.has_value());
+    ASSERT_TRUE(upper.has_value());
+    EXPECT_EQ(decimal64Ticks(*lower), ticks);
+    EXPECT_EQ(decimal64Ticks(*upper), ticks);
+
+    auto ranges = deserializeDateTime64MinMaxRanges(bytes, bytes, type);
+    ASSERT_EQ(ranges.size(), 1);
+    EXPECT_EQ(decimal64Ticks(ranges[0].lower), ticks);
+    EXPECT_EQ(decimal64Ticks(ranges[0].upper), ticks);
+}
+
 TEST(IcebergFieldParseHelpers, NanosecondBoundsAbove1e18OnTimestampAreConverted)
 {
     auto type = std::make_shared<DataTypeDateTime64>(6);

From a0af1c8ac113736a3d58f956440a267b6784d334 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 21:35:40 +0100
Subject: [PATCH 13/16] Broadcast an old-analyzer `global` join without reusing
 a taken alias

A table function on the right is read through `interpretSubquery`. `IN some_table` runs on the initiator, and `IN (SELECT 1)` stays on the replica.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Interpreters/GlobalSubqueriesVisitor.cpp  | 24 ++++---
 src/Storages/IStorageCluster.cpp              | 71 +++++++++++++++++--
 .../test_remote_initiator.py                  | 59 +++++++++++++++
 3 files changed, 139 insertions(+), 15 deletions(-)

diff --git a/src/Interpreters/GlobalSubqueriesVisitor.cpp b/src/Interpreters/GlobalSubqueriesVisitor.cpp
index 7ac4337fd92be..a1261e64f0060 100644
--- a/src/Interpreters/GlobalSubqueriesVisitor.cpp
+++ b/src/Interpreters/GlobalSubqueriesVisitor.cpp
@@ -54,13 +54,16 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names
 
     if (const auto * ast_table_expr = ast->as<ASTTableExpression>())
     {
-        subquery_or_table_name = ast_table_expr->subquery;
-
+        /// `interpretSubquery` reads a table function. A `GLOBAL JOIN s3(...)` is that shape.
         if (ast_table_expr->database_and_table_name)
         {
             subquery_or_table_name = ast_table_expr->database_and_table_name;
             is_table = true;
         }
+        else if (ast_table_expr->table_function)
+            subquery_or_table_name = ast_table_expr->table_function;
+        else
+            subquery_or_table_name = ast_table_expr->subquery;
     }
     else if (ast->as<ASTTableIdentifier>())
     {
@@ -87,14 +90,14 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names
     }
 
     String alias = subquery_or_table_name->tryGetAlias();
-    String external_table_name;
-    if (alias.empty())
-    {
-        auto hash = subquery_or_table_name->getTreeHash(/*ignore_aliases=*/ true);
-        external_table_name = fmt::format("_data_{}", toString(hash));
-    }
-    else
-        external_table_name = alias;
+    const auto hash = subquery_or_table_name->getTreeHash(/*ignore_aliases=*/ true);
+    const String hashed_name = fmt::format("_data_{}", toString(hash));
+    /// The alias is the SQL name. A session temporary table with that name must stay untouched,
+    /// so the broadcast is stored under `_data_{hash}` and the alias is only `AS alias`.
+    const bool alias_taken = !alias.empty()
+        && (external_tables.contains(alias)
+            || static_cast<bool>(getContext()->tryResolveStorageID(StorageID("", alias), Context::ResolveExternal)));
+    const String external_table_name = (alias.empty() || alias_taken) ? hashed_name : alias;
 
     /** We replace the subquery with the name of the temporary table.
         * It is in this form, the request will go to the remote server.
@@ -115,6 +118,7 @@ void GlobalSubqueriesMatcher::Data::addExternalStorage(ASTPtr & ast, const Names
     if (auto * ast_table_expr = ast->as<ASTTableExpression>())
     {
         ast_table_expr->subquery.reset();
+        ast_table_expr->table_function.reset();
         ast_table_expr->database_and_table_name = database_and_table_name;
 
         ast_table_expr->children.clear();
diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp
index e1294e3c0c628..9e7dc440d2787 100644
--- a/src/Storages/IStorageCluster.cpp
+++ b/src/Storages/IStorageCluster.cpp
@@ -371,7 +371,64 @@ bool queryHasRightOrFullJoin(const ASTPtr & query)
     return false;
 }
 
-void rewriteASTJoinsToGlobal(ASTPtr & query)
+bool astContainsTableIdentifier(const ASTPtr & node)
+{
+    if (!node)
+        return false;
+
+    if (node->as<ASTTableIdentifier>())
+        return true;
+
+    for (const auto & child : node->children)
+        if (astContainsTableIdentifier(child))
+            return true;
+
+    return false;
+}
+
+/// A named table in `IN` is read where the client connected. `IN (SELECT 1)` has no table and stays on the replica.
+bool inRightSideIsInitiatorLocal(const ASTPtr & rhs)
+{
+    if (!rhs)
+        return false;
+
+    if (rhs->as<ASTIdentifier>() || rhs->as<ASTTableIdentifier>())
+        return true;
+
+    if (rhs->as<ASTSubquery>() || rhs->as<ASTSelectQuery>() || rhs->as<ASTSelectWithUnionQuery>())
+        return astContainsTableIdentifier(rhs);
+
+    return false;
+}
+
+void rewriteInitiatorLocalInToGlobalIn(ASTPtr & node)
+{
+    if (!node)
+        return;
+
+    if (auto * function = node->as<ASTFunction>())
+    {
+        if (isNameOfLocalInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2
+            && inRightSideIsInitiatorLocal(function->arguments->children[1]))
+            function->name = getGlobalInFunctionNameForLocalInFunctionName(function->name);
+    }
+
+    for (auto & child : node->children)
+    {
+        if (astIsNestedSelect(child))
+            continue;
+        rewriteInitiatorLocalInToGlobalIn(child);
+    }
+}
+
+enum class GlobalInRewrite : uint8_t
+{
+    None,
+    InitiatorLocal,
+    All,
+};
+
+void rewriteASTJoinsToGlobal(ASTPtr & query, GlobalInRewrite in_rewrite)
 {
     ASTSelectQuery * select_query = query->as<ASTSelectQuery>();
     if (!select_query)
@@ -381,7 +438,7 @@ void rewriteASTJoinsToGlobal(ASTPtr & query)
             if (union_query->list_of_selects)
             {
                 for (auto & child : union_query->list_of_selects->children)
-                    rewriteASTJoinsToGlobal(child);
+                    rewriteASTJoinsToGlobal(child, in_rewrite);
             }
         }
         return;
@@ -402,14 +459,17 @@ void rewriteASTJoinsToGlobal(ASTPtr & query)
         }
     }
 
-    rewriteASTInFunctionsToGlobalIn(query);
+    if (in_rewrite == GlobalInRewrite::All)
+        rewriteASTInFunctionsToGlobalIn(query);
+    else if (in_rewrite == GlobalInRewrite::InitiatorLocal)
+        rewriteInitiatorLocalInToGlobalIn(query);
 }
 
 }
 
 void IStorageCluster::rewriteASTForGlobalJoin(ASTPtr & query)
 {
-    rewriteASTJoinsToGlobal(query);
+    rewriteASTJoinsToGlobal(query, GlobalInRewrite::All);
 }
 
 /// The initiator interpreter usually does this already: `IStorageCluster::isRemote` is true, so
@@ -421,7 +481,8 @@ namespace
 {
 void broadcastOldAnalyzerGlobalJoin(ASTPtr & query_to_send, const SelectQueryInfo & query_info, const ContextPtr & context)
 {
-    IStorageCluster::rewriteASTForGlobalJoin(query_to_send);
+    /// `IN (SELECT 1)` stays on the replica. `IN some_table` is read on the initiator.
+    rewriteASTJoinsToGlobal(query_to_send, GlobalInRewrite::InitiatorLocal);
 
     auto prepared_sets = query_info.prepared_sets ? query_info.prepared_sets : std::make_shared<PreparedSets>();
     TemporaryTablesMapping external_tables;
diff --git a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
index 22612fee96018..be776423bf398 100644
--- a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
+++ b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
@@ -158,3 +158,62 @@ def execute_spark_query(query: str):
                 allow_experimental_analyzer=0
             """)
         assert res == "1\t1\t2\n"
+
+    # The alias `t2` is already a temporary table with a different row. The join must use the Memory row.
+    # One client session, so the temporary table is visible to the SELECT.
+    res = instance.exec_in_container(
+        [
+            "clickhouse-client",
+            "--multiquery",
+            "--query",
+            f"""
+            CREATE TEMPORARY TABLE t2 (tag INT, number2 INT);
+            INSERT INTO t2 VALUES (1, 9);
+            SELECT t1.tag, t1.number, t2.number2
+            FROM {TABLE_NAME} AS t1
+            JOIN {TABLE2_NAME} AS t2 USING (tag)
+            SETTINGS
+                object_storage_remote_initiator=1,
+                object_storage_remote_initiator_cluster='cluster_simple',
+                object_storage_cluster_join_mode='global',
+                allow_experimental_analyzer=0
+            """,
+        ]
+    )
+    assert res == "1\t1\t2\n"
+
+    res = instance.query(f"""
+        SELECT t1.tag, t1.number
+        FROM {TABLE_NAME} AS t1
+        JOIN numbers(2) AS t2 ON t1.number = t2.number + 1
+        SETTINGS
+            object_storage_remote_initiator=1,
+            object_storage_remote_initiator_cluster='cluster_simple',
+            object_storage_cluster_join_mode='global',
+            allow_experimental_analyzer=0
+        """)
+    assert res == "1\t1\n"
+
+    res = instance.query(f"""
+        SELECT tag, number
+        FROM {TABLE_NAME}
+        WHERE tag IN (SELECT tag FROM {TABLE2_NAME})
+        SETTINGS
+            object_storage_remote_initiator=1,
+            object_storage_remote_initiator_cluster='cluster_simple',
+            object_storage_cluster_join_mode='global',
+            allow_experimental_analyzer=0
+        """)
+    assert res == "1\t1\n"
+
+    res = instance.query(f"""
+        SELECT tag, number
+        FROM {TABLE_NAME}
+        WHERE number IN (SELECT 1)
+        SETTINGS
+            object_storage_remote_initiator=1,
+            object_storage_remote_initiator_cluster='cluster_simple',
+            object_storage_cluster_join_mode='global',
+            allow_experimental_analyzer=0
+        """)
+    assert res == "1\t1\n"

From 21f471997e9e4b87f50008ff78b5ccf5e7ae36c4 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 22:01:08 +0100
Subject: [PATCH 14/16] Keep both Iceberg `timestamp` bounds in one precision

If either bound is below `1e16`, both stay microseconds. Otherwise a bound at or above `1e18` makes both nanoseconds.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../DataLakes/IDataLakeMetadata.cpp           | 30 ++++++++----------
 .../Iceberg/IcebergFieldParseHelpers.cpp      | 31 +++++++++++++------
 .../Iceberg/IcebergFieldParseHelpers.h        |  5 +--
 .../Iceberg/ManifestFileIterator.cpp          |  6 ++--
 .../gtest_iceberg_field_parse_helpers.cpp     | 31 +++++++++++++++++++
 5 files changed, 71 insertions(+), 32 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
index c0f6ff1101727..4252cfa274166 100644
--- a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
@@ -2,7 +2,6 @@
 #include <Storages/ObjectStorage/StorageObjectStorageSource.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h>
-#include <base/unaligned.h>
 #include <Core/TypeId.h>
 #include <DataTypes/DataTypeNullable.h>
 #include <DataTypes/DataTypesDecimal.h>
@@ -158,23 +157,20 @@ DataFileMetaInfo::DataFileMetaInfo(
                 String right_str;
                 if (i_bounds->second.first.tryGet(left_str) && i_bounds->second.second.tryGet(right_str))
                 {
-                    auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true);
-                    auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false);
-                    if (left && right && !DB::Range::less(*right, *left))
+                    const auto non_nullable_type = removeNullable(type);
+                    if (WhichDataType(non_nullable_type).isDateTime64() && getDecimalScale(*non_nullable_type) == 6)
                     {
-                        const auto non_nullable_type = removeNullable(type);
-                        bool ambiguous_timestamp = false;
-                        if (WhichDataType(non_nullable_type).isDateTime64() && getDecimalScale(*non_nullable_type) == 6
-                            && left_str.size() == sizeof(Int64) && right_str.size() == sizeof(Int64))
-                        {
-                            const Int64 left_ticks = unalignedLoadLittleEndian<Int64>(left_str.data());
-                            const Int64 right_ticks = unalignedLoadLittleEndian<Int64>(right_str.data());
-                            /// `[1e16, 1e18)` can be microseconds or nanoseconds, so the constant-column
-                            /// shortcut must not substitute either reading.
-                            ambiguous_timestamp = Iceberg::isAmbiguousIcebergTimestampBound(left_ticks)
-                                || Iceberg::isAmbiguousIcebergTimestampBound(right_ticks);
-                        }
-                        if (!ambiguous_timestamp)
+                        /// The pair has one precision. Two ranges are the ambiguous middle band,
+                        /// so the constant-column shortcut must not pick either reading.
+                        const auto timestamp_ranges = Iceberg::deserializeDateTime64MinMaxRanges(left_str, right_str, type);
+                        if (timestamp_ranges.size() == 1)
+                            hyperrectangle = DB::Range(timestamp_ranges[0].lower, true, timestamp_ranges[0].upper, true);
+                    }
+                    else
+                    {
+                        auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true);
+                        auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false);
+                        if (left && right && !DB::Range::less(*right, *left))
                             hyperrectangle = DB::Range(*left, true, *right, true);
                     }
                 }
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index e08ec74c85861..78bf640c6fa41 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -241,17 +241,19 @@ std::vector<DateTime64BoundRange> deserializeDateTime64MinMaxRanges(
         return ranges;
     }
 
-    /// Both ends must accept a unit. A lower bound below `1e16` and an upper bound
-    /// of `1e18` produce no range, and min/max pruning skips the column.
-    const bool allow_microseconds = icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_nanoseconds_lower_threshold)
-        && icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_nanoseconds_lower_threshold);
-    const bool allow_nanoseconds = !icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_microseconds_only_threshold)
-        && !icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_microseconds_only_threshold);
-
-    if (allow_microseconds)
+    /// One precision for the pair. A bound below `1e16` keeps both as microseconds,
+    /// so `1e15` and `1e18` stay that range instead of collapsing to one microsecond.
+    /// Otherwise a bound at or above `1e18` converts both from nanoseconds.
+    const bool one_below_1e16 = icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_microseconds_only_threshold)
+        || icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_microseconds_only_threshold);
+    const bool one_at_or_above_1e18 = !icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_nanoseconds_lower_threshold)
+        || !icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_nanoseconds_lower_threshold);
+
+    if (one_below_1e16)
+    {
         appendRange(ranges, *lower_ticks, *upper_ticks, scale);
-
-    if (allow_nanoseconds)
+    }
+    else if (one_at_or_above_1e18)
     {
         appendRange(
             ranges,
@@ -259,6 +261,15 @@ std::vector<DateTime64BoundRange> deserializeDateTime64MinMaxRanges(
             nanosecondsToMicrosecondsForBound(*upper_ticks, false),
             scale);
     }
+    else
+    {
+        appendRange(ranges, *lower_ticks, *upper_ticks, scale);
+        appendRange(
+            ranges,
+            nanosecondsToMicrosecondsForBound(*lower_ticks, true),
+            nanosecondsToMicrosecondsForBound(*upper_ticks, false),
+            scale);
+    }
 
     return ranges;
 }
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
index f0fd586d9d5cb..3eb4bffdd22b1 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
@@ -60,8 +60,9 @@ struct DateTime64BoundRange
 };
 
 /// `DateTime64(6)` min/max used for file pruning.
-/// `|ticks| < 1e16` is microseconds. `|ticks| >= 1e18` is nanoseconds.
-/// A value between them can be either, so both ranges are returned.
+/// Both bounds use one precision. If either has `|ticks| < 1e16`, both are
+/// microseconds. Otherwise, if either has `|ticks| >= 1e18`, both are nanoseconds.
+/// When both sit in `[1e16, 1e18)`, both ranges are returned.
 /// `DateTime64(9)` returns the raw nanosecond range.
 std::vector<DateTime64BoundRange> deserializeDateTime64MinMaxRanges(
     const std::string & lower_str,
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
index 4465d71ce3e80..405e5b06319bd 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
@@ -530,9 +530,9 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                 /// Optional Iceberg fields are `Nullable(DateTime64)`. The outer type is not `DateTime64`.
                 if (WhichDataType(removeNullable(name_and_type.type)).isDateTime64())
                 {
-                    /// `|ticks| < 1e16` is one microsecond range. `|ticks| >= 1e18` is one
-                    /// nanosecond range. Values between them produce both, and the file is
-                    /// pruned only when the predicate misses every range.
+                    /// Both bounds share one precision. If either is below `1e16`, both are
+                    /// microseconds. Otherwise, if either is at or above `1e18`, both are
+                    /// nanoseconds. The open middle band produces both ranges.
                     const auto timestamp_ranges = deserializeDateTime64MinMaxRanges(left_str, right_str, name_and_type.type);
                     if (timestamp_ranges.empty())
                         continue;
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index 1129a9effe102..0b31c2084e49f 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -123,6 +123,37 @@ TEST(IcebergFieldParseHelpers, BoundsBelow1e16StayMicroseconds)
     EXPECT_EQ(decimal64Ticks(ranges[0].upper), ticks);
 }
 
+TEST(IcebergFieldParseHelpers, BoundsThatDifferBy1000ShareOnePrecision)
+{
+    auto type = std::make_shared<DataTypeDateTime64>(6);
+    /// `1e15` microseconds and `1e18` nanoseconds are the same instant. The lower bound is
+    /// below `1e16`, so both stay microseconds and the constant-column shortcut sees a range.
+    constexpr Int64 lower_us = 1'000'000'000'000'000LL;
+    constexpr Int64 upper_times_1000 = 1'000'000'000'000'000'000LL;
+    auto ranges = deserializeDateTime64MinMaxRanges(littleEndianInt64(lower_us), littleEndianInt64(upper_times_1000), type);
+    ASSERT_EQ(ranges.size(), 1);
+    EXPECT_EQ(decimal64Ticks(ranges[0].lower), lower_us);
+    EXPECT_EQ(decimal64Ticks(ranges[0].upper), upper_times_1000);
+    EXPECT_NE(decimal64Ticks(ranges[0].lower), decimal64Ticks(ranges[0].upper));
+
+    /// The same gap before the epoch. `|upper| < 1e16`, so the lower bound is not divided.
+    constexpr Int64 negative_lower = -1'000'000'000'000'000'000LL;
+    constexpr Int64 negative_upper = -1'000'000'000'000'000LL;
+    auto negative = deserializeDateTime64MinMaxRanges(
+        littleEndianInt64(negative_lower), littleEndianInt64(negative_upper), type);
+    ASSERT_EQ(negative.size(), 1);
+    EXPECT_EQ(decimal64Ticks(negative[0].lower), negative_lower);
+    EXPECT_EQ(decimal64Ticks(negative[0].upper), negative_upper);
+
+    /// Neither bound is below `1e16`, and one is at `1e18`, so both are nanoseconds.
+    constexpr Int64 middle = iceberg_timestamp_microseconds_only_threshold;
+    auto nanoseconds = deserializeDateTime64MinMaxRanges(
+        littleEndianInt64(middle), littleEndianInt64(upper_times_1000), type);
+    ASSERT_EQ(nanoseconds.size(), 1);
+    EXPECT_EQ(decimal64Ticks(nanoseconds[0].lower), middle / 1000);
+    EXPECT_EQ(decimal64Ticks(nanoseconds[0].upper), upper_times_1000 / 1000);
+}
+
 TEST(IcebergFieldParseHelpers, NanosecondBoundsAbove1e18OnTimestampAreConverted)
 {
     auto type = std::make_shared<DataTypeDateTime64>(6);

From 94c7664aa50e627386a3f2f835873560299d3d6a Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 22:01:08 +0100
Subject: [PATCH 15/16] Leave `IN (SELECT 1)` on the replica for an
 old-analyzer global join

The interpreter rewrites the same tree that is sent, so it must not turn every subquery into `globalIn`.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/Storages/IStorageCluster.cpp              | 52 ++++---------------
 .../test_remote_initiator.py                  | 21 +++++++-
 2 files changed, 28 insertions(+), 45 deletions(-)

diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp
index 9e7dc440d2787..315f26adf2636 100644
--- a/src/Storages/IStorageCluster.cpp
+++ b/src/Storages/IStorageCluster.cpp
@@ -309,33 +309,6 @@ void stripObjectStorageRemoteInitiatorSettings(ASTPtr & query)
         select_query->setExpression(ASTSelectQuery::Expression::SETTINGS, {});
 }
 
-void rewriteASTInFunctionsToGlobalIn(ASTPtr & node)
-{
-    if (!node)
-        return;
-
-    if (auto * function = node->as<ASTFunction>())
-    {
-        if (isNameOfLocalInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2)
-        {
-            const auto & rhs = function->arguments->children[1];
-            if (rhs
-                && (rhs->as<ASTSubquery>() || rhs->as<ASTSelectQuery>() || rhs->as<ASTSelectWithUnionQuery>()
-                    || rhs->as<ASTIdentifier>() || rhs->as<ASTTableIdentifier>()))
-            {
-                function->name = getGlobalInFunctionNameForLocalInFunctionName(function->name);
-            }
-        }
-    }
-
-    for (auto & child : node->children)
-    {
-        if (astIsNestedSelect(child))
-            continue;
-        rewriteASTInFunctionsToGlobalIn(child);
-    }
-}
-
 bool queryHasRightOrFullJoin(const ASTPtr & query)
 {
     if (!query)
@@ -421,14 +394,10 @@ void rewriteInitiatorLocalInToGlobalIn(ASTPtr & node)
     }
 }
 
-enum class GlobalInRewrite : uint8_t
-{
-    None,
-    InitiatorLocal,
-    All,
-};
-
-void rewriteASTJoinsToGlobal(ASTPtr & query, GlobalInRewrite in_rewrite)
+/// `IN some_table` is read on the initiator. `IN (SELECT 1)` stays on the replica.
+/// `getSampleBlockImpl` sends this same tree, so the interpreter pass must not
+/// turn every subquery into `globalIn` before the broadcast pass runs.
+void rewriteASTJoinsToGlobal(ASTPtr & query)
 {
     ASTSelectQuery * select_query = query->as<ASTSelectQuery>();
     if (!select_query)
@@ -438,7 +407,7 @@ void rewriteASTJoinsToGlobal(ASTPtr & query, GlobalInRewrite in_rewrite)
             if (union_query->list_of_selects)
             {
                 for (auto & child : union_query->list_of_selects->children)
-                    rewriteASTJoinsToGlobal(child, in_rewrite);
+                    rewriteASTJoinsToGlobal(child);
             }
         }
         return;
@@ -459,17 +428,14 @@ void rewriteASTJoinsToGlobal(ASTPtr & query, GlobalInRewrite in_rewrite)
         }
     }
 
-    if (in_rewrite == GlobalInRewrite::All)
-        rewriteASTInFunctionsToGlobalIn(query);
-    else if (in_rewrite == GlobalInRewrite::InitiatorLocal)
-        rewriteInitiatorLocalInToGlobalIn(query);
+    rewriteInitiatorLocalInToGlobalIn(query);
 }
 
 }
 
 void IStorageCluster::rewriteASTForGlobalJoin(ASTPtr & query)
 {
-    rewriteASTJoinsToGlobal(query, GlobalInRewrite::All);
+    rewriteASTJoinsToGlobal(query);
 }
 
 /// The initiator interpreter usually does this already: `IStorageCluster::isRemote` is true, so
@@ -481,8 +447,8 @@ namespace
 {
 void broadcastOldAnalyzerGlobalJoin(ASTPtr & query_to_send, const SelectQueryInfo & query_info, const ContextPtr & context)
 {
-    /// `IN (SELECT 1)` stays on the replica. `IN some_table` is read on the initiator.
-    rewriteASTJoinsToGlobal(query_to_send, GlobalInRewrite::InitiatorLocal);
+    /// Idempotent with `rewriteASTForGlobalJoin`: `IN (SELECT 1)` is still local here.
+    rewriteASTJoinsToGlobal(query_to_send);
 
     auto prepared_sets = query_info.prepared_sets ? query_info.prepared_sets : std::make_shared<PreparedSets>();
     TemporaryTablesMapping external_tables;
diff --git a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
index be776423bf398..b7d61aa5194e3 100644
--- a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
+++ b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py
@@ -206,7 +206,11 @@ def execute_spark_query(query: str):
         """)
     assert res == "1\t1\n"
 
-    res = instance.query(f"""
+    # `IN (SELECT 1)` has no table, so the replica must still see that subquery.
+    # Broadcasting it replaces the text with an external table name.
+    query_id = uuid.uuid4().hex
+    res = instance.query(
+        f"""
         SELECT tag, number
         FROM {TABLE_NAME}
         WHERE number IN (SELECT 1)
@@ -215,5 +219,18 @@ def execute_spark_query(query: str):
             object_storage_remote_initiator_cluster='cluster_simple',
             object_storage_cluster_join_mode='global',
             allow_experimental_analyzer=0
-        """)
+        """,
+        query_id=query_id,
+    )
     assert res == "1\t1\n"
+    for node in started_cluster_iceberg_with_spark.instances.values():
+        node.query("SYSTEM FLUSH LOGS")
+    sent_with_subquery = instance.query(f"""
+        SELECT count()
+        FROM clusterAllReplicas('cluster_simple', system.query_log)
+        WHERE initial_query_id = '{query_id}'
+            AND type = 'QueryFinish'
+            AND is_initial_query = 0
+            AND position(query, 'IN (SELECT 1)') > 0
+        """)
+    assert sent_with_subquery != "0\n"

From 31cd84fede27bef770c83c9d60c1a37c21563829 Mon Sep 17 00:00:00 2001
From: Anton Ivashkin <ianton@live.com>
Date: Mon, 2 Nov 2026 22:12:00 +0100
Subject: [PATCH 16/16] Drop the unused Iceberg timestamp ambiguous-bound
 helper

`deserializeDateTime64MinMaxRanges` already returns two ranges for the middle band.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../ObjectStorage/DataLakes/IDataLakeMetadata.cpp   |  3 +--
 .../DataLakes/Iceberg/IcebergFieldParseHelpers.cpp  |  4 +---
 .../DataLakes/Iceberg/IcebergFieldParseHelpers.h    |  7 -------
 .../DataLakes/Iceberg/IcebergWrites.cpp             | 11 ++++-------
 .../DataLakes/Iceberg/ManifestFileIterator.cpp      |  3 ---
 .../tests/gtest_iceberg_field_parse_helpers.cpp     | 13 -------------
 6 files changed, 6 insertions(+), 35 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
index 4252cfa274166..79a8e4b722769 100644
--- a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp
@@ -160,8 +160,7 @@ DataFileMetaInfo::DataFileMetaInfo(
                     const auto non_nullable_type = removeNullable(type);
                     if (WhichDataType(non_nullable_type).isDateTime64() && getDecimalScale(*non_nullable_type) == 6)
                     {
-                        /// The pair has one precision. Two ranges are the ambiguous middle band,
-                        /// so the constant-column shortcut must not pick either reading.
+                        /// Two ranges are the middle band, so this column is not a constant.
                         const auto timestamp_ranges = Iceberg::deserializeDateTime64MinMaxRanges(left_str, right_str, type);
                         if (timestamp_ranges.size() == 1)
                             hyperrectangle = DB::Range(timestamp_ranges[0].lower, true, timestamp_ranges[0].upper, true);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
index 78bf640c6fa41..81dca86944d0b 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp
@@ -241,9 +241,7 @@ std::vector<DateTime64BoundRange> deserializeDateTime64MinMaxRanges(
         return ranges;
     }
 
-    /// One precision for the pair. A bound below `1e16` keeps both as microseconds,
-    /// so `1e15` and `1e18` stay that range instead of collapsing to one microsecond.
-    /// Otherwise a bound at or above `1e18` converts both from nanoseconds.
+    /// One precision for the pair. Below `1e16` keeps `1e15` and `1e18` as microseconds.
     const bool one_below_1e16 = icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_microseconds_only_threshold)
         || icebergTimestampMagnitudeBelow(*upper_ticks, iceberg_timestamp_microseconds_only_threshold);
     const bool one_at_or_above_1e18 = !icebergTimestampMagnitudeBelow(*lower_ticks, iceberg_timestamp_nanoseconds_lower_threshold)
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
index 3eb4bffdd22b1..e2af51f6b3b63 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.h
@@ -26,13 +26,6 @@ inline bool icebergTimestampMagnitudeBelow(Int64 value, Int64 threshold)
     return value > -threshold && value < threshold;
 }
 
-/// `[iceberg_timestamp_microseconds_only_threshold, iceberg_timestamp_nanoseconds_lower_threshold)`.
-inline bool isAmbiguousIcebergTimestampBound(Int64 ticks)
-{
-    return !icebergTimestampMagnitudeBelow(ticks, iceberg_timestamp_microseconds_only_threshold)
-        && icebergTimestampMagnitudeBelow(ticks, iceberg_timestamp_nanoseconds_lower_threshold);
-}
-
 /// Convert a Field to Int64, accepting Int64 and UInt64 (with overflow check).
 Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name);
 
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
index a6d42b1c51acf..2277bcae28ee5 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
@@ -174,12 +174,6 @@ std::vector<uint8_t> dumpValue(T value)
     return bytes;
 }
 
-std::vector<uint8_t> dumpDateTime64Bound(Int64 ticks)
-{
-    const auto bytes = Iceberg::serializeDateTime64IcebergBound(ticks);
-    return std::vector<uint8_t>(bytes.begin(), bytes.end());
-}
-
 DataTypePtr getTimeTypeOrNull(DataTypePtr type)
 {
     if (type->isNullable())
@@ -358,7 +352,10 @@ std::vector<uint8_t> dumpFieldToBytes(const Field & field, DataTypePtr type)
         case TypeIndex::UInt64:
             return dumpValue(applyVisitor(FieldVisitorConvertToNumber<Int64>(), field));
         case TypeIndex::DateTime64:
-            return dumpDateTime64Bound(getDateTime64ValueForIcebergBounds(field, type));
+        {
+            const auto bytes = Iceberg::serializeDateTime64IcebergBound(getDateTime64ValueForIcebergBounds(field, type));
+            return std::vector<uint8_t>(bytes.begin(), bytes.end());
+        }
         case TypeIndex::String:
         {
             auto value = field.safeGet<String>();
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
index 405e5b06319bd..1d86787154055 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp
@@ -530,9 +530,6 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index)
                 /// Optional Iceberg fields are `Nullable(DateTime64)`. The outer type is not `DateTime64`.
                 if (WhichDataType(removeNullable(name_and_type.type)).isDateTime64())
                 {
-                    /// Both bounds share one precision. If either is below `1e16`, both are
-                    /// microseconds. Otherwise, if either is at or above `1e18`, both are
-                    /// nanoseconds. The open middle band produces both ranges.
                     const auto timestamp_ranges = deserializeDateTime64MinMaxRanges(left_str, right_str, name_and_type.type);
                     if (timestamp_ranges.empty())
                         continue;
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
index 0b31c2084e49f..211ecaa9134bd 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp
@@ -55,16 +55,6 @@ TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept)
     EXPECT_EQ(decimal64Ticks(*sentinel), spark_9999_us);
 }
 
-TEST(IcebergFieldParseHelpers, AmbiguousTimestampBoundsAreTheOpenMiddleBand)
-{
-    EXPECT_FALSE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_microseconds_only_threshold - 1));
-    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_microseconds_only_threshold));
-    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(200'000'000'000'000'000LL));
-    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(-200'000'000'000'000'000LL));
-    EXPECT_TRUE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_nanoseconds_lower_threshold - 1));
-    EXPECT_FALSE(isAmbiguousIcebergTimestampBound(iceberg_timestamp_nanoseconds_lower_threshold));
-}
-
 TEST(IcebergFieldParseHelpers, TimestampBoundsUseBothRangesBetween1e16And1e18)
 {
     auto type = std::make_shared<DataTypeDateTime64>(6);
@@ -106,9 +96,6 @@ TEST(IcebergFieldParseHelpers, BoundsBelow1e16StayMicroseconds)
     /// the integer and does not add a nanosecond range. The constant-column shortcut
     /// may use this point: it is not the ambiguous middle band.
     constexpr Int64 ticks = 86'400'000'000'000LL;
-    EXPECT_FALSE(isAmbiguousIcebergTimestampBound(ticks));
-    EXPECT_LT(ticks, iceberg_timestamp_microseconds_only_threshold);
-
     const auto bytes = littleEndianInt64(ticks);
     auto lower = deserializeFieldFromBinaryRepr(bytes, type, true);
     auto upper = deserializeFieldFromBinaryRepr(bytes, type, false);