feat: Native Delta Lake scan via delta-kernel-rs

.github/workflows/delta_spark_test.yml

@@ -0,0 +1,133 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Delta Lake Native Scan Tests
+
+concurrency:
+  group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }}
+  cancel-in-progress: true
+
+on:
+  push:
+    branches:
+      - main
+    paths-ignore:
+      - "benchmarks/**"
+      - "doc/**"
+      - "docs/**"
+      - "**.md"
+      - "native/core/benches/**"
+      - "native/spark-expr/benches/**"
+      - "spark/src/test/scala/org/apache/spark/sql/benchmark/**"
+  pull_request:
+    paths-ignore:
+      - "benchmarks/**"
+      - "doc/**"
+      - "docs/**"
+      - "**.md"
+      - "native/core/benches/**"
+      - "native/spark-expr/benches/**"
+      - "spark/src/test/scala/org/apache/spark/sql/benchmark/**"
+  workflow_dispatch:
+
+env:
+  RUST_VERSION: stable
+  RUST_BACKTRACE: 1
+
+jobs:
+  build-native:
+    name: Build Native Library
+    runs-on: ubuntu-24.04
+    container:
+      image: amd64/rust
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Setup Rust & Java toolchain
+        uses: ./.github/actions/setup-builder
+        with:
+          rust-version: ${{ env.RUST_VERSION }}
+          jdk-version: 17
+
+      - name: Restore Cargo cache
+        uses: actions/cache/restore@v5
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            native/target
+          key: ${{ runner.os }}-cargo-ci-${{ hashFiles('native/**/Cargo.lock', 'native/**/Cargo.toml') }}-${{ hashFiles('native/**/*.rs') }}
+          restore-keys: |
+            ${{ runner.os }}-cargo-ci-${{ hashFiles('native/**/Cargo.lock', 'native/**/Cargo.toml') }}-
+
+      - name: Build native library
+        run: |
+          cd native && cargo build --profile ci
+        env:
+          RUSTFLAGS: "-Ctarget-cpu=x86-64-v3"
+
+      - name: Save Cargo cache
+        uses: actions/cache/save@v5
+        if: github.ref == 'refs/heads/main'
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            native/target
+          key: ${{ runner.os }}-cargo-ci-${{ hashFiles('native/**/Cargo.lock', 'native/**/Cargo.toml') }}-${{ hashFiles('native/**/*.rs') }}
+
+      - name: Upload native library
+        uses: actions/upload-artifact@v7
+        with:
+          name: native-lib-delta
+          path: native/target/ci/libcomet.so
+          retention-days: 1
+
+  delta-native-suite:
+    needs: build-native
+    strategy:
+      matrix:
+        os: [ubuntu-24.04]
+        java-version: [17]
+        spark-version:
+          - {short: '3.4', full: '3.4.3'}
+          - {short: '3.5', full: '3.5.8'}
+          - {short: '4.0', full: '4.0.1'}
+      fail-fast: false
+    name: delta-native/${{ matrix.os }}/spark-${{ matrix.spark-version.full }}/java-${{ matrix.java-version }}
+    runs-on: ${{ matrix.os }}
+    container:
+      image: amd64/rust
+    env:
+      SPARK_LOCAL_IP: localhost
+    steps:
+      - uses: actions/checkout@v6
+      - name: Setup Rust & Java toolchain
+        uses: ./.github/actions/setup-builder
+        with:
+          rust-version: ${{ env.RUST_VERSION }}
+          jdk-version: ${{ matrix.java-version }}
+      - name: Download native library
+        uses: actions/download-artifact@v8
+        with:
+          name: native-lib-delta
+          path: native/target/debug/
+      - name: Run CometDeltaNativeSuite
+        run: |
+          ./mvnw -Pspark-${{ matrix.spark-version.short }} -pl spark -am test \
+            -Dsuites=org.apache.comet.CometDeltaNativeSuite \
+            -Dmaven.gitcommitid.skip

benchmarks/tpc/create-delta-tables.py

