From e4a353c9bf87e4b14243e06e49029cf158268218 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Wed, 5 Aug 2026 15:47:17 +0200 Subject: [PATCH 1/4] [FLINK-37977][core] Make Variant serializable Variant instances could not be held as member variables of a user-defined function or passed into its constructor. Registering such a function failed with NotSerializableException, because the planner Java-serializes the function instance into the generated code. BinaryVariant already holds nothing but two byte arrays and an offset, so declaring the interface serializable is sufficient. The guarantee belongs on the interface rather than the implementation, since Variant is the only type callers can name: it is the default conversion class for VARIANT and BinaryVariant is internal. --- .../flink/types/variant/BinaryVariant.java | 2 ++ .../apache/flink/types/variant/Variant.java | 10 +++++-- .../types/variant/BinaryVariantTest.java | 30 +++++++++++++++++++ 3 files changed, 40 insertions(+), 2 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java index c5dd0bf3efe683..97c11eedf52b1f 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java @@ -66,6 +66,8 @@ @Internal public final class BinaryVariant implements Variant { + private static final long serialVersionUID = 1L; + private final byte[] value; private final byte[] metadata; // The variant value doesn't use the whole `value` binary, but starts from its `pos` index and diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java b/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java index c0f15788f3f754..2a7483391fa3ba 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java @@ -20,15 +20,21 @@ import org.apache.flink.annotation.PublicEvolving; +import java.io.Serializable; import java.math.BigDecimal; import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.List; -/** Variant represent a semi-structured data. */ +/** + * Variant represent a semi-structured data. + * + *

Instances are serializable so that they can be held as member variables of user-defined + * functions or passed into their constructors. + */ @PublicEvolving -public interface Variant { +public interface Variant extends Serializable { /** Returns true if the variant is a primitive typed value, such as INT, DOUBLE, STRING, etc. */ boolean isPrimitive(); diff --git a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java index 77235e968eebce..dc1a7a6831d677 100644 --- a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java +++ b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java @@ -23,6 +23,10 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.math.BigDecimal; import java.time.Instant; import java.time.LocalDate; @@ -285,4 +289,30 @@ void testGetThrowException() { .isInstanceOf(VariantTypeException.class) .hasMessage("Expected type DOUBLE but got FLOAT"); } + + @Test + void testJavaSerialization() throws Exception { + Variant variant = + builder.object() + .add("i", builder.of(1)) + .add("nested", builder.array().add(builder.of("v")).build()) + .build(); + assertThat(javaRoundTrip(variant)).isEqualTo(variant); + + // a sub-variant is addressed by a position into the value binary of the enclosing document + Variant subVariant = variant.getField("nested"); + assertThat(((BinaryVariant) subVariant).getPos()).isGreaterThan(0); + assertThat(javaRoundTrip(subVariant)).isEqualTo(subVariant); + } + + private static Variant javaRoundTrip(Variant variant) throws Exception { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) { + out.writeObject(variant); + } + try (ObjectInputStream in = + new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { + return (Variant) in.readObject(); + } + } } From b8196466dd76327b82ef7d786a527ec236b70e92 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Wed, 5 Aug 2026 15:47:18 +0200 Subject: [PATCH 2/4] [FLINK-37977][table] Support VARIANT in user-defined functions and process table functions Two gaps kept VARIANT from working in function signatures. DataStructureConverters had no entry for BinaryVariant, even though VariantType advertises the class in its input and output conversion set and ClassDataTypeConverter maps it to VARIANT. Declaring it as an argument or return type therefore passed type extraction and then failed at code generation with "Could not find converter for data type: VARIANT". The converter is the identity, matching how the other internal data structures such as StringData and RoaringBitmapData are registered. hashCodeForType did not handle the type root, so a process table function whose state entry contains a VARIANT field failed with a MatchError. StreamExecProcessTableFunction generates a hash function over the whole state row to detect state changes. BinaryVariant derives equals and hashCode from its contents, so hashing and equality stay consistent for state lookups. VARIANT was previously only exercised as a reflectively extracted argument of one scalar and one aggregate function. A VARIANT return type had no coverage at all, and neither did an explicit @DataTypeHint("VARIANT"). Type inference is now covered for scalar, async scalar, aggregate, table and process table functions, including VARIANT nested in ARRAY, MAP and ROW and the rejection of a non-composite state entry, with end-to-end cases for the hint, the bridge to BinaryVariant and process table function state. --- .../TypeInferenceExtractorTest.java | 101 ++++++++++++++++- .../table/planner/codegen/CodeGenUtils.scala | 2 + .../ProcessTableFunctionSemanticTests.java | 4 +- .../ProcessTableFunctionTestPrograms.java | 36 ++++++ .../stream/ProcessTableFunctionTestUtils.java | 29 +++++ .../runtime/stream/sql/FunctionITCase.java | 104 ++++++++++++++++++ .../conversion/DataStructureConverters.java | 2 + .../data/DataStructureConvertersTest.java | 5 + 8 files changed, 281 insertions(+), 2 deletions(-) diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java index f253819ce2cb13..0c1edfd2ba3968 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java @@ -50,6 +50,7 @@ import org.apache.flink.table.types.utils.DataTypeFactoryMock; import org.apache.flink.types.Row; import org.apache.flink.types.bitmap.Bitmap; +import org.apache.flink.types.variant.Variant; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; @@ -889,7 +890,56 @@ private static Stream functionSpecs() { "Logical type 'BITMAP' does not support a conversion from or to class 'org.apache.flink.table.types.extraction.TypeInferenceExtractorTest$CustomBitmap'."), TestSpec.forScalarFunction("Custom Bitmap", InvalidCustomBitmapTypeFunction2.class) .expectErrorMessage( - "Could not extract a valid type inference for function class 'org.apache.flink.table.types.extraction.TypeInferenceExtractorTest$InvalidCustomBitmapTypeFunction2'.")); + "Could not extract a valid type inference for function class 'org.apache.flink.table.types.extraction.TypeInferenceExtractorTest$InvalidCustomBitmapTypeFunction2'."), + // --- + TestSpec.forScalarFunction("Variant in scalar function", VariantTypeFunction.class) + .expectStaticArgument( + StaticArgument.scalar("v", DataTypes.VARIANT(), false)) + .expectStaticArgument( + StaticArgument.scalar( + "array", DataTypes.ARRAY(DataTypes.VARIANT()), false)) + .expectStaticArgument( + StaticArgument.scalar( + "map", + DataTypes.MAP(DataTypes.INT(), DataTypes.VARIANT()), + false)) + .expectStaticArgument( + StaticArgument.scalar( + "row", + DataTypes.ROW(DataTypes.FIELD("a", DataTypes.VARIANT())), + false)) + .expectOutput(TypeStrategies.explicit(DataTypes.VARIANT())), + // --- + TestSpec.forAsyncScalarFunction( + "Variant in async scalar function", VariantTypeAsyncFunction.class) + .expectStaticArgument( + StaticArgument.scalar("v", DataTypes.VARIANT(), false)) + .expectOutput(TypeStrategies.explicit(DataTypes.VARIANT())), + // --- + TestSpec.forAggregateFunction( + "Variant in aggregate function", VariantTypeAggFunction.class) + .expectStaticArgument( + StaticArgument.scalar("v", DataTypes.VARIANT(), false)) + .expectAccumulator(TypeStrategies.explicit(VariantState.TYPE)) + .expectOutput(TypeStrategies.explicit(DataTypes.VARIANT())), + // --- + TestSpec.forTableFunction( + "Variant in table function", VariantTypeTableFunction.class) + .expectStaticArgument( + StaticArgument.scalar("v", DataTypes.VARIANT(), false)) + .expectOutput( + TypeStrategies.explicit( + DataTypes.ROW(DataTypes.FIELD("v", DataTypes.VARIANT())))), + // --- + TestSpec.forProcessTableFunction(VariantProcessTableFunction.class) + .expectStaticArgument( + StaticArgument.scalar("v", DataTypes.VARIANT(), false)) + .expectState("s", TypeStrategies.explicit(VariantState.TYPE)) + .expectOutput(TypeStrategies.explicit(DataTypes.VARIANT())), + // --- + TestSpec.forProcessTableFunction(InvalidVariantStateProcessTableFunction.class) + .expectErrorMessage( + "State entries must use a mutable, composite data type. But was: VARIANT")); } private static Stream procedureSpecs() { @@ -2723,6 +2773,55 @@ public Bitmap[] call(Object procedureContext, Bitmap bitmap) { } } + @FunctionHint(output = @DataTypeHint("VARIANT")) + private static class VariantTypeFunction extends ScalarFunction { + public Variant eval( + Variant v, + Variant[] array, + Map map, + @DataTypeHint("ROW") Row row) { + return null; + } + } + + private static class VariantTypeAsyncFunction extends AsyncScalarFunction { + public void eval(CompletableFuture f, Variant v) {} + } + + private static class VariantTypeAggFunction extends AggregateFunction { + public void accumulate(VariantState accumulator, Variant v) {} + + @Override + public VariantState createAccumulator() { + return null; + } + + @Override + public Variant getValue(VariantState accumulator) { + return null; + } + } + + @FunctionHint(output = @DataTypeHint("ROW")) + private static class VariantTypeTableFunction extends TableFunction { + public void eval(Variant v) {} + } + + private static class VariantProcessTableFunction extends ProcessTableFunction { + public void eval(@StateHint VariantState s, Variant v) {} + } + + private static class InvalidVariantStateProcessTableFunction + extends ProcessTableFunction { + public void eval(@StateHint Variant s, Variant v) {} + } + + public static class VariantState { + static final DataType TYPE = + DataTypes.STRUCTURED(VariantState.class, DataTypes.FIELD("v", DataTypes.VARIANT())); + public Variant v; + } + @FunctionHint(input = @DataTypeHint(value = "BITMAP", bridgedTo = CustomBitmap.class)) private static class InvalidCustomBitmapTypeFunction1 extends ScalarFunction { public Bitmap eval(Bitmap bitmap) { diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala index aacb8db5a0d4a6..739858689bf555 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala @@ -388,6 +388,8 @@ object CodeGenUtils { } val serTerm = ctx.addReusableObject(serializer, "serializer") s"$term.toObject($serTerm).hashCode()" + case VARIANT => + s"$term.hashCode()" case BITMAP => s"$term.hashCode()" case NULL | SYMBOL | UNRESOLVED => diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java index 2357235b11864b..e32be28c5b1e47 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java @@ -97,6 +97,8 @@ public List programs() { ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY, ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY, ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_TABLE_API, - ProcessTableFunctionTestPrograms.PROCESS_IMPLICIT_CASTS); + ProcessTableFunctionTestPrograms.PROCESS_IMPLICIT_CASTS, + ProcessTableFunctionTestPrograms.PROCESS_VARIANT, + ProcessTableFunctionTestPrograms.PROCESS_VARIANT_STATE); } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java index 327c3167fe53dc..fa089edbf0a5cc 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java @@ -72,6 +72,8 @@ import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UpdatingJoinFunction; import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UpdatingRetractFunction; import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UpdatingUpsertFunction; +import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.VariantFunction; +import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.VariantStateFunction; import org.apache.flink.table.test.program.SinkTestStep; import org.apache.flink.table.test.program.SourceTestStep; import org.apache.flink.table.test.program.TableTestProgram; @@ -928,6 +930,40 @@ public class ProcessTableFunctionTestPrograms { "INSERT INTO sink SELECT * FROM f(columnList1 => NULL, columnList3 => DESCRIPTOR(a, b, c))") .build(); + public static final TableTestProgram PROCESS_VARIANT = + TableTestProgram.of( + "process-variant", + "takes nullable, optional, and not nullable VARIANT arguments") + .setupTemporarySystemFunction("f", VariantFunction.class) + .setupSql(BASIC_VALUES) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(BASE_SINK_SCHEMA) + .consumedValues("+I[{null, null, {\"a\":[1,\"b\"]}}]") + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM f(" + + "variant1 => NULL, " + + "variant3 => PARSE_JSON('{\"a\":[1,\"b\"]}'))") + .build(); + + public static final TableTestProgram PROCESS_VARIANT_STATE = + TableTestProgram.of("process-variant-state", "state entry with a VARIANT field") + .setupTemporarySystemFunction("f", VariantStateFunction.class) + .setupSql(MULTI_VALUES) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(KEYED_BASE_SINK_SCHEMA) + .consumedValues( + "+I[Bob, {VariantScore(v=null), +I[Bob, 12]}]", + "+I[Alice, {VariantScore(v=null), +I[Alice, 42]}]", + "+I[Bob, {VariantScore(v=12), +I[Bob, 99]}]", + "+I[Bob, {VariantScore(v=99), +I[Bob, 100]}]", + "+I[Alice, {VariantScore(v=42), +I[Alice, 400]}]") + .build()) + .runSql("INSERT INTO sink SELECT * FROM f(r => TABLE t PARTITION BY name)") + .build(); + public static final TableTestProgram PROCESS_TIME_CONVERSIONS = TableTestProgram.of( "process-time-conversions", diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java index 89fc127a41334e..8b8a2b626d0fef 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java @@ -41,6 +41,7 @@ import org.apache.flink.types.ColumnList; import org.apache.flink.types.Row; import org.apache.flink.types.RowKind; +import org.apache.flink.types.variant.Variant; import java.time.Duration; import java.time.Instant; @@ -549,6 +550,24 @@ public void eval( } } + /** Testing function. */ + public static class VariantFunction extends AppendProcessTableFunctionBase { + public void eval( + Variant variant1, + @ArgumentHint(isOptional = true) Variant variant2, + @DataTypeHint("VARIANT NOT NULL") Variant variant3) { + collectObjects(variant1, variant2, variant3); + } + } + + /** Testing function. */ + public static class VariantStateFunction extends AppendProcessTableFunctionBase { + public void eval(@StateHint VariantScore s, @ArgumentHint(SET_SEMANTIC_TABLE) Row r) { + collectObjects(s, r); + s.v = Variant.newBuilder().of(r.getFieldAs("score")); + } + } + /** Testing function. */ public static class RequiredTimeFunction extends AppendProcessTableFunctionBase { public void eval(@ArgumentHint({ArgumentTrait.ROW_SEMANTIC_TABLE, REQUIRE_ON_TIME}) Row r) { @@ -1236,6 +1255,16 @@ public String toString() { } } + /** POJO for state. */ + public static class VariantScore { + public Variant v; + + @Override + public String toString() { + return String.format("VariantScore(v=%s)", v); + } + } + private static final Map MODE_SUMMARY = Map.ofEntries( Map.entry("[INSERT, UPDATE_BEFORE, UPDATE_AFTER]", "retract-no-delete"), diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java index c9d31af6e97e95..d993984a1bba6c 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java @@ -59,6 +59,9 @@ import org.apache.flink.table.types.logical.RawType; import org.apache.flink.testutils.junit.utils.TempDirUtils; import org.apache.flink.types.Row; +import org.apache.flink.types.variant.BinaryVariant; +import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantBuilder; import org.apache.flink.util.CloseableIterator; import org.apache.flink.util.CollectionUtil; import org.apache.flink.util.FlinkRuntimeException; @@ -760,6 +763,80 @@ void testRowScalarFunction() throws Exception { assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sourceData); } + @Test + void testVariantScalarFunction() throws Exception { + final VariantBuilder builder = Variant.newBuilder(); + final List sourceData = + List.of( + Row.of("a", builder.object().add("i", builder.of(1)).build()), + Row.of("b", builder.array().add(builder.of("x")).build()), + Row.of("c", builder.ofNull()), + Row.of("d", null)); + + TestCollectionTableFactory.reset(); + TestCollectionTableFactory.initData(sourceData); + + tEnv().executeSql( + "CREATE TABLE TestTable(s STRING, v VARIANT) " + + "WITH ('connector' = 'COLLECTION')"); + + tEnv().createTemporarySystemFunction("VariantScalarFunction", VariantScalarFunction.class); + tEnv().executeSql("INSERT INTO TestTable SELECT s, VariantScalarFunction(v) FROM TestTable") + .await(); + + assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sourceData); + } + + @Test + void testBinaryVariantScalarFunction() throws Exception { + final VariantBuilder builder = Variant.newBuilder(); + final List sourceData = + List.of( + Row.of("a", builder.object().add("i", builder.of(1)).build()), + Row.of("b", builder.of("x")), + Row.of("c", null)); + + TestCollectionTableFactory.reset(); + TestCollectionTableFactory.initData(sourceData); + + tEnv().executeSql( + "CREATE TABLE TestTable(s STRING, v VARIANT) " + + "WITH ('connector' = 'COLLECTION')"); + + tEnv().createTemporarySystemFunction( + "BinaryVariantScalarFunction", BinaryVariantScalarFunction.class); + tEnv().executeSql( + "INSERT INTO TestTable SELECT s, BinaryVariantScalarFunction(v) FROM TestTable") + .await(); + + assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sourceData); + } + + @Test + void testVariantScalarFunctionWithVariantMemberVariable() throws Exception { + final VariantBuilder builder = Variant.newBuilder(); + final Variant constant = builder.object().add("i", builder.of(1)).build(); + + final List sourceData = List.of(Row.of("a", null), Row.of("b", null)); + final List sinkData = List.of(Row.of("a", constant), Row.of("b", constant)); + + TestCollectionTableFactory.reset(); + TestCollectionTableFactory.initData(sourceData); + + tEnv().executeSql( + "CREATE TABLE TestTable(s STRING, v VARIANT) " + + "WITH ('connector' = 'COLLECTION')"); + + // the instance carries a Variant, so it is serialized instead of being identified by class + tEnv().createTemporarySystemFunction( + "VariantConstantFunction", new VariantConstantFunction(constant)); + tEnv().executeSql( + "INSERT INTO TestTable SELECT s, VariantConstantFunction() FROM TestTable") + .await(); + + assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sinkData); + } + @Test void testComplexScalarFunction() throws Exception { final List sourceData = @@ -2030,6 +2107,33 @@ public static class RowScalarFunction extends ScalarFunction { } } + /** Function that takes and returns variants. */ + public static class VariantScalarFunction extends ScalarFunction { + public @DataTypeHint("VARIANT") Variant eval(@DataTypeHint("VARIANT") Variant v) { + return v; + } + } + + /** Function that takes and returns variants through the internal implementation class. */ + public static class BinaryVariantScalarFunction extends ScalarFunction { + public BinaryVariant eval(BinaryVariant v) { + return v; + } + } + + /** Function that returns the variant it was constructed with. */ + public static class VariantConstantFunction extends ScalarFunction { + private final Variant constant; + + public VariantConstantFunction(Variant constant) { + this.constant = constant; + } + + public Variant eval() { + return constant; + } + } + /** Scalar function with argument hint. */ public static class NamedArgumentsScalarFunction extends ScalarFunction { @FunctionHint( diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java index f01a1aca779da7..dbff59b47aa42e 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/DataStructureConverters.java @@ -33,6 +33,7 @@ import org.apache.flink.types.Row; import org.apache.flink.types.bitmap.Bitmap; import org.apache.flink.types.bitmap.RoaringBitmapData; +import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import java.math.BigDecimal; @@ -198,6 +199,7 @@ public final class DataStructureConverters { putConverter(LogicalTypeRoot.RAW, byte[].class, RawByteArrayConverter::create); putConverter(LogicalTypeRoot.RAW, RawValueData.class, identity()); putConverter(LogicalTypeRoot.VARIANT, Variant.class, identity()); + putConverter(LogicalTypeRoot.VARIANT, BinaryVariant.class, identity()); putConverter(LogicalTypeRoot.BITMAP, Bitmap.class, constructor(BitmapBitmapConverter::new)); putConverter(LogicalTypeRoot.BITMAP, RoaringBitmapData.class, identity()); } diff --git a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java index 2c19c579a4ec50..6b6ff55d164b2d 100644 --- a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java +++ b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java @@ -29,6 +29,7 @@ import org.apache.flink.types.Row; import org.apache.flink.types.RowKind; import org.apache.flink.types.bitmap.Bitmap; +import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import org.apache.flink.util.InstantiationUtil; @@ -360,6 +361,10 @@ static List testData() { GenericPojo.class, new GenericPojo<>(LocalDate.ofEpochDay(123))), TestSpec.forDataType(DataTypes.VARIANT()) .convertedTo(Variant.class, Variant.newBuilder().of("hello")), + TestSpec.forDataType(DataTypes.VARIANT().bridgedTo(BinaryVariant.class)) + .convertedTo( + BinaryVariant.class, + (BinaryVariant) Variant.newBuilder().of("hello")), TestSpec.forDataType(DataTypes.BITMAP()) .convertedTo(Bitmap.class, Bitmap.fromArray(new int[] {1, 2, 3})), From 93dbd68e0f363bac8f1437f6c53cbbe1067dcab3 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Wed, 5 Aug 2026 17:54:10 +0200 Subject: [PATCH 3/4] [FLINK-37977][tests] Cover VARIANT in a view and simplify the serialization round-trip A view is stored as expanded SQL, so selecting from a view over a VARIANT expression exercises unparsing the validated node and re-parsing the result. Use CommonTestUtils.createCopySerializable instead of hand-rolling the Java serialization round-trip. --- .../types/variant/BinaryVariantTest.java | 21 ++++------------ .../runtime/stream/sql/FunctionITCase.java | 24 +++++++++++++++++++ 2 files changed, 28 insertions(+), 17 deletions(-) diff --git a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java index dc1a7a6831d677..f9fd6d5dafb828 100644 --- a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java +++ b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java @@ -18,15 +18,13 @@ package org.apache.flink.types.variant; +import org.apache.flink.core.testutils.CommonTestUtils; + import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; import java.math.BigDecimal; import java.time.Instant; import java.time.LocalDate; @@ -297,22 +295,11 @@ void testJavaSerialization() throws Exception { .add("i", builder.of(1)) .add("nested", builder.array().add(builder.of("v")).build()) .build(); - assertThat(javaRoundTrip(variant)).isEqualTo(variant); + assertThat(CommonTestUtils.createCopySerializable(variant)).isEqualTo(variant); // a sub-variant is addressed by a position into the value binary of the enclosing document Variant subVariant = variant.getField("nested"); assertThat(((BinaryVariant) subVariant).getPos()).isGreaterThan(0); - assertThat(javaRoundTrip(subVariant)).isEqualTo(subVariant); - } - - private static Variant javaRoundTrip(Variant variant) throws Exception { - ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - try (ObjectOutputStream out = new ObjectOutputStream(bytes)) { - out.writeObject(variant); - } - try (ObjectInputStream in = - new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { - return (Variant) in.readObject(); - } + assertThat(CommonTestUtils.createCopySerializable(subVariant)).isEqualTo(subVariant); } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java index d993984a1bba6c..c42273cde52ed6 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java @@ -787,6 +787,30 @@ void testVariantScalarFunction() throws Exception { assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sourceData); } + @Test + void testVariantScalarFunctionInView() throws Exception { + final VariantBuilder builder = Variant.newBuilder(); + final List sourceData = + List.of( + Row.of("a", builder.object().add("i", builder.of(1)).build()), + Row.of("b", null)); + + TestCollectionTableFactory.reset(); + TestCollectionTableFactory.initData(sourceData); + + tEnv().executeSql( + "CREATE TABLE TestTable(s STRING, v VARIANT) " + + "WITH ('connector' = 'COLLECTION')"); + + tEnv().createTemporarySystemFunction("VariantScalarFunction", VariantScalarFunction.class); + // a view is stored as expanded SQL, so this also covers unparsing a VARIANT expression + tEnv().executeSql( + "CREATE VIEW VariantView AS SELECT s, VariantScalarFunction(v) AS v FROM TestTable"); + tEnv().executeSql("INSERT INTO TestTable SELECT s, v FROM VariantView").await(); + + assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sourceData); + } + @Test void testBinaryVariantScalarFunction() throws Exception { final VariantBuilder builder = Variant.newBuilder(); From f87f4994bfc5ee394b2eb3c7b3f6fded21694337 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Thu, 6 Aug 2026 08:38:12 +0200 Subject: [PATCH 4/4] [FLINK-37977] Move test --- .../runtime/stream/sql/FunctionITCase.java | 24 --------- .../planner/catalog/CatalogViewITCase.scala | 53 +++++++++++++++++++ 2 files changed, 53 insertions(+), 24 deletions(-) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java index c42273cde52ed6..d993984a1bba6c 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java @@ -787,30 +787,6 @@ void testVariantScalarFunction() throws Exception { assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sourceData); } - @Test - void testVariantScalarFunctionInView() throws Exception { - final VariantBuilder builder = Variant.newBuilder(); - final List sourceData = - List.of( - Row.of("a", builder.object().add("i", builder.of(1)).build()), - Row.of("b", null)); - - TestCollectionTableFactory.reset(); - TestCollectionTableFactory.initData(sourceData); - - tEnv().executeSql( - "CREATE TABLE TestTable(s STRING, v VARIANT) " - + "WITH ('connector' = 'COLLECTION')"); - - tEnv().createTemporarySystemFunction("VariantScalarFunction", VariantScalarFunction.class); - // a view is stored as expanded SQL, so this also covers unparsing a VARIANT expression - tEnv().executeSql( - "CREATE VIEW VariantView AS SELECT s, VariantScalarFunction(v) AS v FROM TestTable"); - tEnv().executeSql("INSERT INTO TestTable SELECT s, v FROM VariantView").await(); - - assertThat(TestCollectionTableFactory.getResult()).isEqualTo(sourceData); - } - @Test void testBinaryVariantScalarFunction() throws Exception { final VariantBuilder builder = Variant.newBuilder(); diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/catalog/CatalogViewITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/catalog/CatalogViewITCase.scala index c61f5985fe9690..27ace8083498df 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/catalog/CatalogViewITCase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/catalog/CatalogViewITCase.scala @@ -21,11 +21,13 @@ import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Schema, Table import org.apache.flink.table.api.config.ExecutionConfigOptions import org.apache.flink.table.api.internal.TableEnvironmentImpl import org.apache.flink.table.functions.ScalarFunction +import org.apache.flink.table.planner.catalog.CatalogViewITCase.VariantIdentityFunction import org.apache.flink.table.planner.factories.TableFactoryHarness import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory import org.apache.flink.table.planner.utils.TableITCaseBase import org.apache.flink.testutils.junit.extensions.parameterized.{ParameterizedTestExtension, Parameters} import org.apache.flink.types.Row +import org.apache.flink.types.variant.Variant import org.apache.flink.util.CollectionUtil import org.assertj.core.api.Assertions.{assertThatList, assertThatThrownBy} @@ -307,6 +309,52 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends TableITCaseBase { TestCollectionTableFactory.RESULT) } + @TestTemplate + def testCreateViewWithVariant(): Unit = { + val builder = Variant.newBuilder() + val sourceData = List( + toRow("a", builder.`object`().add("i", builder.of(1)).build()), + toRow("b", builder.array().add(builder.of("x")).build()), + toRow("c", null)) + + TestCollectionTableFactory.initData(sourceData.asJava) + + val sourceDDL = + """ + |CREATE TABLE T1( + | s varchar, + | v variant + |) with ( + | 'connector' = 'COLLECTION' + |) + """.stripMargin + + val sinkDDL = + """ + |CREATE TABLE T2( + | s varchar, + | v variant + |) with ( + | 'connector' = 'COLLECTION' + |) + """.stripMargin + + // the view is stored as expanded SQL, so reading it also covers unparsing a VARIANT expression + val viewDDL = + """ + |CREATE VIEW T3 AS SELECT s, variantIdentity(v) AS v FROM T1 + """.stripMargin + + tableEnv.createTemporarySystemFunction("variantIdentity", classOf[VariantIdentityFunction]) + tableEnv.executeSql(sourceDDL) + tableEnv.executeSql(sinkDDL) + tableEnv.executeSql(viewDDL) + + tableEnv.sqlQuery("SELECT s, v FROM T3").executeInsert("T2").await() + assertThatList(sourceData.asJava).containsExactlyInAnyOrderElementsOf( + TestCollectionTableFactory.RESULT) + } + private def buildTableDescriptor(): TableDescriptor = { val tableDescriptor: TableDescriptor = TableFactoryHarness .newBuilder() @@ -558,4 +606,9 @@ object CatalogViewITCase { def parameters(): java.util.Collection[Boolean] = { util.Arrays.asList(true, false) } + + /** Testing function that takes and returns a variant. */ + class VariantIdentityFunction extends ScalarFunction { + def eval(v: Variant): Variant = v + } }