[FLINK-37913][table] Add built-in OBJECT_OF function #26704

raminqaf · 2025-06-19T16:29:06Z

What is the purpose of the change

This pull request implements the OBJECT_OF function as specified in FLIP-520 to enable creation of structured types in SQL and Table API. The function allows users to construct structured objects from key-value pairs without requiring UDFs, making structured type creation SQL-serializable and improving the overall usability of structured types in Flink.

Brief change log

Added OBJECT_OF function definition to BuiltInFunctionDefinitions
Implemented ObjectOfInputTypeStrategy for input validation (odd argument count, string keys, unique field names)
Implemented ObjectOfTypeStrategy for output type inference (creates structured types)
Added ObjectOfFunction runtime implementation that creates RowData from key-value pairs
Enhanced Expressions.java with objectOf() methods for Table API support
Added Python API support with object_of() function in expressions.py
Updated SQL functions documentation with OBJECT_OF function details

Verifying this change

This change added tests and can be verified as follows:

Added unit tests for ObjectOfInputTypeStrategy to validate argument count, type validation, and field name uniqueness
Added integration tests in StructuredFunctionsITCase to test end-to-end functionality
Verified SQL usage: SELECT OBJECT_OF('com.example.User', 'name', 'Bob', 'age', 42)
Verified Table API usage: objectOf(User.class, "name", "Bob", "age", 42)

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 (additions to Expressions.java and BuiltInFunctionDefinitions)
The serializers: no
The runtime per-record code paths (performance sensitive): no
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? yes
If yes, how is the feature documented? docs / JavaDocs (updated SQL functions documentation, comprehensive JavaDocs for all new classes and methods, Python API documentation)

flinkbot · 2025-06-19T16:32:33Z

CI report:

40c0cd2 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

raminqaf · 2025-06-19T16:33:44Z

...on/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java

+        if (logicalType.is(LogicalTypeFamily.CHARACTER_STRING)) {
+            return DataTypes.STRING().nullable();
+        }


@twalthr I was not 100% that this is the correct approach! I noticed in the output of the tests that strings were used as CHAR(n), so I had to convert them to STRING(). I couldn't find any util class that does this conversion, so I added the logic here.

gustavodemorais · 2025-06-19T18:29:46Z

I think there's a checkstyle violation making the CI fail. Make sure to run ./mvnw checkstyle:check locally

snuyanzin · 2025-06-19T19:04:26Z

...lanner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java

+                                "NestedConstructor(Type1Constructor(f0, f1), Type2Constructor(15, 'Alice')) = "
+                                        + "OBJECT_OF('"
+                                        + NestedType.class.getName()
+                                        + "', 'n1', OBJECT_OF('"
+                                        + Type1.class.getName()
+                                        + "', 'a', 42, 'b', 'Bob'), "
+                                        + "'n2', OBJECT_OF('"
+                                        + Type2.class.getName()
+                                        + "', 'a', 15, 'b', 'Alice'))",


not sure if it was missed during PTF implementation
will it work with record s?
Asking since we have a dedicated folder for java17 tests
for instance https://github.com/apache/flink/blob/1c34ca011cacdbb3b0f48b485eac89dd913d29bf/flink-tests-java17/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoRecordSerializerUpgradeTestSpecifications.java

Good question. So far we don't have any tests for records because the language level was too low.

snuyanzin · 2025-06-19T19:25:27Z

docs/data/sql_functions.yml

@@ -1178,6 +1178,22 @@ valueconstruction:
      ```sql
      f(columns => DESCRIPTOR(`col1`, `col2`), on_time => DESCRIPTOR(`ts`))
      ```
+  - sql: OBJECT_OF(className, [key, value [, key, value , ...]])


keep in mind that Chinese version also should be updated (could be in English for now)

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java

...-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java

snuyanzin · 2025-06-19T20:32:14Z

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java

+        if (!logicalType.is(LogicalTypeFamily.CHARACTER_STRING)) {
+            throw new ValidationException(
+                    "The field key at position "
+                            + keyIndex
+                            + " must be a STRING/VARCHAR type, but was "
+                            + logicalType.asSummaryString()
+                            + ".");
+        }


