dwsmith1983 commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3826287506
########## 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: Went with decline rather than per-file routing. The claim gate now compares the full lowercased URI authority (userinfo included, so abfss containers count as distinct) across all selected files, with a native check as backstop, and file order is never changed on the native side. One residual worth flagging: the shared object-store cache key in core drops userinfo, so a DV sidecar on a different container of the same account could still resolve through the wrong store. That's in the shared store layer and predates this PR, so I left it as a core follow-up rather than changing cache semantics here. -- 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]
