This is an automated email from the ASF dual-hosted git repository.
yuqi1129 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 eb988d1f98 [#14284] fix(core): Reject nested column changes on managed
tables and validate Lance changes before applying them (#14287)
eb988d1f98 is described below
commit eb988d1f98aa2544e0580379e7af5f514e3e23a3
Author: Qi Yu <[email protected]>
AuthorDate: Tue Oct 27 21:47:23 2026 +0800
[#14284] fix(core): Reject nested column changes on managed tables and
validate Lance changes before applying them (#14287)
### What changes were proposed in this pull request?
- `ManagedTableOperations` rejects column changes on nested fields
(`fieldName.length > 1`).
- `LanceTableOperations` validates every change of a request before it
opens the dataset.
### Why are the changes needed?
Managed tables store columns by their top-level name only. A nested
`addColumn(["s", "x"])` created a bogus top-level column `s.x`. On
Lance, a nested rename or drop was applied to the dataset before the
metadata update failed. Lance also validated each change only when it
reached it, so a request with an invalid later change had already
applied its earlier changes to the dataset. In both cases, the dataset
and the metadata no longer agreed.
Fixes #14284
### Does this PR introduce _any_ user-facing change?
Nested column changes on managed and Lance tables, and Lance requests
that contain an unsupported change, now fail before anything is changed.
### How was this patch tested?
New
`TestManagedTableOperations.testAlterTableRejectsNestedColumnChanges`
and
`TestLanceTableOperations.testHandleLanceTableChangeValidatesBeforeChangingDataset`,
both fail without the fix. Both test classes pass.
---
.../lakehouse/lance/LanceTableOperations.java | 84 +++++++++++++---------
.../lakehouse/lance/TestLanceTableOperations.java | 30 ++++++++
.../gravitino/catalog/ManagedTableOperations.java | 6 ++
.../catalog/TestManagedTableOperations.java | 21 ++++++
4 files changed, 109 insertions(+), 32 deletions(-)
diff --git
a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
index e93913af95..0e43d84284 100644
---
a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
+++
b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
@@ -849,36 +849,16 @@ public class LanceTableOperations extends
ManagedTableOperations {
* @return the new version id of the Lance dataset after applying the changes
*/
long handleLanceTableChange(Table table, TableChange[] changes) {
+ // Reject what Lance cannot apply before opening the dataset, so a change
that fails these
+ // checks does not leave the changes before it applied to the dataset but
not to the metadata.
+ validateLanceTableChanges(changes);
String location = table.properties().get(Table.PROPERTY_LOCATION);
Map<String, String> storageOptions =
LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties,
table.properties());
try (Dataset dataset = openDataset(location, storageOptions)) {
for (TableChange change : changes) {
if (change instanceof TableChange.AddColumn addColumn) {
- String[] fieldName = addColumn.fieldName();
- Preconditions.checkArgument(
- fieldName.length == 1,
- "Lance only supports adding top-level columns: %s",
- String.join(".", fieldName));
- String columnName = fieldName[0];
- Preconditions.checkArgument(
- addColumn.isNullable(),
- "Lance only supports adding nullable columns because existing
rows are backfilled "
- + "with null: %s",
- columnName);
- Preconditions.checkArgument(
-
TableChange.ColumnPosition.defaultPos().equals(addColumn.getPosition()),
- "Lance only supports appending new columns: %s",
- columnName);
- Preconditions.checkArgument(
- !addColumn.isAutoIncrement(),
- "Lance does not support adding auto-increment columns: %s",
- columnName);
- Preconditions.checkArgument(
- addColumn.getDefaultValue() == null
- || addColumn.getDefaultValue().equals(DEFAULT_VALUE_NOT_SET),
- "Lance does not support default values when adding columns: %s",
- columnName);
+ String columnName = addColumn.fieldName()[0];
Field field =
LanceDataTypeConverter.CONVERTER.toArrowField(
@@ -886,7 +866,7 @@ public class LanceTableOperations extends
ManagedTableOperations {
LanceDataTypeConverter.checkFileFormatVersion(field,
dataset.getLanceFileFormatVersion());
dataset.addColumns(List.of(field));
} else if (change instanceof TableChange.DeleteColumn deleteColumn) {
- dataset.dropColumns(List.of(String.join(".",
deleteColumn.fieldName())));
+ dataset.dropColumns(List.of(deleteColumn.fieldName()[0]));
} else if (change instanceof TableChange.AddIndex addIndex) {
IndexType indexType = IndexType.valueOf(addIndex.getType().name());
IndexParams indexParams = getIndexParamsByIndexType(indexType);
@@ -902,16 +882,10 @@ public class LanceTableOperations extends
ManagedTableOperations {
.build());
} else if (change instanceof TableChange.RenameColumn renameColumn) {
ColumnAlteration lanceColumnAlter =
- new ColumnAlteration.Builder(String.join(".",
renameColumn.fieldName()))
+ new ColumnAlteration.Builder(renameColumn.fieldName()[0])
.rename(renameColumn.getNewName())
.build();
dataset.alterColumns(List.of(lanceColumnAlter));
- } else {
- // Currently, only column add/drop/rename and index addition are
supported.
- // TODO: Support change column type once we have a clear knowledge
about the means of
- // castTo in Lance.
- throw new UnsupportedOperationException(
- "Unsupported changes to lance table: " +
change.getClass().getSimpleName());
}
}
return dataset.getVersion().getId();
@@ -923,6 +897,52 @@ public class LanceTableOperations extends
ManagedTableOperations {
}
}
+ private static void validateLanceTableChanges(TableChange[] changes) {
+ for (TableChange change : changes) {
+ if (change instanceof TableChange.AddColumn addColumn) {
+ String[] fieldName = addColumn.fieldName();
+ Preconditions.checkArgument(
+ fieldName.length == 1,
+ "Lance only supports adding top-level columns: %s",
+ String.join(".", fieldName));
+ String columnName = fieldName[0];
+ Preconditions.checkArgument(
+ addColumn.isNullable(),
+ "Lance only supports adding nullable columns because existing rows
are backfilled "
+ + "with null: %s",
+ columnName);
+ Preconditions.checkArgument(
+
TableChange.ColumnPosition.defaultPos().equals(addColumn.getPosition()),
+ "Lance only supports appending new columns: %s",
+ columnName);
+ Preconditions.checkArgument(
+ !addColumn.isAutoIncrement(),
+ "Lance does not support adding auto-increment columns: %s",
+ columnName);
+ Preconditions.checkArgument(
+ addColumn.getDefaultValue() == null
+ || addColumn.getDefaultValue().equals(DEFAULT_VALUE_NOT_SET),
+ "Lance does not support default values when adding columns: %s",
+ columnName);
+ } else if (change instanceof TableChange.DeleteColumn
+ || change instanceof TableChange.RenameColumn) {
+ String[] fieldName = ((TableChange.ColumnChange) change).fieldName();
+ // Gravitino stores a Lance table's columns as top-level columns only,
so a nested change
+ // applied to the dataset could not be recorded in the metadata.
+ Preconditions.checkArgument(
+ fieldName.length == 1,
+ "Lance only supports dropping or renaming top-level columns: %s",
+ String.join(".", fieldName));
+ } else if (!(change instanceof TableChange.AddIndex)) {
+ // Currently, only column add/drop/rename and index addition are
supported.
+ // TODO: Support change column type once we have a clear knowledge
about the means of
+ // castTo in Lance.
+ throw new UnsupportedOperationException(
+ "Unsupported changes to lance table: " +
change.getClass().getSimpleName());
+ }
+ }
+ }
+
Dataset openDataset(String location) {
return openDataset(location, Map.of());
}
diff --git
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
index 3afa32e3c2..2196782da0 100644
---
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
+++
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
@@ -1114,6 +1114,36 @@ public class TestLanceTableOperations {
verify(dataset, never()).addColumns(anyList());
}
+ @Test
+ public void testHandleLanceTableChangeValidatesBeforeChangingDataset() {
+ Table table = mock(Table.class);
+ when(table.properties()).thenReturn(Map.of(Table.PROPERTY_LOCATION,
"location"));
+ Dataset dataset = mock(Dataset.class);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset("location",
Map.of());
+
+ // Each request starts with a valid change, which must not reach the
dataset when a later
+ // change in the same request is rejected.
+ List<TableChange> rejectedChanges =
+ List.of(
+ TableChange.renameColumn(new String[] {"parent", "nested"},
"renamed"),
+ TableChange.deleteColumn(new String[] {"parent", "nested"}, false),
+ TableChange.addColumn(new String[] {"required"},
Types.StringType.get(), false),
+ TableChange.updateColumnComment(new String[] {"col1"}, "comment"));
+ for (TableChange rejected : rejectedChanges) {
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ lanceTableOps.handleLanceTableChange(
+ table,
+ new TableChange[] {
+ TableChange.deleteColumn(new String[] {"col1"}, false),
rejected
+ }));
+ }
+ verify(lanceTableOps, never()).openDataset("location", Map.of());
+ verify(dataset, never()).dropColumns(anyList());
+ verify(dataset, never()).alterColumns(anyList());
+ }
+
@Test
public void testHandleLanceTableChangeRespectsOrder() {
Table table = mock(Table.class);
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
b/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
index e93c8f2cee..63bf737e81 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/ManagedTableOperations.java
@@ -304,6 +304,12 @@ public abstract class ManagedTableOperations implements
TableCatalog {
.collect(Collectors.toList());
for (TableChange.ColumnChange change : columnChanges) {
+ // Columns are stored by their top-level name only. A nested field name
would otherwise be
+ // joined into a new top-level column such as "a.b", or not found at all.
+ Preconditions.checkArgument(
+ change.fieldName().length == 1,
+ "Managed tables only support changes to top-level columns: %s",
+ DOT.join(change.fieldName()));
if (change instanceof TableChange.AddColumn addColumn) {
String columnName = DOT.join(addColumn.getFieldName());
boolean exists = newColumns.stream().anyMatch(col ->
col.name().equals(columnName));
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
b/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
index cd0ce98ff8..8f8e921d86 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestManagedTableOperations.java
@@ -362,6 +362,27 @@ public class TestManagedTableOperations {
Arrays.stream(renamedMiddle.columns()).map(Column::name).toArray(String[]::new));
}
+ @Test
+ public void testAlterTableRejectsNestedColumnChanges() {
+ NameIdentifier tableIdent = createThreeColumnTable("nested_changes");
+ List<TableChange> nestedChanges =
+ List.of(
+ TableChange.addColumn(new String[] {"col1", "nested"},
Types.StringType.get()),
+ TableChange.renameColumn(new String[] {"col1", "nested"},
"renamed"),
+ TableChange.deleteColumn(new String[] {"col1", "nested"}, true),
+ TableChange.updateColumnComment(new String[] {"col1", "nested"},
"comment"));
+
+ for (TableChange change : nestedChanges) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
tableOperations.alterTable(tableIdent, change));
+ }
+ Assertions.assertArrayEquals(
+ new String[] {"col1", "col2", "col3"},
+ Arrays.stream(tableOperations.loadTable(tableIdent).columns())
+ .map(Column::name)
+ .toArray(String[]::new));
+ }
+
private NameIdentifier createThreeColumnTable(String tableName) {
NameIdentifier tableIdent =
NameIdentifierUtil.ofTable(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
tableName);
