This is an automated email from the ASF dual-hosted git repository. jerryshao pushed a commit to branch branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 65c67ee1de6f10b315287ffd85a13f508d16a68a Author: Qi Yu <[email protected]> AuthorDate: Mon Nov 2 22:33:25 2026 +0800 [#14490] fix(spark): report unsupported PostgreSQL write types before execution (#14491) ### What changes were proposed in this pull request? Validate PostgreSQL UUID and External types before creating a Spark JDBC write builder. Report a column-specific unsupported-operation error and document the limitation. This implements explicit early refusal rather than new native-type write support. ### Why are the changes needed? Spark represents PostgreSQL UUID and External column types as strings. JDBC writes bind them as VARCHAR, so PostgreSQL rejects values including NULL with late JDBC batch/type errors. Fixes #14490 ### Does this PR introduce _any_ user-facing change? Affected writes fail before JDBC execution with an actionable message. Ordinary PostgreSQL write types and reads keep their existing behavior. ### How was this patch tested? Spark 3.5 unit tests and targeted Spark 4.0 write-validation tests, including ordinary write-builder construction. All selected tests passed on apache/gravitino main (90da2776a): 251 tests, zero failures. Spotless passed. (cherry picked from commit 89a71fe4811fbaa8e0380098f0382dba13e6cffb) --- docs/spark-connector/spark-connector.md | 8 ++ .../spark/connector/jdbc/SparkJdbcTable.java | 33 +++++++- .../jdbc/TestSparkJdbcTableWriteValidation.java | 98 ++++++++++++++++++++++ 3 files changed, 138 insertions(+), 1 deletion(-) diff --git a/docs/spark-connector/spark-connector.md b/docs/spark-connector/spark-connector.md index 36bd7e384e..979f87273e 100644 --- a/docs/spark-connector/spark-connector.md +++ b/docs/spark-connector/spark-connector.md @@ -113,3 +113,11 @@ Gravitino spark connector support the following datatype mapping between Spark a For Gravitino `UUID` type, Spark connector represents it as `StringType` because Spark has no native UUID type. This behavior is consistent with Spark built-in PostgreSQL JDBC mapping (`uuid` -> `StringType`). ::: + +### PostgreSQL write limitations + +Spark JDBC reads PostgreSQL UUID and External columns as strings. Writes to tables containing +these columns are rejected before JDBC execution, including writes of NULL values and +INSERT OVERWRITE. The error identifies the unsupported column. Spark binds these columns +as VARCHAR, which does not preserve the PostgreSQL target type. Use a native PostgreSQL +writer for these tables. Other PostgreSQL column types retain their existing write behavior. diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java index 3de807c368..940cd43cdf 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java @@ -19,20 +19,26 @@ package org.apache.gravitino.spark.connector.jdbc; +import java.util.Locale; import java.util.Map; +import org.apache.gravitino.rel.Column; import org.apache.gravitino.rel.Table; +import org.apache.gravitino.rel.types.Types; import org.apache.gravitino.spark.connector.PropertiesConverter; import org.apache.gravitino.spark.connector.SparkTransformConverter; import org.apache.gravitino.spark.connector.SparkTypeConverter; import org.apache.gravitino.spark.connector.utils.GravitinoTableInfoHelper; import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.write.LogicalWriteInfo; +import org.apache.spark.sql.connector.write.WriteBuilder; import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTable; import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog; import org.apache.spark.sql.types.StructType; public class SparkJdbcTable extends JDBCTable { - private GravitinoTableInfoHelper gravitinoTableInfoHelper; + private final GravitinoTableInfoHelper gravitinoTableInfoHelper; + private final Table gravitinoTable; public SparkJdbcTable( Identifier identifier, @@ -43,6 +49,7 @@ public class SparkJdbcTable extends JDBCTable { SparkTransformConverter sparkTransformConverter, SparkTypeConverter sparkTypeConverter) { super(identifier, jdbcTable.schema(), jdbcTable.jdbcOptions()); + this.gravitinoTable = gravitinoTable; this.gravitinoTableInfoHelper = new GravitinoTableInfoHelper( false, @@ -53,6 +60,30 @@ public class SparkJdbcTable extends JDBCTable { sparkTypeConverter); } + /** + * Validates PostgreSQL write types before Spark creates a JDBC writer. + * + * @param info the write information supplied by Spark + * @return the JDBC write builder for supported column types + */ + @Override + public WriteBuilder newWriteBuilder(LogicalWriteInfo info) { + if (jdbcOptions().url().toLowerCase(Locale.ROOT).startsWith("jdbc:postgresql:")) { + for (Column column : gravitinoTable.columns()) { + if (column.dataType() instanceof Types.UUIDType + || column.dataType() instanceof Types.ExternalType) { + throw new UnsupportedOperationException( + "Spark JDBC writes to PostgreSQL column '" + + column.name() + + "' of type " + + column.dataType().simpleString() + + " are not supported because Spark binds this type as VARCHAR; use a native PostgreSQL writer"); + } + } + } + return super.newWriteBuilder(info); + } + @Override public String name() { return gravitinoTableInfoHelper.name(); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestSparkJdbcTableWriteValidation.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestSparkJdbcTableWriteValidation.java new file mode 100644 index 0000000000..83eb356876 --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestSparkJdbcTableWriteValidation.java @@ -0,0 +1,98 @@ +/* + * 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.gravitino.spark.connector.jdbc; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Collections; +import org.apache.gravitino.rel.Column; +import org.apache.gravitino.rel.Table; +import org.apache.gravitino.rel.types.Type; +import org.apache.gravitino.rel.types.Types; +import org.apache.gravitino.spark.connector.SparkTransformConverter; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.write.LogicalWriteInfo; +import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions; +import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTable; +import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import scala.collection.immutable.HashMap; + +/** PostgreSQL write validation before Spark creates or executes a JDBC write. */ +public class TestSparkJdbcTableWriteValidation { + + @Test + void testUnsupportedPostgreSqlColumnsFailBeforeWrite() { + for (Type type : + new Type[] { + Types.UUIDType.get(), + Types.ExternalType.of("jsonb"), + Types.ExternalType.of("numeric"), + Types.ExternalType.of("inet"), + Types.ExternalType.of("interval") + }) { + SparkJdbcTable table = table(type, "jdbc:postgresql://localhost/test"); + // Reading remains supported; nullable columns are also rejected on write before any row is + // bound. + Assertions.assertEquals(DataTypes.StringType, table.schema().fields()[0].dataType()); + UnsupportedOperationException error = + Assertions.assertThrows( + UnsupportedOperationException.class, () -> table.newWriteBuilder(null)); + Assertions.assertTrue(error.getMessage().contains("special_value")); + Assertions.assertTrue(error.getMessage().contains(type.simpleString())); + } + } + + @Test + void testOrdinaryPostgreSqlTypesStillCreateWriteBuilder() { + SparkJdbcTable table = table(Types.IntegerType.get(), "jdbc:postgresql://localhost/test"); + LogicalWriteInfo info = mock(LogicalWriteInfo.class); + when(info.options()).thenReturn(new CaseInsensitiveStringMap(Collections.emptyMap())); + Assertions.assertNotNull(table.newWriteBuilder(info)); + } + + private SparkJdbcTable table(Type type, String url) { + Table metadata = mock(Table.class); + when(metadata.columns()) + .thenReturn(new Column[] {Column.of("special_value", type, null, true, false, null)}); + JDBCOptions options = + new JDBCOptions( + new HashMap<String, String>() + .updated("url", url) + .updated("dbtable", "t") + .updated("driver", "org.postgresql.Driver")); + JDBCTable physical = mock(JDBCTable.class); + when(physical.schema()).thenReturn(new StructType()); + when(physical.jdbcOptions()).thenReturn(options); + return new SparkJdbcTable( + Identifier.of(new String[] {"public"}, "t"), + metadata, + physical, + mock(JDBCTableCatalog.class), + JdbcPropertiesConverter.getInstance(), + new SparkTransformConverter(false), + new SparkJdbcTypeConverter()); + } +}
