From a9097cfc83f65bd5459c7d7c7878b6d996a95c49 Mon Sep 17 00:00:00 2001
From: Kanthi Subramanian <subkanthi@gmail.com>
Date: Thu, 17 Sep 2026 05:02:54 +0200
Subject: [PATCH 01/15] 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 <subkanthi@gmail.com>
Date: Fri, 18 Sep 2026 17:45:35 +0200
Subject: [PATCH 02/15] 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 <IO/WriteBufferFromString.h>
 #include <IO/WriteHelpers.h>
 #include <Interpreters/Context.h>
+#include <Interpreters/ActionsDAG.h>
+
+#include <Poco/JSON/Parser.h>
+#include <Poco/JSON/Object.h>
+#include <Poco/JSON/Array.h>
 
 #include <Storages/ObjectStorage/DataLakes/DeltaLake/getSchemaFromSnapshot.h>
 #include <Storages/ObjectStorage/DataLakes/DeltaLake/PartitionPruner.h>
@@ -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<std::string> json_str(static_cast<std::string *>(result));
+
+    DB::Names physical_columns;
+    try
+    {
+        Poco::JSON::Parser parser;
+        auto json = parser.parse(*json_str);
+        auto obj = json.extract<Poco::JSON::Object::Ptr>();
+
+        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<unsigned int>(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<std::string>(static_cast<unsigned int>(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<TableSnapshot>
     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<TableSnapshot>
     };
     mutable std::optional<SchemaInfo> schema;
 
+    /// Clustering columns from the `delta.clustering` domain metadata.
+    /// std::nullopt means not yet fetched; empty vector means not clustered.
+    mutable std::optional<DB::Names> clustering_columns;
+
     struct SnapshotStats
     {
         /// Total number of bytes in table
@@ -121,6 +129,7 @@ class TableSnapshot : public std::enable_shared_from_this<TableSnapshot>
 
     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 <subkanthi@gmail.com>
Date: Fri, 18 Sep 2026 19:28:20 +0200
Subject: [PATCH 03/15] 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 <subkanthi@gmail.com>
Date: Fri, 18 Sep 2026 23:56:27 +0200
Subject: [PATCH 04/15] Revert delta-kernel-rs submodule pointer to 857565ca
 (ClickHouse/v0.19.2-with-panic-safe-ffi)

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 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 <subkanthi@gmail.com>
Date: Fri, 18 Sep 2026 23:56:27 +0200
Subject: [PATCH 05/15] 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 <subkanthi@gmail.com>
Date: Wed, 7 Oct 2026 04:41:09 +0200
Subject: [PATCH 06/15] 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 <subkanthi@gmail.com>
Date: Thu, 8 Oct 2026 17:41:12 +0200
Subject: [PATCH 07/15] 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 <subkanthi@gmail.com>
Date: Thu, 8 Oct 2026 19:37:00 +0200
Subject: [PATCH 08/15] 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 <subkanthi@gmail.com>
Date: Thu, 15 Oct 2026 16:43:27 +0200
Subject: [PATCH 09/15] 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<TableSnapshot>
     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 <subkanthi@gmail.com>
Date: Thu, 15 Oct 2026 20:50:37 +0200
Subject: [PATCH 10/15] 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 <subkanthi@gmail.com>
Date: Fri, 16 Oct 2026 03:36:39 +0200
Subject: [PATCH 11/15] 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 <subkanthi@gmail.com>
Date: Thu, 22 Oct 2026 20:00:17 +0200
Subject: [PATCH 12/15] 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):

From b86ee048626f2ecc2b1f7862c3146c7b746af4a6 Mon Sep 17 00:00:00 2001
From: Kanthi Subramanian <subkanthi@gmail.com>
Date: Tue, 27 Oct 2026 20:44:04 +0100
Subject: [PATCH 13/15] Address review comments

---
 .../DataLakes/DeltaLake/EnginePredicate.cpp   | 14 ++++++++++-
 .../DataLakes/DeltaLake/EnginePredicate.h     |  5 ++++
 .../DataLakes/DeltaLake/TableSnapshot.cpp     | 24 ++++++++++++-------
 tests/integration/test_storage_delta/test.py  |  5 +++-
 4 files changed, 37 insertions(+), 11 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp
index 0655f8d10cb55..552b7b25514ef 100644
--- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp
@@ -99,6 +99,8 @@ struct EngineIteratorData
 
     bool hasException() const { return predicate.exception != nullptr; }
 
+    void markUnsupportedNode() { predicate.has_unsupported_nodes = true; }
+
     const LoggerPtr & log() const { return predicate.log; }
 
     const DB::ActionsDAG::Node * next()
@@ -160,6 +162,12 @@ class  EngineIterator : public ffi::EngineIterator
             {
                 return reinterpret_cast<const void *>(result);
             }
+
+            if (!iterator_data->hasException())
+            {
+                LOG_TEST(iterator_data->log(), "Unsupported node: {}", node->result_name);
+                iterator_data->markUnsupportedNode();
+            }
         }
         catch (...)
         {
@@ -179,7 +187,11 @@ uintptr_t EnginePredicate::visitPredicate(void * data, ffi::KernelExpressionVisi
     EngineIterator engine_iterator(iterator_data);
     auto result = ffi::visit_predicate_and(state, &engine_iterator);
 
-    LOG_TEST(iterator_data.log(), "visitPredicate finished (exception: {})", predicate->hasException());
+    LOG_TEST(
+        iterator_data.log(),
+        "visitPredicate finished (exception: {}, unsupported nodes: {})",
+        predicate->hasException(),
+        predicate->hasUnsupportedNodes());
     return result;
 }
 
diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.h b/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.h
index 4ff17685a2f45..5e3e0c7def5f2 100644
--- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.h
+++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.h
@@ -46,6 +46,10 @@ class EnginePredicate : public ffi::EnginePredicate
             exception = exception_;
     }
 
+    /// Whether some node of the filter could not be translated to a delta-kernel expression.
+    /// Such nodes are dropped from the predicate passed to delta-kernel.
+    bool hasUnsupportedNodes() const { return has_unsupported_nodes; }
+
     const DB::ActionsDAG & getFilterDAG() const { return filter; }
     DB::ContextPtr getContext() const { return context; }
 
@@ -58,6 +62,7 @@ class EnginePredicate : public ffi::EnginePredicate
     /// Exceptions cannot be rethrown as it will cause
     /// panic from rust and server terminate.
     std::exception_ptr & exception;
+    bool has_unsupported_nodes = false;
     /// Context for accessing settings
     DB::ContextPtr context;
 
diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp
index 28c2b2d72daf9..42c295ee532d4 100644
--- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp
@@ -202,7 +202,8 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
                     /* schema */nullptr),
                 "scan");
 
