[rizzabhdwaj]

Which issue does this PR close?

• Closes Implement Aggregate function map_agg #22993

Rationale for this change

DataFusion has no map_agg aggregate function, even though equivalent functionality exists in
other engines (Trino/Presto map_agg, Spark map_from_arrays, PostgreSQL json_object_agg).
Users currently have no way to aggregate (key, value) pairs from many rows into a single
Map value — the map equivalent of array_agg.

What changes are included in this PR?

• Added a new map_agg(key, value) aggregate UDAF in
  datafusion/functions-aggregate/src/map_agg.rs.

  • MapAgg implements AggregateUDFImpl with Signature::any(2, Volatility::Immutable).
  • Returns an Arrow Map with a non-nullable key field and nullable value field.
  • MapAggAccumulator collects key/value batches and concatenates them in evaluate()
    to produce a single-row MapArray scalar.
  • state() / merge_batch() support partial aggregation, including Partial + Final
    aggregation plans.
  • Empty input produces a NULL map, consistent with array_agg.
  • Null keys are rejected with map key cannot be null, matching the behavior of the
    map() UDF.

• Registered map_agg in all_default_aggregate_functions() and exported it via
  expr_fn::map_agg.

• Regenerated
  docs/source/user-guide/sql/aggregate_functions.md.

Testing strategy

Unit tests

Added unit tests in datafusion/functions-aggregate/src/map_agg.rs covering:

• map_agg_builds_map_from_batches — verifies aggregation across multiple batches,
  including nullable values.
• map_agg_empty_group_produces_null_map — verifies empty input produces a NULL map.
• map_agg_null_key_errors — verifies null keys are rejected.
• map_agg_state_merge_roundtrip — verifies state() / merge_batch() preserve the
  aggregate result.
• map_agg_update_batch_rejects_mismatched_lengths — verifies key/value length validation.
• Name, return type, and UDAF registration.

SQL logic tests

Added sqllogictest coverage in:

datafusion/sqllogictest/test_files/map_agg.slt

The tests cover:

• GROUP BY aggregation.
• NULL values.
• NULL key rejection.
• Empty input producing NULL.
• Multi-batch aggregation with batch_size = 1.

For example:

SELECT k, map_agg(v, m)
FROM (
    VALUES
        ('a', 1, 10),
        ('a', 2, 20),
        ('b', 3, 30)
) t(k, v, m)
GROUP BY k
ORDER BY k;


## expected output 
+---+-----------------+
| k | map_agg(v, m)   |
+---+-----------------+
| a | {1: 10, 2: 20}  |
| b | {3: 30}         |
+---+-----------------+

[neilconway]

Rather than starting from scratch, I wonder whether it would be better to resurrect / finish #22996, which seems like it was quite close to being ready to merge. Or if we're going to re-implement it, at least consider the aspects that the previous PR paid attention to: DISTINCT, ORDER BY, behavior for duplicate keys (including docs and tests), and so on.

[rizzabhdwaj]

Notes

This implementation takes inspiration from the previous #22996 implementation and its review discussion, particularly around handling the cases that were called out there.

This PR currently covers:

• ORDER BY within map_agg
• GROUP BY usage
• First-occurrence-wins semantics for duplicate keys
• Tests covering the above behavior

For ordered aggregation, duplicate keys are de-duplicated after applying the aggregate's ORDER BY, so the first occurrence in the resulting order is retained.

[kumarUjjawal]

Thank you @rishabh1244 for working on this.

I left one comment please take a look.

[rizzabhdwaj]

Thanks for reviewing and pointing out required changes , I've pushed the fixes , Whenever you get a chance, could you take a look?

[rizzabhdwaj]

Hey @kumarUjjawal i have done the mentioned changes kindly take some time and look into it.

[kumarUjjawal]

@rizzabhdwaj Please run the local ci setup to make sure CI is green.

[kumarUjjawal]

Both accumulators must handle NULL keys the same way. Please reject them in update_batch with one message, map key cannot be null. Then update the SLT test and the module docs that still say the pair is dropped. The reverse logic is also wrong when the input is not pre-sorted. The simplest fix is to sort by the ORDER BY as given and reverse only on output, like array_agg does. You can also return NotSupported from reverse_expr for now. DISTINCT is ignored right now, so please document it with a test, or reject it. The two accumulators also repeat the same decode and map-build code. Please make them share map_row_to_scalars and build_single_map, and move those helpers into map_agg.rs. In the docs, please say that the result order without ORDER BY is not guaranteed. Also say that NULL keys fail and that empty input returns NULL. Last, please add tests for multi-partition reversed input, window use, and nested values.

[rizzabhdwaj]

Both accumulators must handle NULL keys the same way. Please reject them in update_batch with one message, map key cannot be null. Then update the SLT test and the module docs that still say the pair is dropped. The reverse logic is also wrong when the input is not pre-sorted. The simplest fix is to sort by the ORDER BY as given and reverse only on output, like array_agg does. You can also return NotSupported from reverse_expr for now. DISTINCT is ignored right now, so please document it with a test, or reject it. The two accumulators also repeat the same decode and map-build code. Please make them share map_row_to_scalars and build_single_map, and move those helpers into map_agg.rs. In the docs, please say that the result order without ORDER BY is not guaranteed. Also say that NULL keys fail and that empty input returns NULL. Last, please add tests for multi-partition reversed input, window use, and nested values.

ive made the necessary changes.