should we check if it is not null (or not nullable)?
Otherwise
output for nullable classname is confusing like

SELECT OBJECT_OF(cast(null as string), 'f0', 'a', 'b', 'Alice');

gives

Caused by: org.apache.flink.table.api.ValidationException: Invalid function call: OBJECT_OF(STRING, CHAR(2) NOT NULL, CHAR(1) NOT NULL, CHAR(1) NOT NULL, CHAR(5) NOT NULL) ... Caused by: org.apache.flink.table.api.ValidationException: Could not infer an output type for the given arguments.

which in fact satisfies the required

Supported signatures are: OBJECT_OF(STRING, [STRING, ANY]*...)

Thanks for the feedback! I have addressed the nullability check and added tests.

how will it behave for the case when arg is nullable however we can not resolve during planning like ?

there is a table input with data

class_name1 class_name2 NULL class_name3

and there is a query

SELECT OBJECT_OF(class_name, 'f0', 'value') FROM input

snuyanzin · 2025-06-19T20:33:34Z

...lanner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java

+                        // Test with OBJECT_OF
+                        .testResult(
+                                objectOf(Type1.class, "a", 42, "b", "Bob"),
+                                "OBJECT_OF('" + Type1.class.getName() + "', 'a', 42, 'b', 'Bob')",


seems subquery not supported (not sure if it is related to this PR)
this is ok

SELECT OBJECT_OF('<class_name>', 'f0', 1);

this was not able to success

SELECT OBJECT_OF('<class_name>', 'f0', (SELECT 1));

or did I miss anything?

...-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java

snuyanzin · 2025-06-20T10:05:41Z

...on/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java

+        final String fieldName =
+                callContext
+                        .getArgumentValue(keyIdx, String.class)
+                        .orElseThrow(IllegalStateException::new);


probably better to call CallContext#fail

I looked at the fail method implementation, and it throws a ValidationError. At this point in the code, we have already validated all the inputs in the ObjectOfInputTypeStrategy. Therefore, this case should (ideally) never occur, and if so, it should be an IllegalState.

Therefore, this case should (ideally) never occur, and if so, it should be an IllegalState.

i disagree
the simplest way to make it happen

SELECT OBJECT_OF('class_name', CAST(NULL AS STRING), 'as');

And then user will see just IllegalStateException without any idea what to do about it

i tend to think that classname and all keys should be always non nullable

...-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java

snuyanzin · 2025-06-20T13:27:40Z

...on/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java

+ *   <li>{@code OBJECT_OF('com.example.User', 'name', 'Alice', 'age', 30)} → {@code
+ *       STRUCTURED<com.example.User>(name STRING, age INT)}
+ *   <li>{@code OBJECT_OF('com.example.Point', 'x', 1.5, 'y', 2.0)} → {@code
+ *       STRUCTURED<com.example.Point>(x DOUBLE, y DOUBLE)}
+ * </ul>


if we have validation checking that classname should satisfy jvm rules for class naming
should we have another validation for keys in order to check that field names satisfy rules for jvm field naming?

I'm asking since it seems it is possible to have

OBJECT_OF('com.example.User', '0', 'Alice', '1', 30)

and we can not have fields called 0, and 1 in java ...

same for the reserved keywords which might be used for field names

Good point... We can also validate if the field name has a space, a dot ., or a hyphen -. OBJECT_OF('com.example.User', 'field 1', 'Alice', 'field.2', 30, 'field-3', 60).

If we want to do the validation very specifically, we should also check against reserved JVM keywords (if, void, etc.).

This makes pyFlink validation complicated due to other keywords.

@twalthr what is your opinion on this?

If this kind of validation is needed, I will extract the fieldName validation so we can use it in OBJECT_UPDATE(object, <key>, <value>) too!

SourceVersion#isName ?
like at

flink/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/StructuredType.java

Lines 560 to 568 in f612013

private static @Nullable String checkClassName(@Nullable String className) {

if (className != null && !SourceVersion.isName(className)) {

throw new ValidationException(

String.format(

"Invalid class name '%s'. The class name must comply with JVM identifier rules.",

className));

}

return className;

}

...on/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java

raminqaf commented Jun 19, 2025

View reviewed changes

snuyanzin reviewed Jun 19, 2025

View reviewed changes

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java Outdated Show resolved Hide resolved

snuyanzin reviewed Jun 19, 2025

View reviewed changes

...-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java Outdated Show resolved Hide resolved

snuyanzin reviewed Jun 19, 2025

View reviewed changes

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java Show resolved Hide resolved

snuyanzin reviewed Jun 19, 2025

View reviewed changes

raminqaf force-pushed the FLINK-37913 branch from 5442485 to 6028380 Compare June 20, 2025 07:31

snuyanzin reviewed Jun 20, 2025

View reviewed changes

...-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java Outdated Show resolved Hide resolved

snuyanzin reviewed Jun 20, 2025

View reviewed changes

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java Outdated Show resolved Hide resolved

snuyanzin reviewed Jun 20, 2025

View reviewed changes

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java Outdated Show resolved Hide resolved

snuyanzin reviewed Jun 20, 2025

View reviewed changes

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java Show resolved Hide resolved

snuyanzin reviewed Jun 20, 2025

View reviewed changes

raminqaf added 7 commits June 20, 2025 14:22

[FLINK-37913][table] Add built-in OBJECT_OF function

920f2e6

[FLINK-37913][table] Fix checkstyle error

4a74618

[FLINK-37913][table] Iterate through odd elements

73555ea

[FLINK-37913][table] Validate first argumante for null

29a9380

[FLINK-37913][table] Rename variable and simplified iterator

78e4bdb

[FLINK-37913][table] nit changes

24e11e2

[FLINK-37913][table] Add more docs on java class path

f629d8e

raminqaf force-pushed the FLINK-37913 branch from c8e07b7 to f629d8e Compare June 20, 2025 12:24

snuyanzin reviewed Jun 20, 2025

View reviewed changes

...-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java Outdated Show resolved Hide resolved

snuyanzin reviewed Jun 20, 2025

View reviewed changes

...c/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java Outdated Show resolved Hide resolved

raminqaf added 2 commits June 20, 2025 14:57

[FLINK-37913][table] Fix typos

42fd488

[FLINK-37913][table] Extract method

03682df

snuyanzin reviewed Jun 20, 2025

View reviewed changes

[FLINK-37913][table] Add missing JavaDocs

d7424b8

snuyanzin reviewed Jun 20, 2025

View reviewed changes

...on/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java Show resolved Hide resolved

[FLINK-37913][table] Add validation for null type in field names

40c0cd2

	private static @Nullable String checkClassName(@Nullable String className) {
	if (className != null && !SourceVersion.isName(className)) {
	throw new ValidationException(
	String.format(
	"Invalid class name '%s'. The class name must comply with JVM identifier rules.",
	className));
	}
	return className;
	}

[FLINK-37913][table] Add built-in OBJECT_OF function #26704

Are you sure you want to change the base?

[FLINK-37913][table] Add built-in OBJECT_OF function #26704

Conversation

raminqaf commented Jun 19, 2025

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Uh oh!

flinkbot commented Jun 19, 2025 • 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!

gustavodemorais commented Jun 19, 2025

Uh oh!

snuyanzin Jun 19, 2025 • 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!

Uh oh!

Uh oh!

Uh oh!

snuyanzin Jun 19, 2025 • 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!

snuyanzin Jun 20, 2025 • 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!

Uh oh!

Uh oh!

Uh oh!

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!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

raminqaf Jun 20, 2025 • 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!

Uh oh!

Uh oh!

flinkbot commented Jun 19, 2025 •

edited

Loading

snuyanzin Jun 19, 2025 •

edited

Loading

snuyanzin Jun 19, 2025 •

edited

Loading

snuyanzin Jun 20, 2025 •

edited

Loading

raminqaf Jun 20, 2025 •

edited

Loading