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

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

@@ -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.8.20003.altinityantalya",
         {
             {"iceberg_target_data_file_size_bytes", 536870912, 536870912, "Target file size for Iceberg bin-packing compaction (default 512 MiB)."},

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())
     {
@@ -99,6 +118,12 @@ struct EngineIteratorData
 
     bool hasException() const { return predicate.exception != nullptr; }
 
+    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()
@@ -110,6 +135,7 @@ struct EngineIteratorData
 
     ffi::KernelExpressionVisitorState * state;
     EnginePredicate & predicate;
+    const bool and_context;
 
 private:
     const DB::ActionsDAG::NodeRawConstPtrs & nodes;
@@ -135,30 +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;
+
+                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 (...)
@@ -175,11 +219,15 @@ 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);
 
-    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;
 }
 
@@ -289,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);
 

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;