You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2026/09/19 04:28:05 UTC
[PR] feat: admit string maps in Spark-to-Comet conversion [datafusion-comet]
sunchao opened a new pull request, #6036:
URL: https://github.com/apache/datafusion-comet/pull/6036
## Which issue does this PR close?
Follow-up to #5954, extending the same Spark-to-Comet conversion boundary to `MAP<STRING,STRING>`.
## Rationale for this change
The conversion gate still rejects every map type. Consequently, an explicitly enabled RDD or Parquet input containing a string map cannot feed otherwise supported native filters and projections. For example, a row-backed input with `attributes MAP<STRING,STRING>` falls back even when `RDDScan` conversion is enabled.
The existing Arrow map writer already supports this representation. Admit binary string keys and values through the conversion gate, including nullable maps and values and placement inside supported structs. Existing source opt-ins remain in effect.
## What changes are included in this PR?
- Add the narrow `MapType(StringType, StringType, _)` admission case.
- Document supported nullability and explicit `RDDScan` configuration.
- Cover both reader paths, retained output ownership, input reuse and slices, encoding-failure cleanup, nested structs, native shuffle round trips, and disabled-conversion controls. Other collection types and non-binary collations remain rejected.
## How are these changes tested?
Local validation used Linux x86_64, JDK 21, Spark 4.1.3 and Scala 2.13.17, with a freshly built native debug library.
- Complete `CometArrowStreamSuite` plus the `SparkToColumnar` tests in `CometExecSuite`: **48 passed**, none failed or skipped.
- Clean reactor test compilation, Spotless, Scalastyle, scoped Markdown formatting, and `git diff --check`: passed.
The map integration regression checks Spark result/schema parity and the actual conversion, native filter/projection, and native shuffle operators for RDD and Parquet inputs. Source conversion disabled controls retain fallback. Spark's broader SQL suite is requested through `run-spark-4.1-tests`.
AI assistance: adapted and validated with OpenAI Codex.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: admit string maps in Spark-to-Comet conversion [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: admit string maps in Spark-to-Comet conversion [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao merged PR #6036:
URL: https://github.com/apache/datafusion-comet/pull/6036
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org