This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 3a29bf50b [client] Enum conversion functionality for typed
writer/reader (#3447)
3a29bf50b is described below
commit 3a29bf50b393b8d605f4d27edc62b18459962e1e
Author: Vladyslav Banar <[email protected]>
AuthorDate: Wed Jun 24 16:37:08 2026 +0200
[client] Enum conversion functionality for typed writer/reader (#3447)
* [client] Add support of column aliasing for PojoConverter
* [client] enum conversion functionality for typed writer/reader
* [client] Add support of column aliasing for PojoConverter (#3444)
* [client] Add support of column aliasing for PojoConverter
* [client] test
* [client] updated doc
* [client] fix checks
* [client] address comments
---------
Co-authored-by: valdbanar5 <[email protected]>
(cherry picked from commit 9dd1a56c439e1bfc0527db7d452244478c82d5fc)
* [client] fix test
* [client] fix merge
* [client] address comments
* [client] address comments
* [client] address comments
* [client] fix formatting
* [client] fix for java 8 compatibility
* [client] fix for java 8 compatibility
* [client] re-run the build
* [client] use enum.name() during converting to BinaryString
* [client] adjusted tests to cover additional enum conversion cases
* [client] address comments
---------
Co-authored-by: valdbanar5 <[email protected]>
---
.../fluss/client/converter/ConverterCommons.java | 15 +-
.../client/converter/FlussArrayToPojoArray.java | 11 +-
.../converter/FlussTypeToPojoTypeConverter.java | 38 +-
.../apache/fluss/client/converter/PojoType.java | 13 +-
.../fluss/client/converter/RowToPojoConverter.java | 16 +-
.../client/converter/ConverterCommonsTest.java | 638 +++++++++++++++++++++
.../client/converter/ConvertersTestFixtures.java | 56 +-
.../FlussTypeToPojoTypeConverterTest.java | 287 +++++++++
.../converter/PojoArrayToFlussArrayTest.java | 18 +
.../fluss/client/converter/PojoTypeTest.java | 9 +-
.../client/converter/RowToPojoConverterTest.java | 21 +-
11 files changed, 1095 insertions(+), 27 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java
b/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java
index 82415403f..3e718bf00 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java
@@ -148,6 +148,15 @@ final class ConverterCommons {
}
return;
}
+ if (actual.isEnum()) {
+ if (typeRoot != DataTypeRoot.STRING) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Enum field '%s' must be a STRING column, got
%s",
+ prop.name, typeRoot));
+ }
+ return;
+ }
Set<Class<?>> supported = SUPPORTED_TYPES.get(fieldType.getTypeRoot());
if (supported == null) {
@@ -171,13 +180,17 @@ final class ConverterCommons {
}
static BinaryString toBinaryStringForText(Object v, String fieldName,
DataTypeRoot root) {
- final String s = String.valueOf(v);
+ final String s = objectToString(v);
if (root == DataTypeRoot.CHAR && s.length() != 1) {
throw new
IllegalArgumentException(charLengthExceptionMessage(fieldName, s.length()));
}
return BinaryString.fromString(s);
}
+ private static String objectToString(Object v) {
+ return v instanceof Enum ? ((Enum<?>) v).name() : String.valueOf(v);
+ }
+
static Set<Class<?>> setOf(Class<?>... javaTypes) {
return new HashSet<>(Arrays.asList(javaTypes));
}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussArrayToPojoArray.java
b/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussArrayToPojoArray.java
index 59bf26420..fb4e8f8ed 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussArrayToPojoArray.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussArrayToPojoArray.java
@@ -32,6 +32,7 @@ import java.lang.reflect.Array;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
/** Adapter class for converting Fluss InternalArray to Pojo Array. */
public class FlussArrayToPojoArray {
@@ -115,9 +116,17 @@ public class FlussArrayToPojoArray {
case STRING:
// Default to String when the POJO element type is unspecified
(Object[])
final Class<?> textTarget = (pojoType == Object.class) ?
String.class : pojoType;
+ final Map<String, Object> enumConstantsMap =
+ textTarget.isEnum()
+ ?
FlussTypeToPojoTypeConverter.buildEnumConstantsMap(textTarget)
+ : null;
return (arr, i) ->
FlussTypeToPojoTypeConverter.convertTextValue(
- elementType, fieldName, textTarget,
arr.getString(i));
+ elementType,
+ fieldName,
+ textTarget,
+ arr.getString(i),
+ enumConstantsMap);
case BINARY:
case BYTES:
return InternalArray::getBytes;
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverter.java
b/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverter.java
index 2b693f4e3..6fb09985f 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverter.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverter.java
@@ -31,9 +31,28 @@ import java.time.LocalDate;
import java.time.LocalTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
/** Shared utilities for Fluss type and Pojo type. */
public class FlussTypeToPojoTypeConverter {
+
+ /**
+ * Builds a map of enum name to enum constant for the given enum class.
+ *
+ * @param enumClass the enum class
+ * @return an immutable map from enum constant name to enum constant
+ */
+ static Map<String, Object> buildEnumConstantsMap(Class<?> enumClass) {
+ Map<String, Object> map = new HashMap<>();
+ for (Object constant : enumClass.getEnumConstants()) {
+ map.put(((Enum<?>) constant).name(), constant);
+ }
+ return Collections.unmodifiableMap(map);
+ }
+
/**
* Converts a text value (CHAR/STRING) read from an InternalRow into the
target Java type
* declared by the POJO property.
@@ -45,12 +64,18 @@ public class FlussTypeToPojoTypeConverter {
* @param fieldName The field name
* @param pojoType The pojo type
* @param s The BinaryString read from the row
+ * @param enumConstantsMap (optional) Pre-built enum constants map for
enum types; ignored for
+ * non-enum types
* @return Converted Java value (String or Character)
* @throws IllegalArgumentException if the target type is unsupported or
constraints are
* violated
*/
static Object convertTextValue(
- DataType fieldType, String fieldName, Class<?> pojoType,
BinaryString s) {
+ DataType fieldType,
+ String fieldName,
+ Class<?> pojoType,
+ BinaryString s,
+ Map<String, Object> enumConstantsMap) {
if (s == null) {
return null;
}
@@ -74,6 +99,17 @@ public class FlussTypeToPojoTypeConverter {
ConverterCommons.charLengthExceptionMessage(fieldName,
v.length()));
}
return v.charAt(0);
+ } else if (pojoType.isEnum()) {
+ Map<String, Object> enumMap =
+ enumConstantsMap != null ? enumConstantsMap :
buildEnumConstantsMap(pojoType);
+ Object enumConstant = enumMap.get(v);
+ if (enumConstant == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Could not parse value for enum %s. Expected
one of: %s",
+ pojoType,
Arrays.toString(pojoType.getEnumConstants())));
+ }
+ return enumConstant;
}
throw new IllegalArgumentException(
String.format(
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java
b/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java
index bf57a3299..a81f523d8 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java
@@ -126,8 +126,7 @@ final class PojoType<T> {
field.getGenericType(),
publicField ? field : null,
getter,
- setter,
- mappedColumnName));
+ setter));
}
return new PojoType<>(pojoClass, ctor, props);
@@ -298,12 +297,6 @@ final class PojoType<T> {
/** The generic type of the field (e.g. {@code Map<String,
AddressPojo>}). */
final Type genericType;
- /**
- * The name of the column in the Fluss table. This may differ from
'name' if a @ColumnName
- * annotation is present. Used for looking up the property by table
column name.
- */
- final String mappedName;
-
@Nullable final Field publicField;
@Nullable final Method getter;
@Nullable final Method setter;
@@ -314,12 +307,10 @@ final class PojoType<T> {
Type genericType,
@Nullable Field publicField,
@Nullable Method getter,
- @Nullable Method setter,
- String mappedName) {
+ @Nullable Method setter) {
this.name = Objects.requireNonNull(name, "name");
this.type = Objects.requireNonNull(type, "type");
this.genericType = Objects.requireNonNull(genericType,
"genericType");
- this.mappedName = Objects.requireNonNull(mappedName, "mappedName");
this.publicField = publicField;
this.getter = getter;
this.setter = setter;
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/converter/RowToPojoConverter.java
b/fluss-client/src/main/java/org/apache/fluss/client/converter/RowToPojoConverter.java
index a54b8d97d..3897f5057 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/converter/RowToPojoConverter.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/converter/RowToPojoConverter.java
@@ -141,9 +141,19 @@ public final class RowToPojoConverter<T> {
return InternalRow::getDouble;
case CHAR:
case STRING:
- return (row, pos) ->
- FlussTypeToPojoTypeConverter.convertTextValue(
- fieldType, prop.name, prop.type,
row.getString(pos));
+ {
+ final Map<String, Object> enumConstantsMap =
+ prop.type.isEnum()
+ ?
FlussTypeToPojoTypeConverter.buildEnumConstantsMap(prop.type)
+ : null;
+ return (row, pos) ->
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ fieldType,
+ prop.name,
+ prop.type,
+ row.getString(pos),
+ enumConstantsMap);
+ }
case BINARY:
case BYTES:
return InternalRow::getBytes;
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/converter/ConverterCommonsTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/converter/ConverterCommonsTest.java
new file mode 100644
index 000000000..a3e0c5a58
--- /dev/null
+++
b/fluss-client/src/test/java/org/apache/fluss/client/converter/ConverterCommonsTest.java
@@ -0,0 +1,638 @@
+/*
+ * 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.fluss.client.converter;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ConverterCommons}. */
+public class ConverterCommonsTest {
+
+ // ==================== validatePojoMatchesTable Tests ====================
+
+ @Test
+ public void validatePojoMatchesTableWithExactMatch() {
+ RowType table =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .build();
+
+ PojoType<AllFieldsPojo> pojoType = PojoType.of(AllFieldsPojo.class);
+ ConverterCommons.validatePojoMatchesTable(pojoType, table);
+ }
+
+ @Test
+ public void validatePojoMatchesTableWithTypeIncompatibility() {
+ RowType table =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.INT())
+ .build();
+
+ PojoType<AllFieldsPojo> pojoType = PojoType.of(AllFieldsPojo.class);
+ assertThatThrownBy(() ->
ConverterCommons.validatePojoMatchesTable(pojoType, table))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("incompatible with Fluss type");
+ }
+
+ @Test
+ public void validatePojoMatchesTableWithMissingField() {
+ RowType table =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .field("age", DataTypes.INT())
+ .build();
+
+ PojoType<AllFieldsPojo> pojoType = PojoType.of(AllFieldsPojo.class);
+ assertThatThrownBy(() ->
ConverterCommons.validatePojoMatchesTable(pojoType, table))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must exactly match");
+ }
+
+ // ==================== validatePojoMatchesProjection Tests
====================
+
+ @Test
+ public void validatePojoMatchesProjectionWithSubset() {
+ RowType projection =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .build();
+
+ PojoType<AllFieldsPojo> pojoType = PojoType.of(AllFieldsPojo.class);
+ ConverterCommons.validatePojoMatchesProjection(pojoType, projection);
+ }
+
+ @Test
+ public void validatePojoMatchesProjectionWithSingleField() {
+ RowType projection = RowType.builder().field("id",
DataTypes.INT()).build();
+
+ PojoType<AllFieldsPojo> pojoType = PojoType.of(AllFieldsPojo.class);
+ ConverterCommons.validatePojoMatchesProjection(pojoType, projection);
+ }
+
+ @Test
+ public void validatePojoMatchesProjectionWithMissingField() {
+ RowType projection =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("missingField", DataTypes.STRING())
+ .build();
+
+ PojoType<AllFieldsPojo> pojoType = PojoType.of(AllFieldsPojo.class);
+ assertThatThrownBy(
+ () ->
ConverterCommons.validatePojoMatchesProjection(pojoType, projection))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void validatePojoMatchesProjectionWithTypeIncompatibility() {
+ RowType projection =
+ RowType.builder()
+ .field("id", DataTypes.STRING())
+ .field("name", DataTypes.STRING())
+ .build();
+
+ PojoType<AllFieldsPojo> pojoType = PojoType.of(AllFieldsPojo.class);
+ assertThatThrownBy(
+ () ->
ConverterCommons.validatePojoMatchesProjection(pojoType, projection))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ // ==================== validateProjectionSubset Tests ====================
+
+ @Test
+ public void validateProjectionSubsetAllFieldsInTable() {
+ RowType tableSchema =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .field("age", DataTypes.INT())
+ .build();
+
+ RowType projection =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .build();
+
+ ConverterCommons.validateProjectionSubset(projection, tableSchema);
+ }
+
+ @Test
+ public void validateProjectionSubsetSingleFieldInTable() {
+ RowType tableSchema =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .build();
+
+ RowType projection = RowType.builder().field("id",
DataTypes.INT()).build();
+
+ ConverterCommons.validateProjectionSubset(projection, tableSchema);
+ }
+
+ @Test
+ public void validateProjectionSubsetWithFieldNotInTable() {
+ RowType tableSchema =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .build();
+
+ RowType projection =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("unknown", DataTypes.STRING())
+ .build();
+
+ assertThatThrownBy(() ->
ConverterCommons.validateProjectionSubset(projection, tableSchema))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ // ==================== validateCompatibility Tests ====================
+
+ @Test
+ public void compatibilityBooleanWithBoolean() {
+ PojoType<BooleanPojo> pojoType = PojoType.of(BooleanPojo.class);
+ PojoType.Property prop = pojoType.getProperty("flag");
+ ConverterCommons.validateCompatibility(DataTypes.BOOLEAN(), prop);
+ }
+
+ @Test
+ public void compatibilityTinyintWithByte() {
+ PojoType<BytePojo> pojoType = PojoType.of(BytePojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.TINYINT(), prop);
+ }
+
+ @Test
+ public void compatibilitySmallintWithShort() {
+ PojoType<ShortPojo> pojoType = PojoType.of(ShortPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.SMALLINT(), prop);
+ }
+
+ @Test
+ public void compatibilityIntegerWithInteger() {
+ PojoType<IntPojo> pojoType = PojoType.of(IntPojo.class);
+ PojoType.Property prop = pojoType.getProperty("id");
+ ConverterCommons.validateCompatibility(DataTypes.INT(), prop);
+ }
+
+ @Test
+ public void compatibilityBigintWithLong() {
+ PojoType<LongPojo> pojoType = PojoType.of(LongPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.BIGINT(), prop);
+ }
+
+ @Test
+ public void compatibilityFloatWithFloat() {
+ PojoType<FloatPojo> pojoType = PojoType.of(FloatPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.FLOAT(), prop);
+ }
+
+ @Test
+ public void compatibilityDoubleWithDouble() {
+ PojoType<DoublePojo> pojoType = PojoType.of(DoublePojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.DOUBLE(), prop);
+ }
+
+ @Test
+ public void compatibilityCharWithString() {
+ PojoType<StringPojo> pojoType = PojoType.of(StringPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.CHAR(10), prop);
+ }
+
+ @Test
+ public void compatibilityCharWithCharacter() {
+ PojoType<CharacterPojo> pojoType = PojoType.of(CharacterPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.CHAR(1), prop);
+ }
+
+ @Test
+ public void compatibilityStringWithString() {
+ PojoType<StringPojo> pojoType = PojoType.of(StringPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.STRING(), prop);
+ }
+
+ @Test
+ public void compatibilityStringWithCharacter() {
+ PojoType<CharacterPojo> pojoType = PojoType.of(CharacterPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ ConverterCommons.validateCompatibility(DataTypes.STRING(), prop);
+ }
+
+ @Test
+ public void compatibilityBinaryWithByteArray() {
+ PojoType<ByteArrayPojo> pojoType = PojoType.of(ByteArrayPojo.class);
+ PojoType.Property prop = pojoType.getProperty("bytes");
+ ConverterCommons.validateCompatibility(DataTypes.BINARY(100), prop);
+ }
+
+ @Test
+ public void compatibilityBytesWithByteArray() {
+ PojoType<ByteArrayPojo> pojoType = PojoType.of(ByteArrayPojo.class);
+ PojoType.Property prop = pojoType.getProperty("bytes");
+ ConverterCommons.validateCompatibility(DataTypes.BYTES(), prop);
+ }
+
+ @Test
+ public void compatibilityDecimalWithBigDecimal() {
+ PojoType<BigDecimalPojo> pojoType = PojoType.of(BigDecimalPojo.class);
+ PojoType.Property prop = pojoType.getProperty("amount");
+ ConverterCommons.validateCompatibility(DataTypes.DECIMAL(10, 2), prop);
+ }
+
+ @Test
+ public void compatibilityDateWithLocalDate() {
+ PojoType<LocalDatePojo> pojoType = PojoType.of(LocalDatePojo.class);
+ PojoType.Property prop = pojoType.getProperty("date");
+ ConverterCommons.validateCompatibility(DataTypes.DATE(), prop);
+ }
+
+ @Test
+ public void compatibilityTimeWithLocalTime() {
+ PojoType<LocalTimePojo> pojoType = PojoType.of(LocalTimePojo.class);
+ PojoType.Property prop = pojoType.getProperty("time");
+ ConverterCommons.validateCompatibility(DataTypes.TIME(), prop);
+ }
+
+ @Test
+ public void compatibilityTimestampNtzWithLocalDateTime() {
+ PojoType<LocalDateTimePojo> pojoType =
PojoType.of(LocalDateTimePojo.class);
+ PojoType.Property prop = pojoType.getProperty("timestamp");
+ ConverterCommons.validateCompatibility(DataTypes.TIMESTAMP(), prop);
+ }
+
+ @Test
+ public void compatibilityTimestampLtzWithInstant() {
+ PojoType<InstantPojo> pojoType = PojoType.of(InstantPojo.class);
+ PojoType.Property prop = pojoType.getProperty("timestamp");
+ ConverterCommons.validateCompatibility(DataTypes.TIMESTAMP_LTZ(),
prop);
+ }
+
+ @Test
+ public void compatibilityTimestampLtzWithOffsetDateTime() {
+ PojoType<OffsetDateTimePojo> pojoType =
PojoType.of(OffsetDateTimePojo.class);
+ PojoType.Property prop = pojoType.getProperty("timestamp");
+ ConverterCommons.validateCompatibility(DataTypes.TIMESTAMP_LTZ(),
prop);
+ }
+
+ @Test
+ public void compatibilityArrayWithArrayType() {
+ PojoType<IntArrayPojo> pojoType = PojoType.of(IntArrayPojo.class);
+ PojoType.Property prop = pojoType.getProperty("items");
+
ConverterCommons.validateCompatibility(DataTypes.ARRAY(DataTypes.INT()), prop);
+ }
+
+ @Test
+ public void compatibilityArrayWithListType() {
+ PojoType<ListPojo> pojoType = PojoType.of(ListPojo.class);
+ PojoType.Property prop = pojoType.getProperty("items");
+
ConverterCommons.validateCompatibility(DataTypes.ARRAY(DataTypes.INT()), prop);
+ }
+
+ @Test
+ public void compatibilityMapType() {
+ PojoType<MapPojo> pojoType = PojoType.of(MapPojo.class);
+ PojoType.Property prop = pojoType.getProperty("mapping");
+ ConverterCommons.validateCompatibility(
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), prop);
+ }
+
+ @Test
+ public void compatibilityRowWithNestedPojo() {
+ PojoType<NestedPojo> pojoType = PojoType.of(NestedPojo.class);
+ PojoType.Property prop = pojoType.getProperty("nested");
+ ConverterCommons.validateCompatibility(
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING())),
+ prop);
+ }
+
+ @Test
+ public void compatibilityEnumWithString() {
+ PojoType<EnumPojo> pojoType = PojoType.of(EnumPojo.class);
+ PojoType.Property prop = pojoType.getProperty("status");
+ ConverterCommons.validateCompatibility(DataTypes.STRING(), prop);
+ }
+
+ @Test
+ public void incompatibilityBooleanWithString() {
+ PojoType<StringPojo> pojoType = PojoType.of(StringPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ assertThatThrownBy(() ->
ConverterCommons.validateCompatibility(DataTypes.BOOLEAN(), prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("incompatible");
+ }
+
+ @Test
+ public void incompatibilityIntegerWithString() {
+ PojoType<StringPojo> pojoType = PojoType.of(StringPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ assertThatThrownBy(() ->
ConverterCommons.validateCompatibility(DataTypes.INT(), prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("incompatible");
+ }
+
+ @Test
+ public void incompatibilityArrayWithString() {
+ PojoType<StringPojo> pojoType = PojoType.of(StringPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ assertThatThrownBy(
+ () ->
+ ConverterCommons.validateCompatibility(
+ DataTypes.ARRAY(DataTypes.INT()),
prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be an array or Collection");
+ }
+
+ @Test
+ public void incompatibilityMapWithString() {
+ PojoType<StringPojo> pojoType = PojoType.of(StringPojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ assertThatThrownBy(
+ () ->
+ ConverterCommons.validateCompatibility(
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT()), prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be a Map");
+ }
+
+ @Test
+ public void incompatibilityRowWithArray() {
+ PojoType<IntArrayPojo> pojoType = PojoType.of(IntArrayPojo.class);
+ PojoType.Property prop = pojoType.getProperty("items");
+ assertThatThrownBy(
+ () ->
+ ConverterCommons.validateCompatibility(
+ DataTypes.ROW(DataTypes.FIELD("id",
DataTypes.INT())),
+ prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be a POJO class");
+ }
+
+ @Test
+ public void incompatibilityRowWithList() {
+ PojoType<ListPojo> pojoType = PojoType.of(ListPojo.class);
+ PojoType.Property prop = pojoType.getProperty("items");
+ assertThatThrownBy(
+ () ->
+ ConverterCommons.validateCompatibility(
+ DataTypes.ROW(DataTypes.FIELD("id",
DataTypes.INT())),
+ prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be a POJO class");
+ }
+
+ @Test
+ public void incompatibilityEnumWithInt() {
+ PojoType<EnumPojo> pojoType = PojoType.of(EnumPojo.class);
+ PojoType.Property prop = pojoType.getProperty("status");
+ assertThatThrownBy(() ->
ConverterCommons.validateCompatibility(DataTypes.INT(), prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be a STRING column");
+ }
+
+ @Test
+ public void incompatibilityBigintWithByte() {
+ PojoType<BytePojo> pojoType = PojoType.of(BytePojo.class);
+ PojoType.Property prop = pojoType.getProperty("value");
+ assertThatThrownBy(() ->
ConverterCommons.validateCompatibility(DataTypes.BIGINT(), prop))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("incompatible");
+ }
+
+ @Test
+ public void convertStringToBinaryStringForCharTypeWithWrongLengthThrows() {
+ assertThatThrownBy(
+ () ->
+ ConverterCommons.toBinaryStringForText(
+ "Hello", "testField",
DataTypes.CHAR(1).getTypeRoot()))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void convertBooleanToBinaryString() {
+ BinaryString result =
+ ConverterCommons.toBinaryStringForText(
+ true, "field", DataTypes.STRING().getTypeRoot());
+ assertThat(result.toString()).isEqualTo("true");
+ }
+
+ @Test
+ public void convertNullValueToBinaryString() {
+ BinaryString result =
+ ConverterCommons.toBinaryStringForText(
+ null, "field", DataTypes.STRING().getTypeRoot());
+ assertThat(result.toString()).isEqualTo("null");
+ }
+
+ // ==================== Test POJOs ====================
+
+ /** Test POJO with multiple fields. */
+ public static class AllFieldsPojo {
+ public Integer id;
+ public String name;
+
+ public AllFieldsPojo() {}
+ }
+
+ /** Test POJO with Boolean field. */
+ public static class BooleanPojo {
+ public Boolean flag;
+
+ public BooleanPojo() {}
+ }
+
+ /** Test POJO with Byte field. */
+ public static class BytePojo {
+ public Byte value;
+
+ public BytePojo() {}
+ }
+
+ /** Test POJO with Short field. */
+ public static class ShortPojo {
+ public Short value;
+
+ public ShortPojo() {}
+ }
+
+ /** Test POJO with Integer field. */
+ public static class IntPojo {
+ public Integer id;
+
+ public IntPojo() {}
+ }
+
+ /** Test POJO with Long field. */
+ public static class LongPojo {
+ public Long value;
+
+ public LongPojo() {}
+ }
+
+ /** Test POJO with Float field. */
+ public static class FloatPojo {
+ public Float value;
+
+ public FloatPojo() {}
+ }
+
+ /** Test POJO with Double field. */
+ public static class DoublePojo {
+ public Double value;
+
+ public DoublePojo() {}
+ }
+
+ /** Test POJO with String field. */
+ public static class StringPojo {
+ public String value;
+
+ public StringPojo() {}
+ }
+
+ /** Test POJO with Character field. */
+ public static class CharacterPojo {
+ public Character value;
+
+ public CharacterPojo() {}
+ }
+
+ /** Test POJO with byte[] field. */
+ public static class ByteArrayPojo {
+ public byte[] bytes;
+
+ public ByteArrayPojo() {}
+ }
+
+ /** Test POJO with BigDecimal field. */
+ public static class BigDecimalPojo {
+ public BigDecimal amount;
+
+ public BigDecimalPojo() {}
+ }
+
+ /** Test POJO with LocalDate field. */
+ public static class LocalDatePojo {
+ public LocalDate date;
+
+ public LocalDatePojo() {}
+ }
+
+ /** Test POJO with LocalTime field. */
+ public static class LocalTimePojo {
+ public LocalTime time;
+
+ public LocalTimePojo() {}
+ }
+
+ /** Test POJO with LocalDateTime field. */
+ public static class LocalDateTimePojo {
+ public LocalDateTime timestamp;
+
+ public LocalDateTimePojo() {}
+ }
+
+ /** Test POJO with Instant field. */
+ public static class InstantPojo {
+ public Instant timestamp;
+
+ public InstantPojo() {}
+ }
+
+ /** Test POJO with OffsetDateTime field. */
+ public static class OffsetDateTimePojo {
+ public OffsetDateTime timestamp;
+
+ public OffsetDateTimePojo() {}
+ }
+
+ /** Test POJO with Integer[] field. */
+ public static class IntArrayPojo {
+ public Integer[] items;
+
+ public IntArrayPojo() {}
+ }
+
+ /** Test POJO with List field. */
+ public static class ListPojo {
+ public List<Integer> items;
+
+ public ListPojo() {}
+ }
+
+ /** Test POJO with Map field. */
+ public static class MapPojo {
+ public Map<String, Integer> mapping;
+
+ public MapPojo() {}
+ }
+
+ /** Test POJO with nested POJO field. */
+ public static class NestedPojo {
+ public AllFieldsPojo nested;
+
+ public NestedPojo() {}
+ }
+
+ /** Test POJO with Enum field. */
+ public static class EnumPojo {
+ public StatusEnum status;
+
+ public EnumPojo() {}
+ }
+
+ /** Test enum for compatibility testing. */
+ public enum StatusEnum {
+ ACTIVE,
+ inactive;
+
+ @Override
+ public String toString() {
+ return "StatusEnum{}";
+ }
+ }
+}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/converter/ConvertersTestFixtures.java
b/fluss-client/src/test/java/org/apache/fluss/client/converter/ConvertersTestFixtures.java
index 0a90c522d..8852e0fea 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/converter/ConvertersTestFixtures.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/converter/ConvertersTestFixtures.java
@@ -57,7 +57,11 @@ public final class ConvertersTestFixtures {
.field("offsetDateTimeField", DataTypes.TIMESTAMP_LTZ())
.field("arrayField", DataTypes.ARRAY(DataTypes.INT()))
.field("mapField", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT()))
+ .field("enumField", DataTypes.STRING())
.field("string_with_column_name", DataTypes.STRING())
+ .field("enumList", DataTypes.ARRAY(DataTypes.STRING()))
+ .field("enumValueMap", DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING()))
+ .field("enumKeyMap", DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING()))
.build();
}
@@ -82,10 +86,15 @@ public final class ConvertersTestFixtures {
public OffsetDateTime offsetDateTimeField;
public Integer[] arrayField;
public Map<String, Integer> mapField;
+ public StatusEnum enumField;
@ColumnName("string_with_column_name")
public String stringWithColumnName;
+ public List<StatusEnum> enumList;
+ public Map<String, StatusEnum> enumValueMap;
+ public Map<StatusEnum, String> enumKeyMap;
+
public TestPojo() {}
public TestPojo(
@@ -106,7 +115,11 @@ public final class ConvertersTestFixtures {
OffsetDateTime offsetDateTimeField,
Integer[] arrayField,
Map<String, Integer> mapField,
- String stringWithColumnName) {
+ StatusEnum enumField,
+ String stringWithColumnName,
+ List<StatusEnum> enumList,
+ Map<String, StatusEnum> enumValueMap,
+ Map<StatusEnum, String> enumKeyMap) {
this.booleanField = booleanField;
this.byteField = byteField;
this.shortField = shortField;
@@ -124,7 +137,11 @@ public final class ConvertersTestFixtures {
this.offsetDateTimeField = offsetDateTimeField;
this.arrayField = arrayField;
this.mapField = mapField;
+ this.enumField = enumField;
this.stringWithColumnName = stringWithColumnName;
+ this.enumList = enumList;
+ this.enumValueMap = enumValueMap;
+ this.enumKeyMap = enumKeyMap;
}
public static TestPojo sample() {
@@ -151,7 +168,21 @@ public final class ConvertersTestFixtures {
put("test_2", 2);
}
},
- "string value");
+ StatusEnum.OK,
+ "string value",
+ Arrays.asList(StatusEnum.error, StatusEnum.OK),
+ new HashMap<String, StatusEnum>() {
+ {
+ put("key1", StatusEnum.OK);
+ put("key2", StatusEnum.error);
+ }
+ },
+ new HashMap<StatusEnum, String>() {
+ {
+ put(StatusEnum.OK, "value1");
+ put(StatusEnum.error, "value2");
+ }
+ });
}
@Override
@@ -179,8 +210,12 @@ public final class ConvertersTestFixtures {
&& Objects.equals(timestampLtzField,
testPojo.timestampLtzField)
&& Objects.equals(offsetDateTimeField,
testPojo.offsetDateTimeField)
&& Objects.equals(stringWithColumnName,
testPojo.stringWithColumnName)
+ && Objects.equals(enumField, testPojo.enumField)
&& Arrays.equals(arrayField, testPojo.arrayField)
- && Objects.equals(mapField, testPojo.mapField);
+ && Objects.equals(mapField, testPojo.mapField)
+ && Objects.equals(enumList, testPojo.enumList)
+ && Objects.equals(enumKeyMap, testPojo.enumKeyMap)
+ && Objects.equals(enumValueMap, testPojo.enumValueMap);
}
@Override
@@ -201,7 +236,11 @@ public final class ConvertersTestFixtures {
timestampField,
timestampLtzField,
stringWithColumnName,
+ enumField,
offsetDateTimeField,
+ enumList,
+ enumKeyMap,
+ enumValueMap,
mapField);
result = 31 * result + Arrays.hashCode(bytesField);
result = 31 * result + Arrays.hashCode(arrayField);
@@ -485,4 +524,15 @@ public final class ConvertersTestFixtures {
public ListOfRowPojo() {}
}
+
+ /** Enum to test enum conversion. */
+ public enum StatusEnum {
+ OK,
+ error;
+
+ @Override
+ public String toString() {
+ return "StatusEnum{}";
+ }
+ }
}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverterTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverterTest.java
new file mode 100644
index 000000000..970194e42
--- /dev/null
+++
b/fluss-client/src/test/java/org/apache/fluss/client/converter/FlussTypeToPojoTypeConverterTest.java
@@ -0,0 +1,287 @@
+/*
+ * 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.fluss.client.converter;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link FlussTypeToPojoTypeConverter}. */
+public class FlussTypeToPojoTypeConverterTest {
+
+ // ==================== convertTextValue Tests ====================
+
+ @Test
+ public void testConvertTextValueToString() {
+ BinaryString binaryStr = BinaryString.fromString("Hello");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(), "field", String.class, binaryStr,
null);
+ assertThat(result).isEqualTo("Hello");
+ }
+
+ @Test
+ public void testConvertTextValueToStringFromChar() {
+ BinaryString binaryStr = BinaryString.fromString("A");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.CHAR(1), "field", String.class, binaryStr,
null);
+ assertThat(result).isEqualTo("A");
+ }
+
+ @Test
+ public void testConvertTextValueToCharacter() {
+ BinaryString binaryStr = BinaryString.fromString("X");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(), "field", Character.class,
binaryStr, null);
+ assertThat(result).isEqualTo('X');
+ }
+
+ @Test
+ public void testConvertTextValueToCharacterFromChar() {
+ BinaryString binaryStr = BinaryString.fromString("Z");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.CHAR(1), "field", Character.class,
binaryStr, null);
+ assertThat(result).isEqualTo('Z');
+ }
+
+ @Test
+ public void testConvertTextValueNullReturnsNull() {
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(), "field", String.class, null, null);
+ assertThat(result).isNull();
+ }
+
+ @Test
+ public void testConvertTextValueCharWithLengthOneIsValid() {
+ BinaryString binaryStr = BinaryString.fromString("M");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.CHAR(1), "field", String.class, binaryStr,
null);
+ assertThat(result).isEqualTo("M");
+ }
+
+ @Test
+ public void testConvertTextValueCharWithWrongLengthThrows() {
+ BinaryString binaryStr = BinaryString.fromString("Hello");
+ assertThatThrownBy(
+ () ->
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.CHAR(1),
+ "myField",
+ String.class,
+ binaryStr,
+ null))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testConvertTextValueCharacterWithWrongLengthThrows() {
+ BinaryString binaryStr = BinaryString.fromString("AB");
+ assertThatThrownBy(
+ () ->
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.CHAR(1),
+ "charField",
+ Character.class,
+ binaryStr,
+ null))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testConvertTextValueEmptyStringToCharacterThrows() {
+ BinaryString binaryStr = BinaryString.fromString("");
+ assertThatThrownBy(
+ () ->
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(),
+ "emptyField",
+ Character.class,
+ binaryStr,
+ null))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testConvertTextValueToEnumWithMatchingValue() {
+ BinaryString binaryStr = BinaryString.fromString("ACTIVE");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(),
+ "status",
+ StatusEnum.class,
+ binaryStr,
+ buildConstantMap());
+ assertThat(result).isEqualTo(StatusEnum.ACTIVE);
+ }
+
+ @Test
+ public void testConvertTextValueToEnumWithLowercaseInput() {
+ BinaryString binaryStr = BinaryString.fromString("finished");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(),
+ "status",
+ StatusEnum.class,
+ binaryStr,
+ buildConstantMap());
+ assertThat(result).isEqualTo(StatusEnum.finished);
+ }
+
+ @Test
+ public void testConvertTextValueToEnumWithInvalidValueThrows() {
+ BinaryString binaryStr = BinaryString.fromString("UNKNOWN");
+ assertThatThrownBy(
+ () ->
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(),
+ "status",
+ StatusEnum.class,
+ binaryStr,
+ buildConstantMap()))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testConvertTextValueUnsupportedTypeThrows() {
+ BinaryString binaryStr = BinaryString.fromString("value");
+ assertThatThrownBy(
+ () ->
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(),
+ "wrongField",
+ Integer.class,
+ binaryStr,
+ buildConstantMap()))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void
testConvertTextValueToEnumWithMatchingValueAndWithoutConstantMap() {
+ BinaryString binaryStr = BinaryString.fromString("ACTIVE");
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTextValue(
+ DataTypes.STRING(), "status", StatusEnum.class,
binaryStr, null);
+ assertThat(result).isEqualTo(StatusEnum.ACTIVE);
+ }
+
+ // ==================== convertDateValue Tests ====================
+
+ @Test
+ public void testConvertDateValueEpoch() {
+ LocalDate result = FlussTypeToPojoTypeConverter.convertDateValue(0);
+ assertThat(result).isEqualTo(LocalDate.of(1970, 1, 1));
+ }
+
+ @Test
+ public void testConvertDateValuePositiveOffset() {
+ LocalDate result =
FlussTypeToPojoTypeConverter.convertDateValue(18993);
+ assertThat(result).isEqualTo(LocalDate.of(2022, 1, 1));
+ }
+
+ // ==================== convertTimeValue Tests ====================
+
+ @Test
+ public void testConvertTimeValueMidnight() {
+ LocalTime result = FlussTypeToPojoTypeConverter.convertTimeValue(0);
+ assertThat(result).isEqualTo(LocalTime.MIDNIGHT);
+ }
+
+ @Test
+ public void testConvertTimeValueNoon() {
+ long millisOfDay = 12 * 60 * 60 * 1000;
+ LocalTime result = FlussTypeToPojoTypeConverter.convertTimeValue((int)
millisOfDay);
+ assertThat(result).isEqualTo(LocalTime.of(12, 0, 0));
+ }
+
+ // ==================== convertTimestampNtzValue Tests ====================
+
+ @Test
+ public void testConvertTimestampNtzValue() {
+ TimestampNtz ts =
TimestampNtz.fromLocalDateTime(LocalDateTime.of(2023, 1, 15, 10, 30));
+ Object result =
FlussTypeToPojoTypeConverter.convertTimestampNtzValue(ts);
+ assertThat(result).isEqualTo(LocalDateTime.of(2023, 1, 15, 10, 30));
+ }
+
+ @Test
+ public void testConvertTimestampNtzValueWithNanoseconds() {
+ TimestampNtz ts = TimestampNtz.fromMillis(1000L, 500000); // 1 second
+ 500000 nanos
+ Object result =
FlussTypeToPojoTypeConverter.convertTimestampNtzValue(ts);
+ assertThat(result).isInstanceOf(LocalDateTime.class);
+ LocalDateTime ldt = (LocalDateTime) result;
+ assertThat(ldt.getNano()).isEqualTo(500000);
+ }
+
+ // ==================== convertTimestampLtzValue Tests ====================
+
+ @Test
+ public void testConvertTimestampLtzValueToInstant() {
+ TimestampLtz ts = TimestampLtz.fromEpochMillis(1000L);
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTimestampLtzValue(ts,
"field", Instant.class);
+ assertThat(result).isEqualTo(Instant.ofEpochMilli(1000L));
+ }
+
+ @Test
+ public void testConvertTimestampLtzValueToOffsetDateTime() {
+ TimestampLtz ts = TimestampLtz.fromEpochMillis(0L);
+ Object result =
+ FlussTypeToPojoTypeConverter.convertTimestampLtzValue(
+ ts, "field", OffsetDateTime.class);
+ assertThat(result).isEqualTo(OffsetDateTime.of(1970, 1, 1, 0, 0, 0, 0,
ZoneOffset.UTC));
+ }
+
+ private static Map<String, Object> buildConstantMap() {
+ return
FlussTypeToPojoTypeConverter.buildEnumConstantsMap(StatusEnum.class);
+ }
+
+ // ==================== Helper Enum ====================
+
+ /** Test enum for String-to-Enum conversion tests. */
+ public enum StatusEnum {
+ ACTIVE,
+ INACTIVE,
+ PENDING,
+ finished,
+ ;
+
+ @Override
+ public String toString() {
+ return "StatusEnum{}";
+ }
+ }
+}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoArrayToFlussArrayTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoArrayToFlussArrayTest.java
index 2c21e7222..a1787bd61 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoArrayToFlussArrayTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoArrayToFlussArrayTest.java
@@ -70,6 +70,7 @@ public class PojoArrayToFlussArrayTest {
.field(
"mapArray",
DataTypes.ARRAY(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())))
+ .field("enumArray",
DataTypes.ARRAY(DataTypes.STRING()))
.build();
PojoToRowConverter<ArrayPojo> writer =
PojoToRowConverter.of(ArrayPojo.class, table, table);
@@ -242,6 +243,12 @@ public class PojoArrayToFlussArrayTest {
assertThat(resultMap2).containsEntry("test_3", 3);
assertThat(resultMap2).containsEntry("test_4", 4);
+ // Verify enum array
+ InternalArray enumArray = row.getArray(22);
+ assertThat(enumArray.size()).isEqualTo(3);
+ assertThat(enumArray.getString(0).toString()).isEqualTo("PENDING");
+ assertThat(enumArray.getString(1).toString()).isEqualTo("ACTIVE");
+ assertThat(enumArray.getString(2).toString()).isEqualTo("finished");
}
@Test
@@ -314,6 +321,7 @@ public class PojoArrayToFlussArrayTest {
public Instant[] timestampLtzArray;
public Integer[][] nestedIntArray;
public Map<String, Integer>[] mapArray;
+ public StatusEnum[] enumArray;
public ArrayPojo() {}
@@ -368,6 +376,8 @@ public class PojoArrayToFlussArrayTest {
}
}
};
+ pojo.enumArray =
+ new StatusEnum[] {StatusEnum.PENDING, StatusEnum.ACTIVE,
StatusEnum.finished};
return pojo;
}
}
@@ -434,4 +444,12 @@ public class PojoArrayToFlussArrayTest {
public SimpleArrayPojo() {}
}
+
+ /** Test enum for String-to-Enum conversion tests. */
+ public enum StatusEnum {
+ ACTIVE,
+ INACTIVE,
+ PENDING,
+ finished,
+ }
}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java
index 2d261dc50..1e70a7ed9 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java
@@ -68,19 +68,16 @@ class PojoTypeTest {
assertThat(pojoType.getProperty("user_id"))
.isNotNull()
- .hasFieldOrPropertyWithValue("name", "userId")
- .hasFieldOrPropertyWithValue("mappedName", "user_id");
+ .hasFieldOrPropertyWithValue("name", "userId");
assertThat(pojoType.getProperty("first_name"))
.isNotNull()
- .hasFieldOrPropertyWithValue("name", "firstName")
- .hasFieldOrPropertyWithValue("mappedName", "first_name");
+ .hasFieldOrPropertyWithValue("name", "firstName");
// Fields without @ColumnName should map to themselves
assertThat(pojoType.getProperty("email"))
.isNotNull()
- .hasFieldOrPropertyWithValue("name", "email")
- .hasFieldOrPropertyWithValue("mappedName", "email");
+ .hasFieldOrPropertyWithValue("name", "email");
}
@Test
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/converter/RowToPojoConverterTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/converter/RowToPojoConverterTest.java
index 0d849a61e..cf09ab031 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/converter/RowToPojoConverterTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/converter/RowToPojoConverterTest.java
@@ -44,7 +44,26 @@ public class RowToPojoConverterTest {
ConvertersTestFixtures.TestPojo pojo =
ConvertersTestFixtures.TestPojo.sample();
GenericRow row = writer.toRow(pojo);
- assertThat(row.getFieldCount()).isEqualTo(18);
+ assertThat(row.getFieldCount()).isEqualTo(22);
+
+ ConvertersTestFixtures.TestPojo back = scanner.fromRow(row);
+ assertThat(back).isEqualTo(pojo);
+ }
+
+ @Test
+ public void testRoundTripFullSchemaWithLowerCaseEnum() {
+ RowType table = ConvertersTestFixtures.fullSchema();
+ RowType projection = table;
+
+ PojoToRowConverter<ConvertersTestFixtures.TestPojo> writer =
+ PojoToRowConverter.of(ConvertersTestFixtures.TestPojo.class,
table, projection);
+ RowToPojoConverter<ConvertersTestFixtures.TestPojo> scanner =
+ RowToPojoConverter.of(ConvertersTestFixtures.TestPojo.class,
table, projection);
+
+ ConvertersTestFixtures.TestPojo pojo =
ConvertersTestFixtures.TestPojo.sample();
+ pojo.enumField = ConvertersTestFixtures.StatusEnum.error;
+ GenericRow row = writer.toRow(pojo);
+ assertThat(row.getFieldCount()).isEqualTo(22);
ConvertersTestFixtures.TestPojo back = scanner.fromRow(row);
assertThat(back).isEqualTo(pojo);
