[
https://issues.apache.org/jira/browse/FLINK-40582?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ramin Gharib updated FLINK-40582:
---------------------------------
Description:
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
{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}}. A variant null alone gives \{{VARIANT}}.
- Every type in the result is nullable, so the string never contains \{{NOT
NULL}}.
- SQL NULL gives NULL.
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
{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}}).
- SQL NULL rows are skipped, so a group of only NULL rows gives NULL.
- 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.
was:
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.
> 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
> Priority: Major
>
> 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
> {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}}. A variant null alone gives \{{VARIANT}}.
> - Every type in the result is nullable, so the string never contains \{{NOT
> NULL}}.
> - SQL NULL gives NULL.
> 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
> {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}}).
> - SQL NULL rows are skipped, so a group of only NULL rows gives NULL.
> - 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
(v10.3.26#10030026)
