snapshot caching + incremental loading (#300)

**Support incremental snapshot loading**

When a Delta table is read multiple times within a session, leverage kernel
functionality to load only incremental changes when moving forward in time
(e.g. from v5002 -> v5003). Previously, loading v5003 immediately after loading
v5002 would still require reading metadata for version [0-5002] first. With
this change DuckDB Delta and delta-kernel will begin with cached metadata and only
read and apply new log entries atop that cache.

Incremental/cached loading is not currently supported for backward time travel,
and falls back to a fresh load.

**Also included**

- delta-kernel-rs bumped to v0.21.0 (required for increment load API), includes
  some FFI changes around ffi::Handle wrappers, EngineError and OptionalValue
  support
- build changes avoid downloading 8GB of unused test data in cargo build
- delta kernel logging: allow `SET delta_kernel_logging = false`; the
  underlying kernel event trace will remain active, but duckdb-delta will drop
  inbound events if false
- several test setup/cleanup guards where delta kernel logging state was
  lingering between tests

**Variant encoding**

variant_legacy_encoding is now unconditionally set to true at extension load
time, enabling backwards-compatible parsing of Variant columns written as
STRUCT(value, metadata).

**Test notes**

- test/sql/main/writing/incremental_snapshot.test — verifies that v0→v1→v2
  incremental loads don't re-read earlier log files, and backward time travel
  correctly builds fresh
- test/sql/main/writing/transaction_multi_insert.test — multi-INSERT within a single transaction

Author: benfleis

.github/workflows/MainDistributionPipeline.yml

@@ -19,8 +19,8 @@ jobs:
     with:
       # Main config
       extension_name: &ext_name delta
-      duckdb_version: &duckdb_ref v1.5.1
-      ci_tools_version: &ci_tools_ref v1.5.0
+      duckdb_version: &duckdb_ref v1.5.2
+      ci_tools_version: &ci_tools_ref v1.5-variegata
       enable_rust: true
       rust_logs: true
       exclude_archs: 'wasm_mvp;wasm_eh;wasm_threads;windows_amd64_rtools;windows_amd64_mingw;linux_amd64_musl'

benchmark/micro/snapshot_performance/delta_scan.benchmark

@@ -10,7 +10,7 @@ require delta
 require parquet
 
 run
-SELECT COUNT(*) FROM delta_scan('./data/generated/delta_rs_tpch_sf1_100_splits/lineitem/delta_lake')
+SELECT COUNT(*) FROM delta_scan('./data/generated/tpch_sf1_90commits/lineitem/delta_lake')
 
 result I
-6001215
+6001215

benchmark/micro/snapshot_performance/delta_scan_filter.benchmark

@@ -10,7 +10,7 @@ require delta
 require parquet
 
 run
-SELECT COUNT(*) FROM delta_scan('./data/generated/delta_rs_tpch_sf1_100_splits/lineitem/delta_lake') where l_orderkey is not null
+SELECT COUNT(*) FROM delta_scan('./data/generated/tpch_sf1_90commits/lineitem/delta_lake') where l_orderkey is not null
 
 result I
-6001215
+6001215

benchmark/micro/snapshot_performance/snapshot_no_pin.benchmark

@@ -10,10 +10,10 @@ require delta
 require parquet
 
 load
-ATTACH './data/generated/delta_rs_tpch_sf1_100_splits/lineitem/delta_lake' as lineitem_no_pin (TYPE delta);
+ATTACH './data/generated/tpch_sf1_90commits/lineitem/delta_lake' as lineitem_no_pin (TYPE delta);
 
 run
 SELECT COUNT(*) FROM lineitem_no_pin
 
 result I
-6001215
+6001215

benchmark/micro/snapshot_performance/snapshot_no_pin_filter.benchmark

@@ -10,10 +10,10 @@ require delta
 require parquet
 
 load
-ATTACH './data/generated/delta_rs_tpch_sf1_100_splits/lineitem/delta_lake' as lineitem_no_pin (TYPE delta);
+ATTACH './data/generated/tpch_sf1_90commits/lineitem/delta_lake' as lineitem_no_pin (TYPE delta);
 
 run
 SELECT COUNT(*) FROM lineitem_no_pin where l_orderkey is not null
 
 result I
-6001215
+6001215

benchmark/micro/snapshot_performance/snapshot_pin.benchmark

@@ -10,10 +10,10 @@ require delta
 require parquet
 
 load
-ATTACH './data/generated/delta_rs_tpch_sf1_100_splits/lineitem/delta_lake' as lineitem_pin (TYPE delta, PIN_SNAPSHOT);
+ATTACH './data/generated/tpch_sf1_90commits/lineitem/delta_lake' as lineitem_pin (TYPE delta, PIN_SNAPSHOT);
 
 run
 SELECT COUNT(*) FROM lineitem_pin
 
 result I
-6001215
+6001215

benchmark/micro/snapshot_performance/snapshot_pin_filter.benchmark

@@ -10,10 +10,10 @@ require delta
 require parquet
 
 load
-ATTACH './data/generated/delta_rs_tpch_sf1_100_splits/lineitem/delta_lake' as lineitem_pin (TYPE delta, PIN_SNAPSHOT);
+ATTACH './data/generated/tpch_sf1_90commits/lineitem/delta_lake' as lineitem_pin (TYPE delta, PIN_SNAPSHOT);
 
 run
 SELECT COUNT(*) FROM lineitem_pin where l_orderkey is not null
 
 result I
-6001215
+6001215

scripts/data_generator/delta_rs_generator/__init__.py

@@ -32,7 +32,8 @@ def generate_test_data_delta_rs_multi(base_path, path, init, tables, splits = 1)
 
         # Then we write the parquet files
         for table in tables:
-            total_count = con.sql(f"select count(*) from ({table['query']})").fetchall()[0][0]
+            query = table['query']
+            total_count = con.sql(f"select count(*) from ({query})").fetchall()[0][0]
             # At least 1 tuple per file
             if total_count < splits:
                 splits = total_count

scripts/data_generator/generate_test_data.py

@@ -91,6 +91,13 @@
     for table in ["customer","lineitem","nation","orders","part","partsupp","region","supplier"]:
         con.query(f"create table duckdb_out.{table} as from {table}")
 
+# TPC-H sf with 100 commits
+init = "call dbgen(sf=1);"
+tables = ["customer","lineitem","nation","orders","part","partsupp","region","supplier"]
+queries = [f"from {x}" for x in tables]
+tables = [{'name': x[0], 'query':x[1]} for x in zip(tables,queries)]
+generate_test_data_delta_rs_multi(BASE_PATH, "tpch_sf1_90commits", init, tables, 90)
+
 ################################################
 ### TPC-DS
 ################################################

src/delta_extension.cpp

@@ -14,6 +14,7 @@
 #include "duckdb/main/extension_helper.hpp"
 #include "duckdb/storage/storage_extension.hpp"
 #include "duckdb/main/config.hpp"
+#include "duckdb/catalog/catalog_entry/table_function_catalog_entry.hpp"
 
 namespace duckdb {
 
@@ -106,6 +107,11 @@ static void LoadInternal(ExtensionLoader &loader) {
 	auto &config = DBConfig::GetConfig(loader.GetDatabaseInstance());
 	StorageExtension::Register(config, "delta", make_shared_ptr<DeltaStorageExtension>());
 
+	// NOTE: variant_legacy_encoding now refers to __delta_only_variant_encoding_enabled which is strictly internal
+	// and used to signal parquet variant legacy support. It's not fundamentally optional, and thus controlled here
+	// only.
+	config.options.variant_legacy_encoding = true;
+
 	config.AddExtensionOption("delta_scan_explain_files_filtered",
 	                          "Adds the filtered files to the explain output. Warning: this may impact performance of "
 	                          "delta scan during explain analyze queries.",