parthchandra commented on code in PR #4952: URL: https://github.com/apache/datafusion-comet/pull/4952#discussion_r3607131603
########## spark/src/main/scala/org/apache/comet/rules/DeltaIntegration.scala: ########## @@ -0,0 +1,199 @@ +/* + * 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. + */ + +package org.apache.comet.rules + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan} +import org.apache.spark.sql.execution.datasources.HadoopFsRelation + +import org.apache.comet.serde.CometOperatorSerde +import org.apache.comet.util.ClassLoaders + +/** + * Reflection-based bridge to the optional `contrib/delta/` integration. + * + * On default builds the contrib classes don't exist on the classpath, so the reflective class + * lookups fail and every method here returns the "not handled" sentinel. On builds compiled with + * `-Pcontrib-delta` (Maven) + `--features contrib-delta` (Cargo), the contrib classes are present + * and the lookups resolve, dispatching the call into the contrib helpers. + * + * Keeping this bridge as one small file in core lets the Delta detection block in `CometScanRule` + * and the serde dispatch in `CometExecRule` stay ~10 lines each -- exactly the shape Parth's Review Comment: The change to `CometExecRule` is small but this itself is pure Delta code in core. We really need to follow the same pattern we established of discovering using the ServiceLoader mechanism. `PlanDataInjector` (in operators.scala:72-127, merged in Part 1 / #4700) already uses ServiceLoader discovery. PR #4633 (Lance) also uses the same pattern with `CometScanContrib` I would recommend we replace this with a generic `CometScanContrib` trait in a new file `spark/src/main/scala/org/apache/comet/rules/CometScanContrib.scala` (in core). This trait can cover both V1 and V2 scans: ``` trait CometScanContrib { /** V1 scan hook. Return Some(plan) to claim, None to pass. */ def tryTransformV1( plan: SparkPlan, session: SparkSession, scanExec: FileSourceScanExec, relation: HadoopFsRelation): Option[SparkPlan] = None /** V2 scan hook. Return Some(plan) to claim, None to pass. */ def tryTransformV2(scanExec: BatchScanExec): Option[SparkPlan] = None } object CometScanContrib extends Logging { private lazy val contribs: Seq[CometScanContrib] = { // Built-in contribs (Parquet, Iceberg) can be registered here. // Contrib-gated ones (Delta, Lance) discovered via ServiceLoader. val discovered = try { ServiceLoader.load(classOf[CometScanContrib], getClass.getClassLoader).asScala.toSeq } catch { case NonFatal(e) => logWarning("Failed to load contrib CometScanContrib services", e) Seq.empty } discovered } def tryTransformV1(...): Option[SparkPlan] = { contribs.view.flatMap(_.tryTransformV1(...)).headOption } def tryTransformV2(...): Option[SparkPlan] = { contribs.view.flatMap(_.tryTransformV2(...)).headOption } } ``` This is format-agnostic code. The Delta contrib then has: - `contrib/delta/src/.../DeltaScanRuleContrib.scala` implementing `CometScanContrib` - `contrib/delta/resources/META-INF/services/org.apache.comet.rules.CometScanContrib` naming it This also subsumes PR #4633's `CometScanContrib` — the Lance PR would implement `tryTransformV2` on the same trait rather than defining a separate one. ########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## @@ -125,15 +119,18 @@ case class CometScanRule(session: SparkSession) case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) => withFallbackReason(scan, "Comet Scan is not enabled") + // V1 scans go through `transformV1Scan` which itself first delegates to any Review Comment: With a generic `CometScanContrib.tryTransformV1()`, this reordering is unnecessary. The outer `transformScan` match keeps its current order (metadata-colum guard before `FileSourceScanExec`). Each contrib's `tryTransformV1` can decide internally whether it can handle metadata columns. ########## spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala: ########## @@ -265,6 +265,16 @@ case class CometExecRule(session: SparkSession) // spotless:on private def transform(plan: SparkPlan): SparkPlan = { def convertNode(op: SparkPlan): SparkPlan = op match { + // Delta scan marker produced by the optional contrib/delta integration. Matched by type + // (no compile-time dependency on the contrib) -- present only when -Pcontrib-delta was + // activated. The marker wraps the original, link-bearing scan, so the produced exec's + // originalPlan keeps its logicalLink with no workaround. If conversion declines, the marker + // itself falls back to the vanilla Spark Delta scan, so leaving it in the plan is safe. + case scan if DeltaIntegration.isDeltaScanMarker(scan) => Review Comment: We can clean this up too - Define a marker trait and Delta can implement it ``` /** Marker trait for contrib scan nodes that carry their own serde handler. */ trait CometContribScanMarker { this: SparkPlan => def scanHandler: CometOperatorSerde[_ <: SparkPlan] } ``` then the match becomes ``` case marker: CometContribScanMarker => convertToComet(marker, marker.scanHandler).getOrElse(marker) ``` ########## native/proto/src/proto/operator.proto: ########## @@ -55,6 +55,10 @@ message Operator { CsvScan csv_scan = 115; ShuffleScan shuffle_scan = 116; BroadcastNestedLoopJoin broadcast_nested_loop_join = 117; + // Delta Lake scan. Wire format used by `contrib/delta/`. Only decoded when + // core is built with `--features contrib-delta`; in default builds the + // dispatcher arm is `#[cfg]`-stubbed out so the contrib has zero runtime cost. + DeltaScan delta_scan = 118; Review Comment: If this gets tricky to implement we can consider assigning field ids for contrib scans, but I feel that the approach outlined here is feasible. ########## native/proto/src/proto/operator.proto: ########## @@ -55,6 +55,10 @@ message Operator { CsvScan csv_scan = 115; ShuffleScan shuffle_scan = 116; BroadcastNestedLoopJoin broadcast_nested_loop_join = 117; + // Delta Lake scan. Wire format used by `contrib/delta/`. Only decoded when + // core is built with `--features contrib-delta`; in default builds the + // dispatcher arm is `#[cfg]`-stubbed out so the contrib has zero runtime cost. + DeltaScan delta_scan = 118; Review Comment: Currently both this PR and PR #4633 claim field number 118 (delta_scan = 118 vs lance_scan = 118). More fundamentally, adding a new oneof variant for every contrib scan type means the core proto file must change every time a new format is added — violating the goal that core doesn't know about contrib implementations. Spark Connect solves this at `relations.proto:109` with `google.protobuf.Any extension = 998`;. We should do the same. We can replace this change with a permanent extension point: ``` // One-time addition to the oneof, never changes again: google.protobuf.Any contrib_scan = 200; ``` Then DeltaScan and LanceScan message definitions still live in operator.proto (or preferably in separate proto files under contrib/) — they're just packed into `Any` on the JVM side and unpacked by `type_url` on the Rust side. ########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## @@ -165,9 +169,29 @@ case class CometScanRule(session: SparkSession) scanExec.relation match { case r: HadoopFsRelation => + // Try the optional Delta contrib first. When this build wasn't compiled with + // `-Pcontrib-delta`, the bridge returns None and we fall through to the + // vanilla scan path. When the Delta classes are on the classpath, the contrib + // either claims the scan (returning a CometScanExec marker) or declines via + // its own `withFallbackReason` fallback message. + DeltaIntegration.transformV1IfDelta(plan, session, scanExec, r) match { Review Comment: This will change to `CometScanContrib.tryTransformV1(plan, session, scanExec, r)`. We can also remove the re-applied metadataCols check — it's now redundant because the outer guard still runs first and the Delta implementation decides whether to handle metadata columns or not. -- 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]
