raminqaf commented on code in PR #26638:
URL: https://github.com/apache/flink/pull/26638#discussion_r2131553523
##########
docs/content.zh/docs/dev/table/types.md:
##########
@@ -1281,15 +1293,51 @@ For some classes an annotation is required in order to
map the class to a data t
to assign a fixed precision and scale for `java.math.BigDecimal`).
{{< /tab >}}
{{< tab "Python" >}}
+```python
+Not supported.
+```
{{< /tab >}}
{{< /tabs >}}
**Declaration**
{{< tabs "c5e5527b-b09d-4dc5-9549-8fd2bfc7cc2a" >}}
-{{< tab "Java" >}}
+{{< tab "Java/Scala" >}}
+Structured types are usually declared via their implementation classes:
+
```java
-class User {
+// A simple POJO that qualifies as a structured type.
+// Note: Without a fully assigning constructor, the order of fields will be
alphabetical.
+// The final data type will be:
+// STRUCTURED<'com.myorg.Customer', active BOOLEAN, id INT NOT NULL, name
STRING, properties MAP<STRING, STRING>>
+class Customer {
+ public int id;
+ public String name;
+ public Map<String, String> properties;
+ public boolean active;
+}
+
+// A POJO with a fully assigning constructor defining the field order.
+// The final data type will be:
+// STRUCTURED<'com.myorg.Customer', id INT NOT NULL, name STRING, properties
MAP<STRING, STRING>, active BOOLEAN>
Review Comment:
Question: if the field `id` was defined as `public Integer id` would the
structured type be
```
STRUCTURED<'com.myorg.Customer', id INT, name STRING, properties MAP<STRING,
STRING>, active BOOLEAN>
```
##########
docs/content.zh/docs/dev/table/types.md:
##########
@@ -1232,34 +1232,46 @@ equivalent to `ROW<myField INT, myOtherField BOOLEAN>`.
### User-Defined Data Types
-{{< tabs "udf" >}}
-{{< tab "Java/Scala" >}}
-<span class="label label-danger">Attention</span> User-defined data types are
not fully supported yet. They are
-currently (as of Flink 1.11) only exposed as unregistered structured types in
parameters and return types of functions.
+#### `STRUCTURED`
-A structured type is similar to an object in an object-oriented programming
language. It contains
-zero, one or more attributes. Each attribute consists of a name and a type.
+Data type for a user-defined object.
-There are two kinds of structured types:
+Compared to `ROW`, which may also be considered a "struct-like" type,
structured types are distinguishable even if they
+contain the same set of fields. For example, `Visit(amount DOUBLE)` is
distinct from `Interaction(amount DOUBLE)` due
+its identifier.
-- Types that are stored in a catalog and are identified by a _catalog
identifier_ (like `cat.db.MyType`). Those
-are equal to the SQL standard definition of structured types.
+Similar to classes in object-oriented programming languages, structured types
are identified by a class name and contain
+zero, one or more attributes. Each attribute consists of a name and a type. A
type cannot be defined in such a way that
+one of its attribute types (transitively) refers to itself.
-- Anonymously defined, unregistered types (usually reflectively extracted)
that are identified by
-an _implementation class_ (like `com.myorg.model.MyType`). Those are useful
when programmatically
-defining a table program. They enable reusing existing JVM classes without
manually defining the
-schema of a data type again.
+Structured types are internally converted by the system into suitable data
structures. Serialization and equality checks
+are managed by the system based on the logical type.
-#### Registered Structured Types
+{{< tabs "udt" >}}
+{{< tab "SQL" >}}
+```sql
+STRUCTURED<'c', n0 t0, n1 t1, ...>
+STRUCTURED<'c', n0 t0, n1 t1 'd1', ...>
+```
+The type can be declared using `STRUCTURED<'c', n0 t0 'd0', n1 t1 'd1', ...>`
where `c` is the class name, `n` is the
+unique name of a field, `t` is the logical type of a field, `d` is the
description of a field.
Review Comment:
```suggestion
unique name of a field, `t` is the logical type of a field, `d` is the
optional description of a field.
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java:
##########
Review Comment:
Do we want to have IT tests for nested STRUCTURED type?
##########
docs/content.zh/docs/dev/table/types.md:
##########
@@ -1232,34 +1232,46 @@ equivalent to `ROW<myField INT, myOtherField BOOLEAN>`.
### User-Defined Data Types
-{{< tabs "udf" >}}
-{{< tab "Java/Scala" >}}
-<span class="label label-danger">Attention</span> User-defined data types are
not fully supported yet. They are
-currently (as of Flink 1.11) only exposed as unregistered structured types in
parameters and return types of functions.
+#### `STRUCTURED`
-A structured type is similar to an object in an object-oriented programming
language. It contains
-zero, one or more attributes. Each attribute consists of a name and a type.
+Data type for a user-defined object.
-There are two kinds of structured types:
+Compared to `ROW`, which may also be considered a "struct-like" type,
structured types are distinguishable even if they
+contain the same set of fields. For example, `Visit(amount DOUBLE)` is
distinct from `Interaction(amount DOUBLE)` due
+its identifier.
-- Types that are stored in a catalog and are identified by a _catalog
identifier_ (like `cat.db.MyType`). Those
-are equal to the SQL standard definition of structured types.
+Similar to classes in object-oriented programming languages, structured types
are identified by a class name and contain
+zero, one or more attributes. Each attribute consists of a name and a type. A
type cannot be defined in such a way that
Review Comment:
```suggestion
zero, one or more attributes. Each attribute has a name, a type, and an
optional description. A type cannot be defined in such a way that
```
##########
docs/content.zh/docs/dev/table/types.md:
##########
@@ -1281,15 +1293,51 @@ For some classes an annotation is required in order to
map the class to a data t
to assign a fixed precision and scale for `java.math.BigDecimal`).
{{< /tab >}}
{{< tab "Python" >}}
+```python
+Not supported.
+```
{{< /tab >}}
{{< /tabs >}}
**Declaration**
{{< tabs "c5e5527b-b09d-4dc5-9549-8fd2bfc7cc2a" >}}
-{{< tab "Java" >}}
+{{< tab "Java/Scala" >}}
+Structured types are usually declared via their implementation classes:
+
```java
-class User {
+// A simple POJO that qualifies as a structured type.
Review Comment:
I don't know how important that would be, but having a tiny example with the
description doesn't hurt. WDYT?
##########
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -2489,6 +2489,34 @@ SqlTypeNameSpec ExtendedSqlRowTypeName() :
}
}
+/**
+ * Parse inline structured type such as STRUCTURED<'className', name1 type1
'comment', name2 type2>.
+ */
+SqlTypeNameSpec SqlStructuredTypeName() :
+{
+ final SqlNode className;
+ final List<SqlIdentifier> fieldNames = new ArrayList<SqlIdentifier>();
+ final List<SqlDataTypeSpec> fieldTypes = new ArrayList<SqlDataTypeSpec>();
+ final List<SqlCharStringLiteral> comments = new
ArrayList<SqlCharStringLiteral>();
Review Comment:
Maybe staying consistent here would be good. In the docs (also in some parts
of the code), this field is named `description`:
```suggestion
final List<SqlCharStringLiteral> descriptions = new
ArrayList<SqlCharStringLiteral>();
```
This should apply to all the `comment` names in this file
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java:
##########
@@ -33,4 +35,10 @@ public interface ExtendedRelTypeFactory extends
RelDataTypeFactory {
/** Creates a RAW type such as {@code RAW('org.my.Class', 'sW3Djsds...')}.
*/
RelDataType createRawType(String className, String serializerString);
+
+ /**
+ * Creates a STRUCTURED type such as {@code STRUCTURED('org.my.Class',
name STRING, age INT)}.
+ */
+ RelDataType createStructuredType(
+ String className, List<RelDataType> typeList, List<String>
fieldNameList);
Review Comment:
Question: Why is the comment/description field not present here? I think
this because we don't really need them, and they are just for clarity, written
by the user?
```suggestion
/**
* Creates a STRUCTURED type such as {@code STRUCTURED('org.my.Class',
name STRING 'full name of the person', age INT)}.
*/
RelDataType createStructuredType(
String className, List<RelDataType> typeList, List<String>
fieldNameList, List<String> fieldCommentList);
```
##########
docs/content.zh/docs/dev/table/types.md:
##########
@@ -1301,35 +1349,27 @@ class User {
// enrich the extraction with forcing using RAW types
public @DataTypeHint("RAW") Class<?> modelClass;
}
-
-DataTypes.of(User.class);
```
-**Bridging to JVM Types**
-
-| Java Type | Input | Output | Remarks
|
-|:-------------------------------------|:-----:|:------:|:----------------------------------------|
-|*class* | X | X | Originating class or
subclasses (for input) or <br>superclasses (for output). *Default* |
-|`org.apache.flink.types.Row` | X | X | Represent the
structured type as a row. |
-|`org.apache.flink.table.data.RowData` | X | X | Internal data
structure. |
-
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-case class User(
-
- // extract fields automatically
- age: Int,
- name: String,
+Or via explicit declaration:
+```java
+DataTypes.STRUCTURED(Class, DataTypes.FIELD(n0, t0), DataTypes.FIELD(n1, t1),
...);
+DataTypes.STRUCTURED(String, DataTypes.FIELD(n0, t0), DataTypes.FIELD(n1, t1),
...);
Review Comment:
Shouldn't this be?
```suggestion
DataTypes.STRUCTURED(String.class, DataTypes.FIELD(n0, t0),
DataTypes.FIELD(n1, t1), ...);
```
And what is the difference between this line and the one before it?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.planner.functions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.types.logical.StructuredType;
+
+import java.util.stream.Stream;
+
+/** Tests for functions dealing with {@link StructuredType}. */
+public class StructuredFunctionsITCase extends BuiltInFunctionTestBase {
+
+ @Override
+ Stream<TestSetSpec> getTestSetSpecs() {
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS)
+ .onFieldsWithData(14, "Bob")
+ .andDataTypes(DataTypes.INT(), DataTypes.STRING())
+ .withFunction(Type1.Type1Constructor.class)
+ .withFunction(Type2.Type2Constructor.class)
+ // Same value from function
+ .testSqlResult(
+ "Type1Constructor(f0, f1) =
Type1Constructor(14, 'Bob')",
+ true,
+ DataTypes.BOOLEAN())
+ // Same value from CAST
+ .testSqlResult(
+ "Type1Constructor(f0, f1) = CAST((14, 'Bob')
AS "
+ + Type1.TYPE
+ + ")",
+ true,
+ DataTypes.BOOLEAN())
+ // Different value from function
+ .testSqlResult(
+ "Type1Constructor(f0, f1) =
Type1Constructor(15, 'Alice')",
+ false,
+ DataTypes.BOOLEAN())
+ // Different value from CAST
+ .testSqlResult(
+ "Type1Constructor(f0, f1) = CAST((15, 'Alice')
AS "
+ + Type1.TYPE
+ + ")",
+ false,
+ DataTypes.BOOLEAN())
+ // Different class name from function
+ .testSqlValidationError(
+ "Type1Constructor(f0, f1) =
Type2Constructor(14, 'Bob')",
+ "Incompatible structured types")
+ // Different class name from CAST
+ .testSqlValidationError(
+ "Type1Constructor(f0, f1) = CAST((14, 'Bob')
AS "
+ + Type2.TYPE
+ + ")",
+ "Incompatible structured types")
+ // Same class name but different fields
+ .testSqlValidationError(
+ "Type1Constructor(f0, f1) = CAST((14, 'Bob')
AS STRUCTURED<'"
+ + Type1.class.getName()
+ + "', a BOOLEAN, b BOOLEAN>)",
+ "Cannot apply '=' to arguments"));
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Helpers
+ //
--------------------------------------------------------------------------------------------
+
+ /** Structured type Type1. */
+ public static class Type1 {
+ public static final String TYPE =
Review Comment:
nit: can be private
```suggestion
private static final String TYPE =
```
Same with Type2
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StructuredFunctionsITCase.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.planner.functions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.types.logical.StructuredType;
+
+import java.util.stream.Stream;
+
+/** Tests for functions dealing with {@link StructuredType}. */
+public class StructuredFunctionsITCase extends BuiltInFunctionTestBase {
+
+ @Override
+ Stream<TestSetSpec> getTestSetSpecs() {
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS)
+ .onFieldsWithData(14, "Bob")
+ .andDataTypes(DataTypes.INT(), DataTypes.STRING())
+ .withFunction(Type1.Type1Constructor.class)
+ .withFunction(Type2.Type2Constructor.class)
+ // Same value from function
+ .testSqlResult(
+ "Type1Constructor(f0, f1) =
Type1Constructor(14, 'Bob')",
+ true,
+ DataTypes.BOOLEAN())
+ // Same value from CAST
+ .testSqlResult(
+ "Type1Constructor(f0, f1) = CAST((14, 'Bob')
AS "
+ + Type1.TYPE
+ + ")",
+ true,
+ DataTypes.BOOLEAN())
+ // Different value from function
+ .testSqlResult(
+ "Type1Constructor(f0, f1) =
Type1Constructor(15, 'Alice')",
+ false,
+ DataTypes.BOOLEAN())
+ // Different value from CAST
+ .testSqlResult(
+ "Type1Constructor(f0, f1) = CAST((15, 'Alice')
AS "
+ + Type1.TYPE
+ + ")",
+ false,
+ DataTypes.BOOLEAN())
+ // Different class name from function
+ .testSqlValidationError(
+ "Type1Constructor(f0, f1) =
Type2Constructor(14, 'Bob')",
+ "Incompatible structured types")
+ // Different class name from CAST
+ .testSqlValidationError(
+ "Type1Constructor(f0, f1) = CAST((14, 'Bob')
AS "
+ + Type2.TYPE
+ + ")",
+ "Incompatible structured types")
+ // Same class name but different fields
+ .testSqlValidationError(
+ "Type1Constructor(f0, f1) = CAST((14, 'Bob')
AS STRUCTURED<'"
+ + Type1.class.getName()
+ + "', a BOOLEAN, b BOOLEAN>)",
+ "Cannot apply '=' to arguments"));
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Helpers
+ //
--------------------------------------------------------------------------------------------
+
+ /** Structured type Type1. */
+ public static class Type1 {
+ public static final String TYPE =
+ "STRUCTURED<'" + Type1.class.getName() + "', a INT, b STRING>";
+
+ public Integer a;
Review Comment:
Question: Changing the type to `int` on both classes will cause the tests to
fail. Why is that?
```suggestion
public int a;
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java:
##########
@@ -928,37 +928,88 @@ public static <T> UnresolvedDataType
RAW(TypeInformation<T> typeInformation) {
}
/**
- * Data type of a user-defined object structured type. Structured types
contain zero, one or
- * more attributes. Each attribute consists of a name and a type. A type
cannot be defined so
- * that one of its attribute types (transitively) uses itself.
- *
- * <p>There are two kinds of structured types. Types that are stored in a
catalog and are
- * identified by an {@link ObjectIdentifier} or anonymously defined,
unregistered types (usually
- * reflectively extracted) that are identified by an implementation {@link
Class}.
- *
- * <p>This method helps in manually constructing anonymous, unregistered
types. This is useful
- * in cases where the reflective extraction using {@link
DataTypes#of(Class)} is not applicable.
- * However, {@link DataTypes#of(Class)} is the recommended way of creating
inline structured
- * types as it also considers {@link DataTypeHint}s.
- *
- * <p>Structured types are converted to internal data structures by the
runtime. The given
- * implementation class is only used at the edges of the table ecosystem
(e.g. when bridging to
- * a function or connector). Serialization and equality ({@code
hashCode/equals}) are handled by
- * the runtime based on the logical type. An implementation class must
offer a default
- * constructor with zero arguments or a full constructor that assigns all
attributes.
- *
- * <p>Note: A caller of this method must make sure that the {@link
- * DataType#getConversionClass()} of the given fields matches with the
attributes of the given
- * implementation class, otherwise an exception might be thrown during
runtime.
+ * Data type of a user-defined object structured type. Structured types
are identified by a
+ * class name and contain zero, one or more attributes. Each attribute
consists of a name and a
+ * type. A type cannot be defined in such a way that one of its attribute
types (transitively)
+ * refers to itself.
+ *
+ * <p>Compared to {@link #ROW(Field...)}, which may also be considered a
"struct-like" type,
+ * structured types are distinguishable even if they contain the same set
of fields. For
+ * example, "Visit(amount DOUBLE)" is distinct from "Interaction(amount
DOUBLE)" due its
+ * identifier.
+ *
+ * <p>This method allows for manually constructing an inline structured
type. This is useful in
+ * cases where the reflective extraction using {@link DataTypes#of(Class)}
is not applicable.
+ * However, {@link DataTypes#of(Class)} is the recommended approach for
creating inline
+ * structured types, as it also considers {@link DataTypeHint}s.
+ *
+ * <p>Structured types are internally converted by the system into
suitable data structures.
+ * Serialization and equality checks (e.g. {@code hashCode/equals}) are
managed by the system
+ * based on the logical type.
+ *
+ * <p>If an optional implementation class is provided, the system will
convert a structured
+ * object to a JVM object at the edges of the table ecosystem (e.g. when
bridging to a function
+ * or connector). The implementation class must provide either a
zero-argument constructor or a
+ * full constructor that assigns all attributes. The class name does not
need to be resolvable
+ * in the classpath; it may be used solely to distinguish between objects
with identical
+ * attribute sets. However, in Table API and UDF calls, the system will
attempt to resolve the
+ * class name to an actual implementation class. If resolution fails,
{@link Row} is used as a
+ * fallback.
+ *
+ * <p>Note: The caller of this method must ensure that the {@link
DataType#getConversionClass()}
+ * of each field matches the corresponding attribute in the implementation
class. Otherwise, a
+ * runtime exception may be thrown.
*
* @see DataTypes#of(Class)
* @see StructuredType
*/
public static <T> DataType STRUCTURED(Class<T> implementationClass,
Field... fields) {
// some basic validation of the class to prevent common mistakes
validateStructuredClass(implementationClass);
+ return
buildStructuredType(StructuredType.newBuilder(implementationClass), fields);
+ }
- final StructuredType.Builder builder =
StructuredType.newBuilder(implementationClass);
+ /**
+ * Data type of a user-defined object structured type. Structured types
are identified by a
+ * class name and contain zero, one or more attributes. Each attribute
consists of a name and a
+ * type. A type cannot be defined in such a way that one of its attribute
types (transitively)
+ * refers to itself.
+ *
+ * <p>Compared to {@link #ROW(Field...)}, which may also be considered a
"struct-like" type,
+ * structured types are distinguishable even if they contain the same set
of fields. For
+ * example, "Visit(amount DOUBLE)" is distinct from "Interaction(amount
DOUBLE)" due its
+ * identifier.
+ *
+ * <p>This method allows for manually constructing an inline structured
type. This is useful in
+ * cases where the reflective extraction using {@link DataTypes#of(Class)}
is not applicable.
+ * However, {@link DataTypes#of(Class)} is the recommended approach for
creating inline
+ * structured types, as it also considers {@link DataTypeHint}s.
+ *
+ * <p>Structured types are internally converted by the system into
suitable data structures.
+ * Serialization and equality checks (e.g. {@code hashCode/equals}) are
managed by the system
+ * based on the logical type.
+ *
+ * <p>If an optional implementation class is provided, the system will
convert a structured
+ * object to a JVM object at the edges of the table ecosystem (e.g. when
bridging to a function
+ * or connector). The implementation class must provide either a
zero-argument constructor or a
+ * full constructor that assigns all attributes. The class name does not
need to be resolvable
+ * in the classpath; it may be used solely to distinguish between objects
with identical
+ * attribute sets. However, in Table API and UDF calls, the system will
attempt to resolve the
+ * class name to an actual implementation class. If resolution fails,
{@link Row} is used as a
+ * fallback.
+ *
+ * <p>Note: The caller of this method must ensure that the {@link
DataType#getConversionClass()}
+ * of each field matches the corresponding attribute in the implementation
class. Otherwise, a
+ * runtime exception may be thrown.
+ *
+ * @see DataTypes#of(Class)
+ * @see StructuredType
+ */
+ public static <T> DataType STRUCTURED(String className, Field... fields) {
+ return buildStructuredType(StructuredType.newBuilder(className),
fields);
Review Comment:
Do you think it would be a good idea to fail fast here with a simple
validation like `className.isBlank() || className.isEmpty()`?
Or is it a valid statement to have
```sql
STRUCTURED<``, f0 t0>
STRUCTURED<` `, f0 t0>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
