fix(delta): code review hardening + CI workflow registration

Code review follow-ups (#27-#35):
- Replace unsafe `.get` with safe `flatMap(find)` in row-tracking Project builder
- Replace catch-all `Throwable` with `NonFatal` in Delta reflection paths
- Guard synthetic column names against user-table collisions
- Short-circuit row-tracking rewrite when enableRowTracking=false
- Document why per-file PD splitting is correctness-required
- Centralise `PhysicalNameMetadataKey` constant in DeltaReflection
- Use `pd.copy` for Spark 3.x/4.0 cross-compilation
- Make dev scripts honor DELTA_DIR/DELTA_WORKDIR/TMPDIR env vars

Operational improvements:
- Fall back on Parquet encryption (not wired through Delta exec path)
- Recompute DPP pruning at execution time so AQE broadcast results
  actually prune partitions instead of reading all files
- Add CometDeltaRowTrackingSuite (unmaterialised + post-MERGE tests)

CI/DevOps (per PR feedback):
- Register all Delta test suites in pr_build_linux.yml and pr_build_macos.yml
- Add `permissions: contents: read` to delta_regression_test.yml
- Add DeltaReadFromS3Suite to check-suites.py ignore list
- Fix scalastyle locale violations

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: schenksj

.github/workflows/delta_regression_test.yml

@@ -45,6 +45,9 @@ on:
   # manual trigger
   workflow_dispatch:
 
+permissions:
+  contents: read
+
 env:
   RUST_VERSION: stable
   RUST_BACKTRACE: 1

.github/workflows/pr_build_linux.yml

@@ -290,6 +290,11 @@ jobs:
               org.apache.spark.sql.comet.ParquetEncryptionITCase
               org.apache.comet.exec.CometNativeReaderSuite
               org.apache.comet.CometIcebergNativeSuite
+              org.apache.comet.CometDeltaNativeSuite
+              org.apache.comet.CometDeltaColumnMappingSuite
+              org.apache.comet.CometDeltaAdvancedSuite
+              org.apache.comet.CometDeltaRowTrackingSuite
+              org.apache.comet.CometFuzzDeltaSuite
           - name: "csv"
             value: |
               org.apache.comet.csv.CometCsvNativeReadSuite

.github/workflows/pr_build_macos.yml

@@ -171,6 +171,11 @@ jobs:
               org.apache.spark.sql.comet.ParquetEncryptionITCase
               org.apache.comet.exec.CometNativeReaderSuite
               org.apache.comet.CometIcebergNativeSuite
+              org.apache.comet.CometDeltaNativeSuite
+              org.apache.comet.CometDeltaColumnMappingSuite
+              org.apache.comet.CometDeltaAdvancedSuite
+              org.apache.comet.CometDeltaRowTrackingSuite
+              org.apache.comet.CometFuzzDeltaSuite
           - name: "csv"
             value: |
               org.apache.comet.csv.CometCsvNativeReadSuite

dev/ci/check-suites.py

@@ -35,6 +35,7 @@ def file_to_class_name(path: Path) -> str | None:
         "org.apache.comet.parquet.ParquetReadSuite", # abstract
         "org.apache.comet.parquet.ParquetReadFromS3Suite", # manual test suite
         "org.apache.comet.IcebergReadFromS3Suite", # manual test suite
+        "org.apache.comet.DeltaReadFromS3Suite", # manual test suite
         "org.apache.spark.sql.comet.CometPlanStabilitySuite", # abstract
         "org.apache.spark.sql.comet.ParquetDatetimeRebaseSuite", # abstract
         "org.apache.comet.exec.CometColumnarShuffleSuite" # abstract

dev/run-delta-regression-targeted.sh

@@ -30,7 +30,8 @@
 set -uo pipefail
 
 REPO_ROOT="$(cd "$(dirname "$0")/.." && pwd)"
-DELTA_DIR="/private/var/folders/80/_42csh0j4gqgbny139xqwpym0000gn/T/delta-regression-3.3.2"
+DELTA_VERSION="${DELTA_VERSION:-3.3.2}"
+DELTA_DIR="${DELTA_DIR:-${DELTA_WORKDIR:-${TMPDIR:-/tmp}/delta-regression-${DELTA_VERSION}}}"
 export JAVA_HOME="${JAVA_HOME:-$HOME/jdks/jdk-17.0.18+8/Contents/Home}"
 export SPARK_LOCAL_IP=127.0.0.1
 export RUST_BACKTRACE=1

dev/run-delta-test.sh

@@ -26,7 +26,8 @@
 # target/delta-regression-logs/test-<timestamp>.log (relative to this repo).
 set -euo pipefail
 REPO_ROOT="$(cd "$(dirname "$0")/.." && pwd)"
-DELTA_DIR="/private/var/folders/80/_42csh0j4gqgbny139xqwpym0000gn/T/delta-regression-3.3.2"
+DELTA_VERSION="${DELTA_VERSION:-3.3.2}"
+DELTA_DIR="${DELTA_DIR:-${DELTA_WORKDIR:-${TMPDIR:-/tmp}/delta-regression-${DELTA_VERSION}}}"
 export JAVA_HOME="${JAVA_HOME:-$HOME/jdks/jdk-17.0.18+8/Contents/Home}"
 export SPARK_LOCAL_IP=127.0.0.1
 export RUST_BACKTRACE=1

spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala

@@ -84,7 +84,7 @@ class CometSparkSessionExtensions
                 "false")
             }
           } catch {
-            case _: Throwable => // delta-spark not on classpath; ignore
+            case scala.util.control.NonFatal(_) => // delta-spark not on classpath; ignore
           }
           plan
         }

