This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0b352a2c67 [#10920] fix(flink): Fix MULTISET type mismatch when
writing to Paimon table via Gravitino Flink Connector (#10927)
0b352a2c67 is described below
commit 0b352a2c67062e5d537d5135f82ac1ea539731d4
Author: geyanggang <[email protected]>
AuthorDate: Mon May 11 10:42:45 2026 +0800
[#10920] fix(flink): Fix MULTISET type mismatch when writing to Paimon
table via Gravitino Flink Connector (#10927)
### What changes were proposed in this pull request?
Use `ExternalType` to preserve Paimon's MULTISET type information
through Gravitino's type system, instead of converting it to `MapType`
which causes a schema mismatch error.
Changes:
- **Paimon catalog backend (`TypeUtils.java`)**: `visit(MultisetType)`
now returns `ExternalType` with the original SQL string instead of
`MapType`. Added `EXTERNAL` case in `GravitinoToPaimonTypeVisitor` to
restore `MultisetType` using `DataTypeJsonParser`.
- **Flink Connector (`TypeUtils.java`)**: `toGravitinoType` MULTISET
case now returns `ExternalType`. Added `EXTERNAL` case in `toFlinkType`
using Flink's `LogicalTypeParser` to restore the original Flink type.
- **Unit test**: Added `testMultisetTypeConversion` to verify round-trip
conversion.
### Why are the changes needed?
When a Paimon table contains a `MULTISET<STRING>` column and is accessed
through the Gravitino Flink Connector, writing to the table fails with:
IllegalArgumentException: Flink schema and store schema are not the same
store schema: field_multiset MULTISET<STRING> Flink schema:
field_multiset MAP<STRING, INT NOT NULL>
Root cause: Gravitino's type system has no native MULTISET type. The
Paimon catalog backend converted `MULTISET<T>` to `MapType<T, INT>`,
which was then converted to Flink's `MAP<T, INT NOT NULL>`. Paimon's
`FlinkTableFactory` performs a strict schema comparison and rejects the
mismatch.
Fix: #10920
### Does this PR introduce _any_ user-facing change?
No. This is a bug fix that restores correct behavior for Paimon tables
with MULTISET columns.
### How was this patch tested?
- Added unit test `testMultisetTypeConversion` in `TestTypeUtils`
covering both `toGravitinoType(MULTISET)` and
`toFlinkType(ExternalType)` round-trip.
- All existing unit tests pass for both `flink-connector` and
`catalog-lakehouse-paimon` modules.
---
.../catalog/lakehouse/paimon/utils/TypeUtils.java | 25 ++++++---
.../gravitino/flink/connector/utils/TypeUtils.java | 15 +++++
.../test/paimon/FlinkPaimonCatalogIT.java | 65 ++++++++++++++++++++++
.../flink/connector/utils/TestTypeUtils.java | 32 +++++++++++
4 files changed, 129 insertions(+), 8 deletions(-)
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/TypeUtils.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/TypeUtils.java
index 0ee547f21e..b3543c595f 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/TypeUtils.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/TypeUtils.java
@@ -28,6 +28,7 @@ import org.apache.paimon.types.BooleanType;
import org.apache.paimon.types.CharType;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeDefaultVisitor;
+import org.apache.paimon.types.DataTypeJsonParser;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DateType;
import org.apache.paimon.types.DecimalType;
@@ -192,14 +193,10 @@ public class TypeUtils {
@Override
public Type visit(MultisetType multisetType) {
- // Unlike a Java Set, MultisetType allows for multiple instances for
each of its
- // elements with a common subtype. And a conversion is possible through
a map
- // that assigns each value to an integer to represent the multiplicity
of the values.
- // For example, a `MULTISET<INT>` is converted to a `MAP<Integer,
Integer>`, the key of the
- // map represents the elements of the Multiset and the value represents
the multiplicity of
- // the elements in the Multiset.
- return Types.MapType.of(
- multisetType.getElementType().accept(this), Types.IntegerType.get(),
false);
+ // Gravitino's type system does not have a native MULTISET type. We use
ExternalType to
+ // preserve the original Paimon MULTISET type information, so it can be
correctly restored
+ // when converting back to Paimon or Flink types.
+ return Types.ExternalType.of(multisetType.asSQLString());
}
@Override
@@ -280,6 +277,18 @@ public class TypeUtils {
builder.field(field.name(), dataTypeWithNullable,
field.comment());
});
return builder.build();
+ case EXTERNAL:
+ Types.ExternalType externalType = (Types.ExternalType) type;
+ String catalogString = externalType.catalogString();
+ if (catalogString.startsWith("MULTISET<") &&
catalogString.endsWith(">")) {
+ // Restore MULTISET type: extract element type SQL string and
parse it
+ String elementTypeSql =
+ catalogString.substring("MULTISET<".length(),
catalogString.length() - 1);
+ DataType elementDataType =
DataTypeJsonParser.parseAtomicTypeSQLString(elementTypeSql);
+ return new MultisetType(elementDataType);
+ }
+ throw new UnsupportedOperationException(
+ String.format("Paimon does not support Gravitino external type:
%s", catalogString));
default:
throw new UnsupportedOperationException(
String.format(
diff --git
a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/utils/TypeUtils.java
b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/utils/TypeUtils.java
index e988d082b1..2b032dac7e 100644
---
a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/utils/TypeUtils.java
+++
b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/utils/TypeUtils.java
@@ -30,7 +30,10 @@ import org.apache.flink.table.types.logical.CharType;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeParser;
+import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.gravitino.rel.types.Type;
import org.apache.gravitino.rel.types.Types;
@@ -130,6 +133,10 @@ public class TypeUtils {
case NULL:
return Types.NullType.get();
case MULTISET:
+ MultisetType multisetType = (MultisetType) logicalType;
+ // Gravitino's type system does not have a native MULTISET type. We
use ExternalType to
+ // preserve the original type information so it can be correctly
restored.
+ return Types.ExternalType.of(multisetType.asSerializableString());
case STRUCTURED_TYPE:
case UNRESOLVED:
case DISTINCT_TYPE:
@@ -236,6 +243,14 @@ public class TypeUtils {
return DataTypes.INTERVAL(DataTypes.YEAR());
case INTERVAL_DAY:
return DataTypes.INTERVAL(DataTypes.DAY());
+ case EXTERNAL:
+ Types.ExternalType externalType = (Types.ExternalType) gravitinoType;
+ String catalogString = externalType.catalogString();
+ // Parse the external catalog type string back to Flink LogicalType.
+ // This is used to restore types like MULTISET that Gravitino doesn't
natively support.
+ LogicalType parsedType =
+ LogicalTypeParser.parse(catalogString,
TypeUtils.class.getClassLoader());
+ return TypeConversions.fromLogicalToDataType(parsedType);
default:
throw new UnsupportedOperationException("Not support " +
gravitinoType.toString());
}
diff --git
a/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonCatalogIT.java
b/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonCatalogIT.java
index adb8bc836e..2002909914 100644
---
a/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonCatalogIT.java
+++
b/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonCatalogIT.java
@@ -18,18 +18,29 @@
*/
package org.apache.gravitino.flink.connector.integration.test.paimon;
+import static
org.apache.gravitino.flink.connector.integration.test.utils.TestUtils.toFlinkPhysicalColumn;
+import static org.junit.jupiter.api.Assertions.fail;
+
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.types.Row;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.flink.connector.integration.test.FlinkCommonIT;
+import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
import org.apache.gravitino.rel.expressions.distributions.Strategy;
+import org.apache.gravitino.rel.types.Types;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -224,4 +235,58 @@ public abstract class FlinkPaimonCatalogIT extends
FlinkCommonIT {
Assertions.assertEquals(
numCatalogs, tableEnv.listCatalogs().length, "The created catalog
should be dropped.");
}
+
+ @Test
+ public void testMultisetType() {
+ String databaseName = "test_multiset_type_db";
+ String tableName = "test_multiset_table";
+
+ doWithSchema(
+ currentCatalog(),
+ databaseName,
+ catalog -> {
+ // Create a table with MULTISET column using ExternalType through
Gravitino API.
+ // This simulates the scenario where a Paimon table with MULTISET
type is registered
+ // in Gravitino (e.g., created by native Paimon and then managed by
Gravitino).
+ Column[] columns =
+ new Column[] {
+ Column.of("id", Types.LongType.get(), "id"),
+ Column.of(
+ "field_multiset",
Types.ExternalType.of("MULTISET<STRING>"), "multiset field")
+ };
+ catalog
+ .asTableCatalog()
+ .createTable(
+ NameIdentifier.of(databaseName, tableName),
+ columns,
+ "test multiset table",
+ ImmutableMap.of());
+
+ // Verify that getTable through Gravitino Flink Connector returns
MULTISET type
+ Optional<org.apache.flink.table.catalog.Catalog> flinkCatalog =
+ tableEnv.getCatalog(catalog.name());
+ Assertions.assertTrue(flinkCatalog.isPresent());
+ try {
+ CatalogBaseTable table =
+ flinkCatalog.get().getTable(new ObjectPath(databaseName,
tableName));
+ Assertions.assertNotNull(table);
+
+ org.apache.flink.table.catalog.Column[] expected =
+ new org.apache.flink.table.catalog.Column[] {
+ org.apache.flink.table.catalog.Column.physical("id",
DataTypes.BIGINT())
+ .withComment("id"),
+ org.apache.flink.table.catalog.Column.physical(
+ "field_multiset",
DataTypes.MULTISET(DataTypes.STRING()))
+ .withComment("multiset field")
+ };
+ org.apache.flink.table.catalog.Column[] actual =
+
toFlinkPhysicalColumn(table.getUnresolvedSchema().getColumns());
+ Assertions.assertArrayEquals(expected, actual);
+ } catch (TableNotExistException e) {
+ fail(e);
+ }
+ },
+ true,
+ supportDropCascade());
+ }
}
diff --git
a/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/utils/TestTypeUtils.java
b/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/utils/TestTypeUtils.java
index a7e94533b2..9d0c71ec84 100644
---
a/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/utils/TestTypeUtils.java
+++
b/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/utils/TestTypeUtils.java
@@ -33,6 +33,7 @@ import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.NullType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.SmallIntType;
@@ -253,4 +254,35 @@ public class TestTypeUtils {
UnsupportedOperationException.class,
() -> TypeUtils.toFlinkType(Types.TimestampType.withTimeZone(10)));
}
+
+ @Test
+ public void testMultisetTypeConversion() {
+ // MULTISET<STRING> (VARCHAR(MAX)) should be preserved as ExternalType
+ Assertions.assertEquals(
+ Types.ExternalType.of("MULTISET<VARCHAR(2147483647)>"),
+ TypeUtils.toGravitinoType(new MultisetType(new
VarCharType(Integer.MAX_VALUE))));
+
+ // MULTISET<INT> should be preserved as ExternalType
+ Assertions.assertEquals(
+ Types.ExternalType.of("MULTISET<INT>"),
+ TypeUtils.toGravitinoType(new MultisetType(new IntType())));
+
+ // MULTISET<BIGINT> should be preserved as ExternalType
+ Assertions.assertEquals(
+ Types.ExternalType.of("MULTISET<BIGINT>"),
+ TypeUtils.toGravitinoType(new MultisetType(new BigIntType())));
+
+ // ExternalType with MULTISET should be converted back to Flink MULTISET
+ Assertions.assertEquals(
+ DataTypes.MULTISET(DataTypes.STRING()),
+
TypeUtils.toFlinkType(Types.ExternalType.of("MULTISET<VARCHAR(2147483647)>")));
+
+ Assertions.assertEquals(
+ DataTypes.MULTISET(DataTypes.INT()),
+ TypeUtils.toFlinkType(Types.ExternalType.of("MULTISET<INT>")));
+
+ Assertions.assertEquals(
+ DataTypes.MULTISET(DataTypes.BIGINT()),
+ TypeUtils.toFlinkType(Types.ExternalType.of("MULTISET<BIGINT>")));
+ }
}
