feat: admit string maps in Spark-to-Comet conversion

[sunchao]

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.

[rich7420]

@sunchao thanks for the patch, LGTM

[viirya]

Thanks for the follow-up to #5954 — the shape of the change is right, and the CometArrowStreamSuite additions are unusually thorough (nonzero child offsets, zeroing the source buffers before asserting, ownership after closing the output, and an allocator-leak check on the encoding-failure path). I verified the core mechanics and they hold up: the JVM-side Arrow schema from Utils.toArrowField matches what native serde.rs builds (entries/key/value, sorted=false, non-null key), and the collation exclusion genuinely works.

My main question is about the shape of the gate rather than this line being wrong.

The narrow case contradicts a sibling gate over the same machinery. CometLocalTableScanExec overrides isTypeSupported only for NullType/interval types and otherwise falls through to DataTypeSupport's recursive rule, so it already admits arbitrary maps — including MAP<STRING,INT>, MAP<INT,STRING> and MAP<STRING,ARRAY<STRING>>, all of which this PR explicitly asserts are unsupported. Both operators mix in CometNativeArrowSource and convert through RowArrowReader → ArrowWriter, and ArrowWriter.createFieldWriter recurses into ListVector/MapVector children for every writer it knows. So the conversion machinery is not what's limiting us here.

That leaves two readings, and I can't tell which is intended: either the restriction is purely incremental, in which case dropping the override and inheriting the recursive rule would fix the inconsistency and pre-empt a long tail of one-line PRs; or some of those types are actually broken on this path, in which case CometLocalTableScanExec has a latent bug and the reason belongs in a code comment. Could you say which?

Test coverage regressed against the array test it mirrors. The array test loops v1 ∈ {"", format}, so it reaches the BatchScanExec (DSv2) branch of shouldApplySparkToColumnar. The map test never sets USE_V1_SOURCE_LIST, so it only exercises V1 FileSourceScanExec and RDDScanExec. That branch is separate in CometExecRule, and the existing "SparkToColumnar over BatchScan" test has no map either. JSON coverage was also dropped without explanation.

No duplicate-key coverage. spark.sql.mapKeyDedupPolicy=LAST_WIN lets duplicate keys survive into execution, and Arrow Map doesn't enforce uniqueness. Since the test already exercises partitionValues['hour'], it'd be worth confirming native map_extract picks the same entry Spark does.

Two nearly identical suite tests. The new test is close to line-for-line with the array one (same conf block, same conversions.size == 1, same shuffle assertions, same disabled-control). Worth parameterizing — and if the gate goes recursive, they collapse into one loop naturally.

Smaller things are inline. Nothing here is a correctness objection to the added line itself; I'd just like the design question answered before this lands, since each additional narrow case makes the eventual cleanup larger.

[viirya]

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:

// `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.

[viirya]

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.

[viirya]

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.

[viirya]

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.

[viirya]

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.