You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Au-Miner (via GitHub)" <gi...@apache.org> on 2026/05/12 11:06:24 UTC
[PR] [FLINK-36847][table] Fix enum field conversion in Table API to DataStream [flink]
Au-Miner opened a new pull request, #28144:
URL: https://github.com/apache/flink/pull/28144
## What is the purpose of the change
Fix `TableEnvironment.toDataStream(..., clazz)` for POJOs that contain enum fields. The change makes enum fields extractable and convertible when the table side represents them as string values.
## Brief change log
- Update `DataTypeExtractor#extractDataTypeOrError` to recognize enum classes via `extractEnumType`, and extend `CharType`/`VarCharType` conversion checks so enum classes can be bridged to string logical types.
- Extend `DataStructureConverters#getConverterInternal` with enum handling for `CHAR`/`VARCHAR` and add `StringEnumConverter` to convert between `StringData` and enum values via `Enum.name()` / `Enum.valueOf()`.
## Verifying this change
Updated `DataTypeExtractorTest` to cover plain enums, enums with additional fields, and POJOs containing enum fields, asserting that they are extracted as `STRING` bridged to the enum class. Updated `DataStructureConvertersTest` to verify round-trip conversion between `StringData` and enum values for string logical types.
## 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)`: no
- 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? no
- If yes, how is the feature documented? not applicable
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36847][table] Fix enum field conversion in Table API to DataStream [flink]
Posted by "Au-Miner (via GitHub)" <gi...@apache.org>.
Au-Miner commented on code in PR #28144:
URL: https://github.com/apache/flink/pull/28144#discussion_r3266546840
##########
flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/StringEnumConverter.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.data.conversion;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+/**
+ * Converter for {@link CharType}/{@link VarCharType} of {@link Enum} external type. The enum is
+ * represented as its {@link Enum#name()}.
+ */
+@Internal
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class StringEnumConverter implements DataStructureConverter<StringData, Enum> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Class<? extends Enum> enumClass;
+
+ public StringEnumConverter(Class<? extends Enum> enumClass) {
+ this.enumClass = enumClass;
+ }
+
+ @Override
+ public StringData toInternal(Enum external) {
+ return StringData.fromString(external.name());
+ }
+
+ @Override
+ public Enum toExternal(StringData internal) {
+ return Enum.valueOf(enumClass, internal.toString());
Review Comment:
Referring to BitmapBitmapConverter, explicitly throwing exceptions
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36847][table] Fix enum field conversion in Table API to DataStream [flink]
Posted by "Au-Miner (via GitHub)" <gi...@apache.org>.
Au-Miner commented on code in PR #28144:
URL: https://github.com/apache/flink/pull/28144#discussion_r3266541224
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java:
##########
@@ -544,6 +550,15 @@ else if (clazz == java.time.Period.class) {
return ClassDataTypeConverter.extractDataType(clazz).orElse(null);
}
+ /** Maps an {@link Enum} class to STRING bridged to the enum class. */
+ private @Nullable DataType extractEnumType(Type type) {
+ final Class<?> clazz = toClass(type);
+ if (clazz == null || !clazz.isEnum()) {
+ return null;
+ }
+ return DataTypes.STRING().bridgedTo(clazz);
Review Comment:
There are annotations in [notes](https://github.com/apache/flink/pull/28144/changes#diff-7c1b17f4ff0ededfc43d477dd06738f3a96e62952a3eb28a1dde7da63966e023R29) to explain the current naming convention
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36847][table] Fix enum field conversion in Table API to DataStream [flink]
Posted by "featzhang (via GitHub)" <gi...@apache.org>.
featzhang commented on code in PR #28144:
URL: https://github.com/apache/flink/pull/28144#discussion_r3228549765
##########
flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java:
##########
@@ -595,6 +598,13 @@ private static Map<LocalDate, Integer> createLocalDateMultiset() {
// Structured types
// --------------------------------------------------------------------------------------------
+ private enum ConverterEnum {
+ FIRST,
+ SECOND
+ }
Review Comment:
The test covers `DetailedEnum` with additional fields, which is good. It verifies that the enum constant name is used regardless of extra state.
One missing test case: enum constant name mismatches. For example:
```java
DataStructureConverter converter = StringEnumConverter.create(
DataTypes.STRING().bridgedTo(ConverterEnum.class));
// Should throw IllegalArgumentException
converter.toExternal(StringData.fromString("INVALID"));
```
This would verify the error path and document the expected behavior when data doesn't match the enum.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36847][table] Fix enum field conversion in Table API to DataStream [flink]
Posted by "featzhang (via GitHub)" <gi...@apache.org>.
featzhang commented on code in PR #28144:
URL: https://github.com/apache/flink/pull/28144#discussion_r3228547863
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java:
##########
@@ -544,6 +550,15 @@ else if (clazz == java.time.Period.class) {
return ClassDataTypeConverter.extractDataType(clazz).orElse(null);
}
+ /** Maps an {@link Enum} class to STRING bridged to the enum class. */
+ private @Nullable DataType extractEnumType(Type type) {
+ final Class<?> clazz = toClass(type);
+ if (clazz == null || !clazz.isEnum()) {
+ return null;
+ }
+ return DataTypes.STRING().bridgedTo(clazz);
Review Comment:
The enum extraction returns `STRING` bridged to the enum class, which is clean.
One consideration: what if the enum has custom `toString()` methods? The converter uses `Enum.name()` which is always the constant name, but users might expect `toString()`.
For example:
```java
enum Status {
ACTIVE { public String toString() { return "active"; } },
INACTIVE { public String toString() { return "inactive"; } }
}
```
Using `name()` would give "ACTIVE", but `toString()` gives "active". The current behavior is consistent with standard serialization practices, but worth documenting if this becomes a user question.
##########
flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/data/conversion/StringEnumConverter.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.data.conversion;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+/**
+ * Converter for {@link CharType}/{@link VarCharType} of {@link Enum} external type. The enum is
+ * represented as its {@link Enum#name()}.
+ */
+@Internal
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class StringEnumConverter implements DataStructureConverter<StringData, Enum> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Class<? extends Enum> enumClass;
+
+ public StringEnumConverter(Class<? extends Enum> enumClass) {
+ this.enumClass = enumClass;
+ }
+
+ @Override
+ public StringData toInternal(Enum external) {
+ return StringData.fromString(external.name());
+ }
+
+ @Override
+ public Enum toExternal(StringData internal) {
+ return Enum.valueOf(enumClass, internal.toString());
Review Comment:
The converter uses `Enum.valueOf()` which throws `IllegalArgumentException` if the string doesn't match any constant name.
Should the converter handle invalid enum values gracefully? For example, if the table contains "INVALID_VALUE" but the enum only has FIRST/SECOND/THIRD.
Current behavior: exception propagates to the runtime
Alternative: return null or a default value
Depending on Flink's error handling policy for data conversion, this might be fine. Just worth noting that malformed data will cause job failures rather than null values.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36847][table] Fix enum field conversion in Table API to DataStream [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28144:
URL: https://github.com/apache/flink/pull/28144#issuecomment-4429902718
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8a76c8b808834282574ccdaab4e3617407d2375a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8a76c8b808834282574ccdaab4e3617407d2375a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8a76c8b808834282574ccdaab4e3617407d2375a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36847][table] Fix enum field conversion in Table API to DataStream [flink]
Posted by "Au-Miner (via GitHub)" <gi...@apache.org>.
Au-Miner commented on code in PR #28144:
URL: https://github.com/apache/flink/pull/28144#discussion_r3266554027
##########
flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java:
##########
@@ -595,6 +598,13 @@ private static Map<LocalDate, Integer> createLocalDateMultiset() {
// Structured types
// --------------------------------------------------------------------------------------------
+ private enum ConverterEnum {
+ FIRST,
+ SECOND
+ }
Review Comment:
Added `testStringEnumConverterRejectsInvalidEnumName` and `testStringEnumConverterUsesEnumName` for testing these two special cases
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org