Antalya 26.6 - Read clustering columns from delta lake and use it for data skipping

.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

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

src/Core/Settings.cpp

@@ -7461,6 +7461,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.

src/Core/SettingsChangesHistory.cpp

@@ -48,12 +48,14 @@ 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."},
         });
 
         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`."},

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,13 +54,15 @@ 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
 {
     extern const Event DeltaLakePartitionPrunedFiles;
     extern const Event DeltaLakeSnapshotInitializations;
     extern const Event DeltaLakeScannedFiles;
+    extern const Event DeltaLakeClusteringPredicateFallbacks;
 }
 
 namespace DB::FailPoints
@@ -96,13 +103,15 @@ 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_,
         size_t list_batch_size_,
         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,13 +120,15 @@ 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_)
         , log(log_)
         , 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,13 +679,15 @@ 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;
     const LoggerPtr log;
     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,13 +1007,15 @@ DB::ObjectIterator TableSnapshot::iterate(
         schema->table_schema,
         schema->physical_names_map,
         schema->partition_columns,
+        *clustering_columns,
         object_storage,
         filter_dag,
         callback,
         list_batch_size,
         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

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

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