feat: Iceberg parity improvements + code review fixes for Delta native scan

- Add LRU cache to DeltaPlanDataInjector (matching Iceberg pattern)
- Add ImmutableSQLMetric, planning metrics (total_files, dv_files),
  explicit doExecuteColumnar() override, convertBlock() preserving
  @transient fields
- Add DPP support via doPrepare() + deferred partition pruning
- Add filesystem scheme validation in scan rule
- Consolidate castPartitionString into DeltaReflection (fixes divergent
  date/timestamp/decimal handling between planning and execution)
- Fix DV unwrap_or_default → proper error on kernel mismatch
- Fix silent predicate decode failure → log warning
- Fix ColumnMappingFilterRewriter silent fallback → log warning
- Fix fieldIndex/indexWhere inconsistency in DPP partition resolution
- Fix ThreadLocal cleanup in createExec (try-finally)
- Add unsafe safety docs, DV filter loop invariant comment
- Replace flaky Thread.sleep in timestamp test with deterministic mtime
- Split test suite into 3 focused suites + shared CometDeltaTestBase trait
- Fix license header typo, outdated comments, error message clarity

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

Author: schenksj

native/core/src/delta/error.rs

@@ -54,6 +54,9 @@ pub enum DeltaError {
 
     #[error("delta kernel error: {0}")]
     Kernel(#[from] delta_kernel::Error),
+
+    #[error("{0}")]
+    Internal(String),
 }
 
 pub type DeltaResult<T> = std::result::Result<T, DeltaError>;

native/core/src/delta/jni.rs

@@ -11,7 +11,7 @@
 // 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 success or implied.  See the License for the
+// KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
 
@@ -105,14 +105,21 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_planDeltaScan(
         // BoundReference index-to-name resolution.
         let kernel_predicate = _predicate_proto.and_then(|bytes| {
             use prost::Message;
-            datafusion_comet_proto::spark_expression::Expr::decode(bytes.as_slice())
-                .ok()
-                .map(|expr| {
+            match datafusion_comet_proto::spark_expression::Expr::decode(bytes.as_slice()) {
+                Ok(expr) => Some(
                     crate::delta::predicate::catalyst_to_kernel_predicate_with_names(
                         &expr,
                         &col_names,
-                    )
-                })
+                    ),
+                ),
+                Err(e) => {
+                    log::warn!(
+                        "Failed to decode predicate for Delta file pruning: {e}; \
+                         scanning all files"
+                    );
+                    None
+                }
+            }
         });
 
         let plan =
@@ -238,6 +245,9 @@ fn read_string_array(
     let mut result = Vec::with_capacity(len);
     for i in 0..len {
         let obj = arr.get_element(env, i)?;
+        // SAFETY: get_element returns a valid local JObject reference that we
+        // immediately convert to JString. The array is String[], so the cast
+        // is valid. The env lifetime outlives this scope.
         let jstr = unsafe { JString::from_raw(env, obj.into_raw()) };
         result.push(jstr.try_to_string(env)?);
     }
@@ -256,7 +266,9 @@ fn map_get_string(
     match jmap.get(env, &key_jobj)? {
         None => Ok(None),
         Some(value) => {
-            // Safe: `Map<String, String>::get` only ever returns a String.
+            // SAFETY: Map<String, String>::get always returns a String. The
+            // JObject reference is valid because JMap::get returned it from the
+            // current env frame. We consume the local ref via into_raw().
             let jstr = unsafe { JString::from_raw(env, value.into_raw()) };
             Ok(Some(jstr.try_to_string(env)?))
         }

native/core/src/delta/scan.rs

@@ -240,14 +240,12 @@ pub fn plan_delta_scan_with_predicate(
     // 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)?
-                .unwrap_or_default()
+                .ok_or_else(|| DeltaError::Internal(
+                    format!("DV has_vector() true but get_row_indexes() returned None for {}", r.path),
+                ))?
         } else {
             Vec::new()
         };

native/core/src/execution/operators/delta_dv_filter.rs

@@ -247,6 +247,8 @@ impl DeltaDvFilterStream {
         // that fall inside [batch_start, batch_end).
         let mut mask_buf: Vec<bool> = vec![true; batch_rows as usize];
         let mut dropped: usize = 0;
+        // Loop is safe: next_delete_idx < deleted.len() is checked by the while
+        // condition, and deleted is sorted ascending by the kernel contract.
         while self.next_delete_idx < self.deleted.len() {
             let d = self.deleted[self.next_delete_idx];
             if d >= batch_end {

native/core/src/execution/planner.rs

@@ -1358,7 +1358,7 @@ impl PhysicalPlanner {
                 let common = scan
                     .common
                     .as_ref()
-                    .ok_or_else(|| GeneralError("DeltaScan missing common data".into()))?;
+                    .ok_or_else(|| GeneralError("DeltaScan proto missing 'common' field (Scala serialization error)".into()))?;
 
                 let required_schema: SchemaRef =
                     convert_spark_types_to_arrow_schema(common.required_schema.as_slice());
@@ -1415,7 +1415,19 @@ impl PhysicalPlanner {
                 let data_filters: Result<Vec<Arc<dyn PhysicalExpr>>, ExecutionError> = common
                     .data_filters
                     .iter()
-                    .map(|expr| self.create_expr(expr, Arc::clone(&required_schema)))
+                    .map(|expr| {
+                        let filter =
+                            self.create_expr(expr, Arc::clone(&required_schema))?;
+                        if has_column_mapping {
+                            let mut rewriter = ColumnMappingFilterRewriter {
+                                logical_to_physical: &logical_to_physical,
+                                data_schema: &data_schema,
+                            };
+                            Ok(filter.rewrite(&mut rewriter).data()?)
+                        } else {
+                            Ok(filter)
+                        }
+                    })
                     .collect();
 
                 let object_store_options: HashMap<String, String> = common
@@ -1459,39 +1471,21 @@ impl PhysicalPlanner {
                     .tasks
                     .first()
                     .map(|t| t.file_path.clone())
-                    .ok_or_else(|| GeneralError("DeltaScan has no tasks".into()))?;
+                    .ok_or_else(|| GeneralError("DeltaScan has no tasks after split-mode injection (check DeltaPlanDataInjector)".into()))?;
                 let (object_store_url, _) = prepare_object_store_with_configs(
                     self.session_ctx.runtime_env(),
                     one_file,
                     &object_store_options,
                 )?;
 
-                // When column mapping is active, required_schema also needs physical
-                // names so init_datasource_exec's name-matching logic works against the
-                // physical data_schema.
-                let read_required_schema = if has_column_mapping {
-                    let new_fields: Vec<_> = required_schema
-                        .fields()
-                        .iter()
-                        .map(|f| {
-                            if let Some(physical) = logical_to_physical.get(f.name()) {
-                                Arc::new(Field::new(
-                                    physical,
-                                    f.data_type().clone(),
-                                    f.is_nullable(),
-                                ))
-                            } else {
-                                Arc::clone(f)
-                            }
-                        })
-                        .collect();
-                    Arc::new(Schema::new(new_fields))
-                } else {
-                    Arc::clone(&required_schema)
-                };
-
+                // Keep required_schema in LOGICAL names (Spark's convention).
+                // data_schema uses physical names (when column mapping is active).
+                // DataFusion's schema adapter bridges the gap: it matches file
+                // columns against data_schema by name and produces the
+                // required_schema output shape, injecting nulls for missing
+                // columns (schema evolution).
                 let delta_exec = init_datasource_exec(
-                    read_required_schema,
+                    Arc::clone(&required_schema),
                     Some(data_schema),
                     Some(partition_schema),
                     object_store_url,
@@ -1518,33 +1512,9 @@ impl PhysicalPlanner {
                         delta_exec
                     };
 
-                // Phase 4: when column mapping is active, the output has PHYSICAL
-                // column names. Add a ProjectionExec to rename back to logical.
-                let final_exec = if has_column_mapping {
-                    let physical_to_logical: HashMap<String, String> = logical_to_physical
-                        .iter()
-                        .map(|(l, p)| (p.clone(), l.clone()))
-                        .collect();
-                    let input_schema = final_exec.schema();
-                    let rename_exprs: Result<Vec<(Arc<dyn PhysicalExpr>, String)>, ExecutionError> = input_schema
-                        .fields()
-                        .iter()
-                        .enumerate()
-                        .map(|(idx, f)| {
-                            let col: Arc<dyn PhysicalExpr> =
-                                Arc::new(Column::new(f.name(), idx));
-                            let logical = physical_to_logical
-                                .get(f.name())
-                                .cloned()
-                                .unwrap_or_else(|| f.name().clone());
-                            Ok((col, logical))
-                        })
-                        .collect();
-                    let rename_exprs = rename_exprs?;
-                    Arc::new(ProjectionExec::try_new(rename_exprs, final_exec)?) as Arc<dyn ExecutionPlan>
-                } else {
-                    final_exec
-                };
+                // No rename projection needed: required_schema already uses
+                // logical names, and DataFusion's schema adapter handles the
+                // physical→logical mapping internally via data_schema.
 
                 Ok((
                     vec![],
@@ -2999,6 +2969,45 @@ fn expr_to_columns(
     Ok((left_field_indices, right_field_indices))
 }
 
+/// Rewrites Column references in a PhysicalExpr from logical names/indices
+/// (as in required_schema) to physical names/indices (as in data_schema).
+/// Used by the Delta scan path when column mapping is active so that pushed-down
+/// data filters match the DataSourceExec's base schema (physical column names).
+struct ColumnMappingFilterRewriter<'a> {
+    logical_to_physical: &'a HashMap<String, String>,
+    data_schema: &'a SchemaRef,
+}
+
+impl TreeNodeRewriter for ColumnMappingFilterRewriter<'_> {
+    type Node = Arc<dyn PhysicalExpr>;
+
+    fn f_down(
+        &mut self,
+        node: Self::Node,
+    ) -> datafusion::common::Result<Transformed<Self::Node>> {
+        if let Some(column) = node.as_any().downcast_ref::<Column>() {
+            if let Some(physical_name) = self.logical_to_physical.get(column.name()) {
+                if let Some(idx) = self
+                    .data_schema
+                    .fields()
+                    .iter()
+                    .position(|f| f.name() == physical_name)
+                {
+                    return Ok(Transformed::yes(Arc::new(Column::new(physical_name, idx))));
+                }
+                log::warn!(
+                    "Column mapping: physical name '{}' for logical '{}' not found in data_schema; \
+                     filter may fail at execution time",
+                    physical_name, column.name()
+                );
+            }
+            Ok(Transformed::no(node))
+        } else {
+            Ok(Transformed::no(node))
+        }
+    }
+}
+
 /// A physical join filter rewritter which rewrites the column indices in the expression
 /// to use the new column indices. See `rewrite_physical_expr`.
 struct JoinFilterRewriter<'a> {

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

@@ -137,4 +137,48 @@ object DeltaReflection extends Logging {
       case _: Exception => None
     }
   }
+
+  /**
+   * Convert a Delta partition value string to a Catalyst-internal representation. Delta stores
+   * partition values as strings in add actions; this converts them to the correct type for
+   * predicate evaluation.
+   */
+  def castPartitionString(str: Option[String], dt: org.apache.spark.sql.types.DataType): Any = {
+    import org.apache.spark.sql.catalyst.util.DateTimeUtils
+    import org.apache.spark.sql.types._
+    import org.apache.spark.unsafe.types.UTF8String
+    str match {
+      case None | Some(null) => null
+      case Some(s) =>
+        try {
+          dt match {
+            case StringType => UTF8String.fromString(s)
+            case IntegerType => s.toInt
+            case LongType => s.toLong
+            case ShortType => s.toShort
+            case ByteType => s.toByte
+            case FloatType => s.toFloat
+            case DoubleType => s.toDouble
+            case BooleanType => s.toBoolean
+            case DateType =>
+              DateTimeUtils
+                .stringToDate(UTF8String.fromString(s))
+                .getOrElse(null)
+            case _: TimestampType =>
+              DateTimeUtils
+                .stringToTimestamp(UTF8String.fromString(s), java.time.ZoneOffset.UTC)
+                .getOrElse(null)
+            case d: DecimalType =>
+              val dec =
+                org.apache.spark.sql.types.Decimal(new java.math.BigDecimal(s))
+              dec.changePrecision(d.precision, d.scale)
+              dec
+            case _ => UTF8String.fromString(s)
+          }
+        } catch {
+          case _: NumberFormatException | _: IllegalArgumentException =>
+            null
+        }
+    }
+  }
 }

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

@@ -396,6 +396,26 @@ case class CometScanRule(session: SparkSession)
     if (!isSchemaSupported(scanExec, SCAN_NATIVE_DELTA_COMPAT, r)) {
       return None
     }
+
+    // Validate filesystem schemes from the scan's input files.
+    val supportedSchemes =
+      Set("file", "s3", "s3a", "gs", "gcs", "abfss", "abfs", "wasbs", "wasb", "oss")
+    val inputFiles = scanExec.relation.location.inputFiles
+    if (inputFiles.nonEmpty) {
+      val schemes = inputFiles
+        .map(f => new java.net.URI(f).getScheme)
+        .filter(_ != null)
+        .toSet
+      val unsupported = schemes -- supportedSchemes
+      if (unsupported.nonEmpty) {
+        withInfo(
+          scanExec,
+          s"Native Delta scan does not support filesystem schemes: " +
+            unsupported.mkString(", "))
+        return None
+      }
+    }
+
     Some(CometScanExec(scanExec, session, SCAN_NATIVE_DELTA_COMPAT))
   }