-            if (engine_predicate_exception && enable_clustering_predicate_fallback && !clustering_columns.empty())
+            const bool full_translation_failed = engine_predicate_exception || predicate->hasUnsupportedNodes();
+            if (full_translation_failed && 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).
@@ -232,7 +233,18 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
                 {
                     std::exception_ptr fallback_exception;
                     auto fallback_predicate = getEnginePredicate(restricted_dag, fallback_exception, nullptr);
-                    if (!fallback_exception)
+
+                    /// The predicate is translated inside `ffi::scan`, so its outcome is known only afterwards.
+                    KernelScan fallback_scan;
+                    fallback_scan = KernelUtils::unwrapResult(
+                        ffi::scan(
+                            kernel_snapshot_state->snapshot.get(),
+                            kernel_snapshot_state->engine.get(),
+                            fallback_predicate.get(),
+                            /* schema */nullptr),
+                        "scan");
+
+                    if (!fallback_exception && !fallback_predicate->hasUnsupportedNodes())
                     {
                         LOG_INFO(
                             log,
@@ -242,13 +254,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
 
                         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");
+                        scan = std::move(fallback_scan);
                     }
                 }
             }
diff --git a/tests/integration/test_storage_delta/test.py b/tests/integration/test_storage_delta/test.py
index 59025baa7c3db..fc858942d2f80 100644
--- a/tests/integration/test_storage_delta/test.py
+++ b/tests/integration/test_storage_delta/test.py
@@ -5719,7 +5719,10 @@ def test_liquid_clustering_data_skipping(started_cluster):
 
     # 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(
