andygrove commented on code in PR #5631:
URL: https://github.com/apache/datafusion-comet/pull/5631#discussion_r3917748442
##########
spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala:
##########
@@ -49,6 +49,9 @@ import org.apache.comet.shims.CometTypeShim
import org.apache.comet.vector.CometVector
object Utils extends CometTypeShim with Logging {
+ private val ArrowExtensionNameKey = "ARROW:extension:name"
Review Comment:
Arrow Java already defines this key as
`ArrowType.ExtensionType.EXTENSION_METADATA_KEY_NAME`. Could we use that
instead of a local literal? One less string that has to stay in sync by hand.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -826,15 +827,23 @@ case class CometExecRule(session: SparkSession)
handler: CometOperatorSerde[_]): Option[SparkPlan] = {
val serde = handler.asInstanceOf[CometOperatorSerde[SparkPlan]]
if (isOperatorEnabled(serde, op)) {
+ val dataProducingChildren = op.children.flatMap {
+ case writeFiles: WriteFilesExec => Seq(writeFiles.child)
+ case other => Seq(other)
+ }
+ if (!op.isInstanceOf[CometScanExec] &&
+ (op.output ++ dataProducingChildren.flatMap(_.output)).exists(attr =>
+ containsVariantType(attr.dataType))) {
+ withFallbackReason(
+ op,
+ "Native operators do not support schemas containing type
VariantType")
+ return None
+ }
Review Comment:
Could you say more about why `CometScanExec` is exempted here? As far as I
can tell `CometScanRule` only produces a `CometScanExec` after
`CometScanTypeChecker` has approved every output type, and this PR's own test
asserts that checker rejects `VariantType` and `ArrayType(VariantType)`, so I
could not find a plan where the exemption changes anything.
What makes me want to pin this down is that the same change removes the
other guard on that path. `CometNativeScan.convert` gates on `scanTypes.length
== scan.output.length`, and before this PR a Variant output made that length
check fail and the scan decline with "unsupported Comet operator ... due to
unsupported data types above". Now that `serializeDataType` succeeds for
Variant, that check can never fire for Variant either. So `CometScanExec` ends
up being the one operator where both the old implicit guard and the new
explicit guard disappear in the same commit.
If there is a real case behind the exemption, could we get a comment and a
test that fails without it? If there is not, I would rather drop it. An
unexplained hole in a deny gate tends to become load-bearing later. It also
reads oddly next to `CometBatchScanExec` on the Iceberg path, which is not
exempted.
##########
spark/src/test/scala/org/apache/comet/CometVariantTypeSuite.scala:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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
+
+import java.util.Collections
+
+import scala.collection.mutable.ListBuffer
+import scala.jdk.CollectionConverters._
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.arrow.vector.types.pojo.{ArrowType, Field, FieldType}
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.comet.util.Utils
+import org.apache.spark.sql.types.{ArrayType, BinaryType, StructField,
StructType}
+
+import org.apache.comet.rules.CometScanTypeChecker
+import org.apache.comet.serde.{CometAttributeReference, QueryPlanSerde,
Unsupported}
+
+class CometVariantTypeSuite extends AnyFunSuite {
+ private val storageType = StructType(
+ Seq(
+ StructField("value", BinaryType, nullable = false),
+ StructField("metadata", BinaryType, nullable = false)))
+
+ private def variantField(extensionName: Option[String]): Field = {
+ val metadata = extensionName
+ .map(name => Collections.singletonMap("ARROW:extension:name", name))
+ .getOrElse(Collections.emptyMap[String, String]())
+ val children = Seq(
+ Field.notNullable("value", ArrowType.Binary.INSTANCE),
+ Field.notNullable("metadata", ArrowType.Binary.INSTANCE))
+ new Field(
+ "v",
+ new FieldType(true, ArrowType.Struct.INSTANCE, null, metadata),
+ children.asJava)
+ }
+
+ test("Variant identity requires the canonical Arrow extension marker") {
Review Comment:
The marker name is now a literal in two places that have to agree, here and
`Utils.VariantExtensionName`, and nothing in the PR ties either of them to
arrow-rs. `has_valid_extension_type::<VariantType>()` will not catch drift
either, because arrow-rs's `supports_data_type` accepts any `Struct`. So if
`VariantType::NAME` ever changes, all four tests in this PR still pass while
the real path quietly degrades to `StructType`. That silent degrade is the
exact failure this PR exists to prevent, so it would be good to have something
that fails loudly instead.
Could the Rust test assert `VariantType::NAME == "arrow.parquet.variant"`
directly? That is cheap and pins the upstream side. Beyond that, is a test that
actually crosses the boundary feasible here, serializing a Spark `VariantType`
schema through `QueryPlanSerde.serializeDataType`, running it through native,
and asserting `Utils.fromArrowField` gives back `VariantType`? I realize no
operator can carry Variant yet so this may need a narrow test hook, but the JVM
to native contract is the whole point of the PR, and right now each half is
only tested against a hand-written copy of the other half's assumptions.
For what it is worth I did check that the pieces line up today. arrow-rs
58.4.0 has `NAME = "arrow.parquet.variant"` in
`parquet-variant-compute/src/variant_array.rs`, and Arrow Java 18.3.0's
`SchemaImporter.importField` keeps unregistered extension metadata on the
imported field rather than stripping it, so the marker should survive C Data
import.
--
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]
