fix: address all remaining code review items

Fixes all 14 previously-deferred review findings:

#4  Case-sensitivity in DV column detection: isDeltaDvFilterPattern
    and findAndStripDeltaScanBelow now use equalsIgnoreCase for the
    __delta_internal_is_row_deleted column name match.

#8  S3 key documentation: added comment in JNI documenting the
    Hadoop-style key names that storageOptions carries and how
    extract_storage_config maps them.

#10 Proto comment inaccuracy: updated reserved field number comments
    to describe purpose rather than referencing (now-stale) phase
    numbers. Added field numbering strategy note on DeltaScanCommon.

#11 Module quarantine docs: updated delta/mod.rs doc comment to note
    that create_object_store returns Arc<dyn ObjectStore> from
    object_store_kernel 0.12, and that it never escapes the module.
    Updated public API listing to match current exports.

#12 Optimizer rule double-init: added synchronized double-checked
    locking on the CometDeltaDvConfigRule to prevent concurrent
    threads from racing on the config set.

#14 Incomplete partition type support: castPartitionString now throws
    IllegalArgumentException for unsupported types (STRUCT, ARRAY,
    MAP, etc.) instead of silently converting to UTF8String.

#6  DV materialization clarity: added comment explaining why
    .unwrap_or_default() is safe (get_row_indexes returns Ok(None)
    only if has_vector() lied, which kernel guarantees doesn't happen;
    Err propagates via ?).

#17 Consistent JNI null handling: extracted read_string_array helper
    for reading Java String[] into Vec<String>, consolidating the
    null-check + iteration pattern.

#18-19 Proto field ordering: added numbering strategy comment to
    DeltaScanCommon and DeltaScanTask messages.

#20 Memory note: added comment about potential driver OOM on
    extremely large tables (millions of files) with suggestion for
    future streaming/chunked processing.

Tests: succeeded 35, failed 0, canceled 0, ignored 0, pending 0

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

Author: schenksj

native/core/src/delta/jni.rs

@@ -80,18 +80,12 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_planDeltaScan(
         };
 
         // Phase 2: read column names for BoundReference resolution.
-        let col_names: Vec<String> = if column_names.is_null() {
-            Vec::new()
-        } else {
-            let len = column_names.len(env)?;
-            let mut names = Vec::with_capacity(len);
-            for i in 0..len {
-                let obj = column_names.get_element(env, i)?;
-                let jstr = unsafe { JString::from_raw(env, obj.into_raw()) };
-                names.push(jstr.try_to_string(env)?);
-            }
-            names
-        };
+        // storageOptions map carries Hadoop-style keys (fs.s3a.access.key,
+        // fs.s3a.secret.key, fs.s3a.endpoint, fs.s3a.path.style.access,
+        // fs.s3a.endpoint.region, fs.s3a.session.token) extracted by
+        // NativeConfig.extractObjectStoreOptions on the Scala side.
+        // extract_storage_config below maps these to kernel's DeltaStorageConfig.
+        let col_names = read_string_array(env, &column_names)?;
 
         // Phase 2: deserialize the Catalyst predicate (if provided) for
         // kernel's stats-based file pruning. Empty bytes = no predicate.
@@ -232,6 +226,24 @@ fn extract_storage_config(
     })
 }
 