+        f"SELECT count() FROM {TABLE_NAME} SETTINGS optimize_trivial_count_query = 0",
+        query_id=full_scan_id,
+    )
     instance.query("SYSTEM FLUSH LOGS")
     full_scan_files = int(
         instance.query(

From 8715f489ba6185b380ea420552de34e5f1a92147 Mon Sep 17 00:00:00 2001
From: Kanthi Subramanian <subkanthi@gmail.com>
Date: Wed, 28 Oct 2026 19:06:53 +0100
Subject: [PATCH 14/15] Fix syntax error

---
 src/Core/SettingsChangesHistory.cpp | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
index f7a475c92a83f..1e336d727aba8 100644
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/src/Core/SettingsChangesHistory.cpp
@@ -42,6 +42,8 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         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.8.20003.altinityantalya",
         {
             {"iceberg_target_data_file_size_bytes", 536870912, 536870912, "Target file size for Iceberg bin-packing compaction (default 512 MiB)."},

From 92d9e6bd902654b2d1f0ae7ea9b2da472fd11f54 Mon Sep 17 00:00:00 2001
From: Kanthi Subramanian <subkanthi@gmail.com>
Date: Thu, 29 Oct 2026 00:00:03 +0100
Subject: [PATCH 15/15] Fix unsupported notes truncated by the iterator in
 EnginePredicate

---
 .../DataLakes/DeltaLake/EnginePredicate.cpp   | 93 +++++++++++++------
 .../DataLakes/DeltaLake/TableSnapshot.cpp     | 31 +++++--
 tests/integration/test_storage_delta/test.py  | 37 ++++++--
 3 files changed, 117 insertions(+), 44 deletions(-)

diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp
index 552b7b25514ef..146878c761656 100644
--- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/EnginePredicate.cpp
@@ -69,6 +69,23 @@ namespace
         const auto * nullable = assert_cast<const DB::DataTypeNullable *>(node->result_type.get());
         return nullable->getNestedType();
     }
+
+    /// Whether a node is a constant with value `true`.
+    /// Such nodes are produced e.g. by ActionsDAG::restrictFilterDAGToInputs,
+    /// which replaces non-computable branches of `and` with constant true,
+    /// or by constant folding of the filter.
+    bool isConstantTrueNode(const DB::ActionsDAG::Node * node)
+    {
+        if (!isConstNode(node) || !node->column)
+            return false;
+
+        /// Conjuncts of `and` in a filter DAG are always (Nullable) UInt8.
+        if (getTypeIndex(node) != DB::TypeIndex::UInt8)
+            return false;
+
+        const auto & column = *node->column;
+        return DB::isColumnConst(column) && column.getBool(0);
+    }
 }
 
 std::shared_ptr<EnginePredicate> getEnginePredicate(
@@ -84,9 +101,11 @@ struct EngineIteratorData
     EngineIteratorData(
         ffi::KernelExpressionVisitorState * state_,
         const DB::ActionsDAG::NodeRawConstPtrs & nodes_,
-        EnginePredicate & predicate_)
+        EnginePredicate & predicate_,
+        bool and_context_ = false)
         : state(state_)
         , predicate(predicate_)
+        , and_context(and_context_)
         , nodes(nodes_)
         , it(nodes.begin())
     {
@@ -101,6 +120,10 @@ struct EngineIteratorData
 
     void markUnsupportedNode() { predicate.has_unsupported_nodes = true; }
 
+    /// Whether this iterator feeds `visit_predicate_and`, i.e. its nodes are
+    /// conjuncts of an AND predicate. Only then is it safe to drop constant true.
+    bool isAndContext() const { return and_context; }
+
     const LoggerPtr & log() const { return predicate.log; }
 
     const DB::ActionsDAG::Node * next()
@@ -112,6 +135,7 @@ struct EngineIteratorData
 
     ffi::KernelExpressionVisitorState * state;
     EnginePredicate & predicate;
+    const bool and_context;
 
 private:
     const DB::ActionsDAG::NodeRawConstPtrs & nodes;
@@ -137,36 +161,48 @@ class  EngineIterator : public ffi::EngineIterator
         auto * iterator_data = static_cast<EngineIteratorData *>(data_);
         try
         {
-            if (iterator_data->hasException())
+            while (true)
             {
-                LOG_TEST(iterator_data->log(), "Exception during processing");
-                return nullptr;
-            }
+                if (iterator_data->hasException())
+                {
+                    LOG_TEST(iterator_data->log(), "Exception during processing");
+                    return nullptr;
+                }
 
-            const auto * node = iterator_data->next();
-            if (!node)
-            {
-                LOG_TEST(iterator_data->log(), "Iterator finished");
-                return nullptr;
-            }
+                const auto * node = iterator_data->next();
+                if (!node)
+                {
+                    LOG_TEST(iterator_data->log(), "Iterator finished");
+                    return nullptr;
+                }
 
-            LOG_TEST(
-                iterator_data->log(),
-                "Node name: {}, node type: {}, column type: {}",
-                node->result_name,
-                node->type,
-                node->column ? toString(node->column->getDataType()) : "None");
+                LOG_TEST(
+                    iterator_data->log(),
+                    "Node name: {}, node type: {}, column type: {}",
+                    node->result_name,
+                    node->type,
+                    node->column ? toString(node->column->getDataType()) : "None");
 
-            auto result = getNextImpl(*iterator_data, node);
-            if (result && result != VISITOR_FAILED_OR_UNSUPPORTED)
-            {
-                return reinterpret_cast<const void *>(result);
-            }
+                /// Constant `true` does not change an AND predicate, so it can be
+                /// dropped instead of failing the translation (see isConstantTrueNode).
+                if (iterator_data->isAndContext() && isConstantTrueNode(node))
+                    continue;
 
-            if (!iterator_data->hasException())
-            {
-                LOG_TEST(iterator_data->log(), "Unsupported node: {}", node->result_name);
-                iterator_data->markUnsupportedNode();
+                auto result = getNextImpl(*iterator_data, node);
+                if (result && result != VISITOR_FAILED_OR_UNSUPPORTED)
+                {
+                    return reinterpret_cast<const void *>(result);
+                }
+
+                if (!iterator_data->hasException())
+                {
+                    LOG_TEST(iterator_data->log(), "Unsupported node: {}", node->result_name);
+                    iterator_data->markUnsupportedNode();
+                }
+
+                /// An unsupported node must not truncate the rest of the conjuncts:
+                /// the kernel treats nullptr as the end of the iterator, so
+                /// returning here would silently drop all the following nodes.
             }
         }
         catch (...)
@@ -183,7 +219,7 @@ class  EngineIterator : public ffi::EngineIterator
 uintptr_t EnginePredicate::visitPredicate(void * data, ffi::KernelExpressionVisitorState * state)
 {
     auto * predicate = static_cast<EnginePredicate *>(data);
-    EngineIteratorData iterator_data(state, predicate->filter.getOutputs(), *predicate);
+    EngineIteratorData iterator_data(state, predicate->filter.getOutputs(), *predicate, /* and_context_ */ true);
     EngineIterator engine_iterator(iterator_data);
     auto result = ffi::visit_predicate_and(state, &engine_iterator);
 
@@ -301,7 +337,8 @@ uintptr_t EngineIterator::getNextImpl(EngineIteratorData & iterator_data, const
                 EngineIteratorData current_iterator_data(
                         iterator_data.state,
                         node->children,
-                        iterator_data.predicate);
+                        iterator_data.predicate,
+                        /* and_context_ */ true);
 
                 EngineIterator current_engine_iterator(current_iterator_data);
 
diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp
index 42c295ee532d4..dd18e4c2727f1 100644
--- a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp
@@ -246,15 +246,32 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
 
                     if (!fallback_exception && !fallback_predicate->hasUnsupportedNodes())
                     {
-                        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 = std::move(fallback_scan);
+                        if (engine_predicate_exception)
+                        {
+                            /// The original scan's predicate may be in an unreliable state
+                            /// after a translation exception; use the restricted predicate.
+                            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, ", "));
+
+                            scan = std::move(fallback_scan);
+                        }
+                        else
+                        {
+                            /// The original scan already contains every translatable conjunct,
+                            /// including all clustering-column conjuncts of the restricted
+                            /// predicate, so keeping it is never weaker than the fallback scan.
+                            LOG_INFO(
+                                log,
+                                "Full predicate translation has unsupported nodes; keeping the original "
+                                "scan, which already includes the clustering-column conjuncts "
+                                "(clustering columns: {})",
+                                fmt::join(clustering_columns, ", "));
+                        }
                     }
                 }
             }