spark/src/main/scala/org/apache/comet/delta/DeltaReflection.scala

@@ -161,6 +161,9 @@ object DeltaReflection extends Logging {
     }
   }
 
+  /** StructField metadata key under which Delta stores the column-mapping physical name. */
+  val PhysicalNameMetadataKey: String = "delta.columnMapping.physicalName"
+
   /** Property key for the physical column name Delta materialises row IDs into. */
   val MaterializedRowIdColumnProp: String =
     "delta.rowTracking.materializedRowIdColumnName"
@@ -379,11 +382,11 @@ object DeltaReflection extends Logging {
         val nil = scala.collection.immutable.Nil
         try Option(m.invoke(location, nil, nil))
         catch {
-          case _: Throwable => None
+          case scala.util.control.NonFatal(_) => None
         }
       }
     } catch {
-      case _: Throwable => None
+      case scala.util.control.NonFatal(_) => None
     }
   }
 

spark/src/main/scala/org/apache/comet/delta/RowTrackingAugmentedFileIndex.scala

@@ -70,6 +70,13 @@ class RowTrackingAugmentedFileIndex(
    * Delegates listing to the underlying FileIndex, then splits each returned `PartitionDirectory`
    * into one-per-file directories, each carrying the original partition values PLUS the per-file
    * baseRowId and defaultRowCommitVersion.
+   *
+   * The per-file split is unavoidable for correctness: `AddFile.baseRowId` is unique per file, so
+   * two files that share a Delta partition cannot share a `PartitionDirectory` once we inject the
+   * per-file synthetic columns. Scheduling parallelism is unaffected -- `FileSourceScanExec`
+   * flattens all PDs' files into `PartitionedFile`s and bin-packs them by `maxSplitBytes`, so PD
+   * granularity only governs how partition values get serialised with each file, not the number
+   * of tasks.
    */
   override def listFiles(
       partitionFilters: scala.collection.Seq[Expression],
@@ -80,7 +87,11 @@ class RowTrackingAugmentedFileIndex(
         val info = infoByFileName.getOrElse(
           fileStatus.getPath.getName,
           DeltaReflection.RowTrackingFileInfo(None, None))
-        PartitionDirectory(augmentPartitionValues(pd.values, info), Seq(fileStatus))
+        // Use `pd.copy(...)` rather than `PartitionDirectory.apply(...)` so this
+        // compiles against both Spark 3.x (files: Seq[FileStatus]) and Spark 4.0
+        // (files: Seq[FileStatusWithMetadata]) without a per-version shim -- we
+        // round-trip the same element type we got from `pd.files`.
+        pd.copy(values = augmentPartitionValues(pd.values, info), files = Seq(fileStatus))
       }
     }
   }

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -404,8 +404,15 @@ case class CometScanRule(session: SparkSession)
     }
     val hadoopConf = Option(hadoopConfOrNull).getOrElse(
       r.sparkSession.sessionState.newHadoopConfWithOptions(r.options))
