parthchandra commented on code in PR #4700:
URL: https://github.com/apache/datafusion-comet/pull/4700#discussion_r3463014642
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -87,14 +88,42 @@ private[comet] trait PlanDataInjector {
/**
* Registry and utilities for injecting per-partition planning data into
operator trees.
*/
-private[comet] object PlanDataInjector {
-
- // Registry of injectors for different operator types
- private val injectors: Seq[PlanDataInjector] = Seq(
- IcebergPlanDataInjector,
- NativeScanPlanDataInjector
- // Future: DeltaPlanDataInjector, HudiPlanDataInjector, etc.
- )
+private[comet] object PlanDataInjector extends Logging {
+
+ // Registry of injectors for different operator types. The contrib/delta
integration's
+ // DeltaPlanDataInjector is appended via one reflective class lookup --
present only when
+ // the contrib was bundled (i.e. -Pcontrib-delta on the Maven build).
Default builds get
+ // the empty Option and an unmodified injectors list, so there's zero
contrib surface at
+ // runtime on default builds.
+ private val injectors: Seq[PlanDataInjector] = {
+ val builtin: Seq[PlanDataInjector] = Seq(IcebergPlanDataInjector,
NativeScanPlanDataInjector)
+ val deltaOpt: Option[PlanDataInjector] =
+ try {
+ // Scala compiles `object Foo` into BOTH `Foo.class` (a
static-forwarder
+ // class with no MODULE$ field) AND `Foo$.class` (the module class that
+ // does have MODULE$). The trailing `$` selects the module class.
+ // scalastyle:off classforname
+ val cls =
Class.forName("org.apache.spark.sql.comet.DeltaPlanDataInjector$")
Review Comment:
This would require every other contrib to have to add their own block of
code. Consider ServiceLoader (META-INF/services/...PlanDataInjector) so
contribs are discoverable without modifying core.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -87,14 +88,42 @@ private[comet] trait PlanDataInjector {
/**
* Registry and utilities for injecting per-partition planning data into
operator trees.
*/
-private[comet] object PlanDataInjector {
-
- // Registry of injectors for different operator types
- private val injectors: Seq[PlanDataInjector] = Seq(
- IcebergPlanDataInjector,
- NativeScanPlanDataInjector
- // Future: DeltaPlanDataInjector, HudiPlanDataInjector, etc.
- )
+private[comet] object PlanDataInjector extends Logging {
+
+ // Registry of injectors for different operator types. The contrib/delta
integration's
+ // DeltaPlanDataInjector is appended via one reflective class lookup --
present only when
+ // the contrib was bundled (i.e. -Pcontrib-delta on the Maven build).
Default builds get
+ // the empty Option and an unmodified injectors list, so there's zero
contrib surface at
+ // runtime on default builds.
+ private val injectors: Seq[PlanDataInjector] = {
+ val builtin: Seq[PlanDataInjector] = Seq(IcebergPlanDataInjector,
NativeScanPlanDataInjector)
+ val deltaOpt: Option[PlanDataInjector] =
+ try {
+ // Scala compiles `object Foo` into BOTH `Foo.class` (a
static-forwarder
+ // class with no MODULE$ field) AND `Foo$.class` (the module class that
+ // does have MODULE$). The trailing `$` selects the module class.
+ // scalastyle:off classforname
+ val cls =
Class.forName("org.apache.spark.sql.comet.DeltaPlanDataInjector$")
+ // scalastyle:on classforname
+ Some(cls.getField("MODULE$").get(null).asInstanceOf[PlanDataInjector])
+ } catch {
+ // Default builds (no -Pcontrib-delta) won't have the class -> silent
None.
+ // This is the only EXPECTED miss, so it's the only quiet one.
+ case _: ClassNotFoundException => None
+ // The class IS on the classpath but couldn't be bound: missing
MODULE$,
+ // access drift (NoSuchField/IllegalAccess), an initializer/linkage
error,
+ // or a CCE on the PlanDataInjector cast. That's a misbuilt contrib
jar, not
+ // a default build -- warn so it's diagnosable, then still decline so
the
+ // rest of the planner stays alive.
+ case e: Throwable =>
Review Comment:
Maybe use ` scala.util.control.NonFatal` instead?
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -786,11 +823,16 @@ abstract class CometNativeExec extends CometExec {
(Map.empty, Map.empty)
}
- case nativeScan: CometNativeScanExec =>
- nativeScan.ensureSubqueriesResolved()
- (
- Map(nativeScan.sourceKey -> nativeScan.commonData),
- Map(nativeScan.sourceKey -> nativeScan.perPartitionData))
+ // Generic path for leaf scans that surface planning data via the
+ // `CometScanWithPlanData` trait. Catches `CometNativeScanExec` and any
contrib
+ // leaf scan (e.g. the Delta contrib's `CometDeltaNativeScanExec`)
without
+ // requiring core to compile-time reference contrib classes.
+ case s: CometScanWithPlanData =>
+ s match {
+ case leaf: CometLeafExec => leaf.ensureSubqueriesResolved()
+ case _ => // no DPP lifecycle to drive
Review Comment:
The iceberg path handles `if (commonData.empty && perPartitionData.empty)`
by returning `(Map.empty, Map.empty)` but this path doesn't. An implementation
returning an empty array will probably cause an exception downstream. Better to
handle it here
##########
spark/src/test/scala/org/apache/spark/sql/comet/CometScanWithPlanDataSuite.scala:
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.spark.sql.comet
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+
+/**
+ * Unit coverage for the core SPI that lets out-of-tree contrib leaf scans
(e.g. the Delta
+ * contrib's `CometDeltaNativeScanExec`) participate in Comet native planning
without core holding
+ * a compile-time reference to them: the [[CometScanWithPlanData]] trait
contract and the
+ * reflective `DeltaPlanDataInjector` slot in [[PlanDataInjector]]'s registry.
+ *
+ * Deliberately does not exercise the per-op injector registry mechanics; that
surface is owned by
+ * `PlanDataInjectorSuite`.
+ */
+class CometScanWithPlanDataSuite extends AnyFunSuite {
Review Comment:
This really needs a test that implements a stub/mock `CometLeafExec with
CometScanWithPlanData `wired through `findAllPlanData`. That would confirm the
new code path works independently of the existing scan classes.
--
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]
