feat: support map inputs for explode

.ai/skills/review-comet-expression-pr/SKILL.md

@@ -30,11 +30,12 @@ bar, and the output format. This skill only covers what is specific to expressio
 
 ## Read the Contributor Guide First
 
-| Doc                                                        | What you need from it                                                                     |
-| ---------------------------------------------------------- | ----------------------------------------------------------------------------------------- |
-| `docs/source/contributor-guide/adding_a_new_expression.md` | The serde contract, support levels, when to set the return type explicitly, shimming      |
-| `docs/source/contributor-guide/sql-file-tests.md`          | The test framework expression PRs are expected to use, and every directive it supports    |
-| `docs/source/contributor-guide/optimizing_expressions.md`  | The benchmark workflow and the no-regression rule, for PRs that change an existing kernel |
+| Doc                                                        | What you need from it                                                                                                                                       |
+| ---------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `docs/source/contributor-guide/adding_a_new_expression.md` | The serde contract, support levels, when to set the return type explicitly, shimming                                                                        |
+| `docs/source/contributor-guide/sql-file-tests.md`          | The test framework expression PRs are expected to use, and every directive it supports                                                                      |
+| `docs/source/contributor-guide/optimizing_expressions.md`  | The benchmark workflow and the no-regression rule, for PRs that change an existing kernel                                                                   |
+| `docs/source/contributor-guide/timezones.md`               | The timestamp label rule, where the session timezone comes from, and how to test it, for datetime expressions, casts, and anything that returns a timestamp |
 
 Hold the diff against these. If the PR does something a guide says to do differently, either the PR
 is wrong or the guide is out of date, and you need to say which.
