Merge pull request #81707 from ClickHouse/bump-delta-kernel

thevar1able · web-flow · commit 50f1402abbca · 2025-06-13T06:12:40.000Z
Bump `delta-kernel-rs` to v0.12.1
diff --git a/contrib/delta-kernel-rs b/contrib/delta-kernel-rs
@@ -1 +1 @@
-Subproject commit 1ee29c80f31547ea3cee2fc453c12b618ef35882
+Subproject commit beb2eb808e8c7ae6992c99809331b7cc2ed85278
diff --git a/contrib/delta-kernel-rs-cmake/CMakeLists.txt b/contrib/delta-kernel-rs-cmake/CMakeLists.txt
@@ -22,10 +22,9 @@ endif()
 
 # Ideally we would disable all features to remove dependencies but:
 # * default-engine: Needed for its s3 client (we can't pass our own s3 client via ffi)
-# * delta_kernel/cloud: s3 client
 clickhouse_import_crate(
     MANIFEST_PATH "${DELTA_KERNEL_RS_SOURCE_DIR}/ffi/Cargo.toml"
-    FEATURES "default-engine, delta_kernel/cloud"
+    FEATURES "default-engine"
 )
 clickhouse_config_crate_flags(delta_kernel_ffi)
 
diff --git a/contrib/rust_vendor b/contrib/rust_vendor
@@ -1 +1 @@
-Subproject commit 3f5cbcac0698db449b5174236dfd7040a5ca1a4d
+Subproject commit f2d7604c8857ea1faa68d7220c67687a2ae84720
diff --git a/rust/workspace/Cargo.lock b/rust/workspace/Cargo.lock
diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp
@@ -17,6 +17,7 @@
 #include "getSchemaFromSnapshot.h"
 #include "PartitionPruner.h"
 #include "KernelUtils.h"
+#include <delta_kernel_ffi.hpp>
 #include <fmt/ranges.h>
 
 namespace fs = std::filesystem;
@@ -386,13 +387,13 @@ void TableSnapshot::initSnapshotImpl() const
     LOG_TRACE(log, "Snapshot version: {}", snapshot_version);
 
     scan = KernelUtils::unwrapResult(ffi::scan(snapshot.get(), engine.get(), /* predicate */{}), "scan");
-    scan_state = ffi::get_global_scan_state(scan.get());
+
     LOG_TRACE(log, "Initialized scan state");
 
     std::tie(table_schema, physical_names_map) = getTableSchemaFromSnapshot(snapshot.get());
     LOG_TRACE(log, "Table logical schema: {}", fmt::join(table_schema.getNames(), ", "));
 
-    read_schema = getReadSchemaFromSnapshot(scan_state.get());
+    read_schema = getReadSchemaFromSnapshot(scan.get());
     LOG_TRACE(log, "Table read schema: {}", fmt::join(read_schema.getNames(), ", "));
 
     partition_columns = getPartitionColumnsFromSnapshot(snapshot.get());
diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h b/src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.h
@@ -61,7 +61,6 @@ class TableSnapshot
     using KernelExternEngine = KernelPointerWrapper<ffi::SharedExternEngine, ffi::free_engine>;
     using KernelSnapshot = KernelPointerWrapper<ffi::SharedSnapshot, ffi::free_snapshot>;
     using KernelScan = KernelPointerWrapper<ffi::SharedScan, ffi::free_scan>;
-    using KernelGlobalScanState = KernelPointerWrapper<ffi::SharedGlobalScanState, ffi::free_global_scan_state>;
 
     const KernelHelperPtr helper;
     const DB::ObjectStoragePtr object_storage;
@@ -70,7 +69,6 @@ class TableSnapshot
     mutable KernelExternEngine engine;
     mutable KernelSnapshot snapshot;
     mutable KernelScan scan;
-    mutable KernelGlobalScanState scan_state;
     mutable size_t snapshot_version;
 
     mutable DB::NamesAndTypesList table_schema;
diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/getSchemaFromSnapshot.cpp b/src/Storages/ObjectStorage/DataLakes/DeltaLake/getSchemaFromSnapshot.cpp
@@ -137,8 +137,6 @@ class SchemaVisitorData
     const LoggerPtr log = getLogger("SchemaVisitor");
 
     using KernelScan = KernelPointerWrapper<ffi::SharedScan, ffi::free_scan>;