+/// Read a Java `String[]` into a `Vec<String>`. Returns empty vec for null arrays.
+fn read_string_array(
+    env: &mut Env,
+    arr: &jni::objects::JObjectArray,
+) -> CometResult<Vec<String>> {
+    if arr.is_null() {
+        return Ok(Vec::new());
+    }
+    let len = arr.len(env)?;
+    let mut result = Vec::with_capacity(len);
+    for i in 0..len {
+        let obj = arr.get_element(env, i)?;
+        let jstr = unsafe { JString::from_raw(env, obj.into_raw()) };
+        result.push(jstr.try_to_string(env)?);
+    }
+    Ok(result)
+}
+
 /// `map.get(key)` for a `java.util.Map<String, String>` surfaced as a
 /// `JMap`. Returns `None` if the key is absent or the value is `null`.
 fn map_get_string(

native/core/src/delta/mod.rs

@@ -20,19 +20,24 @@
 //! This module is intentionally quarantined: it imports `delta_kernel`,
 //! `object_store_kernel` (object_store 0.12), and kernel's own `arrow` (57),
 //! which live in a *separate* dep subtree from Comet's `arrow 58` /
-//! `object_store 0.13`. Nothing typed by those crates leaves this module —
+//! `object_store 0.13`. Nothing typed by those crates leaves this module:
 //! only plain Rust types (`String`, `i64`, `HashMap<String,String>`, our own
 //! `DeltaFileEntry`, etc.) cross the boundary to the rest of Comet.
 //!
-//! The public API is deliberately small:
-//!   - `DeltaStorageConfig`: credential struct (mirrors tantivy4java).
-//!   - `list_delta_files`: log-replay driver that returns the active file
-//!     list for a given snapshot version.
-//!   - `DeltaFileEntry`: per-file metadata with partition values + DV flag +
-//!     record-count stats.
+//! Note: `create_object_store()` returns `Arc<dyn ObjectStore>` from the
+//! `object_store_kernel` crate (0.12). This type only travels within this
+//! module as an opaque trait object fed to kernel's `DefaultEngine`; callers
+//! outside the delta module never see it or downcast it.
 //!
-//! Later phases add expression translation, deletion-vector decoding, and
-//! the `DeltaScanExec` physical operator; those live in sibling files.
+//! Public API:
+//!   - `DeltaStorageConfig`: credential struct for S3/Azure/local access.
+//!   - `plan_delta_scan` / `list_delta_files`: log-replay drivers that
+//!     return the active file list, DV indexes, column mappings, and
+//!     unsupported-feature flags for a given snapshot version.
+//!   - `DeltaFileEntry` / `DeltaScanPlan`: per-file metadata with
+//!     partition values, deleted row indexes, and record-count stats.
+//!   - `catalyst_to_kernel_predicate_with_names`: Catalyst-proto Expr to
+//!     kernel Predicate translator for stats-based file pruning.
 
 pub mod engine;
 pub mod error;

native/core/src/delta/scan.rs

@@ -234,8 +234,16 @@ pub fn plan_delta_scan_with_predicate(
     // and the various storage-type variants transparently. This runs on the
     // driver (same process that's building the scan plan), so we only pay
     // the DV-fetch latency once per query.
+    //
+    // Note: for very large tables (millions of files), this collects all
+    // entries into memory before returning. Consider streaming/chunked
+    // processing if driver OOM becomes an issue at extreme scale.
     let mut entries: Vec<DeltaFileEntry> = Vec::with_capacity(raw.len());
     for r in raw {
+        // get_row_indexes returns Ok(Some(Vec<u64>)) when a DV is present,
+        // Ok(None) when has_vector() lied (shouldn't happen), or Err on I/O
+        // failure. The `?` propagates I/O errors; `unwrap_or_default` handles
+        // the None case defensively (empty = no deletions, safe fallback).
         let deleted_row_indexes = if r.dv_info.has_vector() {
             r.dv_info
                 .get_row_indexes(&engine, &table_root_url)?

native/proto/src/proto/operator.proto

@@ -255,6 +255,10 @@ message IcebergDeleteFile {
 // Common data shared by all partitions in split mode for a Delta native scan
 // (sent once at planning time). Mirrors NativeScanCommon/IcebergScanCommon so
 // the same downstream ParquetSource wiring can be reused.
+//
+// Field numbering note: fields are grouped by concern, not sequential.
+// New fields should use the next available number at the END of the
+// message to maintain wire compatibility with existing serialized data.
 message DeltaScanCommon {
   // Required schema (projection applied to data schema)
   repeated SparkStructField required_schema = 1;
@@ -308,9 +312,9 @@ message DeltaScanTask {
   // Empty means the file has no DV in use. Row indexes are 0-based into the file's
   // physical parquet row space (same semantics as `DvInfo::get_row_indexes`).
   repeated uint64 deleted_row_indexes = 5;
-  // Reserved for future phases:
-  //   6: residual_predicate_idx (Phase 5)
-  //   7: physical_schema_idx (Phase 4 — column mapping)
+  // Reserved for future use:
+  //   6: per-task residual predicate index (predicates kernel couldn't fully evaluate)
+  //   7: per-task physical schema index (for per-file schema evolution)
 }
 
 // Partition column value extracted from a Delta add action. Delta represents

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

@@ -64,19 +64,25 @@ class CometSparkSessionExtensions
     extensions.injectOptimizerRule { session =>
       new org.apache.spark.sql.catalyst.rules.Rule[
         org.apache.spark.sql.catalyst.plans.logical.LogicalPlan] {
+        private val configLock = new Object()
         @volatile private var configured = false
         override def apply(plan: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan)
             : org.apache.spark.sql.catalyst.plans.logical.LogicalPlan = {
           if (!configured) {
-            try {
-              if (CometConf.COMET_DELTA_NATIVE_ENABLED.get(session.sessionState.conf)) {
-                session.conf
-                  .set("spark.databricks.delta.deletionVectors.useMetadataRowIndex", "false")
+            configLock.synchronized {
+              if (!configured) {
+                try {
+                  if (CometConf.COMET_DELTA_NATIVE_ENABLED.get(session.sessionState.conf)) {
+                    session.conf.set(
+                      "spark.databricks.delta.deletionVectors.useMetadataRowIndex",
+                      "false")
+                  }
+                } catch {
+                  case _: Throwable => // delta-spark not on classpath; ignore
+                }
+                configured = true
               }
-            } catch {
-              case _: Throwable => // delta-spark not on classpath; ignore
             }
-            configured = true
           }
           plan
         }

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

@@ -183,14 +183,16 @@ case class CometScanRule(session: SparkSession)
   }
 
   /** Matches `__delta_internal_is_row_deleted = 0` (the filter Delta injects). */
-  private def isDeltaDvFilterPattern(cond: Expression): Boolean = cond match {
-    case EqualTo(attr: AttributeReference, lit: Literal)
-        if attr.name == DeltaReflection.IsRowDeletedColumnName =>
-      lit.value != null && lit.value.toString == "0"
-    case EqualTo(lit: Literal, attr: AttributeReference)
-        if attr.name == DeltaReflection.IsRowDeletedColumnName =>
-      lit.value != null && lit.value.toString == "0"
-    case _ => false
+  private def isDeltaDvFilterPattern(cond: Expression): Boolean = {
+    def isRowDeletedRef(name: String): Boolean =
+      name.equalsIgnoreCase(DeltaReflection.IsRowDeletedColumnName)
+    cond match {
+      case EqualTo(attr: AttributeReference, lit: Literal) if isRowDeletedRef(attr.name) =>
+        lit.value != null && lit.value.toString == "0"
+      case EqualTo(lit: Literal, attr: AttributeReference) if isRowDeletedRef(attr.name) =>
+        lit.value != null && lit.value.toString == "0"
+      case _ => false
+    }
   }
 
   /**
@@ -204,7 +206,7 @@ case class CometScanRule(session: SparkSession)
       userOutput: Seq[Attribute]): Option[SparkPlan] = plan match {
     case scan: FileSourceScanExec
         if DeltaReflection.isDeltaFileFormat(scan.relation.fileFormat) &&
-          scan.output.exists(_.name == DeltaReflection.IsRowDeletedColumnName) =>
+          scan.output.exists(_.name.equalsIgnoreCase(DeltaReflection.IsRowDeletedColumnName)) =>
       Some(rebuildDeltaScanWithoutDvColumn(scan, userOutput))
     case other if other.children.size == 1 =>
       // Single-child wrappers (Project, ColumnarToRow, etc.) Delta may insert between

spark/src/main/scala/org/apache/comet/serde/operator/CometDeltaNativeScan.scala

@@ -385,7 +385,11 @@ object CometDeltaNativeScan extends CometOperatorSerde[CometScanExec] with Loggi
           val dec = org.apache.spark.sql.types.Decimal(new java.math.BigDecimal(s))
           dec.changePrecision(d.precision, d.scale)
           dec
-        case _ => UTF8String.fromString(s)
+        case other =>
+          throw new IllegalArgumentException(
+            s"Unsupported Delta partition column type $other for value '$s'. " +
+              "Supported types: string, int, long, short, byte, float, double, " +
+              "boolean, date, timestamp, decimal.")
       }
   }