Ramin Gharib created FLINK-40582:
------------------------------------
Summary: Add SCHEMA_OF_VARIANT and SCHEMA_OF_VARIANT_AGG built-in
functions
Key: FLINK-40582
URL: https://issues.apache.org/jira/browse/FLINK-40582
Project: Flink
Issue Type: New Feature
Components: Table SQL / API, Table SQL / Runtime
Reporter: Ramin Gharib
Assignee: Ramin Gharib
h2. Motivation
Flink can now parse JSON into \{{VARIANT}}
(\{{PARSE_JSON}}/\{{TRY_PARSE_JSON}}) and cast a
{\{VARIANT}} into a structured target (\{{CAST(v AS ROW<...>)}},
\{{ARRAY<...>}},
{\{MAP<...>}}; FLINK-37925 / FLINK-37926). What is missing is a way to
*discover* the
structure of a semi-structured value so the user knows which target to cast to.
Today
{\{TYPEOF(v)}} only reports the generic \{{VARIANT}} label, not the shape of
the contained
data.
Databricks/Spark solve this with two functions:
-
[schema_of_variant|https://docs.databricks.com/aws/en/sql/language-manual/functions/schema_of_variant]
-
[schema_of_variant_agg|https://docs.databricks.com/aws/en/sql/language-manual/functions/schema_of_variant_agg]
This ticket proposes the Flink equivalents.
h2. SCHEMA_OF_VARIANT
Derives the schema of a single \{{VARIANT}} value and returns it as a
\{{STRING}}.
{code:sql}
SCHEMA_OF_VARIANT(variant) -- returns STRING NOT NULL
{code}
Behavior:
- Introspects the stored value, not the generic type, so it returns the
concrete shape.
- An object renders as \{{ROW<...>}} with fields ordered by name; an array as
\{{ARRAY<T>}}.
- Scalar kinds map to their Flink type: integers -> \{{BIGINT}}, decimals ->
\{{DECIMAL(p, s)}},
approximate -> \{{DOUBLE}}, \{{BOOLEAN}}, \{{STRING}}, \{{DATE}},
\{{TIMESTAMP}}, \{{BYTES}}, etc.
- An array whose elements have no common type shreds to \{{ARRAY<VARIANT>}}.
- A JSON \{{null}} contributes a nullable type; an all-null field is
\{{VARIANT}}.
Examples (Flink type syntax, not Databricks \{{OBJECT<>}}):
{code:sql}
SCHEMA_OF_VARIANT(PARSE_JSON('\{"key": 123, "data": [4, 5]}'))
-- ROW<`data` ARRAY<BIGINT>, `key` BIGINT>
SCHEMA_OF_VARIANT(PARSE_JSON('\{"data": [{"a":"a"}, 5]}'))
-- ROW<`data` ARRAY<VARIANT>>
SCHEMA_OF_VARIANT(CAST(123.4 AS VARIANT))
-- DECIMAL(4, 1)
{code}
The returned string is a valid Flink \{{LogicalType}} serialization, so it
pairs directly
with the VARIANT-to-constructed casts: read the schema, then \{{CAST(v AS <that
type>)}}.
h2. SCHEMA_OF_VARIANT_AGG
Aggregate form. Merges the schemas of every \{{VARIANT}} in a group into one
\{{STRING}}.
{code:sql}
SCHEMA_OF_VARIANT_AGG(variant) -- aggregate, returns STRING NOT NULL
{code}
Merging rules:
- Object fields are unioned by name; a field present in only some rows stays in
the result.
- Same-named fields with different types are coerced to their least common type
(e.g. \{{INT}} + \{{DOUBLE}} -> \{{DOUBLE}}); when no common type exists the
field becomes
\{{VARIANT}} (e.g. \{{STRING}} + \{{ROW<...>}} -> \{{VARIANT}}).
- Result mirrors \{{SCHEMA_OF_VARIANT}} for a single-row group.
Example:
{code:sql}
-- rows: \{"foo": "bar", "wing": {"ding": "dong"}} , \{"wing": 123}
SCHEMA_OF_VARIANT_AGG(v)
-- ROW<`foo` STRING, `wing` VARIANT>
{code}
h2. Notes / open questions
- *Output dialect:* emit Flink's own type serialization (\{{ROW<...>}},
\{{ARRAY<...>}},
\{{MAP<...>}}) rather than the Databricks \{{OBJECT<...>}} form, so the
result is
round-trippable into a Flink cast. Confirm during review.
- *Least-common-type:* reuse the existing type-merging / cast-availability logic
(e.g. \{{LogicalTypeMerging}}) rather than a bespoke lattice.
- *Registration:* new \{{BuiltInFunctionDefinition}} entries alongside
\{{PARSE_JSON}} /
\{{TYPEOF}}; runtime in the VARIANT function helpers.
- *Determinism of field order:* fix an ordering (by name) so plans are stable.
- Could be split into two sub-tasks (scalar first, aggregate second), since the
aggregate
needs an accumulator and the schema-merge logic.
h2. Documentation
Add both to the JSON/variant function reference and the systems function docs.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