@@ -77,6 +78,10 @@ Location: `spark/src/main/scala/org/apache/comet/serde/`
 - [ ] A change that routes a whole class of expressions to the JVM codegen dispatcher lists the new
       shapes it admits and tests one of each, including decimal results whose scale differs from the
       declared type and boolean inputs from sliced batches (#6424, #6425)
+- [ ] A timezone-aware expression serializes `CometTimeZone.nativeId(expr.timeZoneId)`, the
+      timezone Spark stamped on it in a form native code can parse, rather than
+      `SQLConf.get.sessionLocalTimeZone` or the JVM default, and returns
+      `CometTimeZone.supportLevel` from `getSupportLevel` when `nativeId` gives `None`
 
 ### Registration in `QueryPlanSerde.scala`
 
@@ -112,6 +117,13 @@ Location: `native/spark-expr/src/`, registered in `comet_scalar_funcs.rs`.
 - [ ] Comparisons, grouping and set operations treat `-0.0` and `0.0`, and NaN payloads, as Spark
       does, including inside arrays and structs. Where a DataFusion kernel treats them differently,
       the serde falls back (#5507, #5701).
+- [ ] A `TimestampType` result is labelled `"UTC"`, in `return_type()` or `data_type()` and in the
+      arrays the kernel builds. Not the session timezone, and not `None`.
+- [ ] Local-time work uses the session timezone from the proto. DataFusion's own datetime functions
+      take the timezone from the input's label, so wiring one in for a timezone-aware Spark
+      expression evaluates it in UTC unless the session timezone is passed explicitly.
+- [ ] A UTC fast path matches a fixed list of UTC aliases, as `is_utc_timezone` in
+      `extract_date_part.rs` does, and sends every other timezone ID down the general path
 
 Before accepting a hand-written kernel, ask whether the function already exists upstream in
 DataFusion or the `datafusion-spark` crate. Comet prefers wiring an upstream function over carrying
@@ -169,8 +181,10 @@ single file by appending a substring of its name to the suite argument.
 - [ ] Edge cases tested: empty input, overflow, boundary values, negative values
 - [ ] Both literal and column arguments tested, in every combination for multi-argument
       expressions. They take different code paths.
-- [ ] Timezone handling tested for timestamp and datetime expressions, including a non-UTC session
-      timezone and timestamps with and without timezone
+- [ ] Timestamp and datetime expressions are tested as "Testing timezone-sensitive code" in
+      `timezones.md` describes. That means several session timezones through `ConfigMatrix`,
+      including `Etc/UTC` and a zone with DST, both `TIMESTAMP` and `TIMESTAMP_NTZ` inputs, and a
+      result that is compared or fed to another expression rather than only projected.
 - [ ] SQL syntax gated with `MinSparkVersion` when it only parses on newer Spark
 - [ ] For a function from another project, such as an Iceberg transform, the expected values come
       from that project's Java implementation, which is what Spark runs, not from iceberg-rust
@@ -254,3 +268,5 @@ reference in that doc is the only place they are documented.
 5. **Missing `getSupportLevel`**, divergences left undeclared rather than marked `Incompatible`
 6. **Version-specific Spark behavior implemented once**, with no shim
 7. **Name collides with a DataFusion built-in** and no explicit return type
+8. **Timestamp result mislabelled**, with the session timezone or no timezone instead of `"UTC"`.
+   It only shows once the result is compared or fed to another expression.

.ai/skills/review-comet-ffi-pr/SKILL.md

@@ -138,6 +138,11 @@ a new subclass needs no new case as long as `getValueVector` returns an Arrow ve
       `ColumnarBatchArrowReader` decodes dictionary-encoded columns before export, and
       `reconcileStreamSchema` advertises the value type. A new reader that exports a dictionary
       would reach `ScanExec` as is, and only the cast in `build_record_batch` would unpack it.
+- [ ] **Timestamps cross unconverted.** Both directions pass the raw microseconds. JVM producers
+      label `TimestampType` with `CometArrowStream.NATIVE_TIMEZONE`, which is `"UTC"`, and a new
+      producer must use it too rather than the session timezone. A change that shifts values by a
+      timezone at the boundary gives wrong answers. See "How Comet represents timestamps" in
+      `docs/source/contributor-guide/timezones.md`.
 
 ## 5. Memory Accounting
 

.ai/skills/review-comet-iceberg-write-pr/SKILL.md

@@ -111,6 +111,16 @@ only the latest.
       `Float.compare` does not
       ([#6138](https://github.com/apache/datafusion-comet/issues/6138)). Check equality, hashing,
       ordering and rendering for float, double, timestamp, timestamptz, binary and decimal.
+- [ ] **Timestamp partition values are UTC.** Iceberg's `years`, `months`, `days` and `hours`
+      ignore the session timezone. For date and timestamp sources, `PartitionValueCalculator`
+      (`iceberg_partition_value.rs`) computes them with the same kernels in
+      `iceberg_funcs/temporal.rs` that key the sort in front of a clustered write, and those follow
+      iceberg-java's `DateTimeUtil`, including its pre-1970 rounding. A change that moves one of
+      these transforms back to iceberg-rust, or that makes the kernels read the column's timezone
+      label, breaks that agreement. iceberg-rust's `years` and `months` follow the label
+      ([apache/iceberg-rust#3142](https://github.com/apache/iceberg-rust/issues/3142)). Ask for a
+      test in a non-UTC session for a change here. "Partition values" in `iceberg-writes.md` lists
+      where iceberg-rust differs, and `timezones.md` covers how Comet labels timestamps.
 - [ ] **Partition paths use the Java renderers.** Directory names come from
       `CometLocationGenerator` / `partition_to_path` in `iceberg_partition_path.rs`, which follow
       iceberg-java's `partitionToPath`, with `java_float_string` for floats. A PR that calls
@@ -142,6 +152,14 @@ Read the ownership table in the contributor guide before reviewing any change ne
       `SparkWrite.abort` keeps working.
 - [ ] The user-visible exception type is still what Spark's own write path throws on each Spark
       version ([#6143](https://github.com/apache/datafusion-comet/issues/6143)).
+- [ ] **Every open file and every held-back row is reserved.** Files must be opened through
+      `MeteredParquetWriterBuilder`, and rows the writer holds outside iceberg-rust (the
+      `PartitionFeed`s, or anything a new feed holds back) must be added to what `run_write_task` reserves. A
+      builder that constructs `ParquetWriterBuilder` directly leaves its files out of the task's
+      memory reservation, so a wide fanout write grows past the pool instead of failing its task.
+      A storage scheme newly supported for writes needs its entry in
+      `StorageWrites::for_location`: a file reports its flushed row groups until its storage has
+      written them out, which a local file does at once and an object store only part by part.
 
 ## 5. Plan Shape
 
@@ -179,16 +197,16 @@ Read the ownership table in the contributor guide before reviewing any change ne
 
 ## 7. Tests
 
-| Suite                               | Covers                                                                               |
-| ----------------------------------- | ------------------------------------------------------------------------------------ |
-| `CometIcebergWriteActionSuite`      | Both layers end to end, parity with iceberg-java, DML, failure cleanup, AQE re-plans |
-| `CometIcebergWriteDetectionSuite`   | One case per eligibility rule                                                        |
-| `CometIcebergSystemFunctionSuite`   | Native partition transforms keeping partitioned writes native                        |
-| `CometIcebergRewriteActionSuite`    | `rewrite_data_files` through the split plan and the native writer                    |
-| `IcebergWriteProtoTranslationSuite` | Property to `IcebergParquetWriteSettings` translation                                |
-| Rust tests in `iceberg_write.rs`    | Rolling, fanout order, clustered checks, cleanup guard, manifest round trip          |
-| `iceberg_partition_path.rs` tests   | Partition path rendering against iceberg-java                                        |
-| `CometIcebergWriteBenchmark`        | Native versus iceberg-java throughput                                                |
+| Suite                               | Covers                                                                                          |
+| ----------------------------------- | ----------------------------------------------------------------------------------------------- |
+| `CometIcebergWriteActionSuite`      | Both layers end to end, parity with iceberg-java, DML, failure cleanup, AQE re-plans            |
+| `CometIcebergWriteDetectionSuite`   | One case per eligibility rule                                                                   |
+| `CometIcebergSystemFunctionSuite`   | Native partition transforms keeping partitioned writes native                                   |
+| `CometIcebergRewriteActionSuite`    | `rewrite_data_files` through the split plan and the native writer                               |
+| `IcebergWriteProtoTranslationSuite` | Property to `IcebergParquetWriteSettings` translation                                           |
+| Rust tests in `iceberg_write.rs`    | Rolling, fanout order, clustered checks, cleanup guard, manifest round trip, memory reservation |
+| `iceberg_partition_path.rs` tests   | Partition path rendering against iceberg-java                                                   |
+| `CometIcebergWriteBenchmark`        | Native versus iceberg-java throughput                                                           |
 
 Ask specifically:
 

.ai/skills/review-comet-pr/SKILL.md

@@ -47,6 +47,11 @@ change.
 For a new operator, read `docs/source/contributor-guide/adding_a_new_operator.md` alongside this
 skill. There is no dedicated operator review skill yet.
 
+For a PR that deals with timestamps or the session timezone, read
+`docs/source/contributor-guide/timezones.md` as well, whichever areas it touches. Timezone handling
+cuts across the serde, the native kernels, the scans and the JVM/native boundary, so no area skill
+owns it. "Timestamps and timezones" in step 5 says what to check.
+
 If the PR falls outside all of these, for example build, CI, docs only, or release tooling, this
 skill on its own is the review.
 
@@ -176,6 +181,41 @@ the minor version is wrong for every earlier patch. CI builds only the newest pa
 it can't catch that (#6042, #5701). The pull request CI also runs only the default Spark profile, so
 logic that depends on the Spark version needs the matching `run-spark-*` labels (#6156).
 
+### Timestamps and timezones
+
+Timezone bugs are easy to miss in review and in tests. A mislabelled timestamp column passes any
+test that only projects it, and a result computed in the wrong timezone looks plausible. A PR deals
+with timezones if it touches a datetime expression, a cast to or from a timestamp, how a scan reads
+timestamps, the timestamp type at the JVM/native boundary, or anything that reads the session
+timezone. Searching the diff flags most of these PRs:
+
+```shell
+gh pr diff <pr> --repo apache/datafusion-comet | grep -inE 'time_?zone|zoneid|chrono_tz|timestamp(ntz)?type|timestampmicro|timestamp\('
+```
+
+For such a PR, hold the diff against "The invariant" and "Guidelines" in
+`docs/source/contributor-guide/timezones.md`. Look for these first:
+
+- A `TimestampType` value labelled with the session timezone, or with no timezone. Inside a native
+  plan every `TimestampType` value is labelled exactly `"UTC"`, and every `TimestampNTZType` value
+  has no timezone. Check the declared type and the arrays the code builds, not only the values.
+- A timezone taken from the JVM default or the host. An expression uses the `timeZoneId` Spark
+  stamped on it, passed through `CometTimeZone.nativeId`, not `SQLConf.get.sessionLocalTimeZone`.
+- A path gated on a UTC session. `Etc/UTC` is the session default on Ubuntu and Debian images, so
+  check what the gate does with it, and that the output there is still labelled `"UTC"` rather than
+  `"Etc/UTC"`.
+- A timezone applied to a `TimestampNTZType` value, other than to convert it to `TimestampType`.
+- Tests that use a single session timezone, or only project the result. "Testing timezone-sensitive
+  code" in the same page says what to ask for.
+
+`timezones.md` also describes specific code: where the `"UTC"` label is set, which serdes serialize
+a timezone, how `CometTimeZone` rewrites timezone IDs, what `array_with_timezone` does, how the
+scans adapt timestamps, and which expressions go through the codegen dispatcher. A PR that changes
+any of these updates the page in the same PR. The page also documents some known limitations as
+current behavior, such as chrono-tz's DST horizon and the timezone database versions. A fix for one
+of the bugs tracked in [#6335](https://github.com/apache/datafusion-comet/issues/6335) usually
+changes that text too.
+
 ### Configuration
 
 New configs go in `CometConf.scala` and must follow

.github/workflows/pr_build_linux.yml

@@ -460,6 +460,7 @@ jobs:
             value: |
               org.apache.comet.parquet.CometParquetWriterSuite
               org.apache.comet.parquet.CometEmptyRelationParquetWriterSuite
+              org.apache.spark.sql.comet.CometNativeWriteSuite
               org.apache.comet.parquet.ParquetReadV1Suite
               org.apache.comet.parquet.ParquetTimestampLtzAsNtzSuite
               org.apache.spark.sql.comet.ParquetEncryptionITCase
@@ -486,6 +487,7 @@ jobs:
               org.apache.comet.exec.CometShuffle4_0Suite
               org.apache.comet.exec.CometNativeColumnarToRowSuite
               org.apache.comet.exec.CometNativeShuffleSuite
+              org.apache.comet.exec.CometShuffleInputConversionSuite
               org.apache.comet.shuffle.CelebornShufflePartitionPusherSuite
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornShuffleManagerSuite
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornNativeShuffleWriterSuite
@@ -503,6 +505,7 @@ jobs:
               org.apache.comet.exec.CometShuffleManagerSuite
               org.apache.comet.exec.DisableAQECometShuffleSuite
               org.apache.spark.shuffle.comet.CometUnboundedShuffleMemoryAllocatorSuite
+              org.apache.spark.shuffle.comet.CometUnifiedShuffleMemoryAllocatorSuite
               org.apache.spark.shuffle.sort.SpillSorterSuite
           - name: "exec"
             value: |
@@ -563,6 +566,7 @@ jobs:
               org.apache.spark.sql.comet.PlanDataInjectorShuffleLifecycleSuite
               org.apache.spark.sql.comet.CometDecimalArithmeticViewSuite
               org.apache.spark.sql.comet.CometDecimalPromotionSuite
+              org.apache.spark.sql.comet.DecimalPrecisionSuite
               org.apache.spark.sql.comet.CometScanWithPlanDataSuite
               org.apache.spark.sql.comet.util.UtilsSuite
               org.apache.comet.vector.NativeUtilSuite

.github/workflows/pr_build_macos.yml

@@ -164,6 +164,7 @@ jobs:
             value: |
               org.apache.comet.parquet.CometParquetWriterSuite
               org.apache.comet.parquet.CometEmptyRelationParquetWriterSuite
+              org.apache.spark.sql.comet.CometNativeWriteSuite
               org.apache.comet.parquet.ParquetReadV1Suite
               org.apache.comet.parquet.ParquetTimestampLtzAsNtzSuite
               org.apache.spark.sql.comet.ParquetEncryptionITCase
@@ -190,6 +191,7 @@ jobs:
               org.apache.comet.exec.CometShuffle4_0Suite
               org.apache.comet.exec.CometNativeColumnarToRowSuite
               org.apache.comet.exec.CometNativeShuffleSuite
+              org.apache.comet.exec.CometShuffleInputConversionSuite
               org.apache.comet.shuffle.CelebornShufflePartitionPusherSuite
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornShuffleManagerSuite
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornNativeShuffleWriterSuite
@@ -207,6 +209,7 @@ jobs:
               org.apache.comet.exec.CometShuffleManagerSuite
               org.apache.comet.exec.DisableAQECometShuffleSuite
               org.apache.spark.shuffle.comet.CometUnboundedShuffleMemoryAllocatorSuite
+              org.apache.spark.shuffle.comet.CometUnifiedShuffleMemoryAllocatorSuite
               org.apache.spark.shuffle.sort.SpillSorterSuite
           - name: "exec"
             value: |
@@ -267,6 +270,7 @@ jobs:
               org.apache.spark.sql.comet.PlanDataInjectorShuffleLifecycleSuite
               org.apache.spark.sql.comet.CometDecimalArithmeticViewSuite
               org.apache.spark.sql.comet.CometDecimalPromotionSuite
+              org.apache.spark.sql.comet.DecimalPrecisionSuite
               org.apache.spark.sql.comet.CometScanWithPlanDataSuite
               org.apache.spark.sql.comet.util.UtilsSuite
               org.apache.comet.vector.NativeUtilSuite

.github/workflows/spark_sql_writer_tests.yml

@@ -115,6 +115,13 @@ jobs:
           spark-short-version: ${{ inputs.spark-version }}
           skip-native-build: true
 
+      - name: Pre-compile Spark SQL test classes
+        run: |
+          cd apache-spark
+          rm -rf /root/.m2/repository/org/apache/parquet
+          NOLINT_ON_COMPILE=true build/sbt -Dsbt.log.noformat=true -mem 3072 \
+            'sql/Test/compile'
+
       - name: Run Parquet writer tests
         run: |
           cd apache-spark