viirya commented on code in PR #6036:
URL: https://github.com/apache/datafusion-comet/pull/6036#discussion_r4054174214
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -3821,6 +3834,78 @@ class CometExecSuite extends CometTestBase {
}
}
+ test("SparkToColumnar string maps cross RDD and Parquet native boundaries") {
Review Comment:
This test is close to line-for-line with "SparkToColumnar string arrays
cross JSON and Parquet native boundaries" at line 3772 — same conf block, same
`conversions.size == 1` / `supportsColumnar` assertions, same native-shuffle
check, same `limit(1)` early-stop, same disabled-control block. The differences
are the schema, the row fixtures, and the projected expression.
Worth folding into one test parameterized over the collection type. If the
gate becomes recursive per my top-level comment, this falls out naturally as
one more entry in a `dataType` loop rather than a third copy.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometSparkToColumnarExec.scala:
##########
@@ -142,6 +142,7 @@ object CometSparkToColumnarExec extends
CometSink[SparkPlan] with DataTypeSuppor
name: String,
fallbackReasons: ListBuffer[String]): Boolean = dt match {
case ArrayType(StringType, _) => true
+ case MapType(StringType, StringType, _) => true
Review Comment:
This line depends on a Scala/Spark subtlety that nothing here records: in
Spark 4.x `StringType` is a `case object` extending
`StringType(UTF8_BINARY_COLLATION_ID, NoConstraint)`, and its `equals` compares
`collationId` and `constraint`. So this pattern compiles to an object-equality
check and a collated `StringType` instance does not match — which is exactly
what makes the `UTF8_LCASE` assertions below pass.
The hazard is that "tidying" this to `MapType(_: StringType, _: StringType,
_)` looks equivalent, compiles fine, and *silently* drops the collation guard.
Same applies to the `ArrayType(StringType, _)` line above, inherited from #5954.
Could you add something like:
```scala
// `StringType` is the UTF8_BINARY case object; a collated StringType
instance
// is not equal to it, so collated maps/arrays fall through to the reject
case
// below. Do NOT rewrite these as `_: StringType`.
```
Also worth noting: the `case _: ArrayType | _: MapType => false` catch-all
on the next line is order-dependent, and every future supported type has to be
inserted above it. A brief note would help.
##########
docs/source/user-guide/latest/datasources.md:
##########
@@ -31,7 +31,7 @@ Arrow format, allowing the Comet pipeline to take over after
that, but the proce
Comet accelerates Iceberg scans of Parquet files. See the [Iceberg Guide] for
more information.
-[Iceberg Guide]: iceberg.md
+[iceberg guide]: iceberg.md
Review Comment:
Unrelated to this PR — this is prettier normalizing the reference-link label
to lowercase. Harmless (CommonMark matches labels case-insensitively, so
`[Iceberg Guide]` on line 32 still resolves), but it's drive-by churn in a docs
section the PR doesn't otherwise touch. Either drop it or mention it in the
description so reviewers don't have to work out whether the link broke.
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -3821,6 +3834,78 @@ class CometExecSuite extends CometTestBase {
}
}
+ test("SparkToColumnar string maps cross RDD and Parquet native boundaries") {
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("partitionValues", MapType(StringType, StringType))
+ .add("nested", new StructType().add("tags", MapType(StringType,
StringType)))
+ val rows = Seq(
+ Row(1, null, null),
+ Row(2, Map.empty[String, String], Row(null)),
+ Row(3, Map("hour" -> null), Row(Map.empty[String, String])),
+ Row(
+ 4,
+ Map("hour" -> "2026-09-02T22", "" -> "é", "東京" -> "a\u0000b"),
+ Row(Map("nullable" -> null, "x" -> "value"))),
+ Row(5, Map("hour" -> "2026-09-03T00", "long" -> ("東京" * 32768)), null),
+ Row(6, Map("hour" -> "2026-09-03T01"), Row(Map("x" -> "value"))))
+ for (sourceType <- Seq("rdd", "parquet-row", "parquet-columnar")) {
+ val vectorized = sourceType == "parquet-columnar"
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString,
+ "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true",
+ CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
+ CometConf.COMET_BATCH_SIZE.key -> "2",
+ CometConf.COMET_SHUFFLE_MODE.key -> "native",
+ CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true",
+ CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
+ CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key ->
"RDDScan") {
Review Comment:
This conf block never sets `SQLConf.USE_V1_SOURCE_LIST`, so the Parquet
cases here only reach the V1 `FileSourceScanExec` branch of
`shouldApplySparkToColumnar`. The array test at line 3772 loops `v1 ∈ {"",
format}` and therefore also covers the `BatchScanExec` (DSv2 `ParquetScan`)
branch, which is a separate `case` in `CometExecRule`.
So the DSv2 admission path is currently untested for maps — the existing
"SparkToColumnar over BatchScan" test doesn't include a map column either.
Adding `v1` to the loop the same way the array test does would close this. Same
question for JSON: the array test covers it and this one drops it, and the PR
description doesn't say whether that was deliberate.
##########
spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CometArrowStreamSuite.scala:
##########
@@ -599,6 +599,182 @@ class CometArrowStreamSuite extends AnyFunSuite with
Matchers {
}
}
+ for (nullable <- Seq(false, true); valueContainsNull <- Seq(false, true);
+ columnar <- Seq(false, true)) {
+ test(s"string maps preserve slices and ownership:
$nullable/$valueContainsNull/$columnar") {
+ val mapType = MapType(StringType, StringType, valueContainsNull)
+ val schema = StructType(Seq(StructField("tags", mapType, nullable)))
+ val arrowSchema = Utils.toArrowSchema(schema, "UTC")
+ val values: Seq[Seq[(String, String)]] = Seq(
+ Seq("" -> "", "é" -> "東京", "a\u0000b" -> "duplicate", "b" ->
"duplicate"),
Review Comment:
Minor, but the `"duplicate"` labels read as if this row covers duplicate map
*keys* — the keys here (`a\u0000b`, `b`) are distinct and it's the *values*
that repeat. The array test's `"dup", "dup"` at line 365 is genuinely about
repeated elements, so the parallel naming is misleading for maps.
Duplicate keys are the case actually worth having:
`mapKeyDedupPolicy=LAST_WIN` lets them reach execution, Arrow Map doesn't
enforce uniqueness, and the `CometExecSuite` test already does
`partitionValues['hour']` lookups. Adding a row with a repeated key and
asserting the lookup matches Spark would cover the one map-specific semantic
that arrays don't have.
--
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]
