diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp index e062057a99c6a..344de9c04ac56 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergFieldParseHelpers.cpp @@ -28,59 +28,16 @@ namespace Iceberg namespace { -/// 1e16 µs is ~year 2286. Spark sentinels such as `9999-12-31` (`~2.53e17` µs) -/// are spec-correct microseconds above this. -constexpr Int64 microseconds_ambiguous_threshold = 10'000'000'000'000'000LL; -/// 1e18 ns is ~year 2001; customer 2026 ns bounds are `~1.76e18`. -constexpr Int64 nanoseconds_lower_threshold = 1'000'000'000'000'000'000LL; -constexpr Int64 nanoseconds_per_microsecond = 1000; - -bool magnitudeExceeds(Int64 value, Int64 threshold) -{ - return value > threshold || value < -threshold; -} - -/// Convert ns ticks to us ticks without shrinking a [lower, upper] range. -Int64 nanosecondsToMicrosecondsForBound(Int64 nanoseconds, bool lower_bound) -{ - Int64 microseconds = nanoseconds / nanoseconds_per_microsecond; - if (nanoseconds % nanoseconds_per_microsecond == 0) - return microseconds; - - /// C++ integer division truncates toward zero. - if (lower_bound) - { - if (nanoseconds < 0) - --microseconds; - } - else if (nanoseconds > 0) - { - ++microseconds; - } - return microseconds; -} - -/// Iceberg timestamps are 8-byte little-endian (spec Appendix D). -/// Some writers store nanosecond stats on Iceberg `timestamp` (`DateTime64(6)`). -/// `|ticks| <= 1e16` is kept as microseconds, as the spec requires. `|ticks| > 1e18` -/// is converted from nanoseconds. `(1e16, 1e18]` fails open: it collides Spark -/// `9999-12-31` sentinels with 1970–2001 nanoseconds. -std::optional deserializeDateTime64FromBinaryRepr(const std::string & str, const IDataType & type, bool lower_bound) +/// Iceberg timestamps are 8-byte little-endian (spec Appendix D). `DateTime64(6)` (`timestamp` / +/// `timestamptz`) bounds are microseconds and `DateTime64(9)` (`timestamp_ns` / `timestamptz_ns`) +/// bounds are nanoseconds. Both are returned as stored. +std::optional 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 Int64 unscaled = unalignedLoadLittleEndian(str.data()); const UInt32 scale = getDecimalScale(type); - - if (scale == 6 && magnitudeExceeds(unscaled, microseconds_ambiguous_threshold)) - { - if (!magnitudeExceeds(unscaled, nanoseconds_lower_threshold)) - return std::nullopt; - - unscaled = nanosecondsToMicrosecondsForBound(unscaled, lower_bound); - } - return DecimalField(DateTime64(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 5286d5217c3fa..ec96121c11fe7 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_field_parse_helpers.cpp @@ -47,28 +47,20 @@ TEST(IcebergFieldParseHelpers, MicrosecondDateTime64BoundsAreKept) ASSERT_TRUE(negative.has_value()); EXPECT_EQ(decimal64Ticks(*negative), us_1960_01_01); - /// Upper edge of the microsecond band (~year 2286). - constexpr Int64 one_e16 = 10'000'000'000'000'000LL; - auto edge = deserializeFieldFromBinaryRepr(littleEndianInt64(one_e16), type, false); - ASSERT_TRUE(edge.has_value()); - EXPECT_EQ(decimal64Ticks(*edge), one_e16); + /// Spark `9999-12-31` sentinel is spec-correct microseconds. + constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL; + auto sentinel = deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false); + ASSERT_TRUE(sentinel.has_value()); + EXPECT_EQ(decimal64Ticks(*sentinel), spark_9999_us); } -TEST(IcebergFieldParseHelpers, ProvenNanosecondDateTime64BoundsAreConverted) +TEST(IcebergFieldParseHelpers, NanosecondDateTime64BoundsAreKept) { - auto type = std::make_shared(6); + auto type = std::make_shared(9); constexpr Int64 ns_2024_01_01 = 1'704'067'200'123'456'789LL; auto field = deserializeFieldFromBinaryRepr(littleEndianInt64(ns_2024_01_01), type, true); ASSERT_TRUE(field.has_value()); - EXPECT_EQ(decimal64Ticks(*field), ns_2024_01_01 / 1000); -} - -TEST(IcebergFieldParseHelpers, FarFutureMicrosecondDateTime64BoundsFailOpen) -{ - auto type = std::make_shared(6); - /// Spark 9999-12-31 sentinel sits in (1e16, 1e18]. - constexpr Int64 spark_9999_us = 253'402'300'799'999'999LL; - EXPECT_FALSE(deserializeFieldFromBinaryRepr(littleEndianInt64(spark_9999_us), type, false).has_value()); + EXPECT_EQ(decimal64Ticks(*field), ns_2024_01_01); } #endif diff --git a/tests/integration/helpers/iceberg_timestamp_ns.py b/tests/integration/helpers/iceberg_timestamp_ns.py index 351b3cc9eeb1c..1c3a942c55c56 100644 --- a/tests/integration/helpers/iceberg_timestamp_ns.py +++ b/tests/integration/helpers/iceberg_timestamp_ns.py @@ -164,57 +164,13 @@ def check_ns_pruning_datetime64_microseconds(node, table, column, pruned_files, ) == "3\n4\n" -def check_minmax_ns_bounds_on_timestamp_schema(node, table, column, pruned_files): - """Customer min/max over-prune: Iceberg `timestamp` with nanosecond min/max bytes. - - OTEL `time_unix_nano` tables are often declared as Iceberg `timestamp` - (`DateTime64(6)`) while 8-byte lower/upper bounds stay in nanoseconds. - ClickHouse then deserializes those ns bytes as microseconds (year 2299), - so `WHERE col <= toDateTime64('2026-...', 6)` prunes every file while the - parquet scan (microseconds) with `use_iceberg_partition_pruning = 0` is correct. - """ - type_name = node.query( - f"SELECT toTypeName({column}) FROM {table} LIMIT 1", - settings=PRUNE_OFF, - ) - assert "DateTime64(6" in type_name, type_name - assert "DateTime64(9" not in type_name, type_name - - ids = node.query(f"SELECT id FROM {table} ORDER BY id", settings=PRUNE_OFF) - assert ids == "1\n2\n3\n4\n" - # Parquet is timestamp[us] (correct 2024 wall times). Iceberg lower/upper bounds - # are still the original nanosecond int64 bytes, so DateTime64(6) min/max - # interprets them as year 2299 and over-prunes. - micros = node.query( - f"SELECT toUnixTimestamp64Micro({column}) FROM {table} WHERE id = 1", - settings=PRUNE_OFF, - ).strip() - assert micros == "1704067200123456", micros - - # Same predicate shape as the customer query: scale-6 literal near a file's - # nanosecond upper bound. Rows 1 and 2 must survive; pruning must not drop them. - le_after_row2 = datetime64_literal( - "2024-03-15 12:00:00.000001000", scale=6, timezone=None - ) - query = f"SELECT id FROM {table} WHERE {column} <= {le_after_row2} ORDER BY id" - # One data file covering all four timestamps; the predicate overlaps it, so - # min/max pruning must keep the file. Over-pruning currently returns 0 rows. - assert pruned_files(query) == 0 - assert node.query(query, settings=PRUNE_ON) == "1\n2\n" - - count_query = f"SELECT count() FROM {table} WHERE {column} <= {le_after_row2}" - assert pruned_files(count_query) == 0 - assert node.query(count_query, settings=PRUNE_ON) == "2\n" - - def _utc_datetime_to_micros(dt): """Integer microseconds from Unix epoch. Do not use float `timestamp` * 1e6.""" delta = dt - datetime(1970, 1, 1, tzinfo=timezone.utc) return delta.days * 86_400_000_000 + delta.seconds * 1_000_000 + delta.microseconds -# Spec-correct Iceberg `timestamp` microseconds that sit in the ambiguous band -# (1e16, 1e18]: converting them as nanoseconds over-prunes. +# Spec-correct Iceberg `timestamp` microseconds, including the Spark `9999-12-31` sentinel. SPARK_TIMESTAMP_SENTINEL_US = _utc_datetime_to_micros( datetime(9999, 12, 31, 23, 59, 59, 999999, tzinfo=timezone.utc) ) @@ -224,11 +180,11 @@ def _utc_datetime_to_micros(dt): def check_minmax_far_future_us_upper_bound(node, table, column, pruned_files): - """Iceberg `timestamp` with a far-future microsecond *upper* bound must not over-prune. + """Iceberg `timestamp` with a far-future microsecond upper bound must keep matching rows. - Lower bound stays a real 2024 microsecond value. If the decoder treats the - upper bound as nanoseconds, max becomes ~1978, the [min, max] range inverts, - and `WHERE col <= toDateTime64('2024-03-15 ...', 6)` drops the file. + Lower bound stays a real 2024 microsecond value. The upper bound is the Spark + `9999-12-31` sentinel or the `DateTime64` maximum. A predicate on 2024 must + still return the matching rows. """ type_name = node.query( f"SELECT toTypeName({column}) FROM {table} LIMIT 1", @@ -253,117 +209,6 @@ def check_minmax_far_future_us_upper_bound(node, table, column, pruned_files): assert node.query(query, settings=PRUNE_ON) == "1\n2\n" -def _scale_bound_bytes(raw, factor=1000): - if not isinstance(raw, (bytes, bytearray)) or len(raw) != 8: - return raw - value = struct.unpack(" 0, f"no timestamp min/max bounds scaled under {table_location}" - - def _set_bound_bytes(raw, value): if not isinstance(raw, (bytes, bytearray)) or len(raw) != 8: return raw diff --git a/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py index b6e62d970a4f5..c7a9beb9a6b5b 100644 --- a/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py +++ b/tests/integration/test_storage_iceberg_no_spark/test_pruning_datetime64_nanoseconds.py @@ -29,11 +29,9 @@ ROWS, SPARK_TIMESTAMP_SENTINEL_US, check_minmax_far_future_us_upper_bound, - check_minmax_ns_bounds_on_timestamp_schema, check_ns_pruning, check_ns_pruning_datetime64_microseconds, patch_iceberg_table_long_to_timestamp_ns, - patch_iceberg_timestamp_bounds_us_to_ns, patch_iceberg_timestamp_upper_bounds, ) from helpers.iceberg_utils import check_validity_and_get_prunned_files_general @@ -119,13 +117,6 @@ def _create_timestamp_table(started_cluster, table_name, patch, rows=ROWS): shutil.rmtree(work, ignore_errors=True) -def _create_timestamp_table_with_ns_bounds(started_cluster, table_name, rows=ROWS): - """Iceberg v2 `timestamp` (microseconds) whose manifest min/max bytes are nanoseconds.""" - return _create_timestamp_table( - started_cluster, table_name, patch_iceberg_timestamp_bounds_us_to_ns, rows=rows - ) - - def _create_timestamp_table_with_upper_bound_us(started_cluster, table_name, micros): """Iceberg v2 `timestamp` whose upper bounds are spec-correct far-future microseconds.""" def patch(table_location, old_prefix, new_prefix): @@ -240,31 +231,6 @@ def pruned_files(select_expression): check_ns_pruning_datetime64_microseconds(node, table, "value", pruned_files, timezone=timezone) -def test_minmax_pruning_ns_bounds_on_timestamp_schema(started_cluster_iceberg_no_spark): - """Iceberg `timestamp` (`DateTime64(6)`) with nanosecond min/max bytes must not over-prune. - - Customer: `WHERE time_unix_nano <= toDateTime64('2026-08-11 20:02:07.897356', 6)` - returns 0 with min/max pruning and the correct row count without it. - """ - node = started_cluster_iceberg_no_spark.instances["node1"] - s3_filename = _create_timestamp_table_with_ns_bounds( - started_cluster_iceberg_no_spark, "minmax_ns_on_ts" - ) - table = _iceberg_table(s3_filename) - - def pruned_files(select_expression): - return check_validity_and_get_prunned_files_general( - node, - "minmax_ns_on_ts", - PRUNE_OFF, - PRUNE_ON, - "IcebergMinMaxIndexPrunedFiles", - select_expression, - ) - - check_minmax_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files) - - @pytest.mark.parametrize( "upper_bound_us", [SPARK_TIMESTAMP_SENTINEL_US, DATETIME64_MAX_US], diff --git a/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py b/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py index c62e6029b8513..5265d08769e1c 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_minmax_pruning.py @@ -81,6 +81,15 @@ def execute_spark_query(query: str): )) creation_expression = TABLE_NAME + spark_timestamps = spark.sql( + f"SELECT unix_micros(ts), unix_micros(time_struct.b) FROM {TABLE_NAME} ORDER BY tag" + ).collect() + expected_timestamps = "".join(f"{row[0]}\t{row[1]}\n" for row in spark_timestamps) + assert instance.query( + f"SELECT toUnixTimestamp64Micro(ts), toUnixTimestamp64Micro(time_struct.b) " + f"FROM {creation_expression} ORDER BY tag" + ) == expected_timestamps + def check_validity_and_get_prunned_files(select_expression): settings1 = { "use_iceberg_partition_pruning": 0, diff --git a/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py b/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py index 229d0b4b239b8..387e1272831bd 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_pruning_datetime64_nanoseconds.py @@ -18,11 +18,9 @@ PRUNE_OFF, PRUNE_ON, ROWS, - check_minmax_ns_bounds_on_timestamp_schema, check_ns_pruning, check_ns_pruning_datetime64_microseconds, patch_iceberg_table_long_to_timestamp_ns, - patch_iceberg_timestamp_bounds_us_to_ns, ) from helpers.iceberg_utils import ( check_validity_and_get_prunned_files_general, @@ -72,36 +70,6 @@ def _create_spark_ns_table( ) -def _create_spark_timestamp_table_with_ns_bounds(spark, started_cluster, table_name): - """Spark Iceberg `TIMESTAMP` (microseconds) whose manifest min/max bytes are nanoseconds.""" - spark.sql( - f""" - CREATE TABLE {table_name} ( - ts TIMESTAMP, - value TIMESTAMP, - id INT - ) - USING iceberg - OPTIONS('format-version'='2') - """ - ) - values = ", ".join( - f"(TIMESTAMP '{utc.split('.')[0]}.{utc.split('.')[1][:6]}', " - f"TIMESTAMP '{utc.split('.')[0]}.{utc.split('.')[1][:6]}', {row_id})" - for _ts_ns, row_id, utc in ROWS - ) - spark.sql(f"INSERT INTO {table_name} VALUES {values}") - - table_location = Path(f"{WAREHOUSE_PREFIX}/{table_name}") - patch_iceberg_timestamp_bounds_us_to_ns(table_location) - default_upload_directory( - started_cluster, - "s3", - f"/iceberg_data/default/{table_name}/", - f"/iceberg_data/default/{table_name}/", - ) - - @pytest.mark.parametrize("make_literal", FILTER_LITERALS, ids=FILTER_LITERAL_IDS) def test_spark_partition_pruning_datetime64_nanoseconds(started_cluster_iceberg_with_spark, make_literal): """Identity partition pruning on Spark-written Iceberg timestamp_ns must not drop matching rows.""" @@ -159,23 +127,3 @@ def pruned_files(select_expression): ) check_ns_pruning_datetime64_microseconds(node, table, "value", pruned_files, timezone=timezone) - - -def test_spark_minmax_pruning_ns_bounds_on_timestamp_schema(started_cluster_iceberg_with_spark): - """Nanosecond parquet/bounds with Iceberg `timestamp` (DateTime64(6)) must not over-prune.""" - node = started_cluster_iceberg_with_spark.instances["node1"] - spark = started_cluster_iceberg_with_spark.spark_session - table_name = "test_spark_minmax_ns_on_ts_" + get_uuid_str() - _create_spark_timestamp_table_with_ns_bounds( - spark, started_cluster_iceberg_with_spark, table_name - ) - table = get_creation_expression( - "s3", table_name, started_cluster_iceberg_with_spark, table_function=True - ) - - def pruned_files(select_expression): - return check_validity_and_get_prunned_files_general( - node, table_name, PRUNE_OFF, PRUNE_ON, "IcebergMinMaxIndexPrunedFiles", select_expression - ) - - check_minmax_ns_bounds_on_timestamp_schema(node, table, "value", pruned_files)