dwsmith1983 commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3831212662
########## native/core/src/execution/planner/delta_spark_scan.rs: ########## @@ -0,0 +1,145 @@ +// 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. + +//! JVM-planned Delta handler for the generic `OpStruct::ContribScan` dispatcher, feature-gated +//! behind `delta`. +//! +//! delta-spark has already done log replay, snapshot resolution, and partition pruning by the +//! time the scan reaches Comet, so the envelope carries a concrete file list (plus deletion +//! vector descriptors) and the read path reuses the exact same shared parquet scan builder as +//! `NativeScan` -- inheriting row-group stats pruning, page-index pruning, and filter pushdown. +//! Sibling of the kernel-planned handler in `delta_scan.rs`; the two claim different +//! `type_url`s within the same `ContribScan` envelope. + +use std::collections::HashMap; + +use datafusion_comet_proto::spark_operator::{ + ContribScan, DeltaSparkScan, Operator, SparkFilePartition, +}; +use prost::Message; + +use crate::execution::operators::ExecutionError::GeneralError; +use crate::execution::planner::PhysicalPlanner; +use crate::execution::planner::PlanCreationResult; + +/// Type name the JVM-planned Delta contrib claims within the `ContribScan` envelope. The +/// contrib jar packs a `DeltaSparkScan` with a `type_url` of +/// `type.googleapis.com/comet.contrib.delta_spark.DeltaSparkScan`; dispatch keys on the +/// contrib-owned suffix, same convention as the kernel path's `delta_scan.rs`. +const DELTA_SPARK_SCAN_TYPE_NAME: &str = "comet.contrib.delta_spark.DeltaSparkScan"; + +/// Contrib entry point for the `OpStruct::ContribScan` dispatcher. Returns `Some(result)` when +/// the envelope carries a JVM-planned Delta scan, or `None` when the `type_url` belongs to some +/// other contrib. +pub(crate) fn try_plan_contrib_scan( + planner: &PhysicalPlanner, + spark_plan: &Operator, + contrib: &ContribScan, +) -> Option<PlanCreationResult> { + if !contrib.type_url.ends_with(DELTA_SPARK_SCAN_TYPE_NAME) { + return None; + } + Some( + DeltaSparkScan::decode(contrib.value.as_slice()) + .map_err(|e| { + GeneralError(format!( + "Failed to decode DeltaSparkScan from contrib_scan: {e}" + )) + }) + .and_then(|scan| plan_delta_spark_scan(planner, spark_plan, &scan)), + ) +} + +fn plan_delta_spark_scan( + planner: &PhysicalPlanner, + spark_plan: &Operator, + scan: &DeltaSparkScan, +) -> PlanCreationResult { + // Delta data files are plain parquet; the read path deliberately reuses + // the same shared parquet scan builder as NativeScan so Delta inherits + // row-group stats pruning, page-index pruning, and filter pushdown. Only + // the file list arrives in Delta-specific form. Note delta_common's + // column_mapping_mode is informational in M1: the actual field-id + // matching switch is common.use_field_id, same as the Iceberg path. + let common = scan + .common + .as_ref() + .ok_or_else(|| GeneralError("DeltaSparkScan missing common data".into()))?; + + let delta_partition = scan + .file_partition + .as_ref() + .ok_or_else(|| GeneralError("DeltaSparkScan missing file_partition".into()))?; + + let spark_partition = SparkFilePartition { + partitioned_file: delta_partition + .partitioned_file + .iter() + .map(|f| { + f.file + .clone() + .ok_or_else(|| GeneralError("DeltaSparkPartitionedFile missing inner file".into())) + }) + .collect::<Result<Vec<_>, _>>()?, + }; + + let (object_store_url, mut files) = + planner.prepare_scan_store_and_files(common, &spark_partition)?; Review Comment: Added, both sides. At claim time the JVM declines when any two URIs the native side resolves stores for (data files and DV sidecars) share the native store-identity key but differ in userinfo, which is exactly your source/clone container case. The JVM key is a deliberate conservative superset of the native one (it also lowercases host and port and always rewrites s3a to s3), so it can only over-decline, never miss a collision native would hit. The native side got a matching check at store-resolution time, which sees DV paths too; it's deliberately not an extension of the data-file authority check so the legitimate cross-bucket DV shape (data in A, sidecar in B) still claims, which the MinIO suite now proves live. Remaining honest gaps, all narrow: the JVM compares decoded authorities while native parses the url-encoded form, so a userinfo differing only by percent-encoding would pass the gate and hit the native error (not constructible from real container names); the native compar e uses the username without the password component; and the native check is per partition while the JVM gate is whole-scan. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
