You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2025/07/17 09:11:48 UTC
(flink) branch master updated: [FLINK-37913][table] Add built-in OBJECT_OF function
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 71f67174a7e [FLINK-37913][table] Add built-in OBJECT_OF function
71f67174a7e is described below
commit 71f67174a7ea281bf3161b63676af26393fb6073
Author: Ramin Gharib <rg...@confluent.io>
AuthorDate: Thu Jul 17 11:11:42 2025 +0200
[FLINK-37913][table] Add built-in OBJECT_OF function
This closes #26704.
---
docs/data/sql_functions.yml | 18 ++
docs/data/sql_functions_zh.yml | 18 ++
.../docs/reference/pyflink.table/expressions.rst | 2 +
flink-python/pyflink/table/expressions.py | 33 ++++
.../org/apache/flink/table/api/Expressions.java | 66 ++++++++
.../functions/BuiltInFunctionDefinitions.java | 10 ++
.../strategies/ObjectOfInputTypeStrategy.java | 182 +++++++++++++++++++++
.../inference/strategies/ObjectOfTypeStrategy.java | 102 ++++++++++++
.../strategies/SpecificInputTypeStrategies.java | 3 +
.../strategies/SpecificTypeStrategies.java | 3 +
.../strategies/ObjectOfInputTypeStrategyTest.java | 165 +++++++++++++++++++
.../functions/StructuredFunctionsITCase.java | 162 ++++++++++++++++++
.../runtime/functions/scalar/ObjectOfFunction.java | 47 ++++++
13 files changed, 811 insertions(+)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 74e7d98cbf6..95cba6904e5 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -1198,6 +1198,24 @@ valueconstruction:
```sql
f(columns => DESCRIPTOR(`col1`, `col2`), on_time => DESCRIPTOR(`ts`))
```
+ - sql: OBJECT_OF(className, [key, value [, key, value , ...]])
+ table: objectOf(STRING, Object...)
+ description: |
+ Creates a structured object from a list of key-value pairs.
+
+ This function creates an instance of a structured type identified by the given class name.
+ The structured type is created by providing alternating key-value pairs where keys must be
+ string literals and values can be arbitrary expressions.
+
+ Note: The class name is only used for distinguishing two structured types with identical fields.
+ Structured types are internally handled with suitable data structures. Thus, serialization and equality checks are managed by the system.
+ In Table API and UDF calls, the system will attempt to resolve the class name to an actual implementation class.
+ In this case the class name needs to be present in the user classpath. If resolution fails, Row is used as a fallback.
+
+ ```sql
+ -- Creates a User object with complex fields "name", "age", and "address"
+ OBJECT_OF('com.example.User', 'name', 'Bob', 'age', 21, 'address', OBJECT_OF('com.example.Address', 'street', 'primary', 'city', 'Berlin'))
+ ```
- table: NUMERIC.rows
description: Creates a NUMERIC interval of rows (commonly used in window creation).
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index fc1a8da66e9..426ed237b17 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -1280,6 +1280,24 @@ valueconstruction:
```sql
f(columns => DESCRIPTOR(`col1`, `col2`), on_time => DESCRIPTOR(`ts`))
```
+ - sql: OBJECT_OF(className, [key, value [, key, value , ...]])
+ table: objectOf(STRING, Object...)
+ description: |
+ Creates a structured object from a list of key-value pairs.
+
+ This function creates an instance of a structured type identified by the given class name.
+ The structured type is created by providing alternating key-value pairs where keys must be
+ string literals and values can be arbitrary expressions.
+
+ Note: The class name is only used for distinguishing two structured types with identical fields.
+ Structured types are internally handled with suitable data structures. Thus, serialization and equality checks are managed by the system.
+ In Table API and UDF calls, the system will attempt to resolve the class name to an actual implementation class.
+ In this case the class name needs to be present in the user classpath. If resolution fails, Row.class is used as a fallback.
+
+ ```sql
+ -- Creates a User object with complex fields "name", "age", and "address"
+ OBJECT_OF('com.example.User', 'name', 'Bob', 'age', 21, 'address', OBJECT_OF('com.example.Address', 'street', 'primary', 'city', 'Berlin'))
+ ```
- table: NUMERIC.rows
description: 创建一个 NUMERIC 行间隔（通常用于窗口创建）。
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
index 8dacbaa836a..67102aebafa 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -55,6 +55,7 @@ Expressions
row
map_
map_from_arrays
+ object_of
row_interval
pi
e
@@ -261,6 +262,7 @@ advanced type helper functions
Expression.array_intersect
Expression.split
+
time definition functions
-------------------------
diff --git a/flink-python/pyflink/table/expressions.py b/flink-python/pyflink/table/expressions.py
index c0ae819eeca..be6fd946f43 100644
--- a/flink-python/pyflink/table/expressions.py
+++ b/flink-python/pyflink/table/expressions.py
@@ -560,6 +560,39 @@ def map_from_arrays(key, value) -> Expression:
return _binary_op("mapFromArrays", key, value)
+@PublicEvolving()
+def object_of(class_name: str, *args) -> Expression:
+ """
+ Creates a structured object from a list of key-value pairs.
+
+ This function creates an instance of a structured type identified by the given class name.
+ The structured type is created by providing alternating key-value pairs where keys must be
+ string literals and values can be arbitrary expressions.
+
+ Note: The class name is only used for distinguishing two structured types with identical fields.
+ Structured types are internally handled with suitable data structures.
+ Thus, serialization and equality checks are managed by the system.
+
+ In Table API and UDF calls, the system will attempt to resolve
+ the class name to an actual implementation class.
+ In this case the class name needs to be present in the user classpath.
+ If resolution fails, Row class is used as a fallback.
+
+ Examples:
+ ::
+
+ >>> # Creates a User object with name="Alice" and age=30
+ >>> object_of("com.example.User", "name", "Alice", "age", 30)
+
+ :param class_name: The fully qualified class name
+ :param args: Alternating key-value pairs: key1, value1, key2, value2, ...
+ :return: A structured object expression
+
+ .. seealso:: SQL function: OBJECT_OF('com.example.User', 'name', 'Bob', 'age', 25)
+ """
+ return _varargs_op("objectOf", class_name, *args)
+
+
@PublicEvolving()
def row_interval(rows: int) -> Expression:
"""
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
index 4e12194a2cf..a4552b25f5e 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
@@ -624,6 +624,72 @@ public final class Expressions {
objectToExpression(value));
}
+ /**
+ * Creates a structured object from a list of key-value pairs.
+ *
+ * <p>This function creates an instance of a structured type identified by the given class. The
+ * structured type is created by providing alternating key-value pairs where keys must be string
+ * literals and values can be arbitrary expressions.
+ *
+ * <p>Note: The class is only used for distinguishing two structured types with identical
+ * fields. Structured types are internally handled with suitable data structures. Thus,
+ * serialization and equality checks are managed by the system.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Creates a User object with name="Alice" and age=30
+ * objectOf(User.class, "name", "Alice", "age", 30)
+ *
+ * }</pre>
+ *
+ * <p>This function corresponds to the SQL {@code OBJECT_OF} function.
+ *
+ * @param clazz The class representing the structured type
+ * @param fields Alternating key-value pairs: key1, value1, key2, value2, ...
+ * @return A structured object expression
+ * @see #objectOf(String, Object...)
+ */
+ public static ApiExpression objectOf(Class<?> clazz, Object... fields) {
+ return apiCallAtLeastOneArgument(
+ BuiltInFunctionDefinitions.OBJECT_OF, valueLiteral(clazz.getName()), fields);
+ }
+
+ /**
+ * Creates a structured object from a list of key-value pairs.
+ *
+ * <p>This function creates an instance of a structured type identified by the given class name.
+ * The structured type is created by providing alternating key-value pairs where keys must be
+ * string literals and values can be arbitrary expressions.
+ *
+ * <p>Note: The class name is only used for distinguishing two structured types with identical
+ * fields. Structured types are internally handled with suitable data structures. Thus,
+ * serialization and equality checks are managed by the system.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Creates a User object with name="Bob" and age=25
+ * objectOf("com.example.User", "name", "Bob", "age", 25)
+ *
+ * }</pre>
+ *
+ * <p>This function corresponds to the SQL {@code OBJECT_OF} function:
+ *
+ * <pre>{@code
+ * OBJECT_OF('com.example.User', 'name', 'Bob', 'age', 25)
+ * }</pre>
+ *
+ * @param className The fully qualified class name representing the structured type
+ * @param fields Alternating key-value pairs: key1, value1, key2, value2, ...
+ * @return A structured object expression
+ * @see #objectOf(Class, Object...)
+ */
+ public static ApiExpression objectOf(String className, Object... fields) {
+ return apiCallAtLeastOneArgument(
+ BuiltInFunctionDefinitions.OBJECT_OF, valueLiteral(className), fields);
+ }
+
/**
* Creates an interval of rows.
*
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 37f3509f1e4..912f46de2bb 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -1618,6 +1618,16 @@ public final class BuiltInFunctionDefinitions {
.runtimeClass("org.apache.flink.table.runtime.functions.scalar.EltFunction")
.build();
+ public static final BuiltInFunctionDefinition OBJECT_OF =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("OBJECT_OF")
+ .kind(SCALAR)
+ .inputTypeStrategy(SpecificInputTypeStrategies.OBJECT_OF)
+ .outputTypeStrategy(SpecificTypeStrategies.OBJECT_OF)
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.ObjectOfFunction")
+ .build();
+
// --------------------------------------------------------------------------------------------
// Math functions
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java
new file mode 100644
index 00000000000..c5287f1f210
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategy.java
@@ -0,0 +1,182 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Input type strategy for the {@code OBJECT_OF} function that validates argument types and counts.
+ *
+ * <p>This strategy validates the input arguments for the {@code OBJECT_OF} function, ensuring:
+ *
+ * <ul>
+ * <li>The argument count is odd (className + pairs of key-value arguments)
+ * <li>The first argument is a non-null STRING/VARCHAR representing the class name
+ * <li>All key arguments (odd positions after the first) are non-null STRING/VARCHAR types
+ * <li>Field names are unique across all key-value pairs
+ * <li>Value arguments (even positions after the first) can be any type
+ * </ul>
+ *
+ * <p>The expected function signature is: {@code OBJECT_OF(className, key1, value1, key2, value2,
+ * ...)}
+ *
+ * <p>Example valid calls:
+ *
+ * <ul>
+ * <li>{@code OBJECT_OF('com.example.User')} - empty object
+ * <li>{@code OBJECT_OF('com.example.User', 'name', 'Alice')} - single field
+ * <li>{@code OBJECT_OF('com.example.User', 'name', 'Alice', 'age', 30)} - multiple fields
+ * </ul>
+ *
+ * @see org.apache.flink.table.functions.BuiltInFunctionDefinitions#OBJECT_OF
+ * @see ObjectOfTypeStrategy
+ */
+@Internal
+public class ObjectOfInputTypeStrategy implements InputTypeStrategy {
+
+ private static final ArgumentCount AT_LEAST_ONE_ODD =
+ new ArgumentCount() {
+ @Override
+ public boolean isValidCount(final int count) {
+ return count % 2 == 1;
+ }
+
+ @Override
+ public Optional<Integer> getMinCount() {
+ return Optional.of(1);
+ }
+
+ @Override
+ public Optional<Integer> getMaxCount() {
+ return Optional.empty();
+ }
+ };
+
+ private static void validateClassArgument(
+ final CallContext callContext, final DataType firstArgumentDataType) {
+ final LogicalType classArgumentType = firstArgumentDataType.getLogicalType();
+
+ if (!classArgumentType.is(LogicalTypeFamily.CHARACTER_STRING)) {
+ final String errorMessage =
+ String.format(
+ "The first argument must be a non-nullable character string representing the class name, but was %s.",
+ classArgumentType.asSummaryString());
+ throw new ValidationException(errorMessage);
+ }
+
+ final Optional<String> argumentValue = callContext.getArgumentValue(0, String.class);
+ if (argumentValue.isEmpty()) {
+ final String errorMessage =
+ "The first argument must be a non-nullable character string literal representing the class name.";
+ throw new ValidationException(errorMessage);
+ }
+ }
+
+ private static void validateKeyArguments(
+ final CallContext callContext, final List<DataType> argumentDataTypes) {
+ final Set<String> fieldNames = new HashSet<>();
+ for (int i = 1; i < argumentDataTypes.size(); i += 2) {
+ final LogicalType fieldNameLogicalType = argumentDataTypes.get(i).getLogicalType();
+ validateFieldNameArgument(callContext, i, fieldNameLogicalType, fieldNames);
+ }
+ }
+
+ private static void validateFieldNameArgument(
+ final CallContext callContext,
+ final int pos,
+ final LogicalType logicalType,
+ final Set<String> fieldNames) {
+ if (!logicalType.is(LogicalTypeFamily.CHARACTER_STRING)) {
+ final String message =
+ String.format(
+ "The field key at position %d must be a non-nullable character string, but was %s.",
+ pos + 1, logicalType.asSummaryString());
+ throw new ValidationException(message);
+ }
+
+ final Optional<String> argumentValue = callContext.getArgumentValue(pos, String.class);
+ if (argumentValue.isEmpty()) {
+ final String message =
+ String.format(
+ "The field key at position %d must be a non-nullable character string literal.",
+ pos + 1);
+ throw new ValidationException(message);
+ }
+
+ final String fieldName = argumentValue.orElseThrow(IllegalStateException::new);
+
+ if (!fieldNames.add(fieldName)) {
+ final String message =
+ String.format(
+ "The field name '%s' at position %d is repeated.", fieldName, pos + 1);
+ throw new ValidationException(message);
+ }
+ }
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return AT_LEAST_ONE_ODD;
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ final CallContext callContext, final boolean throwOnFailure) {
+ final List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
+ try {
+ validateClassArgument(callContext, argumentDataTypes.get(0));
+ validateKeyArguments(callContext, argumentDataTypes);
+ } catch (ValidationException e) {
+ callContext.fail(throwOnFailure, e.getMessage(), argumentDataTypes);
+ }
+
+ return Optional.of(argumentDataTypes);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(final FunctionDefinition definition) {
+ // OBJECT_OF expects: className, key1, value1, key2, value2, ...
+ // OBJECT_OF(<class name>, [<key>, <value> [, <key>, <value> , ...]] )
+ final List<Signature.Argument> arguments = new ArrayList<>();
+
+ // Class name (required)
+ final Argument classArgument = Argument.of("class name", "STRING");
+ arguments.add(classArgument);
+
+ // Key-value pairs (optional, repeating)
+ arguments.add(Signature.Argument.ofVarying("[STRING, ANY]*"));
+
+ return List.of(Signature.of(arguments));
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java
new file mode 100644
index 00000000000..f5c229cb38a
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ObjectOfTypeStrategy.java
@@ -0,0 +1,102 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.table.types.logical.StructuredType;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.IntStream;
+
+/**
+ * Type strategy for the {@code OBJECT_OF} function that infers the output type as a structured
+ * type.
+ *
+ * <p>This strategy creates a {@link DataTypes#STRUCTURED} type based on the provided class name and
+ * key-value pairs. The function signature is: {@code OBJECT_OF(className, key1, value1, key2,
+ * value2, ...)}
+ *
+ * <p>The strategy performs the following operations:
+ *
+ * <ul>
+ * <li>Extracts the class name from the first argument (must be a non-null string literal)
+ * <li>Processes key-value pairs starting from the second argument
+ * <li>Extracts field names (keys) from odd-positioned arguments (indices 1, 3, 5, ...)
+ * </ul>
+ *
+ * <p><b>Examples:</b>
+ *
+ * <ul>
+ * <li>{@code OBJECT_OF('com.example.User', 'name', 'Alice', 'age', 30)} → {@code
+ * STRUCTURED<com.example.User>(name CHAR(5), 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>
+ *
+ * @see org.apache.flink.table.functions.BuiltInFunctionDefinitions#OBJECT_OF
+ * @see ObjectOfInputTypeStrategy
+ */
+@Internal
+public class ObjectOfTypeStrategy implements TypeStrategy {
+
+ private static DataType toStructuredType(
+ final String className, final CallContext callContext) {
+ final DataTypeFactory dataTypeFactory = callContext.getDataTypeFactory();
+ final ClassLoader classLoader = dataTypeFactory.getClassLoader();
+ final List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
+
+ final Field[] fields =
+ IntStream.iterate(1, i -> i < argumentDataTypes.size(), i -> i + 2)
+ .mapToObj(keyIdx -> toFieldDataType(callContext, keyIdx))
+ .toArray(Field[]::new);
+
+ final Optional<Class<?>> resolveClass = StructuredType.resolveClass(classLoader, className);
+
+ return resolveClass
+ .map(clazz -> DataTypes.STRUCTURED(clazz, fields))
+ .orElse(DataTypes.STRUCTURED(className, fields));
+ }
+
+ private static Field toFieldDataType(final CallContext callContext, final int keyIdx) {
+ final List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
+
+ final String fieldName =
+ callContext
+ .getArgumentValue(keyIdx, String.class)
+ .orElseThrow(IllegalStateException::new);
+
+ final DataType fieldDataType = argumentDataTypes.get(keyIdx + 1);
+
+ return DataTypes.FIELD(fieldName, fieldDataType);
+ }
+
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ final Optional<String> argumentValue = callContext.getArgumentValue(0, String.class);
+
+ return argumentValue.map(className -> toStructuredType(className, callContext));
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index 72b4d2560e4..d10653e8572 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -63,6 +63,9 @@ public final class SpecificInputTypeStrategies {
/** See {@link OverTypeStrategy}. */
public static final InputTypeStrategy OVER = new OverTypeStrategy();
+ /** See {@link ObjectOfInputTypeStrategy}. */
+ public static final InputTypeStrategy OBJECT_OF = new ObjectOfInputTypeStrategy();
+
/** See {@link WindowTimeIndictorInputTypeStrategy}. */
public static InputTypeStrategy windowTimeIndicator(TimestampKind timestampKind) {
return new WindowTimeIndictorInputTypeStrategy(timestampKind);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
index df8ac3eeb0f..49e6adfefed 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
@@ -188,6 +188,9 @@ public final class SpecificTypeStrategies {
}
};
+ /** Type strategy specific for {@link BuiltInFunctionDefinitions#OBJECT_OF}. */
+ public static final TypeStrategy OBJECT_OF = new ObjectOfTypeStrategy();
+
private SpecificTypeStrategies() {
// no instantiation
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategyTest.java
new file mode 100644
index 00000000000..0b9719955ce
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ObjectOfInputTypeStrategyTest.java
@@ -0,0 +1,165 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.utils.DataTypeFactoryMock;
+
+import java.util.stream.Stream;
+
+/** Tests for {@link ObjectOfInputTypeStrategy}. */
+class ObjectOfInputTypeStrategyTest extends InputTypeStrategiesTestBase {
+
+ private static final InputTypeStrategy OBJECT_OF_INPUT_STRATEGY =
+ BuiltInFunctionDefinitions.OBJECT_OF
+ .getTypeInference(new DataTypeFactoryMock())
+ .getInputTypeStrategy();
+ private static final String USER_CLASS_PATH = "com.example.User";
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ // Test valid OBJECT_OF with no fields and values
+ TestSpec.forStrategy("Valid OBJECT_OF with only class", OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(DataTypes.STRING().notNull())
+ .calledWithLiteralAt(0, USER_CLASS_PATH)
+ .expectSignature("f(class name STRING, [STRING, ANY]*...)")
+ .expectArgumentTypes(DataTypes.STRING().notNull()),
+
+ // Test valid number of arguments (odd number >= 1)
+ TestSpec.forStrategy(
+ "Valid OBJECT_OF with class and one field and value",
+ OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING().notNull(),
+ DataTypes.INT())
+ .calledWithLiteralAt(0, USER_CLASS_PATH)
+ .calledWithLiteralAt(1, "field1")
+ .expectArgumentTypes(
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING().notNull(),
+ DataTypes.INT()),
+
+ // Test with structured fields
+ TestSpec.forStrategy(
+ "Valid OBJECT_OF with multiple fields", OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.STRING().notNull(), // implementation class
+ DataTypes.STRING().notNull(), // field1 name
+ DataTypes.STRUCTURED(
+ "c1",
+ DataTypes.FIELD("f1", DataTypes.INT())), // field1 value
+ DataTypes.STRING().notNull(), // field2 name
+ DataTypes.STRUCTURED(
+ "c2",
+ DataTypes.FIELD("f1", DataTypes.FLOAT()))) // field2 value
+ .calledWithLiteralAt(0, USER_CLASS_PATH)
+ .calledWithLiteralAt(1, "field1")
+ .calledWithLiteralAt(3, "field2")
+ .expectArgumentTypes(
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING().notNull(),
+ DataTypes.STRUCTURED("c1", DataTypes.FIELD("f1", DataTypes.INT())),
+ DataTypes.STRING().notNull(),
+ DataTypes.STRUCTURED(
+ "c2", DataTypes.FIELD("f1", DataTypes.FLOAT()))),
+
+ // Invalid test case - class argument is type STRING with value null
+ TestSpec.forStrategy(
+ "Invalid OBJECT_OF with class argument as type STRING with null value",
+ OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(DataTypes.STRING())
+ .calledWithLiteralAt(0, null)
+ .expectErrorMessage(
+ "The first argument must be a non-nullable character string literal representing the class name."),
+
+ // Invalid test case - key is type STRING with value null
+ TestSpec.forStrategy(
+ "Valid OBJECT_OF with key as type STRING with null value",
+ OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.STRING().notNull(), DataTypes.STRING(), DataTypes.INT())
+ .calledWithLiteralAt(0, USER_CLASS_PATH)
+ .calledWithLiteralAt(1, null)
+ .expectErrorMessage(
+ "The field key at position 2 must be a non-nullable character string literal."),
+
+ // Invalid test case - even number of arguments
+ TestSpec.forStrategy(
+ "Invalid OBJECT_OF with even number of arguments",
+ OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.STRING().notNull(), // implementation class
+ DataTypes.STRING().notNull()) // only field name, missing value
+ .calledWithLiteralAt(0, USER_CLASS_PATH)
+ .expectErrorMessage("Invalid number of arguments."),
+
+ // Invalid test case - no arguments
+ TestSpec.forStrategy(
+ "Invalid OBJECT_OF with no arguments", OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes()
+ .expectErrorMessage("Invalid number of arguments."),
+
+ // Invalid test case - class name not a string
+ TestSpec.forStrategy(
+ "OBJECT_OF with non-string class name", OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(DataTypes.INT())
+ .calledWithLiteralAt(0, 72)
+ .expectArgumentTypes(DataTypes.INT())
+ .expectErrorMessage(
+ "The first argument must be a non-nullable character string representing the class name, but was INT."),
+
+ // Invalid test case - field name not a string
+ TestSpec.forStrategy(
+ "OBJECT_OF with non-string field name", OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.STRING().notNull(), // implementation class
+ DataTypes.INT(), // field name (not a string)
+ DataTypes.INT()) // field value
+ .calledWithLiteralAt(0, USER_CLASS_PATH)
+ .calledWithLiteralAt(1, 5)
+ .expectArgumentTypes(DataTypes.STRING(), DataTypes.INT(), DataTypes.INT())
+ .expectErrorMessage(
+ "The field key at position 2 must be a non-nullable character string, but was INT."),
+
+ // Invalid test case - repeated field names
+ TestSpec.forStrategy(
+ "OBJECT_OF with repeated field names", OBJECT_OF_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING().notNull(),
+ DataTypes.INT(),
+ DataTypes.STRING().notNull(),
+ DataTypes.INT())
+ .calledWithLiteralAt(0, USER_CLASS_PATH)
+ .calledWithLiteralAt(1, "field1")
+ .calledWithLiteralAt(3, "field1")
+ .expectArgumentTypes(
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING().notNull(),
+ DataTypes.INT(),
+ DataTypes.STRING().notNull(),
+ DataTypes.INT())
+ .expectErrorMessage("The field name 'field1' at position 4 is repeated."));
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java
index d7d6558143c..af7efaea01d 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java
@@ -19,9 +19,11 @@
package org.apache.flink.table.planner.functions;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.types.logical.StructuredType;
+import org.apache.flink.types.Row;
import java.sql.Timestamp;
import java.util.List;
@@ -34,6 +36,10 @@ public class StructuredFunctionsITCase extends BuiltInFunctionTestBase {
@Override
Stream<TestSetSpec> getTestSetSpecs() {
+ return Stream.of(structuredTypeTestCases(), objectOfTestCases()).flatMap(s -> s);
+ }
+
+ private static Stream<TestSetSpec> structuredTypeTestCases() {
return Stream.of(
TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS)
.onFieldsWithData(14, "Bob")
@@ -107,6 +113,90 @@ public class StructuredFunctionsITCase extends BuiltInFunctionTestBase {
DataTypes.of(NonDefaultType.class).notNull()));
}
+ private static Stream<TestSetSpec> objectOfTestCases() {
+ final Type1 type1 = Type1.of(42, "Bob");
+ final Type2 type2 = Type2.of(15, "Alice");
+ final NestedType nestedType = NestedType.of(type1, type2);
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.OBJECT_OF)
+ .onFieldsWithData(42, "Bob")
+ .andDataTypes(DataTypes.INT(), DataTypes.STRING())
+ .withFunction(Type1.Type1Constructor.class)
+ .withFunction(Type2.Type2Constructor.class)
+ .withFunction(NestedType.NestedConstructor.class)
+ // Test with OBJECT_OF
+ .testResult(
+ Expressions.objectOf(Type1.class, "a", 42, "b", "Bob"),
+ "OBJECT_OF('" + Type1.class.getName() + "', 'a', 42, 'b', 'Bob')",
+ type1,
+ DataTypes.STRUCTURED(
+ Type1.class.getName(),
+ DataTypes.FIELD("a", DataTypes.INT().notNull()),
+ DataTypes.FIELD("b", DataTypes.CHAR(3).notNull())))
+ // Test with nested structured types
+ .testResult(
+ Expressions.objectOf(
+ NestedType.class,
+ "n1",
+ Expressions.objectOf(Type1.class, "a", 42, "b", "Bob"),
+ "n2",
+ Expressions.objectOf(Type2.class, "a", 15, "b", "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'))",
+ nestedType,
+ DataTypes.STRUCTURED(
+ NestedType.class.getName(),
+ DataTypes.FIELD(
+ "n1",
+ DataTypes.STRUCTURED(
+ Type1.class.getName(),
+ DataTypes.FIELD(
+ "a", DataTypes.INT().notNull()),
+ DataTypes.FIELD(
+ "b", DataTypes.CHAR(3).notNull()))),
+ DataTypes.FIELD(
+ "n2",
+ DataTypes.STRUCTURED(
+ Type2.class.getName(),
+ DataTypes.FIELD(
+ "a", DataTypes.INT().notNull()),
+ DataTypes.FIELD(
+ "b",
+ DataTypes.CHAR(5).notNull())))))
+ // Test equal with constructor
+ // TODO: Test disabled due to FLINK-38083
+ // .testSqlResult(
+ // "Type1Constructor(f0, f1) = OBJECT_OF('"
+ // + Type1.class.getName()
+ // + "', 'a', 42, 'b', 'Bob')",
+ // true,
+ // DataTypes.BOOLEAN())
+ // Test OBJECT_OF when class not found
+ .testSqlResult(
+ "OBJECT_OF('not.existing.clazz', 'a', 42, 'b', 'Bob')",
+ Row.of(42, "Bob"),
+ DataTypes.STRUCTURED(
+ "not.existing.clazz",
+ DataTypes.FIELD("a", DataTypes.INT().notNull()),
+ DataTypes.FIELD("b", DataTypes.CHAR(3).notNull())))
+ // Invalid Test - field name is not a string literal
+ .testSqlValidationError(
+ "OBJECT_OF('"
+ + Type1.class.getName()
+ + "', CAST(NULL AS STRING), 42, 'b', 'Bob')",
+ "The field key at position 2 must be a non-nullable character string literal.")
+ // Invalid Test - first argument is type string but null
+ .testSqlValidationError(
+ "OBJECT_OF(CAST(NULL AS STRING), 'a', '12', 'b', 'Alice')",
+ "The first argument must be a non-nullable character string literal representing the class name."));
+ }
+
// --------------------------------------------------------------------------------------------
// Helpers
// --------------------------------------------------------------------------------------------
@@ -119,6 +209,30 @@ public class StructuredFunctionsITCase extends BuiltInFunctionTestBase {
public Integer a;
public String b;
+ public static Type1 of(final Integer a, final String b) {
+ final Type1 t = new Type1();
+ t.a = a;
+ t.b = b;
+ return t;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Type1 that = (Type1) o;
+ return Objects.equals(this.a, that.a) && Objects.equals(this.b, that.b);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(a, b);
+ }
+
public static class Type1Constructor extends ScalarFunction {
public Type1 eval(Integer a, String b) {
final Type1 t = new Type1();
@@ -137,6 +251,30 @@ public class StructuredFunctionsITCase extends BuiltInFunctionTestBase {
public Integer a;
public String b;
+ public static Type2 of(final Integer a, final String b) {
+ final Type2 t = new Type2();
+ t.a = a;
+ t.b = b;
+ return t;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Type2 that = (Type2) o;
+ return Objects.equals(a, that.a) && Objects.equals(b, that.b);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(a, b);
+ }
+
public static class Type2Constructor extends ScalarFunction {
public Type2 eval(Integer a, String b) {
final Type2 t = new Type2();
@@ -157,6 +295,30 @@ public class StructuredFunctionsITCase extends BuiltInFunctionTestBase {
public Type1 n1;
public Type2 n2;
+ public static NestedType of(final Type1 n1, final Type2 n2) {
+ final NestedType t = new NestedType();
+ t.n1 = n1;
+ t.n2 = n2;
+ return t;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ NestedType that = (NestedType) o;
+ return Objects.equals(n1, that.n1) && Objects.equals(n2, that.n2);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(n1, n2);
+ }
+
public static class NestedConstructor extends ScalarFunction {
public NestedType eval(Type1 n1, Type2 n2) {
final NestedType t = new NestedType();
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java
new file mode 100644
index 00000000000..ad5a39f9c5e
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ObjectOfFunction.java
@@ -0,0 +1,47 @@
+/*
+ * 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.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#OBJECT_OF}. */
+@Internal
+public class ObjectOfFunction extends BuiltInScalarFunction {
+
+ public ObjectOfFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.OBJECT_OF, context);
+ }
+
+ public RowData eval(StringData className, Object... fieldNameAndValuePairs) {
+ final int fieldCount = fieldNameAndValuePairs.length / 2;
+ final GenericRowData row = new GenericRowData(fieldCount);
+
+ for (int i = 0; i < fieldCount; i++) {
+ Object fieldValues = fieldNameAndValuePairs[2 * i + 1];
+ row.setField(i, fieldValues);
+ }
+
+ return row;
+ }
+}