yuqi1129 commented on code in PR #12383:
URL: https://github.com/apache/gravitino/pull/12383#discussion_r3757436723
##########
catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java:
##########
@@ -811,6 +837,78 @@ Dataset openDataset(String location, Map<String, String>
storageOptions) {
.build();
}
+ private List<Field> prepareFieldsToAdd(TableChange[] changes) {
+ Preconditions.checkArgument(changes != null && changes.length > 0,
"Changes cannot be empty");
+
+ int addColumnCount = 0;
+ for (TableChange change : changes) {
+ Preconditions.checkArgument(change != null, "Table change cannot be
null");
+ if (change instanceof TableChange.AddColumn) {
+ addColumnCount++;
+ } else if (!(change instanceof TableChange.DeleteColumn)
+ && !(change instanceof TableChange.AddIndex)
+ && !(change instanceof TableChange.RenameColumn)) {
+ throw new UnsupportedOperationException(
+ "Unsupported changes to lance table: " +
change.getClass().getSimpleName());
+ }
+ }
+
+ if (addColumnCount == 0) {
+ return List.of();
+ }
+
+ Preconditions.checkArgument(
+ addColumnCount == changes.length,
+ "Lance AddColumn cannot be combined with other table changes");
+
+ Set<String> columnNames = new HashSet<>();
+ List<Field> fieldsToAdd = new ArrayList<>(changes.length);
+ for (TableChange change : changes) {
Review Comment:
Is it possible to merge the loop above and the loop here?
##########
catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java:
##########
@@ -473,6 +475,168 @@ public void testAlterTablePersistsUpdatedLanceVersion()
throws Exception {
Assertions.assertEquals("9",
storedTable.get().properties().get(LANCE_TABLE_VERSION));
}
+ @Test
+ public void testAlterTableAddsNullableColumnsInSingleLanceCommit() throws
Exception {
Review Comment:
Have you covered the case where we changed Lance to succeed but failed to
write metadata to Gravitino, and then It's still okay when we load the table?
##########
catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java:
##########
@@ -754,10 +766,24 @@ private GenericColumn toGenericColumn(ColumnEntity
columnEntity) {
* @return the new version id of the Lance dataset after applying the changes
*/
long handleLanceTableChange(Table table, TableChange[] changes) {
+ List<Field> fieldsToAdd = prepareFieldsToAdd(changes);
+ validateFieldsToAdd(table, fieldsToAdd);
+ return handleLanceTableChange(table, changes, fieldsToAdd);
+ }
+
+ private long handleLanceTableChange(Table table, TableChange[] changes,
List<Field> fieldsToAdd) {
String location = table.properties().get(Table.PROPERTY_LOCATION);
Map<String, String> storageOptions =
LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties,
table.properties());
try (Dataset dataset = openDataset(location, storageOptions)) {
+ if (!fieldsToAdd.isEmpty()) {
+ // Adding all fields in one call creates one Lance schema version and
backfills existing
+ // rows with null for the new nullable columns.
+ dataset.addColumns(fieldsToAdd);
+ dataset.checkoutLatest();
Review Comment:
How do you handle the case where there exist `addColumn` and `dropColumn` or
`AddIndex` at the same time? Have you already excluded such scenarios?
##########
catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java:
##########
@@ -301,15 +304,24 @@ public Table createTable(
@Override
public Table alterTable(NameIdentifier ident, TableChange... changes)
throws NoSuchSchemaException, TableAlreadyExistsException {
+ List<Field> fieldsToAdd = prepareFieldsToAdd(changes);
+ // A declared or registered table can have an empty stored schema while
its Lance dataset
+ // already contains columns. Hydrate the metadata before applying an
incremental AddColumn.
+ Table loadedTable = fieldsToAdd.isEmpty() ? super.loadTable(ident) :
loadTable(ident);
+ validateFieldsToAdd(loadedTable, fieldsToAdd);
+ long version = handleLanceTableChange(loadedTable, changes, fieldsToAdd);
- Table loadedTable = super.loadTable(ident);
- long version = handleLanceTableChange(loadedTable, changes);
// After making changes to the Lance dataset, we need to update the table
metadata in
// Gravitino. If there's any failure during this process, the code will
throw an exception
// and the update won't be applied in Gravitino.
- TableChange[] metadataChanges = Arrays.copyOf(changes, changes.length + 1);
+ int internalChangeCount = fieldsToAdd.isEmpty() ? 1 : 2;
+ TableChange[] metadataChanges = Arrays.copyOf(changes, changes.length +
internalChangeCount);
metadataChanges[changes.length] =
TableChange.setProperty(LanceConstants.LANCE_TABLE_VERSION,
String.valueOf(version));
+ if (!fieldsToAdd.isEmpty()) {
+ metadataChanges[changes.length + 1] =
Review Comment:
You can use `ArrayUtils.add()` directly with explicitly copying it.
##########
catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java:
##########
@@ -301,15 +304,24 @@ public Table createTable(
@Override
public Table alterTable(NameIdentifier ident, TableChange... changes)
throws NoSuchSchemaException, TableAlreadyExistsException {
+ List<Field> fieldsToAdd = prepareFieldsToAdd(changes);
+ // A declared or registered table can have an empty stored schema while
its Lance dataset
+ // already contains columns. Hydrate the metadata before applying an
incremental AddColumn.
+ Table loadedTable = fieldsToAdd.isEmpty() ? super.loadTable(ident) :
loadTable(ident);
+ validateFieldsToAdd(loadedTable, fieldsToAdd);
+ long version = handleLanceTableChange(loadedTable, changes, fieldsToAdd);
- Table loadedTable = super.loadTable(ident);
- long version = handleLanceTableChange(loadedTable, changes);
// After making changes to the Lance dataset, we need to update the table
metadata in
// Gravitino. If there's any failure during this process, the code will
throw an exception
// and the update won't be applied in Gravitino.
- TableChange[] metadataChanges = Arrays.copyOf(changes, changes.length + 1);
+ int internalChangeCount = fieldsToAdd.isEmpty() ? 1 : 2;
+ TableChange[] metadataChanges = Arrays.copyOf(changes, changes.length +
internalChangeCount);
metadataChanges[changes.length] =
TableChange.setProperty(LanceConstants.LANCE_TABLE_VERSION,
String.valueOf(version));
+ if (!fieldsToAdd.isEmpty()) {
+ metadataChanges[changes.length + 1] =
Review Comment:
Another problem is why the index is `changes.length + 1`, not
`changes.length` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
