From a9097cfc83f65bd5459c7d7c7878b6d996a95c49 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 17 Sep 2026 05:02:54 +0200 Subject: [PATCH 01/12] Added integration test to validate reading of delta lake clustering columns --- tests/integration/test_storage_delta/test.py | 49 ++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/tests/integration/test_storage_delta/test.py b/tests/integration/test_storage_delta/test.py index 39684966f8013..bdf42dee89a40 100644 --- a/tests/integration/test_storage_delta/test.py +++ b/tests/integration/test_storage_delta/test.py @@ -5615,3 +5615,52 @@ def test_delta_kernel_retry_on_stale_token_via_catalog_callback(started_cluster) f"Expected the catalog-callback retry log line to fire for query {retry_query_id}, " f"found {retry_hits} hits — the stale-token retry path was not exercised." ) + + +def test_liquid_clustering(started_cluster): + """Test reading a Delta Lake table that uses liquid clustering. + + Creates a table with CLUSTER BY via Spark SQL, writes data, runs OPTIMIZE, + then reads it from ClickHouse via the delta-kernel-rs path. Liquid + clustering tables require the `v2Checkpoint` and `domainMetadata` reader + features, which are only supported by the delta-kernel-rs path. + """ + instance = started_cluster.instances["node1"] + spark = started_cluster.spark_session + TABLE_NAME = randomize_table_name("test_liquid_clustering") + + # Create a liquid-clustered Delta table via Spark SQL. + # delta-spark 3.3.0 supports CLUSTER BY. + delta_path = f"/{TABLE_NAME}" + spark.sql(f""" + CREATE TABLE delta.`{delta_path}` (a INT, b STRING) + USING DELTA CLUSTER BY (a) + """) + spark.sql(f""" + INSERT INTO delta.`{delta_path}` + SELECT id as a, CAST(id + 1 AS STRING) as b FROM range(100) + """) + spark.sql(f"OPTIMIZE delta.`{delta_path}`") + + files = default_upload_directory( + started_cluster, + "s3", + delta_path, + "", + ) + assert len(files) > 0 + + create_delta_table(instance, "s3", TABLE_NAME, started_cluster) + + # Verify row count and data. + assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100 + assert instance.query(f"SELECT * FROM {TABLE_NAME} ORDER BY a") == instance.query( + "SELECT number as a, toString(number + 1) as b FROM numbers(100)" + ) + + # Verify predicate pushdown on the clustering column. + result = instance.query(f"SELECT a FROM {TABLE_NAME} WHERE a = 42") + assert result.strip() == "42" + + # Verify aggregation works. + assert int(instance.query(f"SELECT sum(a) FROM {TABLE_NAME}")) == sum(range(100)) From 9d30c69e7973b6893f8ebc4fe39c1d9a5168450a Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Fri, 18 Sep 2026 17:45:35 +0200 Subject: [PATCH 02/12] Bump delta-kernel-rs: add get_domain_metadata_internal for delta.* system domains --- contrib/delta-kernel-rs | 2 +- src/Common/ProfileEvents.cpp | 1 + src/Core/Settings.cpp | 6 + .../DataLakes/DeltaLake/TableSnapshot.cpp | 159 ++++++++++++++++++ .../DataLakes/DeltaLake/TableSnapshot.h | 9 + tests/integration/helpers/cluster.py | 5 +- tests/integration/test_storage_delta/test.py | 118 +++++++++++++ 7 files changed, 298 insertions(+), 2 deletions(-) diff --git a/contrib/delta-kernel-rs b/contrib/delta-kernel-rs index 857565caf9ac6..ee74bc176a2ba 160000 --- a/contrib/delta-kernel-rs +++ b/contrib/delta-kernel-rs @@ -1 +1 @@ -Subproject commit 857565caf9ac61511c0e005ee3a32a30877a1a15 +Subproject commit ee74bc176a2ba1ed0c7a9ffcd443e94c76636aed diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 184d4941aa3c4..ea7ee24890892 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -450,6 +450,7 @@ M(DeltaLakePartitionPrunedFiles, "Number of skipped files during DeltaLake partition pruning", ValueType::Number) \ M(DeltaLakeSnapshotInitializations, "Number of times a DeltaLake table snapshot was initialized (loaded from object storage)", ValueType::Number) \ M(DeltaLakeScannedFiles, "Number of files scanned during DeltaLake scan callbacks", ValueType::Number) \ + M(DeltaLakeClusteringPredicateFallbacks, "Number of times the clustering-column predicate fallback was used when full predicate translation to delta-kernel failed", ValueType::Number) \ \ M(SlowRead, "Number of reads from a file that were slow. This indicate system overload. Thresholds are controlled by read_backoff_* settings.", ValueType::Number) \ M(ReadBackoff, "Number of times the number of query processing threads was lowered due to slow reads.", ValueType::Number) \ diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index d0725772d8aa4..17f3921981b39 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7457,6 +7457,12 @@ Enables throwing an exception if there was an error when analyzing scan predicat )", 0) \ DECLARE(Bool, delta_lake_enable_engine_predicate, true, R"( Enables delta-kernel internal data pruning. +)", 0) \ + DECLARE(Bool, delta_lake_clustering_predicate_fallback, true, R"( +When full predicate translation to delta-kernel fails, fall back to pushing only the +conjuncts that reference liquid clustering columns (which always have per-file min/max stats). +This improves file-level data skipping for queries that combine clustering-column filters +with expressions the kernel cannot translate. )", 0) \ DECLARE(NonZeroUInt64, delta_lake_insert_max_rows_in_data_file, 1000000, R"( Defines a rows limit for a single inserted data file in delta lake. diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp index e8ba1daa51a63..4e4ae9afad185 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp @@ -26,6 +26,11 @@ #include #include #include +#include + +#include +#include +#include #include #include @@ -49,6 +54,7 @@ namespace DB::Setting extern const SettingsInt64 delta_lake_snapshot_version; extern const SettingsBool delta_lake_throw_on_engine_predicate_error; extern const SettingsBool delta_lake_enable_engine_predicate; + extern const SettingsBool delta_lake_clustering_predicate_fallback; } namespace ProfileEvents @@ -56,6 +62,7 @@ namespace ProfileEvents extern const Event DeltaLakePartitionPrunedFiles; extern const Event DeltaLakeSnapshotInitializations; extern const Event DeltaLakeScannedFiles; + extern const Event DeltaLakeClusteringPredicateFallbacks; } namespace DB::FailPoints @@ -96,6 +103,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator const TableSchema & table_schema_, const DB::NameToNameMap & physical_names_map_, const DB::Names & partition_columns_, + const DB::Names & clustering_columns_, DB::ObjectStoragePtr object_storage_, const DB::ActionsDAG * filter_, DB::IDataLakeMetadata::FileProgressCallback callback_, @@ -103,6 +111,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator bool enable_expression_visitor_logging_, bool throw_on_engine_predicate_error_, bool enable_engine_predicate_, + bool enable_clustering_predicate_fallback_, UpdateStatsFunc update_stats_func_, LoggerPtr log_) : kernel_snapshot_state(kernel_snapshot_state_) @@ -111,6 +120,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator , read_schema(read_schema_) , expression_schema(table_schema_) , partition_columns(partition_columns_) + , clustering_columns(clustering_columns_) , object_storage(object_storage_) , callback(callback_) , list_batch_size(list_batch_size_) @@ -118,6 +128,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator , enable_expression_visitor_logging(enable_expression_visitor_logging_) , throw_on_engine_predicate_error(throw_on_engine_predicate_error_) , enable_engine_predicate(enable_engine_predicate_) + , enable_clustering_predicate_fallback(enable_clustering_predicate_fallback_) , update_stats_func(update_stats_func_) { if (filter_) @@ -189,6 +200,57 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator predicate.get(), /* schema */nullptr), "scan"); + + if (engine_predicate_exception && enable_clustering_predicate_fallback && !clustering_columns.empty()) + { + /// Full predicate translation failed. Try a fallback: restrict the filter + /// to only reference clustering columns (which always have per-file min/max stats). + /// `restrictFilterDAGToInputs` replaces non-computable AND branches with `true`, + /// giving us a weaker but valid predicate for file-level skipping. + DB::NameSet clustering_inputs; + for (const auto & col : clustering_columns) + clustering_inputs.insert(col); + + const auto * filter_output = filter->getOutputs().at(0); + auto restricted_dag = filter->restrictFilterDAGToInputs(filter_output, clustering_inputs); + + /// Check if the restricted filter is non-trivial (has at least one function node). + /// If restrictFilterDAGToInputs replaced everything with constant `true`, + /// the DAG will have no function nodes. + bool has_function = false; + for (const auto & node : restricted_dag.getNodes()) + { + if (node.type == DB::ActionsDAG::ActionType::FUNCTION) + { + has_function = true; + break; + } + } + + if (has_function) + { + std::exception_ptr fallback_exception; + auto fallback_predicate = getEnginePredicate(restricted_dag, fallback_exception, nullptr); + if (!fallback_exception) + { + LOG_INFO( + log, + "Full predicate translation failed; falling back to clustering-column-only " + "predicate for file-level data skipping (clustering columns: {})", + fmt::join(clustering_columns, ", ")); + + ProfileEvents::increment(ProfileEvents::DeltaLakeClusteringPredicateFallbacks); + + scan = KernelUtils::unwrapResult( + ffi::scan( + kernel_snapshot_state->snapshot.get(), + kernel_snapshot_state->engine.get(), + fallback_predicate.get(), + /* schema */nullptr), + "scan"); + } + } + } } else { @@ -617,6 +679,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator DB::NamesAndTypesList read_schema; DB::NamesAndTypesList expression_schema; DB::Names partition_columns; + DB::Names clustering_columns; const DB::ObjectStoragePtr object_storage; const DB::IDataLakeMetadata::FileProgressCallback callback; const size_t list_batch_size; @@ -624,6 +687,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator const bool enable_expression_visitor_logging; const bool throw_on_engine_predicate_error; const bool enable_engine_predicate; + const bool enable_clustering_predicate_fallback; const UpdateStatsFunc update_stats_func; std::exception_ptr scan_exception; @@ -919,6 +983,7 @@ DB::ObjectIterator TableSnapshot::iterate( const auto & settings = context->getSettingsRef(); std::lock_guard lock(mutex); initOrUpdateSchemaIfChanged(); + initOrUpdateClusteringColumns(); auto state = getKernelSnapshotState(); auto update_stats_func = [self = shared_from_this(), version = state->snapshot_version, this] (SnapshotStats && stats) @@ -942,6 +1007,7 @@ DB::ObjectIterator TableSnapshot::iterate( schema->table_schema, schema->physical_names_map, schema->partition_columns, + *clustering_columns, object_storage, filter_dag, callback, @@ -949,6 +1015,7 @@ DB::ObjectIterator TableSnapshot::iterate( settings[DB::Setting::delta_lake_enable_expression_visitor_logging], settings[DB::Setting::delta_lake_throw_on_engine_predicate_error], settings[DB::Setting::delta_lake_enable_engine_predicate], + settings[DB::Setting::delta_lake_clustering_predicate_fallback], std::move(update_stats_func), log); } @@ -1011,6 +1078,98 @@ const DB::NameToNameMap & TableSnapshot::getPhysicalNamesMap() const return schema->physical_names_map; } +const DB::Names & TableSnapshot::getClusteringColumns() const +{ + std::lock_guard lock(mutex); + initOrUpdateClusteringColumns(); + return *clustering_columns; +} + +void TableSnapshot::initOrUpdateClusteringColumns() const +{ + if (clustering_columns.has_value()) + return; + + auto state = getKernelSnapshotState(); + auto result = KernelUtils::unwrapResult( + ffi::get_domain_metadata_internal( + state->snapshot.get(), + KernelUtils::toDeltaString("delta.clustering"), + state->engine.get(), + KernelUtils::allocateString), + "get_domain_metadata_internal(delta.clustering)"); + + if (!result) + { + LOG_TRACE(log, "No delta.clustering domain metadata found — table is not liquid-clustered"); + clustering_columns.emplace(); + return; + } + + std::unique_ptr json_str(static_cast(result)); + + DB::Names physical_columns; + try + { + Poco::JSON::Parser parser; + auto json = parser.parse(*json_str); + auto obj = json.extract(); + + auto arr = obj->getArray("clusteringColumns"); + if (!arr) + throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "delta.clustering: missing clusteringColumns array"); + + for (size_t i = 0; i < arr->size(); ++i) + { + auto path_arr = arr->getArray(static_cast(i)); + if (!path_arr || path_arr->size() == 0) + continue; + + std::string col_name; + for (size_t j = 0; j < path_arr->size(); ++j) + { + if (!col_name.empty()) + col_name += '.'; + col_name += path_arr->getElement(static_cast(j)); + } + physical_columns.push_back(std::move(col_name)); + } + } + catch (const Poco::Exception & e) + { + throw DB::Exception( + DB::ErrorCodes::BAD_ARGUMENTS, + "Failed to parse delta.clustering domain metadata: {}", e.displayText()); + } + + /// Convert physical column names to logical names using the inverse of physical_names_map. + /// The filter DAG uses logical names, so the clustering column names must match. + DB::Names logical_columns; + if (schema.has_value() && !schema->physical_names_map.empty()) + { + /// Build reverse map: physical → logical + DB::NameToNameMap reverse_map; + for (const auto & [logical, physical] : schema->physical_names_map) + reverse_map[physical] = logical; + + for (const auto & phys : physical_columns) + { + auto it = reverse_map.find(phys); + if (it != reverse_map.end()) + logical_columns.push_back(it->second); + else + logical_columns.push_back(phys); + } + } + else + { + logical_columns = std::move(physical_columns); + } + + LOG_TRACE(log, "Clustering columns: {}", fmt::join(logical_columns, ", ")); + clustering_columns.emplace(std::move(logical_columns)); +} + } #endif diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h index 0a6f193113ac7..fce69afebd86d 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h @@ -58,6 +58,10 @@ class TableSnapshot : public std::enable_shared_from_this const DB::Names & getPartitionColumns() const; const DB::NameToNameMap & getPhysicalNamesMap() const; + /// Get the clustering columns from the `delta.clustering` domain metadata. + /// Returns an empty list if the table is not liquid-clustered. + const DB::Names & getClusteringColumns() const; + DB::ObjectStoragePtr getObjectStorage() const { return object_storage; } private: class Iterator; @@ -106,6 +110,10 @@ class TableSnapshot : public std::enable_shared_from_this }; mutable std::optional schema; + /// Clustering columns from the `delta.clustering` domain metadata. + /// std::nullopt means not yet fetched; empty vector means not clustered. + mutable std::optional clustering_columns; + struct SnapshotStats { /// Total number of bytes in table @@ -121,6 +129,7 @@ class TableSnapshot : public std::enable_shared_from_this void initOrUpdateSnapshot() const TSA_REQUIRES(mutex); void initOrUpdateSchemaIfChanged() const TSA_REQUIRES(mutex); + void initOrUpdateClusteringColumns() const TSA_REQUIRES(mutex); SnapshotStats getSnapshotStats() const TSA_REQUIRES(mutex); SnapshotStats getSnapshotStatsImpl() const TSA_REQUIRES(mutex); diff --git a/tests/integration/helpers/cluster.py b/tests/integration/helpers/cluster.py index 2e81d9e52b21b..108cae7022522 100644 --- a/tests/integration/helpers/cluster.py +++ b/tests/integration/helpers/cluster.py @@ -3795,7 +3795,10 @@ def start(self, connection_timeout=None): common_opts = ["--verbose", "up", "-d"] - images_pull_cmd = self.base_cmd + ["pull"] + # Tolerate pull failures for images that are expected to be present + # locally (e.g. CLICKHOUSE_CI_MIN_TESTED_VERSION tags that only exist + # in a private registry); `docker compose up` will use the local image. + images_pull_cmd = self.base_cmd + ["pull", "--ignore-pull-failures"] # sometimes dockerhub/proxy can be flaky def logging_pulling_images(**kwargs): diff --git a/tests/integration/test_storage_delta/test.py b/tests/integration/test_storage_delta/test.py index bdf42dee89a40..e6dcaad69a8d5 100644 --- a/tests/integration/test_storage_delta/test.py +++ b/tests/integration/test_storage_delta/test.py @@ -5664,3 +5664,121 @@ def test_liquid_clustering(started_cluster): # Verify aggregation works. assert int(instance.query(f"SELECT sum(a) FROM {TABLE_NAME}")) == sum(range(100)) + + +def test_liquid_clustering_data_skipping(started_cluster): + """Verify that filtering on a clustering column skips files at the kernel level. + + Creates a liquid-clustered Delta table with multiple batches so that OPTIMIZE + produces several output files, then checks that a point-lookup on the + clustering column reads fewer files than a full scan. + """ + instance = started_cluster.instances["node1"] + spark = started_cluster.spark_session + TABLE_NAME = randomize_table_name("test_liquid_clustering_ds") + + delta_path = f"/{TABLE_NAME}" + spark.sql(f""" + CREATE TABLE delta.`{delta_path}` (a INT, b STRING) + USING DELTA CLUSTER BY (a) + """) + + # Insert multiple batches to produce several files after OPTIMIZE. + for i in range(5): + spark.sql(f""" + INSERT INTO delta.`{delta_path}` + SELECT id + {i * 1000} as a, CAST(id + {i * 1000} AS STRING) as b + FROM range(1000) + """) + spark.sql(f"OPTIMIZE delta.`{delta_path}`") + + files = default_upload_directory(started_cluster, "s3", delta_path, "") + assert len(files) > 0 + + create_delta_table(instance, "s3", TABLE_NAME, started_cluster) + + total_rows = int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) + assert total_rows == 5000 + + # Full scan — note the total scanned files. + full_scan_id = f"{TABLE_NAME}_full_scan" + instance.query(f"SELECT count() FROM {TABLE_NAME}", query_id=full_scan_id) + instance.query("SYSTEM FLUSH LOGS") + full_scan_files = int( + instance.query( + f"SELECT ProfileEvents['DeltaLakeScannedFiles'] " + f"FROM system.query_log WHERE query_id = '{full_scan_id}' AND type = 'QueryFinish'" + ).strip() + ) + + # Point query on the clustering column — should scan fewer files. + point_query_id = f"{TABLE_NAME}_point_query" + result = instance.query( + f"SELECT a FROM {TABLE_NAME} WHERE a = 42", query_id=point_query_id + ) + assert result.strip() == "42" + instance.query("SYSTEM FLUSH LOGS") + point_scan_files = int( + instance.query( + f"SELECT ProfileEvents['DeltaLakeScannedFiles'] " + f"FROM system.query_log WHERE query_id = '{point_query_id}' AND type = 'QueryFinish'" + ).strip() + ) + + assert point_scan_files < full_scan_files, ( + f"Expected clustering to enable file skipping: point query scanned {point_scan_files} files " + f"but full scan scanned {full_scan_files} files" + ) + + +def test_liquid_clustering_predicate_fallback(started_cluster): + """Verify that the clustering-column predicate fallback fires when the full + predicate cannot be translated, and that file skipping still works. + + Uses a filter combining an untranslatable expression with a simple clustering + column filter. The fallback should push the clustering filter to the kernel + while ClickHouse evaluates the full expression on the remaining rows. + """ + instance = started_cluster.instances["node1"] + spark = started_cluster.spark_session + TABLE_NAME = randomize_table_name("test_liquid_clustering_fb") + + delta_path = f"/{TABLE_NAME}" + spark.sql(f""" + CREATE TABLE delta.`{delta_path}` (a INT, b STRING) + USING DELTA CLUSTER BY (a) + """) + + for i in range(5): + spark.sql(f""" + INSERT INTO delta.`{delta_path}` + SELECT id + {i * 1000} as a, CAST(id + {i * 1000} AS STRING) as b + FROM range(1000) + """) + spark.sql(f"OPTIMIZE delta.`{delta_path}`") + + files = default_upload_directory(started_cluster, "s3", delta_path, "") + assert len(files) > 0 + + create_delta_table(instance, "s3", TABLE_NAME, started_cluster) + + # Query with a clustering filter (a = 42) AND an untranslatable expression + # (cityHash64). The full predicate translation should fail, but the fallback + # should push the `a = 42` part for file-level skipping. + fallback_query_id = f"{TABLE_NAME}_fallback" + result = instance.query( + f"SELECT a FROM {TABLE_NAME} WHERE a = 42 AND cityHash64(b) != 0", + query_id=fallback_query_id, + ) + assert result.strip() == "42" + + instance.query("SYSTEM FLUSH LOGS") + fallback_count = int( + instance.query( + f"SELECT ProfileEvents['DeltaLakeClusteringPredicateFallbacks'] " + f"FROM system.query_log WHERE query_id = '{fallback_query_id}' AND type = 'QueryFinish'" + ).strip() + ) + assert fallback_count >= 1, ( + f"Expected the clustering predicate fallback to fire, but got {fallback_count} hits" + ) From 35afbc22c16378c1b9c365b6a1a9c6e10d77044b Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Fri, 18 Sep 2026 19:28:20 +0200 Subject: [PATCH 03/12] Added delta_lake_clustering_predicate_fallback setting to SettingsChangeHistory --- src/Core/SettingsChangesHistory.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index f093c4054cf68..ab1383c34bd27 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -42,10 +42,12 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya", { {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."}, + {"delta_lake_clustering_predicate_fallback", false, true, "New setting to enable clustering-aware predicate fallback for Delta Lake tables with Liquid Clustering."}, }); addSettingsChanges(settings_changes_history, "26.6", { + {"delta_lake_clustering_predicate_fallback", false, true, "New setting to enable clustering-aware predicate fallback for Delta Lake tables with Liquid Clustering."}, {"analyzer_compatibility_multiple_joins_qualify_column_names", false, false, "New compatibility setting. When enabled, the analyzer mimics the old analyzer's qualified result column names for queries whose FROM clause has two or more JOINs."}, {"analyzer_compatibility_apply_final_to_all_joined_tables", true, false, "Fixed a bug in the analyzer where FINAL on the left-most table of a JOIN was incorrectly applied to the other joined tables as well. previous_value=true so `compatibility` with versions before 26.6 restores the old behavior."}, {"analyzer_compatibility_allow_non_aggregate_in_having", false, false, "New compatibility setting. When enabled, the new analyzer mimics the legacy `HAVING`-to-`WHERE` rewrite for non-aggregate AND-conjuncts instead of raising `NOT_AN_AGGREGATE`."}, From 376279f26f6a100a43f8052cc18d4168e5b1189a Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Fri, 18 Sep 2026 23:56:27 +0200 Subject: [PATCH 04/12] Revert delta-kernel-rs submodule pointer to 857565ca (ClickHouse/v0.19.2-with-panic-safe-ffi) Co-authored-by: Cursor --- contrib/delta-kernel-rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/contrib/delta-kernel-rs b/contrib/delta-kernel-rs index ee74bc176a2ba..857565caf9ac6 160000 --- a/contrib/delta-kernel-rs +++ b/contrib/delta-kernel-rs @@ -1 +1 @@ -Subproject commit ee74bc176a2ba1ed0c7a9ffcd443e94c76636aed +Subproject commit 857565caf9ac61511c0e005ee3a32a30877a1a15 From dd65e97fa503997b8c02d0d50e2da731c9a98dd1 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Fri, 18 Sep 2026 23:56:27 +0200 Subject: [PATCH 05/12] Revert delta-kernel-rs submodule pointer to 857565ca (ClickHouse/v0.19.2-with-panic-safe-ffi) --- contrib/delta-kernel-rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/contrib/delta-kernel-rs b/contrib/delta-kernel-rs index ee74bc176a2ba..857565caf9ac6 160000 --- a/contrib/delta-kernel-rs +++ b/contrib/delta-kernel-rs @@ -1 +1 @@ -Subproject commit ee74bc176a2ba1ed0c7a9ffcd443e94c76636aed +Subproject commit 857565caf9ac61511c0e005ee3a32a30877a1a15 From ae296bba0d5a4abd2504d0939fd2a34e7958a82b Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Wed, 7 Oct 2026 04:41:09 +0200 Subject: [PATCH 06/12] Updated delta-kernel-rs to Altinity fork and updated the commit --- .gitmodules | 2 +- contrib/delta-kernel-rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitmodules b/.gitmodules index 6acf8b86ad428..0e8920dd20845 100644 --- a/.gitmodules +++ b/.gitmodules @@ -353,7 +353,7 @@ url = https://github.com/ClickHouse/wasmtime [submodule "contrib/delta-kernel-rs"] path = contrib/delta-kernel-rs - url = https://github.com/ClickHouse/delta-kernel-rs.git + url = https://github.com/Altinity/delta-kernel-rs.git [submodule "contrib/SHA3IUF"] path = contrib/SHA3IUF url = https://github.com/brainhub/SHA3IUF.git diff --git a/contrib/delta-kernel-rs b/contrib/delta-kernel-rs index 857565caf9ac6..625c2fa0911c5 160000 --- a/contrib/delta-kernel-rs +++ b/contrib/delta-kernel-rs @@ -1 +1 @@ -Subproject commit 857565caf9ac61511c0e005ee3a32a30877a1a15 +Subproject commit 625c2fa0911c550ca5cada52952a7fb3eb687ea4 From edadc97cd5122a12c2b866b5d2d4bea96c8b33e9 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 8 Oct 2026 17:41:12 +0200 Subject: [PATCH 07/12] Fix medium defects --- .../DataLakes/DeltaLake/TableSnapshot.cpp | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp index 4e4ae9afad185..f69b54c51f3d9 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp @@ -250,6 +250,15 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator "scan"); } } + else + { + LOG_DEBUG( + log, + "Clustering predicate fallback: restricted filter DAG is trivial " + "(no function nodes remain after restricting to clustering columns: {}). " + "Skipping fallback — full scan will be used.", + fmt::join(clustering_columns, ", ")); + } } } else @@ -758,6 +767,7 @@ TableSnapshot::SnapshotStats TableSnapshot::getSnapshotStats() const /// Invalidate the cached engine so `initOrUpdateSnapshot` rebuilds with the /// freshened credentials, then re-run the stats scan against the new engine. kernel_snapshot_state.reset(); + clustering_columns.reset(); initOrUpdateSnapshot(); snapshot_stats = getSnapshotStatsImpl(); } From 15b23f394d9895e382419d97202a335be880908f Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 8 Oct 2026 19:37:00 +0200 Subject: [PATCH 08/12] Removed clustering columns reset --- .../DataLakes/DeltaLake/TableSnapshot.cpp | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp index f69b54c51f3d9..4e4ae9afad185 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp @@ -250,15 +250,6 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator "scan"); } } - else - { - LOG_DEBUG( - log, - "Clustering predicate fallback: restricted filter DAG is trivial " - "(no function nodes remain after restricting to clustering columns: {}). " - "Skipping fallback — full scan will be used.", - fmt::join(clustering_columns, ", ")); - } } } else @@ -767,7 +758,6 @@ TableSnapshot::SnapshotStats TableSnapshot::getSnapshotStats() const /// Invalidate the cached engine so `initOrUpdateSnapshot` rebuilds with the /// freshened credentials, then re-run the stats scan against the new engine. kernel_snapshot_state.reset(); - clustering_columns.reset(); initOrUpdateSnapshot(); snapshot_stats = getSnapshotStatsImpl(); } From 8a5d1943fa69d6144d0ef3504e512ea4b4dc9262 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 15 Oct 2026 16:43:27 +0200 Subject: [PATCH 09/12] Removed getClusteringColumns function thats not used --- .../ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp | 7 ------- .../ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h | 4 ---- 2 files changed, 11 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp index 4e4ae9afad185..57676237ebd5c 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp @@ -1078,13 +1078,6 @@ const DB::NameToNameMap & TableSnapshot::getPhysicalNamesMap() const return schema->physical_names_map; } -const DB::Names & TableSnapshot::getClusteringColumns() const -{ - std::lock_guard lock(mutex); - initOrUpdateClusteringColumns(); - return *clustering_columns; -} - void TableSnapshot::initOrUpdateClusteringColumns() const { if (clustering_columns.has_value()) diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h index fce69afebd86d..29b9fc0907555 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h @@ -58,10 +58,6 @@ class TableSnapshot : public std::enable_shared_from_this const DB::Names & getPartitionColumns() const; const DB::NameToNameMap & getPhysicalNamesMap() const; - /// Get the clustering columns from the `delta.clustering` domain metadata. - /// Returns an empty list if the table is not liquid-clustered. - const DB::Names & getClusteringColumns() const; - DB::ObjectStoragePtr getObjectStorage() const { return object_storage; } private: class Iterator; From 35197a8120c160460edea50a35c98a5ab3a26401 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 15 Oct 2026 20:50:37 +0200 Subject: [PATCH 10/12] Replaced BAD_ARGUMENTS with INCORRECT_DATA error code --- .../DataLakes/DeltaLake/TableSnapshot.cpp | 22 +++++++++++++------ 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp index 57676237ebd5c..06665745d10f4 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp @@ -46,6 +46,7 @@ namespace DB::ErrorCodes { extern const int BAD_ARGUMENTS; extern const int DELTA_KERNEL_ERROR; + extern const int INCORRECT_DATA; } namespace DB::Setting @@ -983,7 +984,12 @@ DB::ObjectIterator TableSnapshot::iterate( const auto & settings = context->getSettingsRef(); std::lock_guard lock(mutex); initOrUpdateSchemaIfChanged(); - initOrUpdateClusteringColumns(); + /// Clustering columns are only used by the clustering-predicate fallback in `Iterator::initScanState`. + const bool need_clustering_columns = filter_dag + && settings[DB::Setting::delta_lake_enable_engine_predicate] + && settings[DB::Setting::delta_lake_clustering_predicate_fallback]; + if (need_clustering_columns) + initOrUpdateClusteringColumns(); auto state = getKernelSnapshotState(); auto update_stats_func = [self = shared_from_this(), version = state->snapshot_version, this] (SnapshotStats && stats) @@ -1007,7 +1013,7 @@ DB::ObjectIterator TableSnapshot::iterate( schema->table_schema, schema->physical_names_map, schema->partition_columns, - *clustering_columns, + need_clustering_columns ? *clustering_columns : DB::Names{}, object_storage, filter_dag, callback, @@ -1094,7 +1100,7 @@ void TableSnapshot::initOrUpdateClusteringColumns() const if (!result) { - LOG_TRACE(log, "No delta.clustering domain metadata found — table is not liquid-clustered"); + LOG_TRACE(log, "No delta.clustering domain metadata found for table {} — table is not liquid-clustered", helper->getTableLocation()); clustering_columns.emplace(); return; } @@ -1110,7 +1116,7 @@ void TableSnapshot::initOrUpdateClusteringColumns() const auto arr = obj->getArray("clusteringColumns"); if (!arr) - throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "delta.clustering: missing clusteringColumns array"); + throw DB::Exception(DB::ErrorCodes::INCORRECT_DATA, "delta.clustering: missing clusteringColumns array"); for (size_t i = 0; i < arr->size(); ++i) { @@ -1131,8 +1137,10 @@ void TableSnapshot::initOrUpdateClusteringColumns() const catch (const Poco::Exception & e) { throw DB::Exception( - DB::ErrorCodes::BAD_ARGUMENTS, - "Failed to parse delta.clustering domain metadata: {}", e.displayText()); + DB::ErrorCodes::INCORRECT_DATA, + "Failed to parse delta.clustering domain metadata of table {}: {}. " + "Set delta_lake_clustering_predicate_fallback = 0 to read the table without clustering-based file skipping", + helper->getTableLocation(), e.displayText()); } /// Convert physical column names to logical names using the inverse of physical_names_map. @@ -1159,7 +1167,7 @@ void TableSnapshot::initOrUpdateClusteringColumns() const logical_columns = std::move(physical_columns); } - LOG_TRACE(log, "Clustering columns: {}", fmt::join(logical_columns, ", ")); + LOG_TRACE(log, "Clustering columns for table {}: {}", helper->getTableLocation(), fmt::join(logical_columns, ", ")); clustering_columns.emplace(std::move(logical_columns)); } From c9dbbc8ab0598505b5e42bf9f2f34a50879aa7d8 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Fri, 16 Oct 2026 03:36:39 +0200 Subject: [PATCH 11/12] Removed delta_lake_clustering_predicate_fallback from 26.6 --- src/Core/SettingsChangesHistory.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 1e377c31c3add..c1d84434f9583 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -55,7 +55,6 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() addSettingsChanges(settings_changes_history, "26.6", { - {"delta_lake_clustering_predicate_fallback", false, true, "New setting to enable clustering-aware predicate fallback for Delta Lake tables with Liquid Clustering."}, {"analyzer_compatibility_multiple_joins_qualify_column_names", false, false, "New compatibility setting. When enabled, the analyzer mimics the old analyzer's qualified result column names for queries whose FROM clause has two or more JOINs."}, {"analyzer_compatibility_apply_final_to_all_joined_tables", true, false, "Fixed a bug in the analyzer where FINAL on the left-most table of a JOIN was incorrectly applied to the other joined tables as well. previous_value=true so `compatibility` with versions before 26.6 restores the old behavior."}, {"analyzer_compatibility_allow_non_aggregate_in_having", false, false, "New compatibility setting. When enabled, the new analyzer mimics the legacy `HAVING`-to-`WHERE` rewrite for non-aggregate AND-conjuncts instead of raising `NOT_AN_AGGREGATE`."}, From e8453b94bc8b2967d0878c0705665b42623b5a15 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 22 Oct 2026 20:00:17 +0200 Subject: [PATCH 12/12] Fixed review comments --- src/Core/SettingsChangesHistory.cpp | 6 +++- .../DataLakes/DeltaLake/TableSnapshot.cpp | 34 +++++++++++++++---- tests/integration/helpers/cluster.py | 5 +-- 3 files changed, 33 insertions(+), 12 deletions(-) diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index c1d84434f9583..75b05a8bfbf3c 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -39,6 +39,11 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() /// controls new feature and it's 'true' by default, use 'false' as previous_value). /// It's used to implement `compatibility` setting (see https://github.com/ClickHouse/ClickHouse/issues/35972) /// Note: please check if the key already exists to prevent duplicate entries. + addSettingsChanges(settings_changes_history, "26.6.8.20002.altinityantalya", + { + {"delta_lake_clustering_predicate_fallback", false, true, "New setting to enable clustering-aware predicate fallback for Delta Lake tables with Liquid Clustering."}, + }); + addSettingsChanges(settings_changes_history, "26.6.4.20001.altinityantalya", { {"input_format_parquet_footer_read_size", 65536, 0, "New setting to override the adaptive Parquet footer initial read size with a fixed number of bytes; 0 keeps the adaptive behavior. Before this setting the reader always read a fixed 64 KiB tail."}, @@ -48,7 +53,6 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() { {"object_storage_cluster_fallback_to_local_if_empty", false, false, "New setting"}, {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."}, - {"delta_lake_clustering_predicate_fallback", false, true, "New setting to enable clustering-aware predicate fallback for Delta Lake tables with Liquid Clustering."}, {"data_lake_delete_data_on_drop", false, false, "New setting that unifies dropping of data lake data; the released `iceberg_delete_data_on_drop` is kept as an alias for it."}, {"datalake_ignore_unsupported_table_properties", false, false, "New setting: allow `CREATE TABLE` and `SHOW CREATE TABLE` in a `DataLakeCatalog` to omit unsupported table properties."}, }); diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp index 06665745d10f4..28c2b2d72daf9 100644 --- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp +++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp @@ -1089,14 +1089,34 @@ void TableSnapshot::initOrUpdateClusteringColumns() const if (clustering_columns.has_value()) return; + auto read_domain_metadata = [](const KernelSnapshotState & state) + { + return KernelUtils::unwrapResult( + ffi::get_domain_metadata_internal( + state.snapshot.get(), + KernelUtils::toDeltaString("delta.clustering"), + state.engine.get(), + KernelUtils::allocateString), + "get_domain_metadata_internal(delta.clustering)"); + }; + auto state = getKernelSnapshotState(); - auto result = KernelUtils::unwrapResult( - ffi::get_domain_metadata_internal( - state->snapshot.get(), - KernelUtils::toDeltaString("delta.clustering"), - state->engine.get(), - KernelUtils::allocateString), - "get_domain_metadata_internal(delta.clustering)"); + const auto pre_fingerprint = kernel_state_credentials_fingerprint; + decltype(read_domain_metadata(*state)) result = nullptr; + try + { + result = read_domain_metadata(*state); + } + catch (const DB::Exception & e) + { + if (!tryRefreshAfterStaleTokenError(e, pre_fingerprint, "clustering metadata read")) + throw; + /// Invalidate the cached engine so `initOrUpdateSnapshot` rebuilds with the + /// freshened credentials, then re-read the domain metadata once. + kernel_snapshot_state.reset(); + initOrUpdateSnapshot(); + result = read_domain_metadata(*kernel_snapshot_state); + } if (!result) { diff --git a/tests/integration/helpers/cluster.py b/tests/integration/helpers/cluster.py index 108cae7022522..2e81d9e52b21b 100644 --- a/tests/integration/helpers/cluster.py +++ b/tests/integration/helpers/cluster.py @@ -3795,10 +3795,7 @@ def start(self, connection_timeout=None): common_opts = ["--verbose", "up", "-d"] - # Tolerate pull failures for images that are expected to be present - # locally (e.g. CLICKHOUSE_CI_MIN_TESTED_VERSION tags that only exist - # in a private registry); `docker compose up` will use the local image. - images_pull_cmd = self.base_cmd + ["pull", "--ignore-pull-failures"] + images_pull_cmd = self.base_cmd + ["pull"] # sometimes dockerhub/proxy can be flaky def logging_pulling_images(**kwargs):