-    if (encryptionEnabled(hadoopConf) && !isEncryptionConfigSupported(hadoopConf)) {
-      withInfo(scanExec, s"Native Delta scan does not support encryption")
+    // Parquet encryption: the Delta native scan path does not currently thread
+    // `broadcastedHadoopConfForEncryption` / `encryptedFilePaths` through to
+    // `CometExecRDD` the way `CometNativeScanExec` does, so any encrypted read
+    // would fail at decrypt time. Fall back to Spark+Delta whenever encryption
+    // is enabled, regardless of whether the config is otherwise supported.
+    if (encryptionEnabled(hadoopConf)) {
+      withInfo(
+        scanExec,
+        "Native Delta scan does not yet wire Parquet encryption through to executors")
       return None
     }
     if (!isSchemaSupported(scanExec, SCAN_NATIVE_DELTA_COMPAT, r)) {
@@ -493,6 +500,16 @@ case class CometScanRule(session: SparkSession)
     if (!hasRowIdField) return None
 
     val cfg = DeltaReflection.extractMetadataConfiguration(r).getOrElse(Map.empty)
+    // If the table explicitly has row tracking disabled, there is nothing to rewrite -- the
+    // scan references `row_id` / `row_commit_version` but the table isn't producing them.
+    // Fall back rather than spending reflection effort on per-file baseRowId lookups.
+    if (cfg.get("delta.enableRowTracking").exists(_.equalsIgnoreCase("false"))) {
+      withInfo(
+        scanExec,
+        "Native Delta scan: row-tracking columns requested but table has " +
+          "delta.enableRowTracking=false; falling back.")
+      return Some(None)
+    }
     val rowIdPhysical = cfg.get(DeltaReflection.MaterializedRowIdColumnProp)
     val rowVerPhysical = cfg.get(DeltaReflection.MaterializedRowCommitVersionColumnProp)
 
@@ -559,6 +576,27 @@ case class CometScanRule(session: SparkSession)
     }
     val needSynth = includeRowIdSynth || includeRowVerSynth
 
+    // Guard against collisions: if the underlying table already has columns named
+    // the same as any synthetic we are about to introduce, decline native
+    // acceleration rather than silently shadow the user's data.
+    if (needSynth) {
+      import java.util.Locale
+      val existingNames =
+        (r.dataSchema.fieldNames ++ r.partitionSchema.fieldNames)
+          .map(_.toLowerCase(Locale.ROOT))
+          .toSet
+      val syntheticNames = Seq(RowIndexColName, BaseRowIdColName, DefaultRowCommitVersionColName)
+      val collisions =
+        syntheticNames.filter(n => existingNames.contains(n.toLowerCase(Locale.ROOT)))
+      if (collisions.nonEmpty) {
+        withInfo(
+          scanExec,
+          s"Native Delta scan: table has columns that collide with Comet row-tracking " +
+            s"synthetic columns (${collisions.mkString(", ")}); falling back.")
+        return Some(None)
+      }
+    }
+
     val infoByFileName: Map[String, DeltaReflection.RowTrackingFileInfo] =
       if (needSynth) DeltaReflection.extractRowTrackingInfoByFileName(r.location)
       else Map.empty
@@ -633,22 +671,18 @@ case class CometScanRule(session: SparkSession)
     val cometScan = CometScanExec(newScan, session, SCAN_NATIVE_DELTA_COMPAT)
 
     val projectExprs = origOutput.map { a =>
-      renameMap.get(a.name) match {
-        case Some(phys) if a.name.equalsIgnoreCase(RowIdName) && includeRowIdSynth =>
-          val physAttr = baseNewOutput.find(_.name == phys).get
-          // row_id = coalesce(materialized, base_row_id + row_index)
+      renameMap.get(a.name).flatMap(phys => baseNewOutput.find(_.name == phys)) match {
+        case Some(physAttr) if a.name.equalsIgnoreCase(RowIdName) && includeRowIdSynth =>
           val synth = Add(baseRowIdAttr, rowIndexAttr)
           Alias(Coalesce(Seq(physAttr, synth)), a.name)(
             exprId = a.exprId,
             qualifier = a.qualifier)
-        case Some(phys) if a.name.equalsIgnoreCase(RowCommitVersionName) && includeRowVerSynth =>
-          val physAttr = baseNewOutput.find(_.name == phys).get
-          // row_commit_version = coalesce(materialized, default_row_commit_version)
+        case Some(physAttr)
+            if a.name.equalsIgnoreCase(RowCommitVersionName) && includeRowVerSynth =>
           Alias(Coalesce(Seq(physAttr, defaultVerAttr)), a.name)(
             exprId = a.exprId,
             qualifier = a.qualifier)
-        case Some(phys) =>
-          val physAttr = baseNewOutput.find(_.name == phys).get
+        case Some(physAttr) =>
           Alias(physAttr, a.name)(exprId = a.exprId, qualifier = a.qualifier)
         case None => a
       }