@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Convert TPC-H or TPC-DS Parquet data to Delta Lake tables.
+
+Usage:
+    spark-submit \
+        --packages io.delta:delta-spark_2.12:3.3.2 \
+        --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
+        --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
+        create-delta-tables.py \
+        --benchmark tpch \
+        --parquet-path /path/to/tpch/parquet \
+        --warehouse /path/to/delta-warehouse
+
+    spark-submit \
+        --packages io.delta:delta-spark_2.12:3.3.2 \
+        --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
+        --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
+        create-delta-tables.py \
+        --benchmark tpcds \
+        --parquet-path /path/to/tpcds/parquet \
+        --warehouse /path/to/delta-warehouse
+"""
+
+import argparse
+import os
+import sys
+
+from pyspark.sql import SparkSession
+
+
+TPCH_TABLES = [
+    "customer", "lineitem", "nation", "orders",
+    "part", "partsupp", "region", "supplier"
+]
+
+TPCDS_TABLES = [
+    "call_center", "catalog_page", "catalog_returns", "catalog_sales",
+    "customer", "customer_address", "customer_demographics", "date_dim",
+    "household_demographics", "income_band", "inventory", "item",
+    "promotion", "reason", "ship_mode", "store", "store_returns",
+    "store_sales", "time_dim", "warehouse", "web_page", "web_returns",
+    "web_sales", "web_site"
+]
+
+
+def create_delta_tables(spark, benchmark, parquet_path, warehouse):
+    tables = TPCH_TABLES if benchmark == "tpch" else TPCDS_TABLES
+
+    for table_name in tables:
+        input_path = os.path.join(parquet_path, table_name)
+        output_path = os.path.join(warehouse, table_name)
+
+        if not os.path.exists(input_path) and not input_path.startswith("s3"):
+            print(f"  Skipping {table_name}: {input_path} does not exist")
+            continue
+
+        print(f"  Converting {table_name}: {input_path} -> {output_path}")
+        df = spark.read.parquet(input_path)
+        df.write.format("delta").mode("overwrite").save(output_path)
+        print(f"    {table_name}: {df.count()} rows written")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Convert TPC Parquet data to Delta Lake tables"
+    )
+    parser.add_argument(
+        "--benchmark", required=True, choices=["tpch", "tpcds"],
+        help="Which TPC benchmark to convert"
+    )
+    parser.add_argument(
+        "--parquet-path", required=True,
+        help="Path to the TPC Parquet data directory"
+    )
+    parser.add_argument(
+        "--warehouse", required=True,
+        help="Path to the Delta warehouse directory"
+    )
+    args = parser.parse_args()
+
+    spark = SparkSession.builder \
+        .appName(f"Create Delta {args.benchmark.upper()} Tables") \
+        .getOrCreate()
+
+    print(f"Converting {args.benchmark.upper()} tables from Parquet to Delta...")
+    create_delta_tables(spark, args.benchmark, args.parquet_path, args.warehouse)
+    print("Done.")
+
+    spark.stop()
+
+
+if __name__ == "__main__":
+    main()

benchmarks/tpc/engines/comet-delta-hashjoin.toml

@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[engine]
+name = "comet-delta-hashjoin"
+
+[env]
+required = ["COMET_JAR", "DELTA_JAR", "DELTA_WAREHOUSE"]
+
+[spark_submit]
+jars = ["$COMET_JAR", "$DELTA_JAR"]
+driver_class_path = ["$COMET_JAR", "$DELTA_JAR"]
+
+[spark_conf]
+"spark.driver.extraClassPath" = "$COMET_JAR:$DELTA_JAR"
+"spark.executor.extraClassPath" = "$COMET_JAR:$DELTA_JAR"
+"spark.plugins" = "org.apache.spark.CometPlugin"
+"spark.shuffle.manager" = "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager"
+"spark.comet.expression.Cast.allowIncompatible" = "true"
+"spark.comet.enabled" = "true"
+"spark.comet.exec.enabled" = "true"
+"spark.comet.exec.shuffle.enabled" = "true"
+"spark.comet.scan.deltaNative.enabled" = "true"
+"spark.comet.explainFallback.enabled" = "true"
+"spark.sql.extensions" = "io.delta.sql.DeltaSparkSessionExtension"
+"spark.sql.catalog.spark_catalog" = "org.apache.spark.sql.delta.catalog.DeltaCatalog"
+"spark.sql.join.preferSortMergeJoin" = "false"
+"spark.sql.autoBroadcastJoinThreshold" = "-1"
+
+[tpcbench_args]
+data_path = "$DELTA_WAREHOUSE"

benchmarks/tpc/engines/comet-delta.toml

@@ -0,0 +1,42 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[engine]
+name = "comet-delta"
+
+[env]
+required = ["COMET_JAR", "DELTA_JAR", "DELTA_WAREHOUSE"]
+
+[spark_submit]
+jars = ["$COMET_JAR", "$DELTA_JAR"]
+driver_class_path = ["$COMET_JAR", "$DELTA_JAR"]
+
+[spark_conf]
+"spark.driver.extraClassPath" = "$COMET_JAR:$DELTA_JAR"
+"spark.executor.extraClassPath" = "$COMET_JAR:$DELTA_JAR"
+"spark.plugins" = "org.apache.spark.CometPlugin"
+"spark.shuffle.manager" = "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager"
+"spark.comet.expression.Cast.allowIncompatible" = "true"
+"spark.comet.enabled" = "true"
+"spark.comet.exec.enabled" = "true"
+"spark.comet.scan.deltaNative.enabled" = "true"
+"spark.comet.explainFallback.enabled" = "true"
+"spark.sql.extensions" = "io.delta.sql.DeltaSparkSessionExtension"
+"spark.sql.catalog.spark_catalog" = "org.apache.spark.sql.delta.catalog.DeltaCatalog"
+
+[tpcbench_args]
+data_path = "$DELTA_WAREHOUSE"

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -116,6 +116,7 @@ object CometConf extends ShimCometConf {
 
   val SCAN_NATIVE_DATAFUSION = "native_datafusion"
   val SCAN_NATIVE_ICEBERG_COMPAT = "native_iceberg_compat"
+  val SCAN_NATIVE_DELTA_COMPAT = "native_delta_compat"
   val SCAN_AUTO = "auto"
 
   val COMET_NATIVE_SCAN_IMPL: ConfigEntry[String] = conf("spark.comet.scan.impl")
@@ -153,6 +154,38 @@ object CometConf extends ShimCometConf {
       .checkValue(v => v > 0, "Data file concurrency limit must be positive")
       .createWithDefault(1)
 
+  val COMET_DELTA_NATIVE_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.scan.deltaNative.enabled")
+      .category(CATEGORY_SCAN)
+      .doc(
+        "Whether to enable native Delta Lake table scan using delta-kernel-rs. When " +
+          "enabled, Delta tables are read through Comet's native Parquet reader, with " +
+          "the transaction log replayed by delta-kernel-rs on the driver. Bypasses " +
+          "Spark's DataSource V1/V2 paths for better performance.")
+      .booleanConf
+      .createWithDefault(false)
+
+  val COMET_DELTA_DATA_FILE_CONCURRENCY_LIMIT: ConfigEntry[Int] =
+    conf("spark.comet.scan.deltaNative.dataFileConcurrencyLimit")
+      .category(CATEGORY_SCAN)
+      .doc(
+        "The number of Delta data files to read concurrently within a single task. " +
+          "Higher values improve throughput for tables with many small files by overlapping " +
+          "I/O latency, but increase memory usage. Values between 2 and 8 are suggested.")
+      .intConf
+      .checkValue(v => v > 0, "Data file concurrency limit must be positive")
+      .createWithDefault(1)
+
+  val COMET_DELTA_FALLBACK_ON_UNSUPPORTED_FEATURE: ConfigEntry[Boolean] =
+    conf("spark.comet.scan.deltaNative.fallbackOnUnsupportedFeature")
+      .category(CATEGORY_SCAN)
+      .doc(
+        "When true, Delta tables declaring a reader feature that Comet does not yet " +
+          "support (e.g. rowTracking, variantType) are tagged for fallback to Spark's " +
+          "regular path rather than failing the query. Turn off only for testing.")
+      .booleanConf
+      .createWithDefault(true)
+
   val COMET_CSV_V2_NATIVE_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.scan.csv.v2.enabled")
       .category(CATEGORY_TESTING)