diff --git a/tests/integration/test_storage_delta/test.py b/tests/integration/test_storage_delta/test.py
index fc858942d2f80..3df923a3d394b 100644
--- a/tests/integration/test_storage_delta/test.py
+++ b/tests/integration/test_storage_delta/test.py
@@ -74,6 +74,9 @@ def get_spark(log_dir=None):
             "spark.sql.catalog.spark_catalog.warehouse",
             "/var/lib/clickhouse/user_files/test_storage_delta",
         )
+        .config(
+            "spark.databricks.delta.clusteredTable.enableClusteringTablePreview", "true"
+        )
         .config("spark.driver.memory", "2g")
         .config("spark.executor.memory", "2g")
         .master("local")
@@ -162,7 +165,7 @@ def started_cluster():
             ],
             user_configs=["configs/users.d/users.xml"],
             with_installed_binary=True,
-            image="clickhouse/clickhouse-server",
+            image="altinity/clickhouse-server",
             tag=CLICKHOUSE_CI_MIN_TESTED_VERSION,
             with_minio=True,
             with_azurite=True,
@@ -5651,7 +5654,7 @@ def test_liquid_clustering(started_cluster):
     delta_path = f"/{TABLE_NAME}"
     spark.sql(f"""
         CREATE TABLE delta.`{delta_path}` (a INT, b STRING)
-        USING DELTA CLUSTER BY (a)
+        USING DELTA CLUSTER BY (a, b)
     """)
     spark.sql(f"""
         INSERT INTO delta.`{delta_path}`
@@ -5697,7 +5700,7 @@ def test_liquid_clustering_data_skipping(started_cluster):
     delta_path = f"/{TABLE_NAME}"
     spark.sql(f"""
         CREATE TABLE delta.`{delta_path}` (a INT, b STRING)
-        USING DELTA CLUSTER BY (a)
+        USING DELTA CLUSTER BY (a, b)
     """)
 
     # Insert multiple batches to produce several files after OPTIMIZE.
@@ -5707,6 +5710,18 @@ def test_liquid_clustering_data_skipping(started_cluster):
             SELECT id + {i * 1000} as a, CAST(id + {i * 1000} AS STRING) as b
             FROM range(1000)
         """)
