Implement Spark-compatible `schema_of_variant_agg`

Đang mở
  • Có gì không ổn?
#279 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
30/100
Loại issue
Tính năng
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
rust, sql
Lĩnh vực
data

Hướng nghiên cứu

Start by reading the existing schema_of_variant implementation for the type lattice, mergeSchema and the Spark-type printer, since this aggregate must share them; then look at how other UDAFs are registered under crates/spark/src/function/aggregate/ or the planned variant/ module. The work is blocked in practice on the VARIANT Arrow representation proposed in the parse_json issue, so confirm that landed before starting. Reproduce the listed SQL examples and the VariantEndToEndSuite grouping cases in a new crates/sqllogictest/test_files/variant/schema_of_variant_agg.slt, and document the function with #[user_doc].

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

enhancement

Part of #273 and apache/datafusion-comet#7300.

What the function does

schema_of_variant_agg(v) is an aggregate. It merges the schemas of all non-NULL VARIANT values in the group and
returns the result as a STRING. The output format is the same as schema_of_variant: Spark SQL type syntax with
OBJECT<...> for objects. The argument must be VARIANT. The result is non-nullable and uses the default string
type. Added in Spark 4.0.0.

The buffer starts as VOID (NullType). Each row merges SchemaOfVariant.schemaOf(v) into it with
SchemaOfVariant.mergeSchema, and partial buffers are combined the same way. SQL NULL rows are skipped. The merge
and print rules are the ones in the schema_of_variant issue, so the two should share code.

Examples
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES ('1'), ('2'), ('3') AS tab(j);
-- BIGINT
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES ('{"a": 1}'), ('{"b": true}'), ('{"c": 1.23}') AS tab(j);
-- OBJECT<a: BIGINT, b: BOOLEAN, c: DECIMAL(3,2)>
SELECT schema_of_variant_agg(parse_json('{"a": [1, 2, 3]}'));
-- OBJECT<a: ARRAY<BIGINT>>
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES ('{"a": [1, 2, 3]}'), ('{"a": "banana"}') AS tab(j);
-- OBJECT<a: VARIANT>
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES ('{"a": "banana"}'), ('{"b": "apple"}') AS tab(j);
-- OBJECT<a: STRING, b: STRING>
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES ('{"a": "data"}'), (null) AS tab(j);
-- OBJECT<a: STRING>
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES (null), (null) AS tab(j);
-- VOID
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES ('{"a": null}'), ('{"a": null}') AS tab(j);
-- OBJECT<a: VOID>
SELECT schema_of_variant_agg(parse_json(j)) FROM VALUES
  ('{"hi":[]}'), ('{"hi":[{},{}]}'), ('{"hi":[{"it''s":[{"me":[{"a": 1}]}]}]}') AS tab(j);
-- OBJECT<hi: ARRAY<OBJECT<`it's`: ARRAY<OBJECT<me: ARRAY<OBJECT<a: BIGINT>>>>>>>

These come from the ExpressionDescription and VariantEndToEndSuite."schema_of_variant_agg". The suite builds
the inputs as DataFrames, and the queries above are the equivalent SQL.

Edge cases to test
  • Grouping. VariantEndToEndSuite uses rows id in 0..100 with {"a": [id]} when id % 4 == 0 and
    {"a": ["id"]} otherwise:
    • GROUP BY id % 2 gives OBJECT<a: ARRAY<STRING>> and OBJECT<a: ARRAY<VARIANT>>;
    • % 3 gives OBJECT<a: ARRAY<VARIANT>> three times;
    • % 4 gives OBJECT<a: ARRAY<STRING>> three times and OBJECT<a: ARRAY<BIGINT>> once.
      These exercise the multi-partition merge path.
  • If every input is NULL the result is VOID. With no input rows the buffer is never updated, so by code reading
    the result is also VOID, not NULL, because the function is non-nullable.
  • A merge across rows behaves like a merge across array elements. For example, BIGINT with DOUBLE gives
    DOUBLE, objects are unioned with sorted field names, and incompatible types give VARIANT. See the full table
    in the schema_of_variant issue.
  • Order independence: decimal widening, struct union and VOID absorption are commutative. One exception, by code
    reading, is struct merging through findTightestCommonType, which keeps the first field name when names are
    equal case-insensitively (spark.sql.caseSensitive=false). The result could depend on the order of rows or
    partitions, so do not test that case for determinism.
  • Field-name quoting with backticks follows QuotingUtils.quoteIfNeeded. For example, !special! is quoted and
    STRUCT is not.
  • Malformed input raises MALFORMED_VARIANT (unsorted object fields) or UNKNOWN_PRIMITIVE_TYPE_IN_VARIANT, the
    same as schema_of_variant.
  • Test a literal argument, a column argument, GROUP BY, and use as a window aggregate if datafusion-spark
    supports that for UDAFs.
Differences across Spark versions
  • 3.x has no VARIANT.
  • No differences between 4.0.4, 4.1.3 and 4.2.0.
Implementation notes
  • Spark: SchemaOfVariantAgg (a TypedImperativeAggregate[DataType]) in
    sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala. Spark
    serializes the buffer as DataType.json, which is internal.
  • In DataFusion, write an AggregateUDFImpl / Accumulator whose state is the in-progress Spark type tree. The
    state can be a single Utf8 field, with the type serialized in any internal form. merge_batch applies the same
    mergeSchema as update_batch. Reuse the type lattice and printer from schema_of_variant.
  • DataFusion core and arrow-rs have no equivalent. Use parquet_variant::Variant to decode values. This depends on
    the VARIANT Arrow representation proposed in the parse_json issue.
  • apache/datafusion-comet#5438 tracks native Variant support in Comet.
  • Put it under crates/spark/src/function/aggregate/ or the new variant/ module. Add
    crates/sqllogictest/test_files/variant/schema_of_variant_agg.slt and document the function with #[user_doc].

This issue was generated by an LLM (Claude) from the Apache Spark source code and tests. Examples come from Spark's ExpressionDescription examples, test suites and .sql.out golden files, but please verify them against Spark before relying on them.

Ngôn ngữ chính
Rust
Star
7
Fork
15
Merge trung bình
16 giờ 43 phút
Pull request đã merge (30 ngày)
26

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/datafusion-spark

Tất cả issue của apache/datafusion-spark

Issue tương tự

Thêm issue về Rust

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.