yuqi1129 commented on code in PR #12383:
URL: https://github.com/apache/gravitino/pull/12383#discussion_r3755733128
##########
catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java:
##########
@@ -797,6 +836,35 @@ long handleLanceTableChange(Table table, TableChange[]
changes) {
}
}
+ private long addColumns(
+ Table table, List<Field> fieldsToAdd, String location, Map<String,
String> storageOptions) {
+ List<Field> expectedCurrentFields =
convertColumnsToArrowSchema(table.columns()).getFields();
Review Comment:
The "expected" schema is rebuilt from Gravitino metadata through
`LanceDataTypeConverter` and then compared to the live Lance schema with an
exact positional field comparison (see `fieldMatches` below). But `toGravitino`
-> `fromGravitino` is not round-trip lossless, so several classes of perfectly
consistent tables will fail the pre-check at line 847 and never be able to add
a column again:
- **Timestamp with time zone**: `toGravitino` maps any tz to
`TimestampType.withTimeZone(p)`, and `fromGravitino` always re-emits
`Timestamp(unit, "UTC")` (the converter even carries `// todo: need timeZoneId
for timestamp with time zone`). A dataset with `timestamp[us,
tz=Asia/Shanghai]` never matches.
- **List**: `toGravitino` drops the list child field name, and
`toArrowField` hard-codes it back as `"element"`, while datasets written by
pyarrow / arrow-rs name it `"item"`.
- **Decimal**: `fromGravitino` always emits bit width 128, so a `decimal256`
column never matches.
The resulting `OptimisticLockException` surfaces as HTTP 409, which tells
the client to retry, but the condition is permanent — under the default
`DECLARED_AND_EMPTY` refresh mode a non-empty table is never re-hydrated, so
there's no supported way to clear it.
The same strict comparison is reused for the post-add check at line 851,
which has the mirror problem: if Lance normalizes the written field in any way,
an add that actually succeeded gets rolled back and reported as a failure.
Would it be possible to compare against the schema actually read from the
dataset (e.g. snapshot it before the add and diff that against the post-add
schema), rather than against a re-derived one?
##########
catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java:
##########
@@ -301,9 +307,29 @@ public Table createTable(
@Override
public Table alterTable(NameIdentifier ident, TableChange... changes)
throws NoSuchSchemaException, TableAlreadyExistsException {
+ List<Field> fieldsToAdd = prepareFieldsToAdd(changes);
+ if (!fieldsToAdd.isEmpty()) {
+ // AddColumn must use the same schema hydration as loadTable. A declared
table, or a
+ // registered table with empty stored columns, may already have a real
schema in Lance.
+ loadTable(ident);
+ }
+
+ // Schema hydration may update the entity store, so use a fresh entity
both to validate the
+ // physical schema and as the optimistic-lock snapshot for the metadata
update.
+ TableEntity loadedEntity = loadTableEntity(ident);
+ Table loadedTable = toGenericTable(loadedEntity);
+ validateFieldsToAdd(loadedTable, fieldsToAdd);
+ long version = handleLanceTableChange(loadedTable, changes, fieldsToAdd);
+
+ if (!fieldsToAdd.isEmpty()) {
+ try {
+ return persistAddedColumns(ident, loadedEntity, changes, version);
+ } catch (RuntimeException metadataFailure) {
+ rollbackAddedColumns(loadedTable, fieldsToAdd, metadataFailure);
Review Comment:
This rollback can leave Gravitino metadata listing columns that no longer
exist physically.
`persistAddedColumns` fails the CAS whenever
`!current.equals(expectedEntity)`. With
`lance.schema-refresh-mode=VERSION_CHECK`, a concurrent `loadTable` landing in
the window between `dataset.addColumns()` committing and the CAS running will
hydrate the *new* schema into the entity store and bump `lance.version` via
`repairTableMetadata`. That both (a) makes this CAS fail and (b) means the
store already contains the added columns. We then call `rollbackAddedColumns`,
which physically drops them from the dataset, and rethrow — net result is
metadata referencing columns Lance doesn't have, and under the default
`DECLARED_AND_EMPTY` mode a non-empty table is never re-checked, so it stays
that way.
The dispatcher only takes a READ tree-lock for non-rename changes
(`TableOperationDispatcher.alterTable`), so this interleaving isn't excluded.
One option: make the CAS compare only the fields that actually matter
(columns + `lance.version`) and skip the physical rollback when the stored
schema already reflects the add.
##########
catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java:
##########
@@ -811,6 +879,257 @@ Dataset openDataset(String location, Map<String, String>
storageOptions) {
.build();
}
+ private TableEntity loadTableEntity(NameIdentifier ident) {
+ try {
+ return store.get(ident, Entity.EntityType.TABLE, TableEntity.class);
+ } catch (NoSuchEntityException e) {
+ throw new NoSuchTableException(e, "Table %s does not exist", ident);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to load table " + ident, e);
+ }
+ }
+
+ private Table persistAddedColumns(
+ NameIdentifier ident,
+ TableEntity expectedEntity,
+ TableChange[] changes,
+ long datasetVersion) {
+ try {
+ TableEntity updatedEntity =
+ updateTableWithCasRetry(
+ ident,
+ current -> {
+ if (!current.equals(expectedEntity)) {
+ throw new OptimisticLockException(
+ "Table %s metadata changed while adding Lance columns",
ident);
+ }
+ return appendAddedColumns(current, changes, datasetVersion);
+ });
+ return toGenericTable(updatedEntity);
+ } catch (NoSuchEntityException e) {
+ throw new NoSuchTableException(e, "Table %s does not exist", ident);
+ } catch (EntityAlreadyExistsException e) {
+ throw new IllegalArgumentException("Failed to persist added columns for
table " + ident, e);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to persist added columns for table "
+ ident, e);
+ }
+ }
+
+ private TableEntity appendAddedColumns(
+ TableEntity tableEntity, TableChange[] changes, long datasetVersion) {
+ List<ColumnEntity> updatedColumns =
+ tableEntity.columns().stream()
+ .sorted(Comparator.comparingInt(ColumnEntity::position))
+ .collect(Collectors.toCollection(ArrayList::new));
+ Instant now = Instant.now();
+ AuditInfo columnAuditInfo =
+ AuditInfo.builder()
+ .withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+ .withCreateTime(now)
+ .build();
+ for (TableChange change : changes) {
+ TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
+ updatedColumns.add(
+ ColumnEntity.builder()
+ .withId(idGenerator.nextId())
+ .withName(addColumn.fieldName()[0])
+ .withPosition(updatedColumns.size())
+ .withDataType(addColumn.getDataType())
+ .withComment(addColumn.getComment())
+ .withNullable(true)
+ .withAutoIncrement(false)
+ .withDefaultValue(DEFAULT_VALUE_NOT_SET)
+ .withAuditInfo(columnAuditInfo)
+ .build());
+ }
+
+ Map<String, String> updatedProperties = new
HashMap<>(tableEntity.properties());
+ updatedProperties.put(LanceConstants.LANCE_TABLE_VERSION,
String.valueOf(datasetVersion));
+ // A successful physical addition confirms that the Lance schema has been
written, including
+ // when the dataset was previously a genuinely zero-column declared table.
+ updatedProperties.remove(LanceConstants.LANCE_TABLE_DECLARED);
+
+ return TableEntity.builder()
+ .withId(tableEntity.id())
+ .withName(tableEntity.name())
+ .withNamespace(tableEntity.namespace())
+ .withComment(tableEntity.comment())
+ .withColumns(updatedColumns)
+ .withProperties(updatedProperties)
+ .withPartitioning(tableEntity.partitioning())
+ .withDistribution(tableEntity.distribution())
+ .withSortOrders(tableEntity.sortOrders())
+ .withIndexes(tableEntity.indexes())
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(tableEntity.auditInfo().creator())
+ .withCreateTime(tableEntity.auditInfo().createTime())
+
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
+ .withLastModifiedTime(now)
+ .build())
+ .build();
+ }
+
+ private void rollbackAddedColumns(
+ Table table, List<Field> fieldsToAdd, RuntimeException originalFailure) {
+ String location = table.properties().get(Table.PROPERTY_LOCATION);
+ Map<String, String> storageOptions =
+ LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties,
table.properties());
+ List<Field> expectedFields =
convertColumnsToArrowSchema(table.columns()).getFields();
+ try (Dataset dataset = openDataset(location, storageOptions)) {
+ rollbackAddedColumns(
+ dataset, fieldNames(fieldsToAdd), expectedFields, location,
originalFailure);
+ } catch (RuntimeException rollbackFailure) {
+ addRollbackFailure(originalFailure, rollbackFailure, location);
+ }
+ }
+
+ private void rollbackAddedColumns(
+ Dataset dataset,
+ List<String> addedColumnNames,
+ List<Field> expectedFields,
+ String location,
+ RuntimeException originalFailure) {
+ try {
+ dataset.checkoutLatest();
+ Set<String> currentColumnNames =
+
dataset.getSchema().getFields().stream().map(Field::getName).collect(Collectors.toSet());
+ List<String> columnsToDrop =
+
addedColumnNames.stream().filter(currentColumnNames::contains).toList();
+ if (!columnsToDrop.isEmpty()) {
+ dataset.dropColumns(columnsToDrop);
+ dataset.checkoutLatest();
+ }
+ validateSchemaMatches(expectedFields, dataset.getSchema(), location,
"after rollback");
+ LOG.warn(
+ "Rolled back Lance columns {} at {} after alteration failure",
columnsToDrop, location);
+ } catch (RuntimeException rollbackFailure) {
+ addRollbackFailure(originalFailure, rollbackFailure, location);
+ }
+ }
+
+ private void addRollbackFailure(
+ RuntimeException originalFailure, RuntimeException rollbackFailure,
String location) {
+ originalFailure.addSuppressed(rollbackFailure);
+ LOG.error("Failed to roll back added Lance columns at {}", location,
rollbackFailure);
+ }
+
+ private List<Field> prepareFieldsToAdd(Table table, TableChange[] changes) {
+ List<Field> fieldsToAdd = prepareFieldsToAdd(changes);
+ validateFieldsToAdd(table, fieldsToAdd);
+ return fieldsToAdd;
+ }
+
+ 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) {
+ if (change instanceof TableChange.AddColumn addColumn) {
+ Preconditions.checkArgument(
+ addColumn.fieldName().length == 1,
+ "Lance only supports adding top-level columns: %s",
+ String.join(".", addColumn.fieldName()));
+ String columnName = addColumn.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);
+ Preconditions.checkArgument(
+ columnNames.add(columnName), "Column %s already exists",
columnName);
+ fieldsToAdd.add(
+ LanceDataTypeConverter.CONVERTER.toArrowField(
+ addColumn.fieldName()[0], addColumn.getDataType(), true));
+ }
+ }
+ return fieldsToAdd;
+ }
+
+ private void validateFieldsToAdd(Table table, List<Field> fieldsToAdd) {
+ if (fieldsToAdd.isEmpty()) {
+ return;
+ }
+ Set<String> columnNames =
+ new
HashSet<>(Arrays.stream(table.columns()).map(Column::name).toList());
+ for (Field field : fieldsToAdd) {
+ Preconditions.checkArgument(
+ columnNames.add(field.getName()), "Column %s already exists",
field.getName());
+ }
+ }
+
+ private void validateSchemaMatches(
+ List<Field> expectedFields, Schema actualSchema, String location, String
stage) {
+ if (actualSchema == null || !fieldsMatch(expectedFields,
actualSchema.getFields())) {
+ throw new OptimisticLockException(
+ "Lance schema at %s is inconsistent with Gravitino metadata %s;
expected fields %s but "
+ + "found %s",
+ location,
+ stage,
+ expectedFields,
+ actualSchema == null ? "an unavailable schema" :
actualSchema.getFields());
+ }
+ }
+
+ private boolean fieldsMatch(List<Field> expectedFields, List<Field>
actualFields) {
+ if (expectedFields.size() != actualFields.size()) {
+ return false;
+ }
+ for (int i = 0; i < expectedFields.size(); i++) {
+ if (!fieldMatches(expectedFields.get(i), actualFields.get(i))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private boolean fieldMatches(Field expected, Field actual) {
Review Comment:
Minor / lower severity, but it lands on the same path: column types
Lance/Arrow can't map back to a Gravitino type — notably `fixed_size_list`
vector columns, arguably the flagship Lance type — are stored as
`Types.ExternalType` whose `catalogString` is the serialized Arrow `Field`,
*including its original name*.
`RenameColumn` is supported for Lance tables and rewrites only the
`ColumnEntity` name, leaving the embedded JSON name stale.
`convertColumnsToArrowSchema` at line 841 then hits the EXTERNAL branch of
`toArrowField`, which does
`Preconditions.checkArgument(name.equals(field.getName()), "expected field name
%s but got %s")`. So after renaming a vector column, every subsequent
`AddColumn` on that table dies with a confusing `IllegalArgumentException`
before any Lance work happens.
Either rewrite the embedded name on rename, or override the name when
reconstructing the Arrow field from `ExternalType`.
--
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]
