Fix for nanoseconds in Iceberg

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

@@ -28,59 +28,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 timestamps are 8-byte little-endian (spec Appendix D). `DateTime64(6)` (`timestamp` /
+/// `timestamptz`) bounds are microseconds and `DateTime64(9)` (`timestamp_ns` / `timestamptz_ns`)
+/// bounds are nanoseconds. Both are returned as stored.
+std::optional<Field> deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool /*lower_bound*/)
 {
     if (str.size() != sizeof(Int64))
         return std::nullopt;
 
-    Int64 unscaled = unalignedLoadLittleEndian<Int64>(str.data());
+    const 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);
 }
 

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).
-    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

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)
 )
@@ -224,11 +180,11 @@ def _utc_datetime_to_micros(dt):
 
 
 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",
@@ -253,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

tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py

@@ -29,11 +29,9 @@
     ROWS,
     SPARK_TIMESTAMP_SENTINEL_US,
     check_minmax_far_future_us_upper_bound,
-    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,
     patch_iceberg_timestamp_upper_bounds,
 )
 from helpers.iceberg_utils import check_validity_and_get_prunned_files_general
@@ -119,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):
@@ -240,31 +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)
-
-
 @pytest.mark.parametrize(
     "upper_bound_us",
     [SPARK_TIMESTAMP_SENTINEL_US, DATETIME64_MAX_US],

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,