This is an automated email from the ASF dual-hosted git repository.
diqiu50 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 14f4995f9c [MINOR] refactor(flink): Allow catalogs to customize Flink
type conversion (#12961)
14f4995f9c is described below
commit 14f4995f9ccb1fd8c513db4a27eed694823d2558
Author: Qi Yu <[email protected]>
AuthorDate: Mon Sep 7 20:02:22 2026 +0800
[MINOR] refactor(flink): Allow catalogs to customize Flink type conversion
(#12961)
### What changes were proposed in this pull request?
Add a protected `toFlinkType(Type)` hook to `BaseCatalog` and use it
when building table and view schemas. The default implementation
delegates to `TypeUtils.toFlinkType`.
### Why are the changes needed?
Catalogs need to map native external types to Flink types without
duplicating schema construction. This complements the existing
`toGravitinoType` hook.
### Does this PR introduce _any_ user-facing change?
No. Existing catalogs retain their default type mappings.
### How was this patch tested?
Added unit coverage for default mappings and a catalog override,
including column names, nullability, comments, and preservation of
source type metadata.
Passed Flink 1.18 unit tests and Spotless checks:
```sh
./gradlew :flink-connector:flink-common:spotlessApply
:flink-connector:flink-1.18:test -PskipITs --offline
```
---
.../flink/connector/catalog/BaseCatalog.java | 15 ++++++--
.../flink/connector/catalog/TestBaseCatalog.java | 42 ++++++++++++++++++++++
2 files changed, 54 insertions(+), 3 deletions(-)
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
index 8195aacdea..f2e7fdab9c 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
@@ -141,6 +141,16 @@ public abstract class BaseCatalog extends AbstractCatalog {
return TypeUtils.toGravitinoType(logicalType);
}
+ /**
+ * Converts a Gravitino type to a Flink type, allowing catalog-specific
native type mappings.
+ *
+ * @param type the Gravitino type
+ * @return the corresponding Flink data type
+ */
+ protected DataType toFlinkType(Type type) {
+ return TypeUtils.toFlinkType(type);
+ }
+
@Override
public void open() throws CatalogException {
realCatalog().open();
@@ -1153,12 +1163,11 @@ public abstract class BaseCatalog extends
AbstractCatalog {
* @param columns the Gravitino column definitions
* @return a Flink schema builder populated with the given columns
*/
- protected static org.apache.flink.table.api.Schema.Builder
buildSchemaFromColumns(
- Column[] columns) {
+ protected org.apache.flink.table.api.Schema.Builder
buildSchemaFromColumns(Column[] columns) {
org.apache.flink.table.api.Schema.Builder builder =
org.apache.flink.table.api.Schema.newBuilder();
for (Column column : columns) {
- DataType flinkType = TypeUtils.toFlinkType(column.dataType());
+ DataType flinkType = toFlinkType(column.dataType());
builder
.column(column.name(), column.nullable() ? flinkType.nullable() :
flinkType.notNull())
.withComment(column.comment());
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
index 7c5db161c7..5035c93273 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
@@ -39,6 +39,7 @@ import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.types.DataType;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
@@ -55,6 +56,7 @@ import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.ViewCatalog;
import org.apache.gravitino.rel.ViewChange;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
+import org.apache.gravitino.rel.types.Type;
import org.apache.gravitino.rel.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -62,6 +64,46 @@ import org.mockito.Mockito;
public class TestBaseCatalog {
+ @Test
+ void testDefaultFlinkTypeConversion() {
+ BaseCatalog catalog = new TestableBaseCatalog(null, null);
+ Assertions.assertEquals(DataTypes.INT(),
catalog.toFlinkType(Types.IntegerType.get()));
+ Assertions.assertEquals(DataTypes.STRING(),
catalog.toFlinkType(Types.StringType.get()));
+ Assertions.assertEquals(
+ DataTypes.DECIMAL(10, 2), catalog.toFlinkType(Types.DecimalType.of(10,
2)));
+ }
+
+ @Test
+ void testSchemaUsesCatalogTypeConversion() {
+ Type nativeType = Types.ExternalType.of("native_text");
+ BaseCatalog catalog =
+ new TestableBaseCatalog(null, null) {
+ /** {@inheritDoc} */
+ @Override
+ protected DataType toFlinkType(Type type) {
+ return type.equals(nativeType) ? DataTypes.STRING() :
super.toFlinkType(type);
+ }
+ };
+ org.apache.gravitino.rel.Column[] columns = {
+ org.apache.gravitino.rel.Column.of("text", nativeType, "source comment"),
+ org.apache.gravitino.rel.Column.of("required_text", nativeType, null,
false, false, null),
+ org.apache.gravitino.rel.Column.of("id", Types.IntegerType.get(), null)
+ };
+ Schema schema = catalog.buildSchemaFromColumns(columns).build();
+ Schema.UnresolvedPhysicalColumn text =
+ (Schema.UnresolvedPhysicalColumn) schema.getColumns().get(0);
+ Schema.UnresolvedPhysicalColumn requiredText =
+ (Schema.UnresolvedPhysicalColumn) schema.getColumns().get(1);
+ Schema.UnresolvedPhysicalColumn id =
+ (Schema.UnresolvedPhysicalColumn) schema.getColumns().get(2);
+ Assertions.assertEquals("text", text.getName());
+ Assertions.assertEquals(DataTypes.STRING(), text.getDataType());
+ Assertions.assertEquals("source comment", text.getComment().orElseThrow());
+ Assertions.assertEquals(DataTypes.STRING().notNull(),
requiredText.getDataType());
+ Assertions.assertEquals(DataTypes.INT(), id.getDataType());
+ Assertions.assertEquals(nativeType, columns[0].dataType());
+ }
+
@Test
public void testHiveSchemaChanges() {
Map<String, String> currentProperties = ImmutableMap.of("key", "value",
"key2", "value2");
