Implement Spark-compatible `schema_of_variant_agg`
メンテナーはふだん 1 日以内に返信
まだ誰も着手していません。
評価
調査の方向性
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].
索引モデルが issue の本文から書いたものです。
説明
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.
VariantEndToEndSuiteuses rowsidin 0..100 with{"a": [id]}whenid % 4 == 0and
{"a": ["id"]}otherwise:GROUP BY id % 2givesOBJECT<a: ARRAY<STRING>>andOBJECT<a: ARRAY<VARIANT>>;% 3givesOBJECT<a: ARRAY<VARIANT>>three times;% 4givesOBJECT<a: ARRAY<STRING>>three times andOBJECT<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 alsoVOID, not NULL, because the function is non-nullable. - A merge across rows behaves like a merge across array elements. For example,
BIGINTwithDOUBLEgives
DOUBLE, objects are unioned with sorted field names, and incompatible types giveVARIANT. See the full table
in theschema_of_variantissue. - Order independence: decimal widening, struct union and
VOIDabsorption are commutative. One exception, by code
reading, is struct merging throughfindTightestCommonType, 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
STRUCTis not. - Malformed input raises
MALFORMED_VARIANT(unsorted object fields) orUNKNOWN_PRIMITIVE_TYPE_IN_VARIANT, the
same asschema_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(aTypedImperativeAggregate[DataType]) in
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala. Spark
serializes the buffer asDataType.json, which is internal. - In DataFusion, write an
AggregateUDFImpl/Accumulatorwhose state is the in-progress Spark type tree. The
state can be a singleUtf8field, with the type serialized in any internal form.merge_batchapplies the same
mergeSchemaasupdate_batch. Reuse the type lattice and printer fromschema_of_variant. - DataFusion core and arrow-rs have no equivalent. Use
parquet_variant::Variantto decode values. This depends on
the VARIANT Arrow representation proposed in theparse_jsonissue. - apache/datafusion-comet#5438 tracks native Variant support in Comet.
- Put it under
crates/spark/src/function/aggregate/or the newvariant/module. Add
crates/sqllogictest/test_files/variant/schema_of_variant_agg.sltand 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.
- 主要言語
- Rust
- スター
- 7
- フォーク
- 15
- 平均マージ
- 16時間 43分
- マージ済み PR(30日)
- 26
環境構築
- Dockerfile・Docker Compose ファイルなし
- プルリクエストのテンプレートなし
- コントリビューションガイドを読む
はじめの一歩
- issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
- 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
- リポジトリをフォークし、ブランチを切って変更します。
- issue 番号を参照したプルリクエストを送ります。
apache/datafusion-spark のほかの issue
-
enhancement
難易度 2/5 1〜3時間 初心者へのやさしさ 70/100
apache/datafusion-spark#287 ·
メンテナーはふだん 1 日以内に返信
-
enhancement good first issue
難易度 2/5 1〜3時間 初心者へのやさしさ 80/100
apache/datafusion-spark#266 ·
メンテナーはふだん 1 日以内に返信
-
[Bug] make_dt_interval returns NULL for more than about 1.49 million days, Spark returns the interval対応中かも @Subodh-17 が今日担当しました。 オープン
難易度 2/5 1〜3時間 初心者へのやさしさ 78/100
apache/datafusion-spark#109 ·
メンテナーはふだん 1 日以内に返信
-
enhancement
難易度 5/5 1週間以上 初心者へのやさしさ 18/100
apache/datafusion-spark#299 ·
メンテナーはふだん 1 日以内に返信
-
enhancement
難易度 4/5 3〜5日 初心者へのやさしさ 35/100
apache/datafusion-spark#298 ·
メンテナーはふだん 1 日以内に返信
apache/datafusion-spark の issue をすべて見る
似ている issue
-
comp-rbac
難易度 2/5 1〜3時間 初心者へのやさしさ 68/100
ClickHouse/ClickHouse#131875 ·
メンテナーはふだん 1 日以内に返信
-
bug(python): read_transaction raises TypeError on a Rewrite with rewritten indices対応中かも @Borg93 が今日担当しました。 オープンbug
難易度 2/5 1〜3時間 初心者へのやさしさ 75/100
lance-format/lance#11117 ·
メンテナーはふだん 2 日以内に返信
-
難易度 2/5 1〜3時間 初心者へのやさしさ 70/100
storytold/photocraft#15232 · コメント 1 件 ·
メンテナーはふだん 1 日以内に返信
-
forge: fork preflight reads still send `requireCanonical: false`, which Rootstock rejects (-32603)対応中かも @Ynyesto が今日担当しました。 オープン
難易度 2/5 1〜3時間 初心者へのやさしさ 72/100
foundry-rs/foundry#18485 ·
メンテナーはふだん 1 日以内に返信
-
難易度 2/5 1〜3時間 初心者へのやさしさ 70/100
bondybondbond/rushcut#238 · コメント 1 件 ·