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..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,6 +18,8 @@ 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; @@ -285,4 +287,19 @@ 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(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(CommonTestUtils.createCopySerializable(subVariant)).isEqualTo(subVariant); + } } 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-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 + } } 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})),