From db1bcff8e359e4eee6d042e60501bd26d32e7d90 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 18 Aug 2026 21:12:19 +0200 Subject: [PATCH 01/15] Add Iceberg tests for nanosecond timestamp pruning Identity partition pruning on `timestamp_ns` over-prunes matching files. Min/max pruning also returns empty results when Iceberg `timestamp` (`DateTime64(6)`) lower/upper bounds are stored as nanoseconds, which is the customer `toDateTime64(..., 6)` case. Co-authored-by: Cursor --- .../helpers/iceberg_timestamp_ns.py | 444 ++++++++++++++++++ .../test_pruning_datetime64_nanoseconds.py | 244 ++++++++++ .../test_pruning_datetime64_nanoseconds.py | 181 +++++++ 3 files changed, 869 insertions(+) create mode 100644 tests/integration/helpers/iceberg_timestamp_ns.py create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py create mode 100644 tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py diff --git a/tests/integration/helpers/iceberg_timestamp_ns.py b/tests/integration/helpers/iceberg_timestamp_ns.py new file mode 100644 index 0000000000000..37b9e365fa82d --- /dev/null +++ b/tests/integration/helpers/iceberg_timestamp_ns.py @@ -0,0 +1,444 @@ +"""Helpers to turn Iceberg `long` nanosecond columns into `timestamp_ns`. + +Spark 3.5 + Iceberg 1.8.1 and pyiceberg 0.11 cannot write Iceberg v3 +`timestamp_ns`. Tests write epoch nanoseconds as `long`, then call +`patch_iceberg_table_long_to_timestamp_ns` before ClickHouse reads the table. +""" + +import json +import struct +from pathlib import Path + +import avro.datafile +import avro.io +import pyarrow as pa +import pyarrow.compute as pc +import pyarrow.parquet as pq + +# Epoch nanoseconds and the matching UTC wall time. Values differ in the last 3 +# digits so they collapse together if the reader/writer silently truncates to +# microseconds. +ROWS = [ + (1704067200123456789, 1, "2024-01-01 00:00:00.123456789"), + (1710504000000000001, 2, "2024-03-15 12:00:00.000000001"), + (1719791999999999999, 3, "2024-06-30 23:59:59.999999999"), + (1767205845111111111, 4, "2025-12-31 18:30:45.111111111"), +] + +NS_COLUMN_NAMES = ("ts", "value") + +PRUNE_OFF = { + "use_iceberg_partition_pruning": 0, + "input_format_parquet_bloom_filter_push_down": 0, + "input_format_parquet_filter_push_down": 0, +} +PRUNE_ON = { + "use_iceberg_partition_pruning": 1, + "input_format_parquet_bloom_filter_push_down": 0, + "input_format_parquet_filter_push_down": 0, +} + + +def expected_all_ids_and_nanos(): + return "".join(f"{row_id}\t{ts_ns}\t{ts_ns}\n" for ts_ns, row_id, _utc in ROWS) + + +def select_ids_and_nanos(table): + return ( + f"SELECT id, toUnixTimestamp64Nano(ts), toUnixTimestamp64Nano(value) " + f"FROM {table} ORDER BY id" + ) + + +def from_unix_timestamp64_nano(ns): + return f"fromUnixTimestamp64Nano({ns})" + + +def to_datetime64_utc(ns): + for ts_ns, _row_id, utc in ROWS: + if ts_ns == ns: + return datetime64_literal(utc, scale=9, timezone="UTC") + raise KeyError(ns) + + +def datetime64_literal(utc, scale, timezone="UTC"): + whole, frac = utc.split(".") + frac = frac[:scale].ljust(scale, "0") + value = f"{whole}.{frac}" if scale else whole + if timezone is None: + return f"toDateTime64('{value}', {scale})" + return f"toDateTime64('{value}', {scale}, '{timezone}')" + + +# Both constructors must keep nanosecond precision in the pushed-down predicate. +FILTER_LITERALS = [ + from_unix_timestamp64_nano, + to_datetime64_utc, +] +FILTER_LITERAL_IDS = ["fromUnixTimestamp64Nano", "toDateTime64"] + + +def check_ns_pruning(node, table, column, pruned_files, make_literal): + """Equality / range pruning for a timestamp_ns column, using one SQL literal style.""" + assert node.query(select_ids_and_nanos(table), settings=PRUNE_OFF) == expected_all_ids_and_nanos() + + eq_ns, eq_id, _eq_utc = ROWS[1] + eq_lit = make_literal(eq_ns) + assert ( + pruned_files(f"SELECT id FROM {table} WHERE {column} = {eq_lit} ORDER BY id") + == 3 + ) + assert ( + node.query( + f"SELECT id FROM {table} WHERE {column} = {eq_lit} ORDER BY id", + settings=PRUNE_ON, + ) + == f"{eq_id}\n" + ) + + le_ns = ROWS[1][0] + le_lit = make_literal(le_ns) + assert ( + pruned_files(f"SELECT id FROM {table} WHERE {column} <= {le_lit} ORDER BY id") + == 2 + ) + assert node.query( + f"SELECT id FROM {table} WHERE {column} <= {le_lit} ORDER BY id", + settings=PRUNE_ON, + ) == "1\n2\n" + + ge_ns = ROWS[2][0] + ge_lit = make_literal(ge_ns) + assert ( + pruned_files(f"SELECT id FROM {table} WHERE {column} >= {ge_lit} ORDER BY id") + == 2 + ) + assert node.query( + f"SELECT id FROM {table} WHERE {column} >= {ge_lit} ORDER BY id", + settings=PRUNE_ON, + ) == "3\n4\n" + + assert pruned_files(select_ids_and_nanos(table)) == 0 + + +def check_ns_pruning_datetime64_microseconds(node, table, column, pruned_files, timezone="UTC"): + """Min/max (or partition) pruning with a DateTime64(6) predicate on timestamp_ns. + + Same SQL shape as the customer query (`WHERE col <= toDateTime64(..., 6)`), but + against a well-typed Iceberg `timestamp_ns` column (`DateTime64(9)`). KeyCondition + already rescales DateTime64(6) vs DateTime64(9) bounds, so this currently passes. + """ + assert node.query(select_ids_and_nanos(table), settings=PRUNE_OFF) == expected_all_ids_and_nanos() + + # Truncating row 2 to microseconds is 1ns before the stored value, so only row 1 matches. + le_before_row2 = datetime64_literal(ROWS[1][2], scale=6, timezone=timezone) + assert ( + pruned_files(f"SELECT id FROM {table} WHERE {column} <= {le_before_row2} ORDER BY id") + == 3 + ) + assert node.query( + f"SELECT id FROM {table} WHERE {column} <= {le_before_row2} ORDER BY id", + settings=PRUNE_ON, + ) == "1\n" + + # One microsecond after row 2 includes rows 1 and 2. Same shape as the customer query. + le_after_row2 = datetime64_literal("2024-03-15 12:00:00.000001000", scale=6, timezone=timezone) + assert ( + pruned_files(f"SELECT id FROM {table} WHERE {column} <= {le_after_row2} ORDER BY id") + == 2 + ) + assert node.query( + f"SELECT id FROM {table} WHERE {column} <= {le_after_row2} ORDER BY id", + settings=PRUNE_ON, + ) == "1\n2\n" + + ge_row3 = datetime64_literal(ROWS[2][2], scale=6, timezone=timezone) + assert ( + pruned_files(f"SELECT id FROM {table} WHERE {column} >= {ge_row3} ORDER BY id") + == 2 + ) + assert node.query( + f"SELECT id FROM {table} WHERE {column} >= {ge_row3} ORDER BY id", + settings=PRUNE_ON, + ) == "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 _scale_bound_bytes(raw, factor=1000): + if not isinstance(raw, (bytes, bytearray)) or len(raw) != 8: + return raw + value = struct.unpack(" 0, f"no timestamp min/max bounds scaled under {table_location}" + + +def upgrade_long_to_iceberg_timestamp(obj, iceberg_type="timestamp_ns"): + if isinstance(obj, dict): + if obj.get("name") in NS_COLUMN_NAMES and obj.get("type") == "long": + obj["type"] = iceberg_type + for value in obj.values(): + upgrade_long_to_iceberg_timestamp(value, iceberg_type) + elif isinstance(obj, list): + for value in obj: + upgrade_long_to_iceberg_timestamp(value, iceberg_type) + + +def upgrade_long_to_timestamp_ns(obj): + upgrade_long_to_iceberg_timestamp(obj, "timestamp_ns") + + +def _deep_replace(obj, old, new): + if isinstance(obj, str): + return obj.replace(old, new) + if isinstance(obj, dict): + return {k: _deep_replace(v, old, new) for k, v in obj.items()} + if isinstance(obj, list): + return [_deep_replace(v, old, new) for v in obj] + return obj + + +def rewrite_parquet_int64_to_timestamp(path: Path, unit="ns"): + # Read the file itself; pq.read_table() infers Hive partitions from the + # parent directory name (`ts=/`) and then fails to merge that + # dictionary field with the int64 column in the file. + table = pq.ParquetFile(path).read() + columns = {} + scale = {"ns": 1, "us": 1_000, "ms": 1_000_000, "s": 1_000_000_000}[unit] + for name in table.column_names: + column = table[name] + if name in NS_COLUMN_NAMES and pa.types.is_int64(column.type): + if scale == 1: + columns[name] = column.cast(pa.timestamp("ns")) + else: + columns[name] = pc.divide(column, scale).cast(pa.timestamp(unit)) + else: + columns[name] = column + pq.write_table(pa.table(columns), path) + rewritten = pq.ParquetFile(path).schema_arrow + for name in NS_COLUMN_NAMES: + if name in rewritten.names: + field_type = rewritten.field(name).type + assert pa.types.is_timestamp(field_type), (path, name, field_type) + assert field_type.unit == unit, (path, name, field_type) + + +def rewrite_parquet_int64_to_timestamp_ns(path: Path): + rewrite_parquet_int64_to_timestamp(path, unit="ns") + + +def _meta_text(value): + if isinstance(value, (bytes, bytearray)): + return value.decode("utf-8") + return value + + +def rewrite_avro_long_to_timestamp_ns( + src: Path, dst: Path, old=None, new=None, iceberg_type="timestamp_ns" +): + 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) + if old is not None and new is not None: + records = [_deep_replace(record, old, new) for record in reader] + else: + records = list(reader) + 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 key_str in ("schema", "iceberg.schema"): + iceberg_schema = json.loads(text) + upgrade_long_to_iceberg_timestamp(iceberg_schema, iceberg_type) + text = json.dumps(iceberg_schema, separators=(",", ":")) + elif 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() + + +def patch_iceberg_table_long_to_timestamp_ns( + table_location: Path, + old_prefix=None, + new_prefix=None, + iceberg_type="timestamp_ns", + parquet_unit="ns", +): + """Rewrite Parquet timestamps and Iceberg metadata to `iceberg_type`. + + Manifest lower/upper bounds stay the original 8-byte little-endian nanosecond + int64 values. For the customer min/max over-prune, use Iceberg `timestamp` + (`DateTime64(6)`) with `parquet_unit='us'` so the scan is correct while + bounds are still nanoseconds (year 2299 when decoded as microseconds). + """ + for parquet_file in table_location.rglob("*.parquet"): + rewrite_parquet_int64_to_timestamp(parquet_file, unit=parquet_unit) + + meta_dir = table_location / "metadata" + for metadata_file in meta_dir.glob("*.metadata.json"): + text = metadata_file.read_text() + if old_prefix is not None and new_prefix is not None: + text = text.replace(old_prefix, new_prefix) + meta = json.loads(text) + upgrade_long_to_iceberg_timestamp(meta, iceberg_type) + metadata_file.write_text(json.dumps(meta, separators=(",", ":"))) + + for avro_file in list(meta_dir.glob("*.avro")): + tmp = avro_file.with_suffix(".avro.tmp") + rewrite_avro_long_to_timestamp_ns( + avro_file, tmp, old_prefix, new_prefix, iceberg_type=iceberg_type + ) + tmp.replace(avro_file) 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 new file mode 100644 index 0000000000000..e703492ec35fd --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py @@ -0,0 +1,244 @@ +#!/usr/bin/env python3 +"""Iceberg identity-partition and min/max pruning for timestamp_ns (DateTime64(9)). + +pyiceberg 0.11 cannot write Iceberg v3 (`timestamp_ns`), and the REST catalog +image used here also rejects that type. The table is therefore written as +Iceberg v2 with `long` columns holding epoch nanoseconds, then patched to +`timestamp_ns` and Parquet `timestamp[ns]` before ClickHouse reads it. +""" + +import shutil +import tempfile +import uuid +from pathlib import Path + +import pyarrow as pa +import pytest +from pyiceberg.catalog.sql import SqlCatalog +from pyiceberg.partitioning import PartitionField, PartitionSpec +from pyiceberg.schema import NestedField, Schema +from pyiceberg.transforms import IdentityTransform +from pyiceberg.types import IntegerType, LongType, TimestampType + +from helpers.iceberg_timestamp_ns import ( + FILTER_LITERAL_IDS, + FILTER_LITERALS, + 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 + + +def _ns_schema(): + return Schema( + NestedField(field_id=1, name="ts", field_type=LongType(), required=False), + NestedField(field_id=2, name="value", field_type=LongType(), required=False), + NestedField(field_id=3, name="id", field_type=IntegerType(), required=False), + ) + + +def _arrow_row(ts_ns, row_id): + return pa.Table.from_pydict( + { + "ts": pa.array([ts_ns], type=pa.int64()), + "value": pa.array([ts_ns], type=pa.int64()), + "id": pa.array([row_id], type=pa.int32()), + } + ) + + +def _timestamp_schema(): + return Schema( + NestedField(field_id=1, name="ts", field_type=TimestampType(), required=False), + NestedField(field_id=2, name="value", field_type=TimestampType(), required=False), + NestedField(field_id=3, name="id", field_type=IntegerType(), required=False), + ) + + +def _arrow_timestamp_row(ts_ns, row_id): + ts_us = pa.array([ts_ns // 1000], type=pa.timestamp("us")) + return pa.Table.from_pydict( + { + "ts": ts_us, + "value": ts_us, + "id": pa.array([row_id], type=pa.int32()), + } + ) + + +def _create_timestamp_table_with_ns_bounds(started_cluster, table_name): + """Iceberg v2 `timestamp` (microseconds) whose manifest min/max bytes are nanoseconds.""" + work = Path(tempfile.mkdtemp(prefix="iceberg_ns_bounds_")) + try: + catalog = SqlCatalog( + "ns_bounds", + **{ + "uri": f"sqlite:///{work}/catalog.db", + "warehouse": f"file://{work}/warehouse", + }, + ) + catalog.create_namespace("default") + identifier = f"default.{table_name}" + table = catalog.create_table( + identifier=identifier, + schema=_timestamp_schema(), + partition_spec=PartitionSpec(), + ) + table.append( + pa.concat_tables( + [_arrow_timestamp_row(ts_ns, row_id) for ts_ns, row_id, _utc in ROWS] + ) + ) + table = catalog.load_table(identifier) + + old_prefix = table.location() + table_location = Path(old_prefix.removeprefix("file://")) + s3_key = f"iceberg_ns_bounds_{uuid.uuid4().hex}" + new_prefix = f"s3://{started_cluster.minio_bucket}/{s3_key}" + + patch_iceberg_timestamp_bounds_us_to_ns(table_location, old_prefix, new_prefix) + + uploader = started_cluster.default_s3_uploader + for path in table_location.rglob("*"): + if path.is_file(): + rel = path.relative_to(table_location).as_posix() + uploader.upload_file(str(path), f"{s3_key}/{rel}") + + return f"{s3_key}/" + finally: + shutil.rmtree(work, ignore_errors=True) + + +def _create_v3_table( + started_cluster, table_name, partition_spec, iceberg_type="timestamp_ns", parquet_unit="ns" +): + work = Path(tempfile.mkdtemp(prefix="iceberg_ns_pruning_")) + try: + catalog = SqlCatalog( + "ns_pruning", + **{ + "uri": f"sqlite:///{work}/catalog.db", + "warehouse": f"file://{work}/warehouse", + }, + ) + catalog.create_namespace("default") + identifier = f"default.{table_name}" + table = catalog.create_table( + identifier=identifier, + schema=_ns_schema(), + partition_spec=partition_spec, + ) + for ts_ns, row_id, _utc in ROWS: + table.append(_arrow_row(ts_ns, row_id)) + table = catalog.load_table(identifier) + + old_prefix = table.location() + table_location = Path(old_prefix.removeprefix("file://")) + s3_key = f"iceberg_ns_pruning_{uuid.uuid4().hex}" + new_prefix = f"s3://{started_cluster.minio_bucket}/{s3_key}" + + patch_iceberg_table_long_to_timestamp_ns( + table_location, + old_prefix, + new_prefix, + iceberg_type=iceberg_type, + parquet_unit=parquet_unit, + ) + + uploader = started_cluster.default_s3_uploader + for path in table_location.rglob("*"): + if path.is_file(): + rel = path.relative_to(table_location).as_posix() + uploader.upload_file(str(path), f"{s3_key}/{rel}") + + return f"{s3_key}/" + finally: + shutil.rmtree(work, ignore_errors=True) + + +def _iceberg_table(s3_filename): + return ( + f"icebergS3(s3, filename = '{s3_filename}', format=Parquet, " + f"url = 'http://minio1:9001/root/')" + ) + + +@pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS) +def test_partition_pruning_datetime64_nanoseconds(started_cluster_iceberg_no_spark, make_literal): + """Identity partition pruning on Iceberg timestamp_ns must not drop matching rows.""" + node = started_cluster_iceberg_no_spark.instances["node1"] + partition_spec = PartitionSpec( + PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="ts") + ) + s3_filename = _create_v3_table(started_cluster_iceberg_no_spark, "part_ns", partition_spec) + table = _iceberg_table(s3_filename) + + def pruned_files(select_expression): + return check_validity_and_get_prunned_files_general( + node, "part_ns", PRUNE_OFF, PRUNE_ON, "IcebergPartitionPrunedFiles", select_expression + ) + + check_ns_pruning(node, table, "ts", pruned_files, make_literal) + + +@pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS) +def test_minmax_pruning_datetime64_nanoseconds(started_cluster_iceberg_no_spark, make_literal): + """Min/max file pruning on Iceberg timestamp_ns must not drop matching rows.""" + node = started_cluster_iceberg_no_spark.instances["node1"] + s3_filename = _create_v3_table(started_cluster_iceberg_no_spark, "minmax_ns", PartitionSpec()) + table = _iceberg_table(s3_filename) + + def pruned_files(select_expression): + return check_validity_and_get_prunned_files_general( + node, "minmax_ns", PRUNE_OFF, PRUNE_ON, "IcebergMinMaxIndexPrunedFiles", select_expression + ) + + check_ns_pruning(node, table, "value", pruned_files, make_literal) + + +@pytest.mark.parametrize("timezone", [None, "UTC"], ids=["toDateTime64_6", "toDateTime64_6_UTC"]) +def test_minmax_pruning_datetime64_nanoseconds_microsecond_predicate( + started_cluster_iceberg_no_spark, timezone +): + """DateTime64(6) predicate vs timestamp_ns min/max (customer over-prune).""" + node = started_cluster_iceberg_no_spark.instances["node1"] + s3_filename = _create_v3_table(started_cluster_iceberg_no_spark, "minmax_ns_us", PartitionSpec()) + table = _iceberg_table(s3_filename) + + def pruned_files(select_expression): + return check_validity_and_get_prunned_files_general( + node, "minmax_ns_us", PRUNE_OFF, PRUNE_ON, "IcebergMinMaxIndexPrunedFiles", 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) 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 new file mode 100644 index 0000000000000..229d0b4b239b8 --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""Spark-written Iceberg identity-partition and min/max pruning for timestamp_ns. + +Spark 3.5 + Iceberg 1.8.1 (the integration-test runtime) cannot create or write +Iceberg `timestamp_ns`: Spark TIMESTAMP is microsecond-only, and Iceberg Spark +maps it to `timestamp` / `timestamptz`. This test therefore writes epoch +nanoseconds as BIGINT via Spark, patches the table to `timestamp_ns`, then +checks ClickHouse pruning. +""" + +from pathlib import Path + +import pytest + +from helpers.iceberg_timestamp_ns import ( + FILTER_LITERAL_IDS, + FILTER_LITERALS, + 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, + default_upload_directory, + get_creation_expression, + get_uuid_str, +) + +WAREHOUSE_PREFIX = "/var/lib/clickhouse/user_files/iceberg_data/default" + + +def _create_spark_ns_table( + spark, + started_cluster, + table_name, + partitioned, + iceberg_type="timestamp_ns", + parquet_unit="ns", +): + partition_clause = "PARTITIONED BY (identity(ts))" if partitioned else "" + spark.sql( + f""" + CREATE TABLE {table_name} ( + ts BIGINT, + value BIGINT, + id INT + ) + USING iceberg + {partition_clause} + OPTIONS('format-version'='2') + """ + ) + for ts_ns, row_id, _utc in ROWS: + spark.sql( + f"INSERT INTO {table_name} VALUES ({ts_ns}, {ts_ns}, {row_id})" + ) + + table_location = Path(f"{WAREHOUSE_PREFIX}/{table_name}") + patch_iceberg_table_long_to_timestamp_ns( + table_location, iceberg_type=iceberg_type, parquet_unit=parquet_unit + ) + default_upload_directory( + started_cluster, + "s3", + f"/iceberg_data/default/{table_name}/", + f"/iceberg_data/default/{table_name}/", + ) + + +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.""" + node = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + table_name = "test_spark_part_ns_" + get_uuid_str() + _create_spark_ns_table(spark, started_cluster_iceberg_with_spark, table_name, partitioned=True) + 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, "IcebergPartitionPrunedFiles", select_expression + ) + + check_ns_pruning(node, table, "ts", pruned_files, make_literal) + + +@pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS) +def test_spark_minmax_pruning_datetime64_nanoseconds(started_cluster_iceberg_with_spark, make_literal): + """Min/max file pruning on Spark-written Iceberg timestamp_ns must not drop matching rows.""" + node = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + table_name = "test_spark_minmax_ns_" + get_uuid_str() + _create_spark_ns_table(spark, started_cluster_iceberg_with_spark, table_name, partitioned=False) + 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_ns_pruning(node, table, "value", pruned_files, make_literal) + + +@pytest.mark.parametrize("timezone", [None, "UTC"], ids=["toDateTime64_6", "toDateTime64_6_UTC"]) +def test_spark_minmax_pruning_datetime64_nanoseconds_microsecond_predicate( + started_cluster_iceberg_with_spark, timezone +): + """DateTime64(6) predicate vs Spark-written timestamp_ns min/max (customer over-prune).""" + node = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + table_name = "test_spark_minmax_ns_us_" + get_uuid_str() + _create_spark_ns_table(spark, started_cluster_iceberg_with_spark, table_name, partitioned=False) + 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_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 2802c71edb5a4deef6180abb9fcca9e1cb2bb592 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 18 Aug 2026 21:48:30 +0200 Subject: [PATCH 02/15] Fix Iceberg over-pruning of nanosecond timestamps Identity partition values for `timestamp_ns` arrive as Avro Int64, and some writers store nanosecond min/max bytes on Iceberg `timestamp` (`DateTime64(6)`). Both were compared at the wrong scale and dropped matching files. Write `DateTime64(9)` as `timestamp_ns` so ClickHouse does not recreate that layout. Co-authored-by: Cursor --- .../table-engines/integrations/iceberg.md | 2 + .../Iceberg/IcebergFieldParseHelpers.cpp | 64 +++++++++++ .../DataLakes/Iceberg/IcebergWrites.cpp | 22 +++- .../Iceberg/ManifestFilesPruning.cpp | 36 +++++- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 19 ++- .../test.py | 2 +- .../test_writes_datetime64_nanoseconds.py | 108 ++++++++++++++++++ 7 files changed, 249 insertions(+), 4 deletions(-) create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py diff --git a/docs/en/engines/table-engines/integrations/iceberg.md b/docs/en/engines/table-engines/integrations/iceberg.md index e36e90ffb3fd6..2a61173e30ed7 100644 --- a/docs/en/engines/table-engines/integrations/iceberg.md +++ b/docs/en/engines/table-engines/integrations/iceberg.md @@ -98,6 +98,8 @@ The following table shows how Iceberg data types are mapped to ClickHouse data t | `fixed(N)` | `FixedString(N)` | | | `decimal(P, S)` | `Decimal(P, S)` | | +Writes use the inverse mapping: `DateTime64(9)` is stored as `timestamp_ns`, and `DateTime64` with an explicit timezone and scale 9 as `timestamptz_ns`. `DateTime` and `DateTime64` with scale 6 or less are stored as `timestamp`. + ### Complex types {#complex-types} | Iceberg type | ClickHouse type | diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index b779ea8c61927..35c912e2b3875 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -4,6 +4,7 @@ #include #include +#include #include #include #include @@ -22,6 +23,66 @@ extern const int BAD_ARGUMENTS; namespace Iceberg { +namespace +{ + +/// ~year 2286 as DateTime64(6) microseconds. Spec-correct Iceberg `timestamp` bounds +/// for realistic dates stay below this; nanosecond ticks for dates after ~1970-04 +/// sit well above it (2024 is ~1.7e18). +constexpr Int64 microseconds_implausible_threshold = 10'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). `insertData` +/// memcpy's native DateTime64 ticks, which is correct when the bound scale +/// matches the column. Some writers store nanosecond stats on Iceberg +/// `timestamp` (`DateTime64(6)`); treating those as microseconds yields dates +/// around year 2299 and over-prunes. Rescale those bounds; if they are still +/// implausible, fail open (`nullopt`) so min/max pruning is skipped. +std::optional deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound) +{ + if (str.size() != sizeof(Int64)) + return std::nullopt; + + Int64 unscaled = unalignedLoadLittleEndian(str.data()); + const UInt32 scale = getDecimalScale(type); + + if (scale == 6 && magnitudeExceeds(unscaled, microseconds_implausible_threshold)) + { + unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); + if (magnitudeExceeds(unscaled, microseconds_implausible_threshold)) + return std::nullopt; + } + + return DecimalField(DateTime64(unscaled), scale); +} + +} + Int64 fieldToInt64(const Field & value, std::string_view context, std::string_view arg_name) { if (value.getType() == Field::Types::Int64) @@ -104,6 +165,9 @@ std::vector fieldToInt64Array(const Field & value, std::string_view conte std::optional deserializeFieldFromBinaryRepr(const std::string & str, DataTypePtr expected_type, bool lower_bound) { auto non_nullable_type = removeNullable(expected_type); + if (WhichDataType(non_nullable_type).isDateTime64()) + return deserializeDateTime64FromBinaryRepr(str, *non_nullable_type, lower_bound); + auto column = non_nullable_type->createColumn(); if (WhichDataType(non_nullable_type).isDecimal()) { diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index 5ddc5ff3744b2..c725c66b1be69 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -10,6 +10,7 @@ #include #include #include +#include #include #include #include @@ -189,6 +190,25 @@ Int64 getTimeValueInMicroseconds(const Field & field, DataTypePtr type) throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected Time or Time64, got {}", type->getName()); } +Int64 getDateTime64ValueForIcebergBounds(const Field & field, DataTypePtr type) +{ + if (type->isNullable()) + return getDateTime64ValueForIcebergBounds(field, assert_cast(type.get())->getNestedType()); + + if (!WhichDataType(type).isDateTime64()) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected DateTime64, got {}", type->getName()); + + const auto scale = getDecimalScale(*type); + const auto value = field.safeGet().getValue().value; + if (scale == 9) + return value; + if (scale > 6) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName()); + + /// Iceberg `timestamp` bounds are microseconds. DateTime64(s) for s < 6 stores coarser ticks. + return value * DataTypeDateTime64::getScaleMultiplier(6 - scale).value; +} + std::vector dumpFieldToBytes(const Field & field, DataTypePtr type) { switch (type->getTypeId()) @@ -206,7 +226,7 @@ std::vector dumpFieldToBytes(const Field & field, DataTypePtr type) case TypeIndex::Time64: return dumpValue(getTimeValueInMicroseconds(field, type)); case TypeIndex::DateTime64: - return dumpValue(field.safeGet().getValue().value); + return dumpValue(getDateTime64ValueForIcebergBounds(field, type)); case TypeIndex::String: { auto value = field.safeGet(); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp index 0a02fd75c04fd..38fb66967d75b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -24,6 +25,37 @@ using namespace DB; +namespace +{ + +/// Avro does not map timestamp-nanos to DateTime64, so identity partition values for +/// Iceberg `timestamp_ns` / `timestamptz_ns` (and similarly `Time64`) arrive as Int64. +/// `KeyCondition` compares DateTime64/Time64 predicates as DecimalField with a scale, +/// so wrap the raw ticks before `mayBeTrueInRange`. Calendar transforms stay Int64. +void wrapIntegerPartitionValuesAsDateTime64(Row & partition_values, const DataTypes & data_types) +{ + const size_t size = std::min(partition_values.size(), data_types.size()); + for (size_t i = 0; i < size; ++i) + { + auto & field = partition_values[i]; + if (field.isNull() || field.getType() != Field::Types::Int64) + continue; + + auto type = removeNullable(data_types[i]); + WhichDataType which(type); + if (which.isDateTime64()) + { + field = DecimalField(DateTime64(field.safeGet()), getDecimalScale(*type)); + } + else if (which.isTime64()) + { + field = DecimalField(Time64(field.safeGet()), getDecimalScale(*type)); + } + } +} + +} + namespace DB::Iceberg { @@ -149,7 +181,9 @@ PruningReturnStatus ManifestFilesPruner::canBePruned( { if (partition_key_condition.has_value()) { - const auto & partition_value = entry->parsed_entry->partition_key_value; + Row partition_value = entry->parsed_entry->partition_key_value; + wrapIntegerPartitionValuesAsDateTime64(partition_value, partition_key->data_types); + std::vector index_value(partition_value.begin(), partition_value.end()); for (auto & field : index_value) { diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index c340a93af8be8..e0873b7cd60d1 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -537,8 +538,24 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite case TypeIndex::Date32: return {"date", true}; case TypeIndex::DateTime: - case TypeIndex::DateTime64: return {"timestamp", true}; + case TypeIndex::DateTime64: + { + /// Iceberg `timestamp` is microseconds; `timestamp_ns` is nanoseconds (v3). + /// Mapping all DateTime64 to `timestamp` while dumping unscaled ticks made + /// ClickHouse-written DateTime64(9) recreate ns-bounds-on-us-schema over-pruning. + const auto scale = getDecimalScale(*type); + if (scale == 9) + { + auto date_time64 = std::static_pointer_cast(type); + if (date_time64->hasExplicitTimeZone()) + return {Iceberg::f_timestamptz_ns, true}; + return {Iceberg::f_timestamp_ns, true}; + } + if (scale <= 6) + return {Iceberg::f_timestamp, true}; + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName()); + } case TypeIndex::Time: return {"time", true}; case TypeIndex::Time64: diff --git a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py index 486adf1f2b17a..4f75f9c4c6f0b 100644 --- a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py +++ b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py @@ -209,7 +209,7 @@ def test_export_part_all_iceberg_types(cluster): float f32 Float32 double f64 Float64 date event_dt Date - timestamp ts DateTime64(6) (DateTime / DateTime64 both → "timestamp") + timestamp ts DateTime64(6) (DateTime / DateTime64(scale<=6) → "timestamp"; DateTime64(9) → "timestamp_ns") string name String uuid uid_val UUID diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py new file mode 100644 index 0000000000000..8790d2b4af87c --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +"""ClickHouse Iceberg writes of DateTime64(9) must use timestamp_ns, not timestamp. + +Mapping DateTime64(9) to Iceberg `timestamp` while dumping unscaled nanosecond +min/max ticks recreates the customer over-prune (schema microseconds, stats +nanoseconds). +""" + +import pytest + +from helpers.iceberg_timestamp_ns import ( + FILTER_LITERAL_IDS, + FILTER_LITERALS, + PRUNE_OFF, + PRUNE_ON, + ROWS, + check_ns_pruning, +) +from helpers.iceberg_utils import ( + check_validity_and_get_prunned_files_general, + create_iceberg_table, + get_uuid_str, +) + + +def _insert_ns_rows(instance, table_name): + for ts_ns, row_id, _utc in ROWS: + instance.query( + f"INSERT INTO {table_name} VALUES " + f"(fromUnixTimestamp64Nano({ts_ns}), fromUnixTimestamp64Nano({ts_ns}), {row_id})" + ) + + +def _assert_written_as_datetime64_9(instance, table_name, timezone=None): + describe = instance.query(f"DESCRIBE TABLE {table_name}") + assert "DateTime64(9" in describe, describe + assert "DateTime64(6" not in describe, describe + if timezone == "UTC": + assert "UTC" in describe, describe + + +@pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS) +def test_writes_minmax_pruning_datetime64_nanoseconds( + started_cluster_iceberg_no_spark, make_literal +): + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_minmax_ns_" + get_uuid_str() + create_iceberg_table( + "s3", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(ts DateTime64(9), value DateTime64(9), id Int32)", + format_version=3, + ) + _assert_written_as_datetime64_9(instance, table_name) + _insert_ns_rows(instance, table_name) + + def pruned_files(select_expression): + return check_validity_and_get_prunned_files_general( + instance, table_name, PRUNE_OFF, PRUNE_ON, "IcebergMinMaxIndexPrunedFiles", select_expression + ) + + check_ns_pruning(instance, table_name, "value", pruned_files, make_literal) + + +@pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS) +def test_writes_partition_pruning_datetime64_nanoseconds( + started_cluster_iceberg_no_spark, make_literal +): + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_part_ns_" + get_uuid_str() + create_iceberg_table( + "s3", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(ts DateTime64(9), value DateTime64(9), id Int32)", + format_version=3, + partition_by="ts", + ) + _assert_written_as_datetime64_9(instance, table_name) + _insert_ns_rows(instance, table_name) + + def pruned_files(select_expression): + return check_validity_and_get_prunned_files_general( + instance, table_name, PRUNE_OFF, PRUNE_ON, "IcebergPartitionPrunedFiles", select_expression + ) + + check_ns_pruning(instance, table_name, "ts", pruned_files, make_literal) + + +def test_writes_datetime64_nanoseconds_with_timezone(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_ns_tz_" + get_uuid_str() + create_iceberg_table( + "s3", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(ts DateTime64(9, 'UTC'), id Int32)", + format_version=3, + ) + _assert_written_as_datetime64_9(instance, table_name, timezone="UTC") + instance.query( + f"INSERT INTO {table_name} VALUES (fromUnixTimestamp64Nano({ROWS[0][0]}), 1)" + ) + assert instance.query(f"SELECT id FROM {table_name}") == "1\n" From 5a3173b274c9dac28fec90599b4f1b0d66ec6469 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 18 Aug 2026 22:37:35 +0200 Subject: [PATCH 03/15] Do not treat far-future Iceberg timestamp bounds as nanoseconds `1e16` microseconds is ~year 2286, so spec-correct Spark `9999-12-31` and `DateTime64` max `2299-12-31` bounds were divided by 1000 and over-pruned. Convert only `|ticks| > 1e18` (true nanosecond band); fail open in between. Co-authored-by: Cursor --- .../Iceberg/IcebergFieldParseHelpers.cpp | 28 ++-- .../helpers/iceberg_timestamp_ns.py | 151 ++++++++++++++++++ .../test_pruning_datetime64_nanoseconds.py | 55 ++++++- 3 files changed, 218 insertions(+), 16 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index 35c912e2b3875..22b027a22afeb 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -26,10 +26,12 @@ namespace Iceberg namespace { -/// ~year 2286 as DateTime64(6) microseconds. Spec-correct Iceberg `timestamp` bounds -/// for realistic dates stay below this; nanosecond ticks for dates after ~1970-04 -/// sit well above it (2024 is ~1.7e18). -constexpr Int64 microseconds_implausible_threshold = 10'000'000'000'000'000LL; +/// 1e16 µs is ~year 2286. That is not an Iceberg `timestamp` max: Spark sentinels +/// such as `9999-12-31` (`~2.53e17` µs) and ClickHouse `DateTime64` max +/// `2299-12-31` (`~1.04e16` µs) are spec-correct microseconds above this. +/// 1e18 ns is ~year 2001; customer 2026 ns bounds are `~1.76e18`. +constexpr Int64 microseconds_ambiguous_threshold = 10'000'000'000'000'000LL; +constexpr Int64 nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL; constexpr Int64 nanoseconds_per_microsecond = 1000; bool magnitudeExceeds(Int64 value, Int64 threshold) @@ -57,12 +59,11 @@ Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound) return microseconds; } -/// Iceberg timestamps are 8-byte little-endian (spec Appendix D). `insertData` -/// memcpy's native DateTime64 ticks, which is correct when the bound scale -/// matches the column. Some writers store nanosecond stats on Iceberg -/// `timestamp` (`DateTime64(6)`); treating those as microseconds yields dates -/// around year 2299 and over-prunes. Rescale those bounds; if they are still -/// implausible, fail open (`nullopt`) so min/max pruning is skipped. +/// Iceberg timestamps are 8-byte little-endian (spec Appendix D). +/// Some writers store nanosecond stats on Iceberg `timestamp` (`DateTime64(6)`). +/// Convert only in a true nanosecond band (`|ticks| > 1e18`). Between 1e16 and +/// 1e18 the value may be far-future microseconds (do not convert; fail open so +/// min/max pruning is skipped). `|ticks| <= 1e16` is kept as microseconds. std::optional deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound) { if (str.size() != sizeof(Int64)) @@ -71,11 +72,12 @@ std::optional deserializeDateTime64FromBinaryRepr(const std::string & str Int64 unscaled = unalignedLoadLittleEndian(str.data()); const UInt32 scale = getDecimalScale(type); - if (scale == 6 && magnitudeExceeds(unscaled, microseconds_implausible_threshold)) + if (scale == 6 && magnitudeExceeds(unscaled, microseconds_ambiguous_threshold)) { - unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); - if (magnitudeExceeds(unscaled, microseconds_implausible_threshold)) + if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold)) return std::nullopt; + + unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); } return DecimalField(DateTime64(unscaled), scale); diff --git a/tests/integration/helpers/iceberg_timestamp_ns.py b/tests/integration/helpers/iceberg_timestamp_ns.py index 37b9e365fa82d..351b3cc9eeb1c 100644 --- a/tests/integration/helpers/iceberg_timestamp_ns.py +++ b/tests/integration/helpers/iceberg_timestamp_ns.py @@ -7,6 +7,7 @@ import json import struct +from datetime import datetime, timezone from pathlib import Path import avro.datafile @@ -206,6 +207,52 @@ def check_minmax_ns_bounds_on_timestamp_schema(node, table, column, pruned_files 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. +SPARK_TIMESTAMP_SENTINEL_US = _utc_datetime_to_micros( + datetime(9999, 12, 31, 23, 59, 59, 999999, tzinfo=timezone.utc) +) +DATETIME64_MAX_US = _utc_datetime_to_micros( + datetime(2299, 12, 31, 23, 59, 59, 999999, tzinfo=timezone.utc) +) + + +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. + + 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. + """ + 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" + micros = node.query( + f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1", + settings=PRUNE_OFF, + ).strip() + assert micros == "1704067200123456", micros + + 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" + assert pruned_files(query) == 0 + 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 @@ -317,6 +364,110 @@ def patch_iceberg_timestamp_bounds_us_to_ns( 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 + return struct.pack(" 0, f"no timestamp upper bounds patched under {table_location}" + + def upgrade_long_to_iceberg_timestamp(obj, iceberg_type="timestamp_ns"): if isinstance(obj, dict): if obj.get("name") in NS_COLUMN_NAMES and obj.get("type") == "long": 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 e703492ec35fd..e30843636eec2 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 @@ -21,16 +21,20 @@ from pyiceberg.types import IntegerType, LongType, TimestampType from helpers.iceberg_timestamp_ns import ( + DATETIME64_MAX_US, FILTER_LITERAL_IDS, FILTER_LITERALS, PRUNE_OFF, PRUNE_ON, 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 @@ -72,8 +76,8 @@ def _arrow_timestamp_row(ts_ns, row_id): ) -def _create_timestamp_table_with_ns_bounds(started_cluster, table_name): - """Iceberg v2 `timestamp` (microseconds) whose manifest min/max bytes are nanoseconds.""" +def _create_timestamp_table(started_cluster, table_name, patch): + """Iceberg v2 `timestamp` (microseconds parquet) with a custom manifest-bound patch.""" work = Path(tempfile.mkdtemp(prefix="iceberg_ns_bounds_")) try: catalog = SqlCatalog( @@ -102,7 +106,7 @@ def _create_timestamp_table_with_ns_bounds(started_cluster, table_name): s3_key = f"iceberg_ns_bounds_{uuid.uuid4().hex}" new_prefix = f"s3://{started_cluster.minio_bucket}/{s3_key}" - patch_iceberg_timestamp_bounds_us_to_ns(table_location, old_prefix, new_prefix) + patch(table_location, old_prefix, new_prefix) uploader = started_cluster.default_s3_uploader for path in table_location.rglob("*"): @@ -115,6 +119,23 @@ def _create_timestamp_table_with_ns_bounds(started_cluster, table_name): shutil.rmtree(work, ignore_errors=True) +def _create_timestamp_table_with_ns_bounds(started_cluster, table_name): + """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 + ) + + +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): + patch_iceberg_timestamp_upper_bounds( + table_location, micros, old_prefix, new_prefix + ) + + return _create_timestamp_table(started_cluster, table_name, patch) + + def _create_v3_table( started_cluster, table_name, partition_spec, iceberg_type="timestamp_ns", parquet_unit="ns" ): @@ -242,3 +263,31 @@ def pruned_files(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], + ids=["spark_9999_12_31", "datetime64_2299_12_31"], +) +def test_minmax_pruning_far_future_us_upper_bound_on_timestamp_schema( + started_cluster_iceberg_no_spark, upper_bound_us +): + """Spec-correct far-future Iceberg `timestamp` upper bounds must not be treated as ns.""" + node = started_cluster_iceberg_no_spark.instances["node1"] + s3_filename = _create_timestamp_table_with_upper_bound_us( + started_cluster_iceberg_no_spark, "minmax_far_future_us", upper_bound_us + ) + table = _iceberg_table(s3_filename) + + def pruned_files(select_expression): + return check_validity_and_get_prunned_files_general( + node, + "minmax_far_future_us", + PRUNE_OFF, + PRUNE_ON, + "IcebergMinMaxIndexPrunedFiles", + select_expression, + ) + + check_minmax_far_future_us_upper_bound(node, table, "value", pruned_files) From 381b43d5e2d94bd8bf9554f5dbc782f0ed8eee73 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 18 Aug 2026 22:49:54 +0200 Subject: [PATCH 04/15] Reject DateTime64(9) on Iceberg format versions below 3 `timestamp_ns` is v3-only. Mapping `DateTime64(9)` without checking `iceberg_format_version` wrote invalid v2 metadata (default is 2). Throw instead of emitting v3 types into v1/v2 tables. Co-authored-by: Cursor --- .../table-engines/integrations/iceberg.md | 2 +- .../DataLakes/Iceberg/Compaction.cpp | 4 +- .../DataLakes/Iceberg/MetadataGenerator.cpp | 6 +- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 26 +++-- .../ObjectStorage/DataLakes/Iceberg/Utils.h | 5 +- .../test_writes_datetime64_nanoseconds.py | 107 ++++++++++++++++++ 6 files changed, 135 insertions(+), 15 deletions(-) diff --git a/docs/en/engines/table-engines/integrations/iceberg.md b/docs/en/engines/table-engines/integrations/iceberg.md index 2a61173e30ed7..728c2a4291160 100644 --- a/docs/en/engines/table-engines/integrations/iceberg.md +++ b/docs/en/engines/table-engines/integrations/iceberg.md @@ -98,7 +98,7 @@ The following table shows how Iceberg data types are mapped to ClickHouse data t | `fixed(N)` | `FixedString(N)` | | | `decimal(P, S)` | `Decimal(P, S)` | | -Writes use the inverse mapping: `DateTime64(9)` is stored as `timestamp_ns`, and `DateTime64` with an explicit timezone and scale 9 as `timestamptz_ns`. `DateTime` and `DateTime64` with scale 6 or less are stored as `timestamp`. +Writes use the inverse mapping: `DateTime64(9)` is stored as `timestamp_ns`, and `DateTime64` with an explicit timezone and scale 9 as `timestamptz_ns`, but only when `iceberg_format_version` is 3 or higher. `DateTime` and `DateTime64` with scale 6 or less are stored as `timestamp`. Creating or altering an Iceberg table with `DateTime64` scale 9 on format version 1 or 2 raises an exception. ### Complex types {#complex-types} diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp index 477c22316d9eb..8478e1231a355 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp @@ -401,7 +401,9 @@ static void writeMetadataFiles( auto log = getLogger("IcebergCompaction"); ColumnsDescription columns_description = ColumnsDescription::fromNamesAndTypes(sample_block_->getNamesAndTypes()); - auto [metadata_object, metadata_object_str] = createEmptyMetadataFile(table_path, columns_description, nullptr, nullptr, context); + const auto format_version = plan.initial_metadata_object->getValue(Iceberg::f_format_version); + auto [metadata_object, metadata_object_str] = createEmptyMetadataFile( + table_path, columns_description, nullptr, nullptr, context, format_version); auto current_schema_id = metadata_object->getValue(Iceberg::f_current_schema_id); Poco::JSON::Object::Ptr current_schema; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index 459a78056e0d0..16d6da3d2ccf6 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -320,7 +320,8 @@ void MetadataGenerator::generateAddColumnMetadata(const String & column_name, Da auto last_column_id = metadata_object->getValue(Iceberg::f_last_column_id); metadata_object->set(Iceberg::f_last_column_id, last_column_id + 1); - auto new_type = Iceberg::getIcebergType(type, last_column_id); + auto format_version = metadata_object->getValue(Iceberg::f_format_version); + auto new_type = Iceberg::getIcebergType(type, last_column_id, format_version); Poco::JSON::Object::Ptr new_field = new Poco::JSON::Object; new_field->set(Iceberg::f_id, last_column_id + 1); new_field->set(Iceberg::f_name, column_name); @@ -352,8 +353,9 @@ void MetadataGenerator::generateModifyColumnMetadata(const String & column_name, throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found schema with id {}", current_schema_id); current_schema = deepCopy(current_schema); auto last_column_id = metadata_object->getValue(Iceberg::f_last_column_id); + auto format_version = metadata_object->getValue(Iceberg::f_format_version); - auto new_type = Iceberg::getIcebergType(type, last_column_id); + auto new_type = Iceberg::getIcebergType(type, last_column_id, format_version); auto schema_fields = current_schema->getArray(Iceberg::f_fields); bool found = false; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index e0873b7cd60d1..63835296e4824 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -518,7 +518,7 @@ Poco::JSON::Object::Ptr getMetadataJSONObject( } /// Returns type and required -std::pair getIcebergType(DataTypePtr type, Int32 & iter) +std::pair getIcebergType(DataTypePtr type, Int32 & iter, UInt64 format_version) { switch (type->getTypeId()) { @@ -541,12 +541,18 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite return {"timestamp", true}; case TypeIndex::DateTime64: { - /// Iceberg `timestamp` is microseconds; `timestamp_ns` is nanoseconds (v3). - /// Mapping all DateTime64 to `timestamp` while dumping unscaled ticks made - /// ClickHouse-written DateTime64(9) recreate ns-bounds-on-us-schema over-pruning. + /// Iceberg `timestamp` is microseconds; `timestamp_ns` is nanoseconds and is + /// allowed only in format version 3. Writing it into v1/v2 metadata is invalid. const auto scale = getDecimalScale(*type); if (scale == 9) { + if (format_version < 3) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Iceberg type {} requires format version 3 or higher, got version {}", + type->getName(), + format_version); + auto date_time64 = std::static_pointer_cast(type); if (date_time64->hasExplicitTimeZone()) return {Iceberg::f_timestamptz_ns, true}; @@ -580,7 +586,7 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite Poco::JSON::Object::Ptr field = new Poco::JSON::Object; field->set(Iceberg::f_id, ++iter_fields); field->set(Iceberg::f_name, type_tuple->getNameByPosition(iter_names)); - auto child_type = getIcebergType(element->getNormalizedType(), iter); + auto child_type = getIcebergType(element->getNormalizedType(), iter, format_version); field->set(Iceberg::f_required, child_type.second); field->set(Iceberg::f_type, child_type.first); fields->add(field); @@ -596,7 +602,7 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite field->set(Iceberg::f_type, "list"); field->set(Iceberg::f_element_id, ++iter); - auto child_type = getIcebergType(type_array->getNestedType(), iter); + auto child_type = getIcebergType(type_array->getNestedType(), iter, format_version); field->set(Iceberg::f_required, false); field->set(Iceberg::f_element, child_type.first); field->set(Iceberg::f_element_required, child_type.second); @@ -611,8 +617,8 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite field->set(Iceberg::f_key_id, ++iter); field->set(Iceberg::f_value_id, ++iter); - field->set(Iceberg::f_key, getIcebergType(type_map->getKeyType(), iter).first); - auto value_type = getIcebergType(type_map->getValueType(), iter); + field->set(Iceberg::f_key, getIcebergType(type_map->getKeyType(), iter, format_version).first); + auto value_type = getIcebergType(type_map->getValueType(), iter, format_version); field->set(Iceberg::f_value, value_type.first); field->set(Iceberg::f_value_required, value_type.second); return {field, true}; @@ -620,7 +626,7 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite case TypeIndex::Nullable: { auto type_nullable = std::static_pointer_cast(type); - return {getIcebergType(type_nullable->getNestedType(), iter).first, false}; + return {getIcebergType(type_nullable->getNestedType(), iter, format_version).first, false}; } case TypeIndex::Variant: { @@ -1035,7 +1041,7 @@ std::pair createEmptyMetadataFile( Poco::JSON::Object::Ptr field = new Poco::JSON::Object; field->set(Iceberg::f_id, ++iter_for_initial_columns); field->set(Iceberg::f_name, column.name); - auto type = getIcebergType(column.type, iter); + auto type = getIcebergType(column.type, iter, format_version); field->set(Iceberg::f_required, type.second); field->set(Iceberg::f_type, type.first); column_name_to_source_id[column.name] = iter_for_initial_columns; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h index bdedcdf095bae..9fced7fa5e7ff 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h @@ -92,7 +92,10 @@ Poco::JSON::Object::Ptr getMetadataJSONObject( const std::optional & table_uuid); -std::pair getIcebergType(DataTypePtr type, Int32 & iter); +/// Maps a ClickHouse type to an Iceberg JSON type. `timestamp_ns` / `timestamptz_ns` +/// require `format_version` >= 3; lower versions throw instead of writing v3 types +/// into v1/v2 metadata. +std::pair getIcebergType(DataTypePtr type, Int32 & iter, UInt64 format_version); Poco::Dynamic::Var getAvroType(DataTypePtr type); Poco::Dynamic::Var getAvroLogicalType(DataTypePtr type); diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py index 8790d2b4af87c..29591c3339400 100644 --- a/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py +++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py @@ -19,6 +19,7 @@ from helpers.iceberg_utils import ( check_validity_and_get_prunned_files_general, create_iceberg_table, + get_creation_expression, get_uuid_str, ) @@ -106,3 +107,109 @@ def test_writes_datetime64_nanoseconds_with_timezone(started_cluster_iceberg_no_ f"INSERT INTO {table_name} VALUES (fromUnixTimestamp64Nano({ROWS[0][0]}), 1)" ) assert instance.query(f"SELECT id FROM {table_name}") == "1\n" + + +def _assert_timestamp_ns_requires_v3(error): + assert "BAD_ARGUMENTS" in error, error + assert "LOGICAL_ERROR" not in error, error + assert "format version 3" in error, error + + +@pytest.mark.parametrize("format_version", [1, 2]) +@pytest.mark.parametrize( + "schema", + [ + "(ts DateTime64(9), id Int32)", + "(ts DateTime64(9, 'UTC'), id Int32)", + "(ts Array(DateTime64(9)), id Int32)", + ], + ids=["timestamp_ns", "timestamptz_ns", "nested"], +) +def test_writes_datetime64_nanoseconds_rejected_on_v1_v2( + started_cluster_iceberg_no_spark, format_version, schema +): + """DateTime64(9) is Iceberg v3 timestamp_ns; v1/v2 (default is 2) must not emit it.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_ns_v2_reject_" + get_uuid_str() + error = instance.query_and_get_error( + get_creation_expression( + "s3", + table_name, + started_cluster_iceberg_no_spark, + schema, + format_version, + ) + ) + _assert_timestamp_ns_requires_v3(error) + + +def test_writes_datetime64_nanoseconds_rejected_on_default_format_version( + started_cluster_iceberg_no_spark, +): + """CREATE without iceberg_format_version uses default 2 and must reject DateTime64(9).""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_ns_default_ver_" + get_uuid_str() + bucket = started_cluster_iceberg_no_spark.minio_bucket + error = instance.query_and_get_error( + f""" + DROP TABLE IF EXISTS {table_name}; + CREATE TABLE {table_name} (ts DateTime64(9), id Int32) + ENGINE = IcebergS3(s3, filename = 'var/lib/clickhouse/user_files/iceberg_data/default/{table_name}/', format=Parquet, url = 'http://minio1:9001/{bucket}/'); + """ + ) + _assert_timestamp_ns_requires_v3(error) + + +def test_writes_datetime64_nanoseconds_alter_rejected_on_v2(started_cluster_iceberg_no_spark): + """ALTER ADD / MODIFY must not write timestamp_ns into existing v2 metadata.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_ns_alter_v2_" + get_uuid_str() + create_iceberg_table( + "s3", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(ts DateTime64(6), id Int32)", + format_version=2, + ) + alter_settings = {"allow_insert_into_iceberg": 1} + + add_error = instance.query_and_get_error( + f"ALTER TABLE {table_name} ADD COLUMN ts_ns Nullable(DateTime64(9))", + settings=alter_settings, + ) + _assert_timestamp_ns_requires_v3(add_error) + + modify_error = instance.query_and_get_error( + f"ALTER TABLE {table_name} MODIFY COLUMN ts DateTime64(9)", + settings=alter_settings, + ) + _assert_timestamp_ns_requires_v3(modify_error) + + describe = instance.query(f"DESCRIBE TABLE {table_name}") + assert "DateTime64(6)" in describe, describe + assert "DateTime64(9" not in describe, describe + + +def test_writes_datetime64_nanoseconds_alter_allowed_on_v3(started_cluster_iceberg_no_spark): + """ALTER ADD Nullable(DateTime64(9)) is valid on Iceberg v3.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_ns_alter_v3_" + get_uuid_str() + create_iceberg_table( + "s3", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(id Int32)", + format_version=3, + ) + instance.query( + f"ALTER TABLE {table_name} ADD COLUMN ts_ns Nullable(DateTime64(9))", + settings={"allow_insert_into_iceberg": 1}, + ) + describe = instance.query(f"DESCRIBE TABLE {table_name}") + assert "DateTime64(9" in describe, describe + instance.query( + f"INSERT INTO {table_name} VALUES (1, fromUnixTimestamp64Nano({ROWS[0][0]}))" + ) + assert instance.query(f"SELECT id FROM {table_name}") == "1\n" From 273e50ff03805ee6f35fa52a1af570e7142c8cab Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 4 Sep 2026 11:39:22 +0200 Subject: [PATCH 05/15] Fix build --- .../DataLakes/Iceberg/MetadataGenerator.cpp | 6 ++-- .../tests/gtest_iceberg_type_mapping.cpp | 36 ++++++++++++------- 2 files changed, 28 insertions(+), 14 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index bd375aea6d2e6..eba2cec5a84be 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -253,7 +253,8 @@ bool MetadataGenerator::isAddColumnApplied(const String & column_name, DataTypeP return false; Int32 unused_field_id = metadata_object->getValue(Iceberg::f_last_column_id); - auto expected_type = Iceberg::getIcebergType(type, unused_field_id); + auto format_version = metadata_object->getValue(Iceberg::f_format_version); + auto expected_type = Iceberg::getIcebergType(type, unused_field_id, format_version); auto fields = current_schema->getArray(Iceberg::f_fields); for (UInt32 i = 0; i < fields->size(); ++i) @@ -310,7 +311,8 @@ bool MetadataGenerator::isModifyColumnApplied(const String & column_name, DataTy return false; Int32 unused_field_id = metadata_object->getValue(Iceberg::f_last_column_id); - auto expected_type = Iceberg::getIcebergType(type, unused_field_id); + auto format_version = metadata_object->getValue(Iceberg::f_format_version); + auto expected_type = Iceberg::getIcebergType(type, unused_field_id, format_version); auto fields = current_schema->getArray(Iceberg::f_fields); for (UInt32 i = 0; i < fields->size(); ++i) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp index 0b51a126abc8a..3168cdbf90922 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp @@ -24,7 +24,8 @@ TEST(IcebergTypeMapping, BoolMapsToBoolean) { auto bool_type = DataTypeFactory::instance().get("Bool"); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(bool_type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(bool_type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "boolean"); EXPECT_TRUE(required); @@ -34,7 +35,8 @@ TEST(IcebergTypeMapping, NullableBoolMapsToBoolean) { auto bool_type = makeNullable(DataTypeFactory::instance().get("Bool")); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(bool_type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(bool_type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "boolean"); EXPECT_FALSE(required); @@ -44,7 +46,8 @@ TEST(IcebergTypeMapping, UInt8MapsToInt) { auto type = std::make_shared(); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "int"); EXPECT_TRUE(required); @@ -54,7 +57,8 @@ TEST(IcebergTypeMapping, Int8MapsToInt) { auto type = std::make_shared(); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "int"); } @@ -63,7 +67,8 @@ TEST(IcebergTypeMapping, UInt16MapsToInt) { auto type = std::make_shared(); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "int"); } @@ -72,7 +77,8 @@ TEST(IcebergTypeMapping, Int16MapsToInt) { auto type = std::make_shared(); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "int"); } @@ -81,7 +87,8 @@ TEST(IcebergTypeMapping, Decimal32MapsToDecimal) { auto type = std::make_shared>(9, 2); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "decimal(9, 2)"); EXPECT_TRUE(required); @@ -91,7 +98,8 @@ TEST(IcebergTypeMapping, Decimal64MapsToDecimal) { auto type = std::make_shared>(18, 5); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "decimal(18, 5)"); } @@ -100,7 +108,8 @@ TEST(IcebergTypeMapping, Decimal128MapsToDecimal) { auto type = std::make_shared>(38, 10); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "decimal(38, 10)"); } @@ -109,7 +118,8 @@ TEST(IcebergTypeMapping, NullableDecimalMapsToDecimalNotRequired) { auto type = makeNullable(std::make_shared>(7, 3)); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "decimal(7, 3)"); EXPECT_FALSE(required); @@ -126,9 +136,10 @@ constexpr UInt32 iceberg_max_decimal_precision = 38; void expectIcebergTypeRejected(const DataTypePtr & type) { Int32 iter = 0; + UInt64 format_version = 3; try { - auto [iceberg_type, required] = getIcebergType(type, iter); + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); FAIL() << type->getName() << " has no Iceberg representation but was mapped to " << iceberg_type.toString(); } @@ -146,7 +157,8 @@ TEST(IcebergTypeMapping, EveryDecimalPrecisionWithinSpecLimitMaps) { auto type = createDecimal(precision, 1); Int32 iter = 0; - auto [iceberg_type, required] = getIcebergType(type, iter); + UInt64 format_version = 3; + auto [iceberg_type, required] = getIcebergType(type, iter, format_version); ASSERT_TRUE(iceberg_type.isString()) << "precision " << precision; EXPECT_EQ(iceberg_type.extract(), "decimal(" + std::to_string(precision) + ", 1)"); EXPECT_TRUE(required); From 6f0719e7692589760acb09db801293c90a314aa6 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 1 Oct 2026 16:16:38 +0200 Subject: [PATCH 06/15] Fix Iceberg inserts into tables partitioned by `DateTime64(9)` `getAvroType` required `timestamp-micros` after the #109764 backport, so identity-partitioned `timestamp_ns` writes failed. Co-authored-by: Cursor --- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 8 +++- .../test_writes_datetime64_nanoseconds.py | 41 +++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 93d1d839018db..a1db54dacf391 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -705,12 +705,16 @@ Poco::Dynamic::Var getAvroType(DataTypePtr type, Int32 field_id) } case TypeIndex::DateTime64: { - if (getDecimalScale(*type) != 6) + /// Iceberg `timestamp` is microseconds; `timestamp_ns` is nanoseconds (v3). + /// contrib Avro has no `timestamp-nanos` enum, so it stores the logical type + /// as `NONE` and the value as `long`; other engines still see the annotation. + const auto scale = getDecimalScale(*type); + if (scale != 6 && scale != 9) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName()); Poco::JSON::Object::Ptr timestamp_type = new Poco::JSON::Object; timestamp_type->set("type", "long"); - timestamp_type->set("logicalType", "timestamp-micros"); + timestamp_type->set("logicalType", scale == 9 ? "timestamp-nanos" : "timestamp-micros"); timestamp_type->set("adjust-to-utc", assert_cast(*type).hasExplicitTimeZone()); return timestamp_type; } diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py index 29591c3339400..3b399f0e7eb96 100644 --- a/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py +++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_datetime64_nanoseconds.py @@ -109,6 +109,47 @@ def test_writes_datetime64_nanoseconds_with_timezone(started_cluster_iceberg_no_ assert instance.query(f"SELECT id FROM {table_name}") == "1\n" +@pytest.mark.parametrize( + "schema, timezone", + [ + ("(ts DateTime64(9), id Int32)", None), + ("(ts DateTime64(9, 'UTC'), id Int32)", "UTC"), + ], + ids=["timestamp_ns", "timestamptz_ns"], +) +def test_writes_identity_partitioned_datetime64_nanoseconds_insert( + started_cluster_iceberg_no_spark, schema, timezone +): + """INSERT into a v3 Iceberg table partitioned by DateTime64(9) must succeed. + + Manifest partition fields go through `getAvroType`. Scale 6 is encoded as + Avro `timestamp-micros`; scale 9 must still be writable as `timestamp_ns` + (not rejected as an unsupported Iceberg type). + """ + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_writes_part_ns_insert_" + get_uuid_str() + create_iceberg_table( + "s3", + instance, + table_name, + started_cluster_iceberg_no_spark, + schema, + format_version=3, + partition_by="ts", + ) + _assert_written_as_datetime64_9(instance, table_name, timezone=timezone) + + ts_ns, row_id, _utc = ROWS[0] + instance.query( + f"INSERT INTO {table_name} VALUES (fromUnixTimestamp64Nano({ts_ns}), {row_id})" + ) + assert instance.query(f"SELECT id FROM {table_name}") == f"{row_id}\n" + assert ( + instance.query(f"SELECT toUnixTimestamp64Nano(ts) FROM {table_name}") + == f"{ts_ns}\n" + ) + + def _assert_timestamp_ns_requires_v3(error): assert "BAD_ARGUMENTS" in error, error assert "LOGICAL_ERROR" not in error, error From d08eb59da92e92b6afe0e4e943d99c2b4959a75c Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 7 Oct 2026 10:48:48 +0200 Subject: [PATCH 07/15] Honor `iceberg_timezone_for_timestamptz` for Iceberg `timestamptz_ns` `getSimpleType` hardcoded UTC for nanosecond timestamptz while microsecond `timestamptz` already used the setting. Co-authored-by: Cursor --- .../ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp | 5 ++++- .../Iceberg/tests/gtest_iceberg_schema_processor.cpp | 5 +++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp index f6c989b13ad38..d198fba1f8882 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp @@ -440,7 +440,10 @@ DataTypePtr IcebergSchemaProcessor::getSimpleType(const String & type_name_arg, if (type_name == f_timestamp_ns) return std::make_shared(9); if (type_name == f_timestamptz_ns) - return std::make_shared(9, "UTC"); + { + std::string timezone = context_->getSettingsRef()[Setting::iceberg_timezone_for_timestamptz]; + return std::make_shared(9, timezone); + } if (type_name == f_string || type_name == f_binary) return std::make_shared(); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_schema_processor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_schema_processor.cpp index e13a421eda1f1..8148abae1887c 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_schema_processor.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_schema_processor.cpp @@ -89,6 +89,11 @@ TEST(IcebergSchemaProcessor, GetSimpleTypeTimestamptzNs) { auto type = IcebergSchemaProcessor::getSimpleType("timestamptz_ns", getContext().context); EXPECT_EQ(type->getName(), "DateTime64(9, 'UTC')"); + + auto context = DB::Context::createCopy(getContext().context); + context->setSetting("iceberg_timezone_for_timestamptz", String("Europe/Berlin")); + type = IcebergSchemaProcessor::getSimpleType("timestamptz_ns", context); + EXPECT_EQ(type->getName(), "DateTime64(9, 'Europe/Berlin')"); } TEST(IcebergSchemaProcessor, GetSimpleTypeString) From 1b0e9d6dc93da30069a08a631d7e6980db6038b8 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 7 Oct 2026 11:53:52 +0200 Subject: [PATCH 08/15] Write Iceberg `timestamp_ns` Parquet as not UTC-adjusted Default Parquet still marks every `DateTime64` as `isAdjustedToUTC`. Iceberg writes set the flag from the column timezone so `timestamp_ns` matches `TIMESTAMP(NANOS, false)`. Co-authored-by: Cursor --- src/Formats/FormatSettings.h | 4 ++++ .../Formats/Impl/Parquet/PrepareForWrite.cpp | 8 ++++--- src/Processors/Formats/Impl/Parquet/Write.h | 1 + .../Formats/Impl/ParquetBlockOutputFormat.cpp | 1 + .../tests/gtest_write_parquet_page_index.cpp | 23 +++++++++++++++++++ .../DataLakes/Iceberg/Compaction.cpp | 6 ++++- .../DataLakes/Iceberg/MultipleFileWriter.cpp | 12 +++++----- .../DataLakes/Iceberg/Mutations.cpp | 8 +++++-- 8 files changed, 51 insertions(+), 12 deletions(-) diff --git a/src/Formats/FormatSettings.h b/src/Formats/FormatSettings.h index 745898c0c751b..ee2877b7c2711 100644 --- a/src/Formats/FormatSettings.h +++ b/src/Formats/FormatSettings.h @@ -374,6 +374,10 @@ struct FormatSettings bool output_datetime_as_uint32 = false; bool output_date_as_uint16 = false; bool output_enum_as_byte_array = false; + /// When false (default), DateTime64 is always written as isAdjustedToUTC=true because + /// ClickHouse ticks are UTC. Iceberg sets this so timestamp(_ns) is false and + /// timestamptz(_ns) is true, matching TIMESTAMP(*, hasExplicitTimeZone()). + bool output_timestamp_adjusted_to_utc_from_timezone = false; bool parallel_encoding = true; bool write_page_index = false; diff --git a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp index 7d657c4548dd1..c54459d050473 100644 --- a/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp +++ b/src/Processors/Formats/Impl/Parquet/PrepareForWrite.cpp @@ -419,9 +419,11 @@ void preparePrimitiveColumn(ColumnPtr column, DataTypePtr type, const std::strin } parq::TimestampType tt; - /// (Shouldn't we check the DateTime64's timezone parameter here? No, the actual number - /// in DateTime64 column is always in UTC, regardless of the timezone parameter.) - tt.__set_isAdjustedToUTC(true); + /// ClickHouse ticks are always UTC. By default keep isAdjustedToUTC=true. Iceberg + /// sets output_timestamp_adjusted_to_utc_from_timezone so timestamp(_ns) is + /// TIMESTAMP(*, false) and timestamptz(_ns) is TIMESTAMP(*, true). + tt.__set_isAdjustedToUTC( + options.output_timestamp_adjusted_to_utc_from_timezone ? dt.hasExplicitTimeZone() : true); tt.__set_unit(unit); parq::LogicalType t; t.__set_TIMESTAMP(tt); diff --git a/src/Processors/Formats/Impl/Parquet/Write.h b/src/Processors/Formats/Impl/Parquet/Write.h index e7d657819bb20..da7140200729c 100644 --- a/src/Processors/Formats/Impl/Parquet/Write.h +++ b/src/Processors/Formats/Impl/Parquet/Write.h @@ -25,6 +25,7 @@ struct WriteOptions bool output_datetime_as_uint32 = false; bool output_date_as_uint16 = false; bool output_enum_as_byte_array = false; + bool output_timestamp_adjusted_to_utc_from_timezone = false; /// Note: the meaning of some compression methods here is different from /// wrapReadBufferWithCompressionMethod: diff --git a/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp b/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp index 161da8da9aa42..678b38c67ce52 100644 --- a/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp @@ -49,6 +49,7 @@ ParquetBlockOutputFormat::ParquetBlockOutputFormat(WriteBuffer & out_, SharedHea options.output_datetime_as_uint32 = format_settings.parquet.output_datetime_as_uint32; options.output_date_as_uint16 = format_settings.parquet.output_date_as_uint16; options.output_enum_as_byte_array = format_settings.parquet.output_enum_as_byte_array; + options.output_timestamp_adjusted_to_utc_from_timezone = format_settings.parquet.output_timestamp_adjusted_to_utc_from_timezone; options.data_page_size = format_settings.parquet.data_page_size; options.write_batch_size = format_settings.parquet.write_batch_size; options.write_page_index = format_settings.parquet.write_page_index; diff --git a/src/Processors/tests/gtest_write_parquet_page_index.cpp b/src/Processors/tests/gtest_write_parquet_page_index.cpp index 6fe0841fb8582..18d81d8319372 100644 --- a/src/Processors/tests/gtest_write_parquet_page_index.cpp +++ b/src/Processors/tests/gtest_write_parquet_page_index.cpp @@ -8,6 +8,8 @@ # include # include # include +# include +# include # include # include # include @@ -368,5 +370,26 @@ TEST(Parquet, WriteParquetPageIndexOversizedStringStats) ASSERT_GT(offset_index_location.value().offset, 0); ASSERT_GT(offset_index_location.value().length, 0); } + +TEST(ParquetWrite, DateTime64AdjustedToUTCFromTimezone) +{ + auto ts = std::make_shared(9); + auto ts_tz = std::make_shared(9, "UTC"); + Block sample; + sample.insert({ts->createColumn(), ts, "ts"}); + sample.insert({ts_tz->createColumn(), ts_tz, "ts_tz"}); + + Parquet::WriteOptions default_options; + auto default_schema = Parquet::convertSchema(sample, default_options, std::nullopt); + ASSERT_EQ(default_schema.size(), 3); + EXPECT_TRUE(default_schema[1].logicalType.TIMESTAMP.isAdjustedToUTC); + EXPECT_TRUE(default_schema[2].logicalType.TIMESTAMP.isAdjustedToUTC); + + Parquet::WriteOptions iceberg_options; + iceberg_options.output_timestamp_adjusted_to_utc_from_timezone = true; + auto iceberg_schema = Parquet::convertSchema(sample, iceberg_options, std::nullopt); + EXPECT_FALSE(iceberg_schema[1].logicalType.TIMESTAMP.isAdjustedToUTC); + EXPECT_TRUE(iceberg_schema[2].logicalType.TIMESTAMP.isAdjustedToUTC); +} } #endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp index 6d2aede15108f..e9fec155344e9 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp @@ -371,8 +371,12 @@ static void writeDataFiles( FormatFilterInfoPtr output_format_filter_info = std::make_shared(nullptr, context, column_mapper, nullptr, nullptr); + auto write_format_settings = format_settings; + if (!write_format_settings) + write_format_settings = getFormatSettings(context); + write_format_settings->parquet.output_timestamp_adjusted_to_utc_from_timezone = true; auto output_format = FormatFactory::instance().getOutputFormat( - write_format, *write_buffer, *sample_block, context, format_settings, output_format_filter_info); + write_format, *write_buffer, *sample_block, context, write_format_settings, output_format_filter_info); while (true) { diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MultipleFileWriter.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MultipleFileWriter.cpp index 3414bf04e4c5f..d0fadd5b48f07 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MultipleFileWriter.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MultipleFileWriter.cpp @@ -61,12 +61,12 @@ void MultipleFileWriter::startNewFile() buffer = object_storage->writeObject( StoredObject(storage_path), WriteMode::Rewrite, std::nullopt, DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings()); - if (format_settings) - { - format_settings->parquet.write_page_index = true; - format_settings->parquet.bloom_filter_push_down = true; - format_settings->parquet.filter_push_down = true; - } + if (!format_settings) + format_settings = getFormatSettings(context); + format_settings->parquet.write_page_index = true; + format_settings->parquet.bloom_filter_push_down = true; + format_settings->parquet.filter_push_down = true; + format_settings->parquet.output_timestamp_adjusted_to_utc_from_timezone = true; FormatFilterInfoPtr format_filter_info = std::make_shared(nullptr, context, column_mapper, nullptr, nullptr); output_format = FormatFactory::instance().getOutputFormatParallelIfPossible( write_format, *buffer, *sample_block, context, format_settings, format_filter_info); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp index 2933f50f92963..470162dda0266 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp @@ -217,6 +217,10 @@ static std::optional writeDataFiles( Poco::JSON::Object::Ptr data_schema) { chassert(commands.size() == 1); + auto iceberg_format_settings = format_settings; + if (!iceberg_format_settings) + iceberg_format_settings = getFormatSettings(context); + iceberg_format_settings->parquet.output_timestamp_adjusted_to_utc_from_timezone = true; DataFileWriteResultByPartitionKey delete_data_result; DataFileStatisticsByPartitionKey delete_data_statistics; std::unordered_map, ChunkPartitioner::PartitionKeyHasher> delete_data_write_buffers; @@ -284,7 +288,7 @@ static std::optional writeDataFiles( column_mapper->setStorageColumnEncoding(std::move(field_ids)); FormatFilterInfoPtr format_filter_info = std::make_shared(nullptr, context, column_mapper, nullptr, nullptr); auto output_format = FormatFactory::instance().getOutputFormat( - write_format, *write_buffer, delete_file_sample_block, context, format_settings, format_filter_info); + write_format, *write_buffer, delete_file_sample_block, context, iceberg_format_settings, format_filter_info); delete_data_write_buffers[partition_key] = std::move(write_buffer); delete_data_writers[partition_key] = std::move(output_format); @@ -385,7 +389,7 @@ static std::optional writeDataFiles( ColumnMapperPtr data_column_mapper = createColumnMapper(data_schema); FormatFilterInfoPtr data_format_filter_info = std::make_shared(nullptr, context, data_column_mapper, nullptr, nullptr); auto data_output_format = FormatFactory::instance().getOutputFormat( - write_format, *data_write_buffer, data_block, context, format_settings, data_format_filter_info); + write_format, *data_write_buffer, data_block, context, iceberg_format_settings, data_format_filter_info); update_data_write_buffers[partition_key] = std::move(data_write_buffer); it = update_data_writers.emplace(partition_key, std::move(data_output_format)).first; From 8ddc7a2d2e7f81aaa2a69c9ef5517d9c2d4d9c4c Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 8 Oct 2026 10:22:29 +0200 Subject: [PATCH 09/15] Keep Iceberg `timestamp-nanos` in identity-partition Avro schemas contrib Avro `toJson` drops unknown logical types, so DataFileWriter was writing a bare `long`. Persist the original JSON in `avro.schema` instead. Co-authored-by: Cursor --- src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index e239da4235593..6916b5f3d210c 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -725,6 +725,8 @@ void generateManifestFile( auto adapter = std::make_unique(buf); avro::DataFileWriter writer(std::move(adapter), schema); + /// Avro C++ `toJson` drops unknown logical types such as Iceberg `timestamp-nanos`. + writer.setMetadata("avro.schema", schema_representation); writer.setMetadata(Iceberg::f_schema, json_representation); writer.setMetadata(Iceberg::f_format_version, std::to_string(version)); From f668b9920539d5a6d6fb085f1ac0009048bbc9d6 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 12 Oct 2026 18:09:09 +0200 Subject: [PATCH 10/15] Pass `format_version` to Iceberg `unknown` type unit tests `getIcebergType` now requires `format_version`; the two `Nothing` tests still used two arguments and failed debug builds. Co-authored-by: Cursor --- .../Iceberg/tests/gtest_iceberg_metadata_generator.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp index ab18ed85fe376..6da2044a15565 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp @@ -629,7 +629,7 @@ TEST(IcebergMetadataGenerator, ModifyColumnWideningRecordsTheNewTypeInANewSchema TEST(IcebergMetadataGenerator, GetIcebergTypeNothingProducesUnknown) { Int32 iter = 0; - auto [iceberg_type, required] = Iceberg::getIcebergType(std::make_shared(), iter); + auto [iceberg_type, required] = Iceberg::getIcebergType(std::make_shared(), iter, /* format_version */ 3); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "unknown"); EXPECT_FALSE(required); @@ -639,7 +639,7 @@ TEST(IcebergMetadataGenerator, GetIcebergTypeNothingProducesUnknown) TEST(IcebergMetadataGenerator, GetIcebergTypeNullableNothingProducesUnknown) { Int32 iter = 0; - auto [iceberg_type, required] = Iceberg::getIcebergType(makeNullable(std::make_shared()), iter); + auto [iceberg_type, required] = Iceberg::getIcebergType(makeNullable(std::make_shared()), iter, /* format_version */ 3); ASSERT_TRUE(iceberg_type.isString()); EXPECT_EQ(iceberg_type.extract(), "unknown"); EXPECT_FALSE(required); From 0bab7d0fa74d7176c608eb07ff8f92f6198fa24b Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 12 Oct 2026 18:17:18 +0200 Subject: [PATCH 11/15] Fail open Iceberg min/max pruning when decoded bounds invert Mixed-scale `timestamp` stats can decode a near-epoch ns min as microseconds and a later ns max into a smaller value. An inverted range pruned the whole file; skip that column's hyperrectangle instead. Co-authored-by: Cursor --- src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp | 2 +- .../ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp index e553907b4a041..7052abbf76d80 100644 --- a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp @@ -156,7 +156,7 @@ DataFileMetaInfo::DataFileMetaInfo( { auto left = Iceberg::deserializeFieldFromBinaryRepr(left_str, type, true); auto right = Iceberg::deserializeFieldFromBinaryRepr(right_str, type, false); - if (left && right) + if (left && right && !DB::Range::less(*right, *left)) hyperrectangle = DB::Range(*left, true, *right, true); } } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp index 705f38c561c57..14fd6b1fee670 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp @@ -476,6 +476,9 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index) auto right = deserializeFieldFromBinaryRepr(right_str, name_and_type.type, false); if (!left || !right) continue; + /// Mixed-scale timestamp stats can decode to lower > upper; skip min/max for this column. + if (DB::Range::less(*right, *left)) + continue; hyperrectangles.emplace(column_id, DB::Range(*left, true, *right, true)); } From edcd99ee3d165e8ca1e2b19664c704f5279e1c11 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 13 Oct 2026 12:48:07 +0200 Subject: [PATCH 12/15] Accept Int64 Iceberg `DateTime64(9)` partition values on manifest rewrite contrib Avro has no `timestamp-nanos`, so compaction reads identity partition keys as `Int64`. `getDateTime64ValueForIcebergBounds` required `Decimal64` and threw `BAD_GET` before the snapshot committed. Co-authored-by: Cursor --- .../ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index 7aae481e12df8..6fb1d6ee12018 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -219,6 +219,12 @@ Int64 getDateTime64ValueForIcebergBounds(const Field & field, DataTypePtr type) if (!WhichDataType(type).isDateTime64()) throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected DateTime64, got {}", type->getName()); + /// Manifest rewrite reads the Avro `long` back as Int64 (contrib Avro has no `timestamp-nanos`). + if (field.getType() == Field::Types::Int64) + return field.safeGet(); + if (field.getType() == Field::Types::UInt64) + return static_cast(field.safeGet()); + const auto scale = getDecimalScale(*type); const auto value = field.safeGet().getValue().value; if (scale == 9) From c62589d4ec3c9a46cb85a34ef75cbbadee680a49 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 13 Oct 2026 14:11:02 +0200 Subject: [PATCH 13/15] Fail open Iceberg `timestamp` min/max for near-epoch nanosecond bounds `|ticks| <= 1e16` was kept as microseconds, so a 1-day-after-epoch ns min/max decoded as ~1972 and over-pruned. Skip that column's hyperrectangle below 2024-01-01 UTC so 2024+ pruning stays intact. Co-authored-by: Cursor --- .../Iceberg/IcebergFieldParseHelpers.cpp | 23 ++++-- .../gtest_iceberg_field_parse_helpers.cpp | 72 +++++++++++++++++++ .../helpers/iceberg_timestamp_ns.py | 34 +++++++++ .../test_pruning_datetime64_nanoseconds.py | 31 ++++++-- 4 files changed, 152 insertions(+), 8 deletions(-) create mode 100644 src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index c41bda4e02232..b0259668c5a62 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -33,6 +33,13 @@ namespace constexpr Int64 microseconds_ambiguous_threshold = 10'000'000'000'000'000LL; constexpr Int64 nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL; constexpr Int64 nanoseconds_per_microsecond = 1000; +/// 2024-01-01 00:00:00 UTC as microseconds. Below this, `|ticks|` is also a +/// plausible near-epoch nanosecond count (1.704e15 ns is ~19.7 days). Fail-open +/// that overlap so a 1-day-after-epoch ns min/max is not read as year ~1972. +/// Ticks from this instant through 1e16 stay microseconds so 2024+ Iceberg +/// `timestamp` pruning is unchanged. Residual: 19.7..115 days as ns still collide +/// with 2024..2286 as µs. +constexpr Int64 min_kept_microseconds = 1'704'067'200'000'000LL; bool magnitudeExceeds(Int64 value, Int64 threshold) { @@ -63,7 +70,9 @@ Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound) /// Some writers store nanosecond stats on Iceberg `timestamp` (`DateTime64(6)`). /// Convert only in a true nanosecond band (`|ticks| > 1e18`). Between 1e16 and /// 1e18 the value may be far-future microseconds (do not convert; fail open so -/// min/max pruning is skipped). `|ticks| <= 1e16` is kept as microseconds. +/// min/max pruning is skipped). Below `min_kept_microseconds`, ticks may be +/// near-epoch nanoseconds (also fail open). `(min_kept_microseconds, 1e16]` is +/// kept as microseconds. std::optional deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound) { if (str.size() != sizeof(Int64)) @@ -72,12 +81,18 @@ std::optional deserializeDateTime64FromBinaryRepr(const std::string & str Int64 unscaled = unalignedLoadLittleEndian(str.data()); const UInt32 scale = getDecimalScale(type); - if (scale == 6 && magnitudeExceeds(unscaled, microseconds_ambiguous_threshold)) + if (scale == 6) { - if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold)) + if (!magnitudeExceeds(unscaled, min_kept_microseconds)) return std::nullopt; - unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); + if (magnitudeExceeds(unscaled, microseconds_ambiguous_threshold)) + { + if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold)) + return std::nullopt; + + unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); + } } return DecimalField(DateTime64(unscaled), scale); 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 new file mode 100644 index 0000000000000..7d6e0851c024c --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp @@ -0,0 +1,72 @@ +#include "config.h" + +#if USE_AVRO + +#include + +#include +#include +#include +#include + +using namespace DB; +using namespace DB::Iceberg; + +namespace +{ + +std::string littleEndianInt64(Int64 value) +{ + std::string bytes(sizeof(Int64), '\0'); + unalignedStoreLittleEndian(bytes.data(), value); + return bytes; +} + +Int64 decimal64Ticks(const Field & field) +{ + return static_cast(field.safeGet>().getValue()); +} + +} + +TEST(IcebergFieldParseHelpers, NearEpochNanosecondDateTime64BoundsFailOpen) +{ + auto type = std::make_shared(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, ModernMicrosecondDateTime64BoundsAreKept) +{ + auto type = std::make_shared(6); + /// 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); +} + +TEST(IcebergFieldParseHelpers, ProvenNanosecondDateTime64BoundsAreConverted) +{ + auto type = std::make_shared(6); + 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(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()); +} + +#endif diff --git a/tests/integration/helpers/iceberg_timestamp_ns.py b/tests/integration/helpers/iceberg_timestamp_ns.py index 351b3cc9eeb1c..5e91221fc5955 100644 --- a/tests/integration/helpers/iceberg_timestamp_ns.py +++ b/tests/integration/helpers/iceberg_timestamp_ns.py @@ -223,6 +223,40 @@ 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"), +] + + +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_far_future_us_upper_bound(node, table, column, pruned_files): """Iceberg `timestamp` with a far-future microsecond *upper* bound must not over-prune. 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 e30843636eec2..2c470a2eda21c 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,11 +24,13 @@ DATETIME64_MAX_US, FILTER_LITERAL_IDS, FILTER_LITERALS, + NEAR_EPOCH_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_ns_pruning, check_ns_pruning_datetime64_microseconds, @@ -76,7 +78,7 @@ def _arrow_timestamp_row(ts_ns, row_id): ) -def _create_timestamp_table(started_cluster, table_name, patch): +def _create_timestamp_table(started_cluster, table_name, patch, rows=ROWS): """Iceberg v2 `timestamp` (microseconds parquet) with a custom manifest-bound patch.""" work = Path(tempfile.mkdtemp(prefix="iceberg_ns_bounds_")) try: @@ -96,7 +98,7 @@ def _create_timestamp_table(started_cluster, table_name, patch): ) table.append( pa.concat_tables( - [_arrow_timestamp_row(ts_ns, row_id) for ts_ns, row_id, _utc in ROWS] + [_arrow_timestamp_row(ts_ns, row_id) for ts_ns, row_id, _utc in rows] ) ) table = catalog.load_table(identifier) @@ -119,10 +121,10 @@ def _create_timestamp_table(started_cluster, table_name, patch): shutil.rmtree(work, ignore_errors=True) -def _create_timestamp_table_with_ns_bounds(started_cluster, table_name): +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 + started_cluster, table_name, patch_iceberg_timestamp_bounds_us_to_ns, rows=rows ) @@ -265,6 +267,27 @@ def pruned_files(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) + + @pytest.mark.parametrize( "upper_bound_us", [SPARK_TIMESTAMP_SENTINEL_US, DATETIME64_MAX_US], From 3fcb41a5b761ab36753bf5a11691e525ae5360f0 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 13 Oct 2026 14:30:00 +0200 Subject: [PATCH 14/15] Keep Iceberg `timestamp` min/max at the 2024-01-01 microsecond threshold `magnitudeExceeds` is strict, so `|ticks| == 2024-01-01 UTC` fail-opened while the comment said that instant is kept. Fail-open only below the threshold. Co-authored-by: Cursor --- .../DataLakes/Iceberg/IcebergFieldParseHelpers.cpp | 9 +++++++-- .../tests/gtest_iceberg_field_parse_helpers.cpp | 10 ++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index b0259668c5a62..2052910927d12 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -46,6 +46,11 @@ bool magnitudeExceeds(Int64 value, Int64 threshold) return value > threshold || value < -threshold; } +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) { @@ -71,7 +76,7 @@ Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound) /// Convert only in a true nanosecond band (`|ticks| > 1e18`). Between 1e16 and /// 1e18 the value may be far-future microseconds (do not convert; fail open so /// min/max pruning is skipped). Below `min_kept_microseconds`, ticks may be -/// near-epoch nanoseconds (also fail open). `(min_kept_microseconds, 1e16]` is +/// near-epoch nanoseconds (also fail open). `[min_kept_microseconds, 1e16]` is /// kept as microseconds. std::optional deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound) { @@ -83,7 +88,7 @@ std::optional deserializeDateTime64FromBinaryRepr(const std::string & str if (scale == 6) { - if (!magnitudeExceeds(unscaled, min_kept_microseconds)) + if (magnitudeBelow(unscaled, min_kept_microseconds)) return std::nullopt; if (magnitudeExceeds(unscaled, microseconds_ambiguous_threshold)) 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 7d6e0851c024c..2687fd5687027 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 @@ -42,6 +42,16 @@ TEST(IcebergFieldParseHelpers, NearEpochNanosecondDateTime64BoundsFailOpen) TEST(IcebergFieldParseHelpers, ModernMicrosecondDateTime64BoundsAreKept) { auto type = std::make_shared(6); + /// 2024-01-01 00:00:00 UTC is the first kept microsecond bound (`min_kept_microseconds`). + constexpr Int64 us_2024_01_01 = 1'704'067'200'000'000LL; + auto at_threshold = deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_01_01), type, true); + ASSERT_TRUE(at_threshold.has_value()); + EXPECT_EQ(decimal64Ticks(*at_threshold), us_2024_01_01); + auto negative_threshold = deserializeFieldFromBinaryRepr(littleEndianInt64(-us_2024_01_01), type, true); + ASSERT_TRUE(negative_threshold.has_value()); + EXPECT_EQ(decimal64Ticks(*negative_threshold), -us_2024_01_01); + EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_01_01 - 1), 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); From f884dae7291e6b8de17053ffe7fa9c7c63396af5 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 14 Oct 2026 11:18:30 +0200 Subject: [PATCH 15/15] =?UTF-8?q?Fail=20open=20Iceberg=20`timestamp`=20min?= =?UTF-8?q?/max=20in=20the=20remaining=20ns/=C2=B5s=20overlap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `[1.704e15, 1e16]` is both ~19.7–115 days as nanoseconds and 2024–2286 as microseconds. Convert only `|ticks| > 1e18`; skip min/max pruning otherwise because no integer test splits those scales. Co-authored-by: Cursor --- .../Iceberg/IcebergFieldParseHelpers.cpp | 34 +++---------------- .../gtest_iceberg_field_parse_helpers.cpp | 29 +++++++--------- .../helpers/iceberg_timestamp_ns.py | 33 ++++++++++++++++++ .../test_pruning_datetime64_nanoseconds.py | 23 +++++++++++++ 4 files changed, 73 insertions(+), 46 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index 2052910927d12..b8bb69a04a6ef 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -26,31 +26,15 @@ namespace Iceberg namespace { -/// 1e16 µs is ~year 2286. That is not an Iceberg `timestamp` max: Spark sentinels -/// such as `9999-12-31` (`~2.53e17` µs) and ClickHouse `DateTime64` max -/// `2299-12-31` (`~1.04e16` µs) are spec-correct microseconds above this. /// 1e18 ns is ~year 2001; customer 2026 ns bounds are `~1.76e18`. -constexpr Int64 microseconds_ambiguous_threshold = 10'000'000'000'000'000LL; constexpr Int64 nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL; constexpr Int64 nanoseconds_per_microsecond = 1000; -/// 2024-01-01 00:00:00 UTC as microseconds. Below this, `|ticks|` is also a -/// plausible near-epoch nanosecond count (1.704e15 ns is ~19.7 days). Fail-open -/// that overlap so a 1-day-after-epoch ns min/max is not read as year ~1972. -/// Ticks from this instant through 1e16 stay microseconds so 2024+ Iceberg -/// `timestamp` pruning is unchanged. Residual: 19.7..115 days as ns still collide -/// with 2024..2286 as µs. -constexpr Int64 min_kept_microseconds = 1'704'067'200'000'000LL; bool magnitudeExceeds(Int64 value, Int64 threshold) { return value > threshold || value < -threshold; } -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) { @@ -73,11 +57,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 in a true nanosecond band (`|ticks| > 1e18`). Between 1e16 and -/// 1e18 the value may be far-future microseconds (do not convert; fail open so -/// min/max pruning is skipped). Below `min_kept_microseconds`, ticks may be -/// near-epoch nanoseconds (also fail open). `[min_kept_microseconds, 1e16]` is -/// kept as microseconds. +/// 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. std::optional deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound) { if (str.size() != sizeof(Int64)) @@ -88,16 +70,10 @@ std::optional deserializeDateTime64FromBinaryRepr(const std::string & str if (scale == 6) { - if (magnitudeBelow(unscaled, min_kept_microseconds)) + if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold)) return std::nullopt; - if (magnitudeExceeds(unscaled, microseconds_ambiguous_threshold)) - { - if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold)) - return std::nullopt; - - unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); - } + unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); } return DecimalField(DateTime64(unscaled), scale); 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 2687fd5687027..069a01bed01c1 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 @@ -39,27 +39,22 @@ TEST(IcebergFieldParseHelpers, NearEpochNanosecondDateTime64BoundsFailOpen) EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(-one_day_ns), type, true).has_value()); } -TEST(IcebergFieldParseHelpers, ModernMicrosecondDateTime64BoundsAreKept) +TEST(IcebergFieldParseHelpers, AmbiguousScaleDateTime64BoundsFailOpen) { auto type = std::make_shared(6); - /// 2024-01-01 00:00:00 UTC is the first kept microsecond bound (`min_kept_microseconds`). + /// 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; - auto at_threshold = deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_01_01), type, true); - ASSERT_TRUE(at_threshold.has_value()); - EXPECT_EQ(decimal64Ticks(*at_threshold), us_2024_01_01); - auto negative_threshold = deserializeFieldFromBinaryRepr(littleEndianInt64(-us_2024_01_01), type, true); - ASSERT_TRUE(negative_threshold.has_value()); - EXPECT_EQ(decimal64Ticks(*negative_threshold), -us_2024_01_01); - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(us_2024_01_01 - 1), type, true).has_value()); + 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()); + + /// Band upper edge: 1e16 ns is ~1970-04-26, 1e16 µs is ~year 2286. + constexpr Int64 one_e16 = 10'000'000'000'000'000LL; + EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(one_e16), type, true).has_value()); } TEST(IcebergFieldParseHelpers, ProvenNanosecondDateTime64BoundsAreConverted) diff --git a/tests/integration/helpers/iceberg_timestamp_ns.py b/tests/integration/helpers/iceberg_timestamp_ns.py index 5e91221fc5955..3922f8d37d891 100644 --- a/tests/integration/helpers/iceberg_timestamp_ns.py +++ b/tests/integration/helpers/iceberg_timestamp_ns.py @@ -229,6 +229,12 @@ def _utc_datetime_to_micros(dt): (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. @@ -257,6 +263,33 @@ def check_minmax_near_epoch_ns_bounds_on_timestamp_schema(node, table, column, p 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. 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 2c470a2eda21c..5a78ef2a69005 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 @@ -25,6 +25,7 @@ FILTER_LITERAL_IDS, FILTER_LITERALS, NEAR_EPOCH_NS_ROWS, + OVERLAP_BAND_NS_ROWS, PRUNE_OFF, PRUNE_ON, ROWS, @@ -32,6 +33,7 @@ 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, @@ -288,6 +290,27 @@ def pruned_files(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],