+    # OPTIMIZE targets ~1GB cubes/files by default, which would pack this small
+    # dataset into a single file and leave nothing to skip. Shrink the cube and
+    # file size targets so several files are produced.
+    spark.conf.set("spark.databricks.delta.optimize.maxFileSize", "10240")
+    spark.conf.set(
+        "spark.databricks.delta.optimize.clustering.mergeStrategy.minCubeSize.targetCubeSize",
+        "10240",
+    )
+    spark.conf.set(
+        "spark.databricks.delta.optimize.clustering.mergeStrategy.minCubeSize.threshold",
+        "1024",
+    )
     spark.sql(f"OPTIMIZE delta.`{delta_path}`")
 
     files = default_upload_directory(started_cluster, "s3", delta_path, "")
@@ -5765,14 +5780,16 @@ def test_liquid_clustering_predicate_fallback(started_cluster):
 
     delta_path = f"/{TABLE_NAME}"
     spark.sql(f"""
-        CREATE TABLE delta.`{delta_path}` (a INT, b STRING)
-        USING DELTA CLUSTER BY (a)
+        CREATE TABLE delta.`{delta_path}` (a INT, b STRING, c STRING)
+        USING DELTA CLUSTER BY (a, b)
     """)
 
     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
+            SELECT id + {i * 1000} as a,
+                   CAST(id + {i * 1000} AS STRING) as b,
+                   CAST(id + {i * 1000} + 1 AS STRING) as c
             FROM range(1000)
         """)
     spark.sql(f"OPTIMIZE delta.`{delta_path}`")
@@ -5783,11 +5800,13 @@ def test_liquid_clustering_predicate_fallback(started_cluster):
     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.
+    # referencing non-clustering column `c` (cityHash64 cannot be translated by
+    # the delta-kernel engine predicate). `restrictFilterDAGToInputs` will
+    # replace the non-clustering conjunct with constant true, leaving only
+    # `a = 42` for the fallback predicate to push 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",
+        f"SELECT a FROM {TABLE_NAME} WHERE a = 42 AND cityHash64(c) != 0",
         query_id=fallback_query_id,
     )
     assert result.strip() == "42"