This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 4856ee8d27 [Cherry-pick to branch-1.3] [#14705] fix(spark): accept
table comments, no-op alters and add-column defaults (#14708) (#14731)
4856ee8d27 is described below
commit 4856ee8d2795ecdd95260893ab391c832de0e8f7
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Nov 6 21:54:03 2026 +0800
[Cherry-pick to branch-1.3] [#14705] fix(spark): accept table comments,
no-op alters and add-column defaults (#14708) (#14731)
**Cherry-pick Information:**
- Original commit: 2779f6b0e6449644dea0454291ca87ff45baf399
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
---------
Co-authored-by: Qi Yu <[email protected]>
Co-authored-by: yuqi <[email protected]>
---
.../spark/connector/SparkTableChangeConverter.java | 22 ++-
.../spark/connector/catalog/BaseCatalog.java | 15 +-
.../catalog/TestBaseCatalogTableChanges.java | 180 +++++++++++++++++++++
.../catalog/TestTransformTableChange.java | 3 +
.../connector/SparkTableChangeConverter34.java | 13 ++
.../catalog/TestBaseCatalogDefaults34.java | 68 ++++++++
6 files changed, 296 insertions(+), 5 deletions(-)
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
index 06fef33be3..66e5f4f3e5 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.spark.connector;
import com.google.common.base.Preconditions;
import java.util.function.Supplier;
import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.expressions.Expression;
+import org.apache.gravitino.rel.types.Type;
import org.apache.spark.sql.connector.catalog.TableChange;
public class SparkTableChangeConverter {
@@ -47,12 +49,15 @@ public class SparkTableChangeConverter {
return
org.apache.gravitino.rel.TableChange.removeProperty(removeProperty.property());
} else if (change instanceof TableChange.AddColumn) {
TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
+ Type dataType = sparkTypeConverter.toGravitinoType(addColumn.dataType());
return org.apache.gravitino.rel.TableChange.addColumn(
addColumn.fieldNames(),
- sparkTypeConverter.toGravitinoType(addColumn.dataType()),
+ dataType,
addColumn.comment(),
transformColumnPosition(addColumn.position()),
- addColumn.isNullable());
+ addColumn.isNullable(),
+ false,
+ convertAddColumnDefaultValue(addColumn, dataType));
} else if (change instanceof TableChange.DeleteColumn) {
TableChange.DeleteColumn deleteColumn = (TableChange.DeleteColumn)
change;
return org.apache.gravitino.rel.TableChange.deleteColumn(
@@ -110,6 +115,19 @@ public class SparkTableChangeConverter {
return toGravitinoTableChange(change);
}
+ /**
+ * Converts an added column's default when supported by the Spark version.
Spark 3.3 exposes no
+ * add-column default API, so the base converter leaves the default unset.
+ *
+ * @param addColumn the Spark add-column change
+ * @param dataType the new column's Gravitino type
+ * @return the converted default, or {@link Column#DEFAULT_VALUE_NOT_SET}
when absent
+ */
+ protected Expression convertAddColumnDefaultValue(
+ TableChange.AddColumn addColumn, Type dataType) {
+ return Column.DEFAULT_VALUE_NOT_SET;
+ }
+
private org.apache.gravitino.rel.TableChange.ColumnPosition
transformColumnPosition(
TableChange.ColumnPosition columnPosition) {
if (null == columnPosition) {
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
index 90e2ff8ae1..b10bcfdc53 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
@@ -233,10 +233,14 @@ public abstract class BaseCatalog implements
TableCatalog, SupportsNamespaces, F
.map(structField -> createGravitinoColumn(structField))
.toArray(org.apache.gravitino.rel.Column[]::new);
+ // Spark stores the table comment in the table properties. Take it out
before converting the
+ // properties, because it is passed to Gravitino as the comment, not as a
table property, and
+ // a catalog that accepts no table properties would otherwise refuse it.
+ Map<String, String> sparkProperties =
+ properties == null ? new HashMap<>() : new HashMap<>(properties);
+ String comment = sparkProperties.remove(ConnectorConstants.COMMENT);
Map<String, String> gravitinoProperties =
- propertiesConverter.toGravitinoTableProperties(properties);
- // Spark store comment in properties, we should retrieve it and pass to
Gravitino explicitly.
- String comment = gravitinoProperties.remove(ConnectorConstants.COMMENT);
+ propertiesConverter.toGravitinoTableProperties(sparkProperties);
DistributionAndSortOrdersInfo distributionAndSortOrdersInfo =
sparkTransformConverter.toGravitinoDistributionAndSortOrders(transforms);
@@ -277,6 +281,11 @@ public abstract class BaseCatalog implements TableCatalog,
SupportsNamespaces, F
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws
NoSuchTableException {
+ if (changes == null || changes.length == 0) {
+ // Spark sends no change when every requested change is a no-op, for
example DROP COLUMN IF
+ // EXISTS on an absent column. There is nothing to apply.
+ return loadTable(ident);
+ }
try {
Supplier<Column[]> columns =
Suppliers.memoize(
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogTableChanges.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogTableChanges.java
new file mode 100644
index 0000000000..fbc9d1cfcb
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogTableChanges.java
@@ -0,0 +1,180 @@
+/*
+ * 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.catalog;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.gravitino.spark.connector.PropertiesConverter;
+import org.apache.gravitino.spark.connector.SparkTableChangeConverter;
+import org.apache.gravitino.spark.connector.SparkTransformConverter;
+import org.apache.gravitino.spark.connector.SparkTypeConverter;
+import org.apache.gravitino.spark.connector.jdbc.JdbcPropertiesConverter;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.expressions.Transform;
+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.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.mockito.ArgumentCaptor;
+
+/** Table create and alter requests that Spark sends to a Gravitino catalog. */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class TestBaseCatalogTableChanges {
+
+ private static final Identifier IDENT = Identifier.of(new String[]
{"schema"}, "t");
+ private static final NameIdentifier GRAVITINO_IDENT =
NameIdentifier.of("schema", "t");
+
+ private TableCatalog gravitinoTableCatalog;
+ private org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog;
+ private TestCatalog catalog;
+
+ @BeforeAll
+ void initCatalogManager() {
+ GravitinoCatalogManager.create("user", () -> mock(GravitinoClient.class));
+ }
+
+ @AfterAll
+ void cleanupCatalogManager() {
+ GravitinoCatalogManager.get().close();
+ }
+
+ @BeforeEach
+ void setUp() throws Exception {
+ Catalog gravitinoCatalog = mock(Catalog.class);
+ gravitinoTableCatalog = mock(TableCatalog.class);
+ sparkCatalog =
mock(org.apache.spark.sql.connector.catalog.TableCatalog.class);
+ when(gravitinoCatalog.asTableCatalog()).thenReturn(gravitinoTableCatalog);
+ catalog = new TestCatalog(gravitinoCatalog, sparkCatalog);
+ SparkTypeConverter typeConverter = new SparkTypeConverter();
+ setField("sparkTypeConverter", typeConverter);
+ setField("sparkTableChangeConverter", new
SparkTableChangeConverter(typeConverter));
+ }
+
+ @Test
+ void testCreateTableCommentIsNotTreatedAsTableProperty() throws Exception {
+ // PostgreSQL rejects table options such as comment; pass the table
comment separately.
+ PropertiesConverter jdbcConverter =
JdbcPropertiesConverter.getPGInstance();
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () ->
jdbcConverter.toGravitinoTableProperties(ImmutableMap.of("comment", "c")));
+
+ StructType schema = new StructType().add("c", DataTypes.IntegerType, true,
"column comment");
+ catalog.createGravitinoTable(
+ IDENT, schema, new Transform[0], ImmutableMap.of("comment", "tbl"));
+
+ ArgumentCaptor<Map<String, String>> properties =
ArgumentCaptor.forClass(Map.class);
+ verify(gravitinoTableCatalog)
+ .createTable(
+ eq(GRAVITINO_IDENT),
+ any(Column[].class),
+ eq("tbl"),
+ properties.capture(),
+ any(),
+ any(),
+ any());
+ Assertions.assertTrue(properties.getValue().isEmpty());
+ }
+
+ @Test
+ void testCreateTableDoesNotChangeCallerProperties() throws Exception {
+ Map<String, String> sparkProperties = Collections.singletonMap("comment",
"tbl");
+ StructType schema = new StructType().add("c", DataTypes.IntegerType);
+ catalog.createGravitinoTable(IDENT, schema, new Transform[0],
sparkProperties);
+ // The caller's map is read, never changed.
+ Assertions.assertEquals("tbl", sparkProperties.get("comment"));
+ }
+
+ @Test
+ void testAlterWithoutChangesDoesNotWrite() throws Exception {
+ org.apache.gravitino.rel.Table gravitinoTable =
mock(org.apache.gravitino.rel.Table.class);
+ Table sparkTable = mock(Table.class);
+
when(gravitinoTableCatalog.loadTable(GRAVITINO_IDENT)).thenReturn(gravitinoTable);
+ when(sparkCatalog.loadTable(IDENT)).thenReturn(sparkTable);
+
+ Assertions.assertSame(sparkTable, catalog.alterTable(IDENT));
+
+ verify(gravitinoTableCatalog, never()).alterTable(any(), any());
+ }
+
+ private void setField(String name, Object value) throws Exception {
+ Field field = BaseCatalog.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(catalog, value);
+ }
+
+ private static class TestCatalog extends BaseCatalog {
+
+ private TestCatalog(
+ Catalog gravitinoCatalog,
+ org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog) {
+ this.gravitinoCatalogClient = gravitinoCatalog;
+ this.sparkCatalog = sparkCatalog;
+ this.propertiesConverter = JdbcPropertiesConverter.getPGInstance();
+ this.sparkTransformConverter = new SparkTransformConverter(false);
+ }
+
+ @Override
+ protected org.apache.spark.sql.connector.catalog.TableCatalog
createAndInitSparkCatalog(
+ String name, CaseInsensitiveStringMap options, Map<String, String>
properties) {
+ return sparkCatalog;
+ }
+
+ @Override
+ protected Table createSparkTable(
+ Identifier identifier,
+ org.apache.gravitino.rel.Table gravitinoTable,
+ Table sparkTable,
+ org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog,
+ PropertiesConverter propertiesConverter,
+ SparkTransformConverter sparkTransformConverter,
+ SparkTypeConverter sparkTypeConverter) {
+ return sparkTable;
+ }
+
+ @Override
+ protected PropertiesConverter getPropertiesConverter() {
+ return JdbcPropertiesConverter.getPGInstance();
+ }
+
+ @Override
+ protected SparkTransformConverter getSparkTransformConverter() {
+ return new SparkTransformConverter(false);
+ }
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
index 1dc0e93970..c26d56e2c8 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.spark.connector.catalog;
+import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.TableChange.UpdateComment;
import org.apache.gravitino.spark.connector.ConnectorConstants;
import org.apache.gravitino.spark.connector.SparkTableChangeConverter;
@@ -142,6 +143,8 @@ public class TestTransformTableChange {
Assertions.assertArrayEquals(
sparkAddColumnFirst.fieldNames(), gravitinoAddColumnFirst.fieldName());
+ Assertions.assertEquals(
+ Column.DEFAULT_VALUE_NOT_SET,
gravitinoAddColumnFirst.getDefaultValue());
Assertions.assertTrue(
"string".equalsIgnoreCase(gravitinoAddColumnFirst.getDataType().simpleString()));
Assertions.assertTrue(
diff --git
a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
index fa95206864..489b59b788 100644
---
a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
+++
b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
@@ -74,6 +74,19 @@ public class SparkTableChangeConverter34 extends
SparkTableChangeConverter {
return super.toGravitinoTableChange(change);
}
+ /** {@inheritDoc} */
+ @Override
+ protected Expression convertAddColumnDefaultValue(
+ TableChange.AddColumn addColumn, Type dataType) {
+ // The new column is not in the table yet, so its own type is the target
of its default.
+ return addColumn.defaultValue() == null
+ ? Column.DEFAULT_VALUE_NOT_SET
+ : convertDefaultValue(
+ addColumn.defaultValue().getSql(),
+ addColumn.fieldNames(),
+ () -> new Column[] {Column.of(addColumn.fieldNames()[0],
dataType)});
+ }
+
private Expression convertDefaultValue(
String sql, String[] fieldNames, Supplier<Column[]> columns) {
// Spark uses an empty SQL string for DROP DEFAULT; a string literal
containing no characters
diff --git
a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogDefaults34.java
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogDefaults34.java
index a1c3b0d956..a38adee813 100644
---
a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogDefaults34.java
+++
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogDefaults34.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.spark.connector.catalog;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -40,12 +41,15 @@ import
org.apache.gravitino.spark.connector.SparkTableChangeConverter34;
import org.apache.gravitino.spark.connector.SparkTransformConverter;
import org.apache.gravitino.spark.connector.SparkTypeConverter;
import org.apache.gravitino.spark.connector.SparkTypeConverter34;
+import org.apache.spark.sql.connector.catalog.ColumnDefaultValue;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.connector.expressions.LiteralValue;
import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.apache.spark.unsafe.types.UTF8String;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -244,6 +248,70 @@ public class TestBaseCatalogDefaults34 {
verify(table).columns();
}
+ @Test
+ void testAddColumnDefaultIsWritten() throws Exception {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of("schema",
"table_a");
+ when(sparkCatalog.loadTable(identifier)).thenReturn(mock(Table.class));
+
+ catalog.alterTable(
+ identifier,
+ TableChange.addColumn(
+ new String[] {"c_int"},
+ DataTypes.IntegerType,
+ true,
+ null,
+ null,
+ new ColumnDefaultValue("5", LiteralValue.apply(5,
DataTypes.IntegerType))),
+ TableChange.addColumn(
+ new String[] {"c_str"},
+ DataTypes.StringType,
+ true,
+ "with comment",
+ null,
+ new ColumnDefaultValue(
+ "'abc'", LiteralValue.apply(UTF8String.fromString("abc"),
DataTypes.StringType))),
+ TableChange.addColumn(new String[] {"c_none"}, DataTypes.IntegerType));
+
+ ArgumentCaptor<org.apache.gravitino.rel.TableChange> changes =
+ ArgumentCaptor.forClass(org.apache.gravitino.rel.TableChange.class);
+ verify(gravitinoTableCatalog).alterTable(eq(gravitinoIdentifier),
changes.capture());
+ org.apache.gravitino.rel.TableChange.AddColumn intColumn =
+ (org.apache.gravitino.rel.TableChange.AddColumn)
changes.getAllValues().get(0);
+ org.apache.gravitino.rel.TableChange.AddColumn stringColumn =
+ (org.apache.gravitino.rel.TableChange.AddColumn)
changes.getAllValues().get(1);
+ org.apache.gravitino.rel.TableChange.AddColumn plainColumn =
+ (org.apache.gravitino.rel.TableChange.AddColumn)
changes.getAllValues().get(2);
+ Assertions.assertEquals(Literals.integerLiteral(5),
intColumn.getDefaultValue());
+ Assertions.assertEquals(Literals.stringLiteral("abc"),
stringColumn.getDefaultValue());
+ Assertions.assertEquals("with comment", stringColumn.getComment());
+ Assertions.assertEquals(Types.StringType.get(),
stringColumn.getDataType());
+ Assertions.assertEquals(Column.DEFAULT_VALUE_NOT_SET,
plainColumn.getDefaultValue());
+ // A new column's default does not need the existing table schema.
+ verify(gravitinoTableCatalog, never()).loadTable(gravitinoIdentifier);
+ }
+
+ @Test
+ void testAddColumnUnsupportedDefaultIsRejectedBeforeWriting() {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ catalog.alterTable(
+ identifier,
+ TableChange.addColumn(
+ new String[] {"c"},
+ DataTypes.IntegerType,
+ true,
+ null,
+ null,
+ new ColumnDefaultValue(
+ "abs(-1)", LiteralValue.apply(1,
DataTypes.IntegerType)))));
+ Assertions.assertTrue(error.getMessage().contains("abs(-1)"),
error.getMessage());
+ Mockito.verifyNoInteractions(gravitinoTableCatalog, sparkCatalog);
+ }
+
private static class TestCatalog extends BaseCatalog {
private TestCatalog(
