Add support for MapSort expression in Spark 4.0.0 #1941

Activity

1. andygrove

   added

   enhancementNew feature or requestNew feature or request

   on Jun 26, 2025
2. andygrove

   added this to the 0.10.0 milestone on Jun 27, 2025
3. rishvin commented on Jul 13, 2025

   

   rishvin

   on Jul 13, 2025

   Contributor

   More actions

   I would like to work on this.
4. rishvin commented on Jul 22, 2025

   

   rishvin

   on Jul 22, 2025

   Contributor

   More actions

   Some Updates:
   I have a simple test to start with, which will produce _groupingmapsort.

   val data = Seq(
        |   Map("a" -> 1, "b" -> 2),
        |   Map("a" -> 3, "b" -> 4),
        |   Map("b" -> 2, "a" -> 1)
        | )
   
   val df = data.toDF("map")
   df.groupBy("map").count().show(false)
   

   So far based on my understanding, it looks like - this will require some plumbing at the Parquet reader utils in Scala, because currently we only support primitive types but MapType is a complex type. I made some hack to bypass some type-checking, however, the Native.initColumnReader() expects primitiveTypeId. I have to understand how the MapType would translate to primitiveTypeId. The closest one at the moment seems to be parquet's BINARY physical type. So, thinking of passing BINARY type and annotating with MapKeyValueTypeAnnotation. However, I'm still trying to understand this piece of code, so I might be wrong. I will do some more experiments to have more clarity on this.
5. rishvin commented on Jul 24, 2025

   

   rishvin

   on Jul 24, 2025

   Last edited by rishvin

   Contributor

   More actions

   Some Updates: I have a simple test to start with, which will produce _groupingmapsort.

   val data = Seq(
        |   Map("a" -> 1, "b" -> 2),
        |   Map("a" -> 3, "b" -> 4),
        |   Map("b" -> 2, "a" -> 1)
        | )
   
   val df = data.toDF("map")
   df.groupBy("map").count().show(false)
   

   So far based on my understanding, it looks like - this will require some plumbing at the Parquet reader utils in Scala, because currently we only support primitive types but MapType is a complex type. I made some hack to bypass some type-checking, however, the Native.initColumnReader() expects primitiveTypeId. I have to understand how the MapType would translate to primitiveTypeId. The closest one at the moment seems to be parquet's BINARY physical type. So, thinking of passing BINARY type and annotating with MapKeyValueTypeAnnotation. However, I'm still trying to understand this piece of code, so I might be wrong. I will do some more experiments to have more clarity on this.

   I managed to scan the map-type by setting CometConf.COMET_NATIVE_SCAN_IMPL.key -> native_datafusion . Added map_sort UDF with return type as Map.

   However, it looks like doing a group-by on the map-type is not possible at the moment. Native code throws error message like - Not yet implemented: not yet implemented: Map(...).

   This error seems to be coming from Arrow. So, presently, we cannot use MapType for grouping, we might have to translate map to a more simple type. Will explore more on this.
6. parthchandra commented on Jul 24, 2025

   

   parthchandra

   on Jul 24, 2025

   Contributor

   More actions

   managed to scan the map-type by setting CometConf.COMET_NATIVE_SCAN_IMPL.key -> native_datafusion . Added map_sort UDF with return type as Map.

   Right. native_comet does not support complex types. You can also try with native_iceberg_compat.
   For map_sort you could try an order by instead of a group by in your test query
7. rishvin commented on Jul 25, 2025

   

   rishvin

   on Jul 25, 2025

   Last edited by rishvin

   Contributor

   More actions

   managed to scan the map-type by setting CometConf.COMET_NATIVE_SCAN_IMPL.key -> native_datafusion . Added map_sort UDF with return type as Map.

   Right. native_comet does not support complex types. You can also try with native_iceberg_compat. For map_sort you could try an order by instead of a group by in your test query

   Thanks @parthchandra. We cannot use order by because spark applies MapSort either for groupby or repartitioning using the following rules,
   InsertMapSortInRepartitioningExpression
   InsertMapSortInGroupingExpression

   I doubled-checked applying orderby on mapType in Spark-4.0 and got,

   The `sortorder` does not support ordering on type "MAP<STRING, INT>". SQLSTATE: 42K09;
   

   I'm attempting if we can do some changes to the plan to use string-representation of map instead of map directly during groupby/shuffling. If this doesn't work correctly or turned out to be too convoluted, I will wait for the complex type support.
8. parthchandra commented on Jul 25, 2025

   

   parthchandra

   on Jul 25, 2025

   Contributor

   More actions

   Oh well.
   For Maps, you might want to look at the mapping of Spark's MapType to a Parquet Schema here:
   

   datafusion-comet/common/src/main/scala/org/apache/spark/sql/comet/parquet/CometSparkToParquetSchemaConverter.scala

   Line 254 in 320ce55

   case MapType(keyType, valueType, valueContainsNull) if writeLegacyParquetFormat =>

   
   

   datafusion-comet/common/src/main/scala/org/apache/spark/sql/comet/parquet/CometSparkToParquetSchemaConverter.scala

   Line 287 in 320ce55

   case MapType(keyType, valueType, valueContainsNull) =>

   Essentially a Map type is a repeated group of key and value fields. Note however that both key and value can themselves be complex types as in this test file:

   datafusion-comet/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala

   Line 719 in 320ce55

   def makeParquetFileComplexTypes(

   👍React with 👍1Reacted by Rishab Joshi
9. andygrove

   removed this from the 0.10.0 milestone on Jul 31, 2025
10. rishvin commented on Aug 20, 2025

    

    rishvin

    on Aug 20, 2025

    Contributor

    More actions

    I have something working for HashAgg on map-type integrated with MapSort. Will open a PR soon.

    👍React with 👍1Reacted by Parth Chandra
11. rishvin commented on Aug 25, 2025

    

    rishvin

    on Aug 25, 2025

    Contributor

    More actions

    Opened #2221
12. rishvin commented on Sep 11, 2025

    

    rishvin

    on Sep 11, 2025

    Contributor

    More actions

    Hi @andygrove / @parthchandra: Please see new updates here: #2381 (comment)
13. andygrove

    added

    spark 4.0

    on Jan 14, 2026
14. andygrove commented on Apr 22, 2026

    

    andygrove

    on Apr 22, 2026

    MemberAuthor

    More actions

    Eleven tests in CometColumnarShuffleSuite are currently skipped with assume(!isSpark40Plus) for this issue:

    • columnar shuffle on map [bool]
    • columnar shuffle on map [byte]
    • columnar shuffle on map [short]
    • columnar shuffle on map [int]
    • columnar shuffle on map [long]
    • columnar shuffle on map [float]
    • columnar shuffle on map [double]
    • columnar shuffle on map [date]
    • columnar shuffle on map [timestamp]
    • columnar shuffle on map [decimal]
    • columnar shuffle on map [string]
    • columnar shuffle on map [binary]

    (12 tests total — the array element variant passes.)

    On Spark 4, every failing test shows the same plan: CometShuffleExchangeExec is replaced by a plain Exchange with hashpartitioning(mapsort(_2#N), mapsort(_3#N), ...). Spark 4 wraps each map partitioning key with MapSort so that equal maps normalize to the same canonical key order before hashing. Comet's shuffle partitioning serde does not recognize MapSort, so native shuffle is rejected for every map-keyed shuffle.
15. andygrove

    assigned

    andygrove

    and unassigned

    andygrove

    on Apr 24, 2026
16. andygrove

    closed this as completedin #4076on Apr 28, 2026