Merge pull request ClickHouse#79115 from ClickHouse/update-delta-kernel

Update delta-kernel-rs to latest main

Author: kssenii

contrib/delta-kernel-rs

@@ -108,8 +108,8 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
     {
         scan = KernelUtils::unwrapResult(ffi::scan(snapshot.get(), engine.get(), /* predicate */{}), "scan");
         scan_data_iterator = KernelUtils::unwrapResult(
-            ffi::kernel_scan_data_init(engine.get(), scan.get()),
-            "kernel_scan_data_init");
+            ffi::scan_metadata_iter_init(engine.get(), scan.get()),
+            "scan_metadata_iter_init");
     }
 
     void scanDataFunc()
@@ -118,8 +118,8 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
         while (!shutdown.load())
         {
             bool have_scan_data_res = KernelUtils::unwrapResult(
-                ffi::kernel_scan_data_next(scan_data_iterator.get(), this, visitData),
-                "kernel_scan_data_next");
+                ffi::scan_metadata_next(scan_data_iterator.get(), this, visitData),
+                "scan_metadata_next");
 
             if (have_scan_data_res)
             {
@@ -204,14 +204,10 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
 
     static void visitData(
         void * engine_context,
-        ffi::ExclusiveEngineData * engine_data,
-        const struct ffi::KernelBoolSlice selection_vec,
-        const ffi::CTransforms * transforms)
+        ffi::SharedScanMetadata * scan_metadata)
     {
-        ffi::visit_scan_data(engine_data, selection_vec, transforms, engine_context, Iterator::scanCallback);
-
-        ffi::free_bool_slice(selection_vec);
-        ffi::free_engine_data(engine_data);
+        ffi::visit_scan_metadata(scan_metadata, engine_context, Iterator::scanCallback);
+        ffi::free_scan_metadata(scan_metadata);
     }
 
     static void scanCallback(
@@ -220,6 +216,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
         int64_t size,
         const ffi::Stats * stats,
         const ffi::DvInfo * /* dv_info */,
+        const ffi::Expression * /* transform */,
         const struct ffi::CStringMap * partition_map)
     {
         auto * context = static_cast<TableSnapshot::Iterator *>(engine_context);
@@ -300,7 +297,7 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
 
 private:
     using KernelScan = KernelPointerWrapper<ffi::SharedScan, ffi::free_scan>;
-    using KernelScanDataIterator = KernelPointerWrapper<ffi::SharedScanDataIterator, ffi::free_kernel_scan_data>;
+    using KernelScanDataIterator = KernelPointerWrapper<ffi::SharedScanMetadataIterator, ffi::free_scan_metadata_iter>;
 
     const KernelExternEngine & engine;
     const KernelSnapshot & snapshot;
@@ -395,7 +392,7 @@ void TableSnapshot::initSnapshotImpl() const
     read_schema = getReadSchemaFromSnapshot(scan_state.get());
     LOG_TRACE(log, "Table read schema: {}", fmt::join(read_schema.getNames(), ", "));
 
-    partition_columns = getPartitionColumnsFromSnapshot(scan_state.get());
+    partition_columns = getPartitionColumnsFromSnapshot(snapshot.get());
     LOG_TRACE(log, "Partition columns: {}", fmt::join(partition_columns, ", "));
 }
 

contrib/rust_vendor

@@ -149,13 +149,14 @@ class SchemaVisitorData
  */
 class SchemaVisitor
 {
-    using KernelSharedSchema = KernelPointerWrapper<ffi::SharedSchema, ffi::free_global_read_schema>;
+    using KernelSharedSchema = KernelPointerWrapper<ffi::SharedSchema, ffi::free_schema>;
     using KernelStringSliceIterator = KernelPointerWrapper<ffi::StringSliceIterator, ffi::free_string_slice_data>;
 public:
     static void visitTableSchema(ffi::SharedSnapshot * snapshot, SchemaVisitorData & data)
     {
+        KernelSharedSchema schema(ffi::logical_schema(snapshot));
         auto visitor = createVisitor(data);
-        [[maybe_unused]] size_t result = ffi::visit_snapshot_schema(snapshot, &visitor);
+        [[maybe_unused]] size_t result = ffi::visit_schema(schema.get(), &visitor);
         chassert(result == 0, "Unexpected result: " + DB::toString(result));
     }
 
@@ -170,10 +171,10 @@ class SchemaVisitor
     }
 
     static void visitPartitionColumns(
-        ffi::SharedGlobalScanState * scan_state,
+        ffi::SharedSnapshot * snapshot,
         SchemaVisitorData & data)
     {
-        KernelStringSliceIterator partition_columns_iter(ffi::get_partition_columns(scan_state));
+        KernelStringSliceIterator partition_columns_iter(ffi::get_partition_columns(snapshot));
         while (ffi::string_slice_next(partition_columns_iter.get(), &data, &visitPartitionColumn)) {}
     }
 
@@ -481,10 +482,10 @@ DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedGlobalScanState * sca
     return data.getSchemaResult().names_and_types;
 }
 
-DB::Names getPartitionColumnsFromSnapshot(ffi::SharedGlobalScanState * scan_state)
+DB::Names getPartitionColumnsFromSnapshot(ffi::SharedSnapshot * snapshot)
 {
     SchemaVisitorData data;
-    SchemaVisitor::visitPartitionColumns(scan_state, data);
+    SchemaVisitor::visitPartitionColumns(snapshot, data);
     return data.getPartitionColumns();
 }
 

src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp

@@ -27,7 +27,7 @@ DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedGlobalScanState * sca
 /// Get list of partition columns.
 /// Read schema does not contain partition columns,
 /// therefore partition columns are passed separately.
-DB::Names getPartitionColumnsFromSnapshot(ffi::SharedGlobalScanState * scan_state);
+DB::Names getPartitionColumnsFromSnapshot(ffi::SharedSnapshot * snapshot);
 
 }