AHeise commented on code in PR #29369:
URL: https://github.com/apache/flink/pull/29369#discussion_r4181315118
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java:
##########
@@ -730,6 +730,23 @@ private static boolean supportsCasting(
return allowExplicit
&& mapChildren.get(0).is(CHARACTER_STRING)
&& supportsCasting(sourceType, mapChildren.get(1), true);
+ } else if (targetRoot == VARIANT
Review Comment:
Keying a ROW by its type's field names means `CAST(ROW(id, name) AS
VARIANT)` produces `{"EXPR$0": ..., "EXPR$1": ...}`, even though both operands
are column references. I'd expect that to be the most common way to write it.
The same query in the Table API gives `f0`/`f1`, and `EXPR$N` is a Calcite
naming detail we don't control. Once released, these keys become a data
contract that ends up in sinks.
FLIP-521 does list ROW and STRUCTURED as castable to VARIANT, so CAST itself
is fine. For comparison, Spark rejects `CAST(struct AS VARIANT)` and offers
`to_variant_object`, documented as "maps and structs are converted to variant
objects which are unordered unlike SQL structs".
I'm not asking to change the approach in this PR, but I'd like the `EXPR$N`
keys to be a conscious decision on the ML rather than an emergent one. Options:
reject a ROW constructor without explicit names, document it more prominently,
or add a dedicated object-building function later. WDYT?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ToVariantConverter.java:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
+import org.apache.flink.types.variant.BinaryVariant;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder.FieldEntry;
+import org.apache.flink.types.variant.Variant;
+import org.apache.flink.types.variant.VariantTypeException;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Converts internal data of a SQL type into a {@link Variant}, like {@code
CAST(x AS VARIANT)}.
+ *
+ * <p>An {@code ARRAY} becomes a variant array. A {@code ROW} or {@code
STRUCTURED} value becomes a
+ * variant object keyed by its field names, and a {@code MAP} one keyed by its
character string
+ * keys. A {@code NULL} element, field or map value becomes a variant null. A
leaf is stored like
+ * the cast stores it on its own, see {@link VariantCastUtils}, and a nested
{@code VARIANT} is
+ * embedded as is.
+ *
+ * <p>The converter is created once per type and writes a whole value into one
builder, so a nested
+ * value is encoded in a single pass. It keeps no state between calls.
+ */
+@Internal
+public final class ToVariantConverter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String sourceType;
+ private final ValueWriter writer;
+
+ private ToVariantConverter(String sourceType, ValueWriter writer) {
+ this.sourceType = sourceType;
+ this.writer = writer;
+ }
+
+ /** Creates a converter for a type that {@code LogicalTypeCasts} allows to
cast to VARIANT. */
+ public static ToVariantConverter create(LogicalType type) {
+ return new ToVariantConverter(type.asSummaryString(),
createWriter(type));
+ }
+
+ /**
+ * Converts a value that is not {@code NULL}.
+ *
+ * @throws TableRuntimeException if the value has a {@code NULL} map key,
a timestamp outside
+ * the range of its variant kind, or does not fit into the 16 MiB of a
{@code VARIANT}
+ */
+ public Variant convert(Object value) {
+ // A MAP can hold the same key twice at runtime. The builder keeps the
last one, like the
+ // MAP constructor does.
+ final BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(true);
+ try {
+ writer.write(builder, value);
+ return builder.build();
+ } catch (VariantTypeException e) {
+ if (e !=
BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION) {
+ throw e;
+ }
+ throw new TableRuntimeException(
+ String.format(
+ "Cannot cast a value of type %s to VARIANT. A
VARIANT is limited to "
+ + "16 MiB.",
+ sourceType));
+ }
+ }
+
+ @FunctionalInterface
+ private interface ValueWriter extends Serializable {
+ void write(BinaryVariantInternalBuilder builder, Object value);
+ }
+
+ private static ValueWriter createNullableWriter(LogicalType type) {
+ final ValueWriter writer = createWriter(type);
+ return (builder, value) -> {
+ if (value == null) {
+ builder.appendNull();
+ } else {
+ writer.write(builder, value);
+ }
+ };
+ }
+
+ private static ValueWriter createWriter(LogicalType type) {
Review Comment:
This switch encodes every leaf a second time, next to
`VariantCastUtils.fromXxx`, which `PrimitiveToVariantCastRule` uses. The two
match today; I compared the bytes for TIMESTAMP(3), TIMESTAMP(9), TIME,
CHAR(4), DECIMAL(38, 0) and DECIMAL(10, 4). But any change, such as the
deferred UTF-8 byte path for strings from #29311, now has to happen in two
places, and drift would make `CAST(x AS VARIANT)` differ from `x` inside
`CAST(ARRAY[x] AS VARIANT)`.
Could we keep one source of truth? Either `fromXxx` delegates to append
helpers as commit 2 does for timestamps, or `PrimitiveToVariantCastRule` uses
`ToVariantConverter.create(type)` for primitives too.
##########
docs/content/docs/sql/reference/data-types.md:
##########
@@ -1725,7 +1725,38 @@ CAST(NULL AS VARIANT) -- NULL
CAST(CAST('NaN' AS DOUBLE) AS VARIANT) -- NaN, stored as a DOUBLE
CAST(INTERVAL '2' DAY AS VARIANT) -- fails at validation
CAST(ARRAY[1, NULL] AS ARRAY<VARIANT>) -- [1, NULL], each element a
VARIANT, the NULL stays SQL NULL
-CAST(ARRAY[1, 2] AS VARIANT) -- fails at validation, not
supported yet
+```
+
+A whole `ARRAY`, `MAP`, `ROW`, or `STRUCTURED` value can also be cast into a
single `VARIANT`. An
Review Comment:
nit: FLIP-521 lists ARRAY, ROW and STRUCTURED as castable to and from
VARIANT, but not MAP. VARIANT to MAP already exists, so this is symmetric and
I'm fine with it. It may be worth a line on the ML or a FLIP update so the doc
stays the reference.
##########
flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java:
##########
@@ -537,7 +537,9 @@ public void finishWritingArray(int start,
ArrayList<Integer> offsets) {
// the
// input variant, we can directly copy the binary slice.
public void appendVariant(BinaryVariant v) {
- appendVariantImpl(v.getValue(), v.getMetadata(), v.getPos());
+ // A nested variant, such as a field or an element of another variant,
starts at its own
+ // position in the shared buffer. getValue() would copy it to position
0 instead.
+ appendVariantImpl(v.rawValue(), v.getMetadata(), v.getPos());
Review Comment:
This fix is independent of the cast, and `VariantBuilder` is
`@PublicEvolving`. Every released version with VARIANT ships the bug. Could you
give it its own JIRA so it can be backported?
Related but pre-existing, so a separate ticket: `BinaryVariant#hashCode`
hashes the whole underlying buffer, while `equals` compares the slice and
`pos`. Two equal nested variants can therefore have different hash codes.
Embedding nested values makes such variants more common.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java:
##########
@@ -95,7 +96,7 @@ public CastExecutor<IN, OUT> create(
"public "
+ castExecutorClassName
+ "("
- + ctx.typeSerializers.values().stream()
+ + constructorFields.entrySet().stream()
.map(
entry ->
className(entry.getValue().getClass())
Review Comment:
nit: The constructor parameter type comes from
`entry.getValue().getClass()`, and `ScalarOperatorGens` uses
`getCanonicalName()` through `addReusableObject`. Both break for a lambda, an
anonymous class or a non-public class. This only works because
`ToVariantConverter` is public and final. Maybe let `declareReusableObject`
take the declared type, or state the restriction in its Javadoc.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ToVariantConverter.java:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
+import org.apache.flink.types.variant.BinaryVariant;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder.FieldEntry;
+import org.apache.flink.types.variant.Variant;
+import org.apache.flink.types.variant.VariantTypeException;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Converts internal data of a SQL type into a {@link Variant}, like {@code
CAST(x AS VARIANT)}.
+ *
+ * <p>An {@code ARRAY} becomes a variant array. A {@code ROW} or {@code
STRUCTURED} value becomes a
+ * variant object keyed by its field names, and a {@code MAP} one keyed by its
character string
+ * keys. A {@code NULL} element, field or map value becomes a variant null. A
leaf is stored like
+ * the cast stores it on its own, see {@link VariantCastUtils}, and a nested
{@code VARIANT} is
+ * embedded as is.
+ *
+ * <p>The converter is created once per type and writes a whole value into one
builder, so a nested
Review Comment:
nit: "encoded in a single pass" isn't quite exact. Each
`finishWritingObject`/`finishWritingArray` shifts its data to make room for the
header, so the bytes move once per nesting level. That's still far cheaper than
re-encoding every subtree. Maybe "without re-encoding nested values"?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ToVariantConverter.java:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
+import org.apache.flink.types.variant.BinaryVariant;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder.FieldEntry;
+import org.apache.flink.types.variant.Variant;
+import org.apache.flink.types.variant.VariantTypeException;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Converts internal data of a SQL type into a {@link Variant}, like {@code
CAST(x AS VARIANT)}.
+ *
+ * <p>An {@code ARRAY} becomes a variant array. A {@code ROW} or {@code
STRUCTURED} value becomes a
+ * variant object keyed by its field names, and a {@code MAP} one keyed by its
character string
+ * keys. A {@code NULL} element, field or map value becomes a variant null. A
leaf is stored like
+ * the cast stores it on its own, see {@link VariantCastUtils}, and a nested
{@code VARIANT} is
+ * embedded as is.
+ *
+ * <p>The converter is created once per type and writes a whole value into one
builder, so a nested
+ * value is encoded in a single pass. It keeps no state between calls.
+ */
+@Internal
+public final class ToVariantConverter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String sourceType;
+ private final ValueWriter writer;
+
+ private ToVariantConverter(String sourceType, ValueWriter writer) {
+ this.sourceType = sourceType;
+ this.writer = writer;
+ }
+
+ /** Creates a converter for a type that {@code LogicalTypeCasts} allows to
cast to VARIANT. */
+ public static ToVariantConverter create(LogicalType type) {
+ return new ToVariantConverter(type.asSummaryString(),
createWriter(type));
+ }
+
+ /**
+ * Converts a value that is not {@code NULL}.
+ *
+ * @throws TableRuntimeException if the value has a {@code NULL} map key,
a timestamp outside
+ * the range of its variant kind, or does not fit into the 16 MiB of a
{@code VARIANT}
+ */
+ public Variant convert(Object value) {
+ // A MAP can hold the same key twice at runtime. The builder keeps the
last one, like the
+ // MAP constructor does.
+ final BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(true);
+ try {
+ writer.write(builder, value);
+ return builder.build();
+ } catch (VariantTypeException e) {
+ if (e !=
BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION) {
+ throw e;
+ }
+ throw new TableRuntimeException(
+ String.format(
+ "Cannot cast a value of type %s to VARIANT. A
VARIANT is limited to "
+ + "16 MiB.",
+ sourceType));
+ }
+ }
+
+ @FunctionalInterface
+ private interface ValueWriter extends Serializable {
+ void write(BinaryVariantInternalBuilder builder, Object value);
+ }
+
+ private static ValueWriter createNullableWriter(LogicalType type) {
+ final ValueWriter writer = createWriter(type);
+ return (builder, value) -> {
+ if (value == null) {
+ builder.appendNull();
+ } else {
+ writer.write(builder, value);
+ }
+ };
+ }
+
+ private static ValueWriter createWriter(LogicalType type) {
+ switch (type.getTypeRoot()) {
+ case NULL:
+ return (builder, value) -> builder.appendNull();
+ case BOOLEAN:
+ return (builder, value) -> builder.appendBoolean((Boolean)
value);
+ case TINYINT:
+ return (builder, value) -> builder.appendByte((Byte) value);
+ case SMALLINT:
+ return (builder, value) -> builder.appendShort((Short) value);
+ case INTEGER:
+ return (builder, value) -> builder.appendInt((Integer) value);
+ case BIGINT:
+ return (builder, value) -> builder.appendLong((Long) value);
+ case FLOAT:
+ return (builder, value) -> builder.appendFloat((Float) value);
+ case DOUBLE:
+ return (builder, value) -> builder.appendDouble((Double)
value);
+ case DECIMAL:
+ return (builder, value) ->
+ builder.appendDecimal(((DecimalData)
value).toBigDecimal());
+ case CHAR:
+ case VARCHAR:
+ return (builder, value) ->
builder.appendString(value.toString());
+ case BINARY:
+ case VARBINARY:
+ return (builder, value) -> builder.appendBinary((byte[])
value);
+ case DATE:
+ return (builder, value) -> builder.appendDate((Integer) value);
+ case TIME_WITHOUT_TIME_ZONE:
+ // TIME is millisecond-of-day at runtime, the variant kind
holds microseconds.
+ return (builder, value) -> builder.appendTime((Integer) value
* 1_000L);
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ {
+ final int precision = LogicalTypeChecks.getPrecision(type);
+ return (builder, value) ->
+ VariantCastUtils.appendTimestamp(
+ builder, (TimestampData) value, precision);
+ }
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ {
+ final int precision = LogicalTypeChecks.getPrecision(type);
+ return (builder, value) ->
+ VariantCastUtils.appendTimestampLtz(
+ builder, (TimestampData) value, precision);
+ }
+ case UUID:
+ return (builder, value) ->
builder.appendUuid(UuidUtils.fromBytes((byte[]) value));
+ case VARIANT:
Review Comment:
Embedding a deeply nested VARIANT recurses once per level in
`appendVariantImpl`, so around 2,000 levels overflow the stack. `TRY_CAST`
catches it because the generated try/catch takes `Throwable`, but plain `CAST`
throws a `StackOverflowError`. `ExceptionUtils.isJvmFatalError` doesn't treat
it as fatal, so the task fails over and restarts on the same record forever.
That's a poison pill with no useful error message. `PARSE_JSON` is bounded by
Jackson's nesting limit, but a VARIANT from a UDF using `VariantBuilder`, a
format converter, or a view that wraps a VARIANT per level is not.
I agree `toJson()` has the same problem, but this PR adds a new per-record
path that can hit it. Could we at least catch `StackOverflowError` in `convert`
and rethrow it as a `TableRuntimeException` naming the nesting depth? A depth
limit or an iterative `appendVariantImpl` can be the separate ticket; please
link it here.
##########
docs/content/docs/sql/reference/data-types.md:
##########
@@ -1725,7 +1725,38 @@ CAST(NULL AS VARIANT) -- NULL
CAST(CAST('NaN' AS DOUBLE) AS VARIANT) -- NaN, stored as a DOUBLE
CAST(INTERVAL '2' DAY AS VARIANT) -- fails at validation
CAST(ARRAY[1, NULL] AS ARRAY<VARIANT>) -- [1, NULL], each element a
VARIANT, the NULL stays SQL NULL
-CAST(ARRAY[1, 2] AS VARIANT) -- fails at validation, not
supported yet
+```
+
+A whole `ARRAY`, `MAP`, `ROW`, or `STRUCTURED` value can also be cast into a
single `VARIANT`. An
+`ARRAY` becomes a variant array, and a `MAP`, `ROW`, or `STRUCTURED` value a
variant object. Each
+leaf is stored by the rules above, so the cast is supported only when every
leaf type casts to
+`VARIANT`.
+
+- A `ROW` or `STRUCTURED` value is keyed by its field names. The SQL `ROW`
constructor names its
+ fields `EXPR$0`, `EXPR$1`, and so on, and the Table API `row()` names them
`f0`, `f1`, and so on.
+ To choose the keys, cast to a `ROW` with named fields first, or name each
field with `as()` in the
+ Table API.
+- A `MAP` needs a character string key, which becomes the object key. A `NULL`
key fails the cast.
+ If a key appears twice, the last value is kept.
+- A variant object sorts its keys, so the field order of a `ROW` is not kept.
+- A `NULL` element, field, or map value becomes a variant null, so an array
keeps its length and an
+ object keeps its keys.
+- A nested `VARIANT` is embedded as is.
+- The whole value must fit into the 16 MiB of a `VARIANT`. Any `ARRAY` or
`MAP` can exceed it, and so
+ can a `ROW` with a nested `VARIANT` or with fields whose declared sizes add
up to more. The cast
+ then fails, and `TRY_CAST` returns `NULL` for the whole value.
+- Casting the `VARIANT` back to the original type returns the original value,
since a cast to `ROW`
Review Comment:
The round trip has a few more exceptions than the SQL `NULL` in a `VARIANT`
field:
- The same happens for a SQL `NULL` element of `ARRAY<VARIANT>` and a SQL
`NULL` value of `MAP<STRING, VARIANT>`.
- A MAP with a duplicate key loses the earlier value.
It's also worth one sentence contrasting the two NULL behaviors:
`ARRAY<INT>` to `ARRAY<VARIANT>` keeps a SQL `NULL`, while `ARRAY<INT>` to
`VARIANT` turns it into a variant null. Otherwise users will hit that
difference by surprise.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ConstructedToVariantCastRule.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.casting;
+
+import org.apache.flink.table.runtime.functions.ToVariantConverter;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.types.variant.BinaryVariantUtil;
+import org.apache.flink.types.variant.Variant;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+
+/**
+ * Cast rule from {@link LogicalTypeRoot#ARRAY}, {@link LogicalTypeRoot#MAP},
{@link
+ * LogicalTypeRoot#ROW} and {@link LogicalTypeRoot#STRUCTURED_TYPE} to {@link
+ * LogicalTypeRoot#VARIANT}.
+ *
+ * <p>The generated code calls a {@link ToVariantConverter} created once for
the input type, which
+ * describes how a value is stored. A failure anywhere in the value fails the
whole cast, so {@code
+ * TRY_CAST} returns {@code NULL} rather than a partial result. See {@link
#canFail} for when a cast
+ * can fail.
+ */
+class ConstructedToVariantCastRule extends
AbstractNullAwareCodeGeneratorCastRule<Object, Variant> {
+
+ static final ConstructedToVariantCastRule INSTANCE = new
ConstructedToVariantCastRule();
+
+ /** The largest fixed-size value: a 16-byte decimal with its header and
scale. */
+ private static final long MAX_FIXED_SIZE = 18;
+
+ /** A header byte and a 4-byte size or length, the largest header a value
has. */
+ private static final long MAX_HEADER_SIZE = 1 + BinaryVariantUtil.U32_SIZE;
+
+ private ConstructedToVariantCastRule() {
+
super(CastRulePredicate.builder().predicate(ConstructedToVariantCastRule::matches).build());
+ }
+
+ private static boolean matches(LogicalType input, LogicalType target) {
+ return target.is(LogicalTypeRoot.VARIANT)
+ && (input.is(LogicalTypeRoot.ARRAY)
+ || input.is(LogicalTypeRoot.MAP)
+ || input.is(LogicalTypeRoot.ROW)
+ || input.is(LogicalTypeRoot.STRUCTURED_TYPE))
+ && LogicalTypeCasts.supportsExplicitCast(input, target);
+ }
+
+ /**
+ * Returns whether a value of the input type can fail the cast, so that
{@code TRY_CAST} returns
+ * {@code NULL} for it instead of failing. A value can fail in two ways:
+ *
+ * <ul>
+ * <li>A leaf fails like it does on its own, see {@link
PrimitiveToVariantCastRule#canFail}.
+ * <li>The whole value does not fit into the 16 MiB of a {@code
VARIANT}. Any {@code ARRAY},
+ * {@code MAP} or nested {@code VARIANT} can reach that size. A
{@code ROW} can when the
+ * declared sizes of its fields add up to more.
+ * </ul>
+ *
+ * <p>A {@code MAP} also fails on a {@code NULL} key, but every {@code
MAP} can fail anyway.
+ */
+ @Override
+ public boolean canFail(LogicalType inputLogicalType, LogicalType
targetLogicalType) {
+ // Without an ARRAY, MAP or VARIANT, the size of a value is bounded by
the sum of its
+ // leaves and object headers.
+ long maxSize = 0;
+ final Deque<LogicalType> pending = new ArrayDeque<>();
+ pending.push(inputLogicalType);
+ while (!pending.isEmpty()) {
+ final LogicalType type = pending.pop();
+ if (type.is(LogicalTypeRoot.DISTINCT_TYPE)) {
+ pending.push(((DistinctType) type).getSourceType());
+ continue;
+ }
+ switch (type.getTypeRoot()) {
+ case ARRAY:
+ case MAP:
+ case VARIANT:
+ return true;
+ case ROW:
+ case STRUCTURED_TYPE:
+
LogicalTypeChecks.getFieldTypes(type).forEach(pending::push);
+ break;
+ default:
+ if (PrimitiveToVariantCastRule.INSTANCE.canFail(type,
targetLogicalType)) {
+ return true;
+ }
+ }
+ maxSize += maxOwnSize(type);
+ }
+ return maxSize > BinaryVariantUtil.SIZE_LIMIT;
+ }
+
+ private static long maxOwnSize(LogicalType type) {
+ switch (type.getTypeRoot()) {
+ case ROW:
+ case STRUCTURED_TYPE:
+ // Only the object header. Its fields are counted on their
own. A 4-byte id per
+ // field, and a 4-byte offset per field plus one for the end.
+ return MAX_HEADER_SIZE
+ + BinaryVariantUtil.U32_SIZE
+ * (2L * LogicalTypeChecks.getFieldCount(type)
+ 1);
+ case CHAR:
+ case VARCHAR:
+ return MAX_HEADER_SIZE
+ + (long)
PrimitiveToVariantCastRule.MAX_UTF8_BYTES_PER_CHAR
+ * LogicalTypeChecks.getLength(type);
+ case BINARY:
+ case VARBINARY:
+ return MAX_HEADER_SIZE + LogicalTypeChecks.getLength(type);
+ default:
+ return MAX_FIXED_SIZE;
+ }
+ }
+
+ /* Example generated code for ARRAY<INT>, inside the null check of the
base class. The
+ converter is a field, created once when the code is generated:
+
+ result$1 = toVariantConverter$2.convert(array$0);
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
Review Comment:
Together with the element-wise casts from #29311, `CAST(arr AS
ARRAY<VARIANT>)` for `arr ARRAY<ROW<...>>` becomes valid: each element turns
into a variant object. That path runs `declareReusableObject` inside the nested
code generation of `ArrayToArrayCastRule`, both through `ScalarOperatorGens`
and through the standalone `CastExecutor`, which is the trickier one. The tests
only cover `ARRAY<VARIANT>` to VARIANT. Could you add `ARRAY<ROW<...>>` to
`ARRAY<VARIANT>` and `MAP<STRING, ROW<...>>` to `MAP<STRING, VARIANT>` rows to
`CastRulesTest` and `CastFunctionITCase`?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ToVariantConverter.java:
##########
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
+import org.apache.flink.types.variant.BinaryVariant;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
+import org.apache.flink.types.variant.BinaryVariantInternalBuilder.FieldEntry;
+import org.apache.flink.types.variant.Variant;
+import org.apache.flink.types.variant.VariantTypeException;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Converts internal data of a SQL type into a {@link Variant}, like {@code
CAST(x AS VARIANT)}.
+ *
+ * <p>An {@code ARRAY} becomes a variant array. A {@code ROW} or {@code
STRUCTURED} value becomes a
+ * variant object keyed by its field names, and a {@code MAP} one keyed by its
character string
+ * keys. A {@code NULL} element, field or map value becomes a variant null. A
leaf is stored like
+ * the cast stores it on its own, see {@link VariantCastUtils}, and a nested
{@code VARIANT} is
+ * embedded as is.
+ *
+ * <p>The converter is created once per type and writes a whole value into one
builder, so a nested
+ * value is encoded in a single pass. It keeps no state between calls.
+ */
+@Internal
+public final class ToVariantConverter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String sourceType;
+ private final ValueWriter writer;
+
+ private ToVariantConverter(String sourceType, ValueWriter writer) {
+ this.sourceType = sourceType;
+ this.writer = writer;
+ }
+
+ /** Creates a converter for a type that {@code LogicalTypeCasts} allows to
cast to VARIANT. */
+ public static ToVariantConverter create(LogicalType type) {
+ return new ToVariantConverter(type.asSummaryString(),
createWriter(type));
+ }
+
+ /**
+ * Converts a value that is not {@code NULL}.
+ *
+ * @throws TableRuntimeException if the value has a {@code NULL} map key,
a timestamp outside
+ * the range of its variant kind, or does not fit into the 16 MiB of a
{@code VARIANT}
+ */
+ public Variant convert(Object value) {
+ // A MAP can hold the same key twice at runtime. The builder keeps the
last one, like the
+ // MAP constructor does.
+ final BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(true);
Review Comment:
nit: Every value allocates a new builder (128 bytes, growing by doubling), a
boxed `ArrayList<Integer>` per array, and one `FieldEntry` per field. That's
fine for now; just noting it as the obvious target if the micro benchmark
matters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