-    using KernelGlobalScanState = KernelPointerWrapper<ffi::SharedGlobalScanState, ffi::free_global_scan_state>;
-
 };
 
 /**
@@ -161,10 +159,10 @@ class SchemaVisitor
     }
 
     static void visitReadSchema(
-        ffi::SharedGlobalScanState * scan_state,
+        ffi::SharedScan * scan,
         SchemaVisitorData & data)
     {
-        KernelSharedSchema schema(ffi::get_global_read_schema(scan_state));
+        KernelSharedSchema schema(ffi::scan_physical_schema(scan));
         auto visitor = createVisitor(data);
         [[maybe_unused]] size_t result = ffi::visit_schema(schema.get(), &visitor);
         chassert(result == 0, "Unexpected result: " + DB::toString(result));
@@ -475,10 +473,10 @@ std::pair<DB::NamesAndTypesList, DB::NameToNameMap> getTableSchemaFromSnapshot(f
     return {result.names_and_types, result.physical_names_map};
 }
 
-DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedGlobalScanState * scan_state)
+DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedScan * scan)
 {
     SchemaVisitorData data;
-    SchemaVisitor::visitReadSchema(scan_state, data);
+    SchemaVisitor::visitReadSchema(scan, data);
     return data.getSchemaResult().names_and_types;
 }
 
diff --git a/src/Storages/ObjectStorage/DataLakes/DeltaLake/getSchemaFromSnapshot.h b/src/Storages/ObjectStorage/DataLakes/DeltaLake/getSchemaFromSnapshot.h
@@ -5,6 +5,7 @@
 #if USE_DELTA_KERNEL_RS
 
 #include <Core/NamesAndTypes.h>
+#include <delta_kernel_ffi.hpp>
 
 namespace ffi
 {
@@ -22,7 +23,7 @@ std::pair<DB::NamesAndTypesList, DB::NameToNameMap> getTableSchemaFromSnapshot(f
 
 /// Get read schema.
 /// Represents read schema based on data files.
-DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedGlobalScanState * scan_state);
+DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedScan * scan);
 
 /// Get list of partition columns.
 /// Read schema does not contain partition columns,
diff --git a/tests/integration/test_storage_delta/test.py b/tests/integration/test_storage_delta/test.py
@@ -1332,7 +1332,7 @@ def test_session_token(started_cluster):
     )
 
     assert (
-        "Received DeltaLake kernel error ReqwestError: Error interacting with object store"
+        "Received DeltaLake kernel error ObjectStoreError: Error interacting with object store"
         in instance2.query_and_get_error(
             f"""
     SELECT count() FROM deltaLake(
diff --git a/utils/list-licenses/list-licenses.sh b/utils/list-licenses/list-licenses.sh

Original file line number	Diff line number	Diff line change
`@@ -137,8 +137,6 @@ class SchemaVisitorData`
`137`	`137`	`const LoggerPtr log = getLogger("SchemaVisitor");`
`138`	`138`
`139`	`139`	`using KernelScan = KernelPointerWrapper<ffi::SharedScan, ffi::free_scan>;`
`140`		`- using KernelGlobalScanState = KernelPointerWrapper<ffi::SharedGlobalScanState, ffi::free_global_scan_state>;`
`141`		`-`
`142`	`140`	`};`
`143`	`141`
`144`	`142`	`/**`
`@@ -161,10 +159,10 @@ class SchemaVisitor`
`161`	`159`	`}`
`162`	`160`
`163`	`161`	`static void visitReadSchema(`
`164`		`- ffi::SharedGlobalScanState * scan_state,`
	`162`	`+ ffi::SharedScan * scan,`
`165`	`163`	`SchemaVisitorData & data)`
`166`	`164`	`{`
`167`		`- KernelSharedSchema schema(ffi::get_global_read_schema(scan_state));`
	`165`	`+ KernelSharedSchema schema(ffi::scan_physical_schema(scan));`
`168`	`166`	`auto visitor = createVisitor(data);`
`169`	`167`	`[[maybe_unused]] size_t result = ffi::visit_schema(schema.get(), &visitor);`
`170`	`168`	`chassert(result == 0, "Unexpected result: " + DB::toString(result));`
`@@ -475,10 +473,10 @@ std::pair<DB::NamesAndTypesList, DB::NameToNameMap> getTableSchemaFromSnapshot(f`
`475`	`473`	`return {result.names_and_types, result.physical_names_map};`
`476`	`474`	`}`
`477`	`475`
`478`		`-DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedGlobalScanState * scan_state)`
	`476`	`+DB::NamesAndTypesList getReadSchemaFromSnapshot(ffi::SharedScan * scan)`
`479`	`477`	`{`
`480`	`478`	`SchemaVisitorData data;`
`481`		`- SchemaVisitor::visitReadSchema(scan_state, data);`
	`479`	`+ SchemaVisitor::visitReadSchema(scan, data);`
`482`	`480`	`return data.getSchemaResult().names_and_types;`
`483`	`481`	`}`
`484`	`482`
Original file line number	Diff line number	Diff line change
`@@ -1332,7 +1332,7 @@ def test_session_token(started_cluster):`
`1332`	`1332`	`)`
`1333`	`1333`
`1334`	`1334`	`assert (`
`1335`		`- "Received DeltaLake kernel error ReqwestError: Error interacting with object store"`
	`1335`	`+ "Received DeltaLake kernel error ObjectStoreError: Error interacting with object store"`
`1336`	`1336`	`in instance2.query_and_get_error(`
`1337`	`1337`	`f"""`
`1338`	`1338`	`SELECT count() FROM deltaLake(`