parthchandra commented on code in PR #4952: URL: https://github.com/apache/datafusion-comet/pull/4952#discussion_r3668153430
########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## Review Comment: There is a corner case here. A future V2 contrib that has a table name like `files`, `snapshots` etc (iceberg reserved names) would pass the `isIcebergMetadataTable` check and the contrib will never get called. We could either call transformV2Scan first or explicitly check for an iceberg scan in `isIcebergMetadataTable` ? ########## spark/src/main/scala/org/apache/comet/rules/CometScanContrib.scala: ########## @@ -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. + */ + +package org.apache.comet.rules + +import java.util.ServiceLoader + +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +import org.apache.spark.internal.Logging +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.spark.sql.execution.datasources.v2.BatchScanExec + +import org.apache.comet.serde.CometOperatorSerde + +/** + * Format-agnostic hook that lets an optional, out-of-tree contrib (Delta, Lance, ...) claim a + * scan before Comet's built-in scan handling runs. This is the scan-rule counterpart to + * [[org.apache.spark.sql.comet.PlanDataInjector]]: core holds no compile-time reference to any + * contrib and names none of them -- implementations are discovered at runtime via the JDK + * [[ServiceLoader]], so default builds (which ship no service file) see an empty registry and + * zero contrib surface. + * + * A contrib ships: + * - an implementation of this trait (e.g. `contrib/delta/.../DeltaScanRuleContrib`), and + * - a `META-INF/services/org.apache.comet.rules.CometScanContrib` resource naming it. + * + * Both hooks default to `None` so a contrib overrides only the scan kind(s) it handles. + */ +trait CometScanContrib { Review Comment: Can we add a doc note that an implementation MUST return `None` for a scan it does not own. This is to prevent two contribs from competing for the same scan. ########## spark/src/main/scala/org/apache/comet/rules/CometScanContrib.scala: ########## @@ -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. + */ + +package org.apache.comet.rules + +import java.util.ServiceLoader + +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +import org.apache.spark.internal.Logging +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.spark.sql.execution.datasources.v2.BatchScanExec + +import org.apache.comet.serde.CometOperatorSerde + +/** + * Format-agnostic hook that lets an optional, out-of-tree contrib (Delta, Lance, ...) claim a + * scan before Comet's built-in scan handling runs. This is the scan-rule counterpart to + * [[org.apache.spark.sql.comet.PlanDataInjector]]: core holds no compile-time reference to any + * contrib and names none of them -- implementations are discovered at runtime via the JDK + * [[ServiceLoader]], so default builds (which ship no service file) see an empty registry and + * zero contrib surface. + * + * A contrib ships: + * - an implementation of this trait (e.g. `contrib/delta/.../DeltaScanRuleContrib`), and + * - a `META-INF/services/org.apache.comet.rules.CometScanContrib` resource naming it. + * + * Both hooks default to `None` so a contrib overrides only the scan kind(s) it handles. + */ +trait CometScanContrib { + + /** + * V1 (`FileSourceScanExec`) hook. Return `Some(plan)` to claim the scan -- either a transformed + * plan node or an explicit fallback the contrib produced via `withFallbackReason`. Return + * `None` to pass, letting Comet's generic V1 handling proceed. A claiming contrib is + * responsible for its own metadata-column handling (the generic guard in [[CometScanRule]] runs + * only on the pass path). + */ + def tryTransformV1( + plan: SparkPlan, + session: SparkSession, + scanExec: FileSourceScanExec, + relation: HadoopFsRelation): Option[SparkPlan] = None + + /** + * V2 (`BatchScanExec`) hook. Return `Some(plan)` to claim the scan, `None` to pass. + */ + def tryTransformV2(scanExec: BatchScanExec): Option[SparkPlan] = None +} + +object CometScanContrib extends Logging { + + // Discovered contrib scan handlers. Mirrors `PlanDataInjector.injectors`: the standard JDK + // ServiceLoader forces the provider iterator here so a misbuilt contrib jar (malformed service + // file, or a listed provider that can't be instantiated) surfaces as a warning rather than + // taking down the planner. Default builds carry no service file, so this is empty and there is + // no contrib surface at runtime. There are no built-in (in-core) contribs today; the format + // built-ins (Parquet/Iceberg) keep their existing paths in CometScanRule. + private lazy val contribs: Seq[CometScanContrib] = + try { + ServiceLoader.load(classOf[CometScanContrib], getClass.getClassLoader).asScala.toSeq + } catch { + // NonFatal covers ServiceConfigurationError (not a LinkageError). + case NonFatal(e) => + logWarning("Failed to load contrib CometScanContrib services", e) + Seq.empty + } + + /** + * Offer a scan to each registered contrib in turn and return the first claim. + * + * A contrib that throws is treated as having declined: the failure is logged and the scan + * continues down Comet's built-in path (ultimately vanilla Spark). A contrib's planning work is + * speculative and can fail for reasons entirely outside the query -- an unreachable object + * store, a metadata format newer than the contrib understands, a version-skewed reflective + * lookup -- and none of those should turn a runnable query into a failed one. Logging (rather + * than swallowing silently) keeps an unexpectedly-declining contrib diagnosable. `NonFatal` + * deliberately lets `LinkageError`/`OOM`-class failures through. + */ + private def firstClaim(hook: CometScanContrib => Option[SparkPlan]): Option[SparkPlan] = Review Comment: This is nice. Can we add a small test suite similar to `CometScanWithPlanDataSuite`? Some cases to consider (all running on the default build) - - an empty registry returns None from tryTransformV1/tryTransformV2 - a stub contrib registered through a URLClassLoader that claims a scan is returned - a stub that throws is logged and declined so the next contrib still gets a look Also, can two contribs claim the same scan? ########## native/core/src/execution/planner.rs: ########## @@ -1601,6 +1610,26 @@ impl PhysicalPlanner { )), )) } + OpStruct::ContribScan(contrib) => { + // Extension point for optional, out-of-tree contrib scans (Delta, Lance, ...). The + // concrete scan message is packed into the `ContribScan` envelope on the JVM side; + // here we route by `type_url` to whichever contrib was compiled in. Core names no + // specific contrib: the type_url match + decode lives entirely inside each + // contrib's gated module, so a default build carries zero contrib surface. The arm + // itself is unconditional so a default build that receives a contrib-shaped plan + // from a misconfigured driver gets a clear error instead of a "no match" decode + // failure. + #[cfg(feature = "contrib-delta")] Review Comment: I know this is based on the previous review comment but on deeper thought it may be possible to remove this from core as well by using a generic handler (similar to the jvm side). Could you log a follow up issue for this as well? We have two paths we can consider - 1) a true service loader type system for dynamically discovering and loading an extension. There may be dragons along this path. 2) a statically linked version which builds each crate independently but may need some refactoring at the crate level to avoid circular dependencies. ########## native/proto/src/proto/operator.proto: ########## @@ -279,6 +309,224 @@ message IcebergDeleteFile { repeated int32 equality_ids = 4; } +// ===================================================================================== +// Delta Lake scan messages -- consumed by the optional `contrib/delta/` integration. +// Field numbers must remain stable; older serialised plans encode them on the wire. +// ===================================================================================== + +// Per-scan invariants. Lives at the head of every Delta scan operator payload. +message DeltaScanCommon { Review Comment: Can we log a follow up issue to move this out of core and into `contrib/delta/proto`? We will probably need to add a (manual) pipeline for the contribs. ########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## @@ -165,9 +168,27 @@ case class CometScanRule(session: SparkSession) scanExec.relation match { case r: HadoopFsRelation => + // Give any optional, out-of-tree scan contrib (e.g. Delta) first crack at this scan. On a + // default build no contrib is registered, so this returns None and we fall through to the + // vanilla scan path. A registered contrib either claims the scan (returning its marker + // node) or declines via its own `withFallbackReason` fallback message. A claiming contrib + // owns its own metadata-column handling -- which is why the guard below runs only after + // the contrib has declined. + CometScanContrib.tryTransformV1(plan, session, scanExec, r) match { + case Some(handled) => return handled + case None => // proceed with vanilla logic + } + if (metadataCols(scanExec).nonEmpty) { Review Comment: Can we add a test case that covers metadata for V1 scans? say selecting `_metadata.file_path` for a parquet source. The default build should fall back with this reason. -- 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]
