[FLINK-37977][table] Support VARIANT in user-defined functions and process table functions by raminqaf · Pull Request #28928 · apache/flink

raminqaf · 2026-08-05T13:42:53Z

What is the purpose of the change

VARIANT worked in SQL but not in the Java extension points. Three separate things failed: holding a Variant as a member of a user-defined function could not be registered, declaring BinaryVariant as an argument or return type passed type extraction and then failed at code generation, and a process table function whose state entry contains a VARIANT field failed to plan.

This makes VARIANT usable across UDF and PTF signatures, including as a return type and through an explicit @DataTypeHint("VARIANT").

Brief change log

Variant extends Serializable, so instances can be function members or constructor arguments
Registered the BinaryVariant identity conversion in DataStructureConverters, which VariantType already advertised in its conversion set
Added the VARIANT case to CodeGenUtils.hashCodeForType, used to hash process table function state

Verifying this change

This change added tests and can be verified as follows:

BinaryVariantTest: Java serialization round-trip for scalar, object, array, null and a sub-variant that shares the value binary of its enclosing document
TypeInferenceExtractorTest: VARIANT signatures 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 VARIANT state entry
DataStructureConvertersTest: the BinaryVariant conversion class
FunctionITCase: end-to-end scalar functions for @DataTypeHint("VARIANT"), for BinaryVariant as the conversion class, and for a function instance carrying a Variant member
ProcessTableFunctionSemanticTests: process-variant for nullable, optional and VARIANT NOT NULL scalar arguments, and process-variant-state for a state entry with a VARIANT field

Each of the three fixes has a test that fails without it.

Notes for reviewers

Serializable sits on the Variant interface rather than on BinaryVariant. Variant is the only type callers can name: it is the default conversion class for VARIANT and BinaryVariant is @Internal. Putting it on the implementation would make the guarantee hold only by accident. Bitmap does the opposite, but it gets serializability incidentally through RoaringBitmapData; making Bitmap extends Serializable would be a separate change.

Implementing Value instead was considered and rejected. Its IOReadableWritable.read mutates the instance, which is impossible for an immutable BinaryVariant with final fields, ValueSerializer requires a public nullary constructor that BinaryVariant cannot have, and the Value branch in TypeExtractor.privateGetForClass precedes the VARIANT branch, so it would shadow VariantTypeInfo and replace VariantSerializer.

Narrowing a sub-variant's payload on Java serialization was prototyped and dropped. pos != 0 only arises from getField and getElement, and Java serialization only reaches a Variant held as a function member, which is assigned in driver code, so the case was not worth the extra serialization logic.

Follow-up, not in this PR: passing an untyped NULL to a VARIANT parameter of a scalar function fails in SqlTypeUtil.convertTypeToSpec, which has no branch for SqlTypeName.VARIANT. That is an upstream Calcite gap and CAST(NULL AS VARIANT) works. Process table functions are unaffected.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes, Variant is @PublicEvolving and now extends Serializable
The serializers: no, VariantSerializer and the VARIANT binary format are untouched
The runtime per-record code paths (performance sensitive): no, the new hashCodeForType branch only makes reachable a case that previously failed at planning
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no, it makes an existing type work in existing extension points
If yes, how is the feature documented? not applicable
Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5)

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.

…ocess 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.

flinkbot · 2026-08-05T13:53:17Z

CI report:

f87f499 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

snuyanzin · 2026-08-05T14:59:14Z

    }

+    @Test
+    void testVariantScalarFunction() throws Exception {


can we have at least one test with view?

create
2 select from view

to be sure unparse for variant works ok

Added testVariantScalarFunctionInView

use a semantic test for this

snuyanzin · 2026-08-05T15:01:28Z

+    }
+
+    private static Variant javaRoundTrip(Variant variant) throws Exception {
+        ByteArrayOutputStream bytes = new ByteArrayOutputStream();


why ByteArrayOutputStream is out of try with resources?

Removed the helper and replaced it with CommonTestUtils.createCopySerializable

snuyanzin · 2026-08-05T15:03:05Z

+ */
 @PublicEvolving
-public interface Variant {
+public interface Variant extends Serializable {


why don't we use Value?

Value extends IOReadableWritable, whose read(DataInputView) deserializes into this. ValueSerializer also instantiates via a public nullary constructor, which BinaryVariant cannot offer, since its constructor validates the version byte and the size limit.
It would also add a second serialization path. VARIANT already has VariantTypeInfo and VariantSerializer, and TypeExtractor would resolve to ValueTypeInfo instead, because it checks Value first.

…zation 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.

snuyanzin · 2026-08-06T06:26:11Z

    }

+    @Test
+    void testVariantScalarFunctionInView() throws Exception {


I tend to think this test better to have at CatalogViewITCase

twalthr · 2026-08-06T15:11:49Z

+                                        false))
+                        .expectOutput(TypeStrategies.explicit(DataTypes.VARIANT())),
+                // ---
+                TestSpec.forAsyncScalarFunction(


only test scalar function should be enough.

twalthr · 2026-08-06T15:16:28Z

+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink SELECT * FROM f("
+                                    + "variant1 => NULL, "


add a table arg with a variant column as well

twalthr · 2026-08-06T15:20:03Z

    }

+    @Test
+    void testVariantScalarFunction() throws Exception {


Semantic tests replace ITCases. So let's not over-test this. +1 for removal in this class.

twalthr · 2026-08-06T15:20:29Z

    }

+    @Test
+    void testVariantScalarFunction() throws Exception {


use a semantic test for this

twalthr · 2026-08-06T15:25:41Z

        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());


nit: I'm wondering whether we actually need this I'm not sure if binary variant should be an internal class of the logical type. Take a look at string: string has a string data and binary string, but binary string is internal and not exposed

raminqaf added 2 commits August 5, 2026 15:47

raminqaf force-pushed the FLINK-37977 branch from 3f77b78 to b819646 Compare August 5, 2026 13:50

snuyanzin reviewed Aug 5, 2026

View reviewed changes

snuyanzin reviewed Aug 6, 2026

View reviewed changes

[FLINK-37977] Move test

f87f499

twalthr reviewed Aug 6, 2026

View reviewed changes

Uh oh!

Conversation

raminqaf commented Aug 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Notes for reviewers

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Aug 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

raminqaf Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

raminqaf commented Aug 5, 2026 •

edited

Loading

flinkbot commented Aug 5, 2026 •

edited

Loading

raminqaf Aug 6, 2026 •

edited

Loading