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 b73fdecdb4 [Cherry-pick to branch-1.3] [#12407] fix(lance): Hydrate
empty schema before table alteration (#12895) (#12926)
b73fdecdb4 is described below
commit b73fdecdb44784dccda07baf69b3b7da40ccee8b
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Sep 4 20:38:30 2026 +0800
[Cherry-pick to branch-1.3] [#12407] fix(lance): Hydrate empty schema
before table alteration (#12895) (#12926)
**Cherry-pick Information:**
- Original commit: 1db4894587d1265889d8e26780f762aab30279b9
- Target branch: `branch-1.3`
- Status: ✅ **Conflicts resolved**
Resolved the test-source conflict and verified the backport with
`:catalogs:catalog-lakehouse-generic:test`.
---------
Co-authored-by: Qi Yu <[email protected]>
---
.../lakehouse/lance/LanceTableOperations.java | 141 ++++++++++++---------
.../lakehouse/lance/TestLanceTableOperations.java | 123 ++++++++++++++++++
docs/lakehouse-generic-lance-table.md | 5 +-
3 files changed, 208 insertions(+), 61 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 7edfc468d5..dd7dc2b314 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
@@ -177,64 +177,7 @@ public class LanceTableOperations extends
ManagedTableOperations {
@Override
public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
- Table table = super.loadTable(ident);
- // Spark staged create can write the actual schema only to the Lance
dataset path. Refresh
- // Gravitino metadata when the stored table is declared-only, empty, or
configured to track
- // Lance dataset versions.
- boolean declaredOnly = isDeclaredOnly(table);
- boolean emptySchema = table.columns().length == 0;
- SchemaRefreshMode refreshMode = schemaRefreshMode();
- if (!declaredOnly && !emptySchema && refreshMode ==
SchemaRefreshMode.DECLARED_AND_EMPTY) {
- return table;
- }
- // Empty-schema table that was already confirmed against a stored version:
skip the dataset
- // open. The stored lance.version acts as a "checked at this version"
marker written on the
- // first confirmation. VERSION_CHECK mode does not take this shortcut — it
opens the dataset
- // every time to compare the current version.
- if (!declaredOnly
- && emptySchema
- &&
StringUtils.isNotBlank(table.properties().get(LanceConstants.LANCE_TABLE_VERSION))
- && refreshMode == SchemaRefreshMode.DECLARED_AND_EMPTY) {
- return table;
- }
-
- String location = table.properties().get(Table.PROPERTY_LOCATION);
- if (StringUtils.isBlank(location)) {
- return table;
- }
-
- Map<String, String> storageOptions =
- LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties,
table.properties());
- Column[] columns;
- long datasetVersion;
- try (Dataset dataset = openDataset(location, storageOptions)) {
- datasetVersion = dataset.version();
- if (refreshMode == SchemaRefreshMode.VERSION_CHECK
- && !declaredOnly
- && !isDatasetVersionChanged(table, datasetVersion)) {
- return table;
- }
- columns = extractColumns(dataset.getSchema());
- } catch (Exception e) {
- LOG.debug(
- "Failed to load Lance schema from location {} for table {}. Return
stored metadata.",
- location,
- ident,
- e);
- return table;
- }
-
- if (columns.length == 0) {
- // Dataset is genuinely empty: record the checked version so future
DECLARED_AND_EMPTY loads
- // can skip the dataset open (see the early-return above). Declared
tables are excluded
- // because their lance.declared flag is the authoritative "not yet
written" signal.
- if (!declaredOnly) {
- return recordCheckedEmptyVersion(ident, datasetVersion);
- }
- return table;
- }
-
- return repairTableMetadata(ident, columns, datasetVersion);
+ return loadTableInternal(ident, false);
}
@Override
@@ -302,7 +245,21 @@ public class LanceTableOperations extends
ManagedTableOperations {
public Table alterTable(NameIdentifier ident, TableChange... changes)
throws NoSuchSchemaException, TableAlreadyExistsException {
- Table loadedTable = super.loadTable(ident);
+ // Hydrate an empty stored schema before changing the dataset. Otherwise
this method can write
+ // the latest lance.version while leaving columns empty, making that
incomplete metadata look
+ // like a zero-column schema already confirmed at the same version.
+ Table loadedTable = loadTableInternal(ident, true);
+ boolean unhydratedSchema =
+ isDeclaredOnly(loadedTable)
+ || (loadedTable.columns().length == 0
+ && StringUtils.isBlank(
+
loadedTable.properties().get(LanceConstants.LANCE_TABLE_VERSION)));
+ if (unhydratedSchema) {
+ throw new IllegalStateException(
+ "Cannot alter Lance table "
+ + ident
+ + " because its dataset schema is not initialized or could not
be loaded");
+ }
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
@@ -491,6 +448,72 @@ public class LanceTableOperations extends
ManagedTableOperations {
return new Schema(fields);
}
+ private Table loadTableInternal(NameIdentifier ident, boolean forAlter) {
+ Table table = super.loadTable(ident);
+ // Spark staged create can write the actual schema only to the Lance
dataset path. Refresh
+ // Gravitino metadata when the stored table is declared-only, empty, or
configured to track
+ // Lance dataset versions.
+ boolean declaredOnly = isDeclaredOnly(table);
+ boolean emptySchema = table.columns().length == 0;
+ SchemaRefreshMode refreshMode = schemaRefreshMode();
+ if (!declaredOnly && !emptySchema && refreshMode ==
SchemaRefreshMode.DECLARED_AND_EMPTY) {
+ return table;
+ }
+ // Empty-schema table that was already confirmed against a stored version:
skip the dataset
+ // open during ordinary loads. An alter must recheck it so an externally
initialized schema is
+ // hydrated before the latest dataset version is persisted.
+ if (!forAlter
+ && !declaredOnly
+ && emptySchema
+ &&
StringUtils.isNotBlank(table.properties().get(LanceConstants.LANCE_TABLE_VERSION))
+ && refreshMode == SchemaRefreshMode.DECLARED_AND_EMPTY) {
+ return table;
+ }
+
+ String location = table.properties().get(Table.PROPERTY_LOCATION);
+ if (StringUtils.isBlank(location)) {
+ return table;
+ }
+
+ Map<String, String> storageOptions =
+ LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties,
table.properties());
+ Column[] columns;
+ long datasetVersion;
+ try (Dataset dataset = openDataset(location, storageOptions)) {
+ datasetVersion = dataset.version();
+ if (refreshMode == SchemaRefreshMode.VERSION_CHECK
+ && !declaredOnly
+ && !(forAlter && emptySchema)
+ && !isDatasetVersionChanged(table, datasetVersion)) {
+ return table;
+ }
+ columns = extractColumns(dataset.getSchema());
+ } catch (Exception e) {
+ if (forAlter) {
+ throw new IllegalStateException(
+ "Failed to load Lance schema before altering table " + ident, e);
+ }
+ LOG.debug(
+ "Failed to load Lance schema from location {} for table {}. Return
stored metadata.",
+ location,
+ ident,
+ e);
+ return table;
+ }
+
+ if (columns.length == 0) {
+ // Dataset is genuinely empty: record the checked version so future
DECLARED_AND_EMPTY loads
+ // can skip the dataset open (see the early-return above). Declared
tables are excluded
+ // because their lance.declared flag is the authoritative "not yet
written" signal.
+ if (!declaredOnly) {
+ return recordCheckedEmptyVersion(ident, datasetVersion);
+ }
+ return table;
+ }
+
+ return repairTableMetadata(ident, columns, datasetVersion);
+ }
+
private SchemaRefreshMode schemaRefreshMode() {
return
Optional.ofNullable(catalogProperties.get(LanceConstants.LANCE_SCHEMA_REFRESH_MODE))
.map(mode -> mode.trim().replace('-', '_').toUpperCase())
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 bf99a64814..f5b092fbf5 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
@@ -22,6 +22,7 @@ import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_CREAT
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_SCHEMA_REFRESH_MODE;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_STORAGE_OPTIONS_PREFIX;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_DECLARED;
+import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_REGISTER;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_VERSION;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
@@ -64,6 +65,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.lance.Dataset;
import org.lance.Version;
import org.lance.index.IndexOptions;
@@ -473,6 +476,111 @@ public class TestLanceTableOperations {
Assertions.assertEquals("9",
storedTable.get().properties().get(LANCE_TABLE_VERSION));
}
+ @ParameterizedTest(name = "recordedVersion={0}")
+ @ValueSource(booleans = {false, true})
+ public void
testAlterTableHydratesRegisteredSchemaBeforeRecordingVersion(boolean
recordedVersion)
+ throws Exception {
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("alter-registered-table-" +
recordedVersion).toString();
+ Map<String, String> properties =
+ recordedVersion
+ ? Map.of(
+ Table.PROPERTY_LOCATION,
+ location,
+ LANCE_TABLE_REGISTER,
+ Boolean.TRUE.toString(),
+ LANCE_TABLE_VERSION,
+ "2")
+ : Map.of(
+ Table.PROPERTY_LOCATION, location, LANCE_TABLE_REGISTER,
Boolean.TRUE.toString());
+ AtomicReference<TableEntity> storedTable =
+ new AtomicReference<>(tableEntity(ident, List.of(), properties));
+ stubMutableTable(ident, storedTable);
+ when(idGenerator.nextId()).thenReturn(10L);
+
+ Dataset schemaDataset = mock(Dataset.class);
+ when(schemaDataset.version()).thenReturn(3L);
+ when(schemaDataset.getSchema())
+ .thenReturn(new Schema(List.of(Field.nullable("embedding", new
ArrowType.Utf8()))));
+ Dataset alterDataset = mock(Dataset.class);
+ Version alteredVersion = mock(Version.class);
+ when(alterDataset.getVersion()).thenReturn(alteredVersion);
+ when(alteredVersion.getId()).thenReturn(4L);
+ Mockito.doReturn(schemaDataset, alterDataset)
+ .when(lanceTableOps)
+ .openDataset(location, Map.of());
+
+ Table alteredTable =
+ PrincipalUtils.doAs(
+ new UserPrincipal("tester"),
+ () ->
+ lanceTableOps.alterTable(
+ ident,
+ TableChange.addIndex(
+ Index.IndexType.SCALAR, "embedding_idx", new
String[][] {{"embedding"}})));
+
+ Assertions.assertEquals(1, alteredTable.columns().length);
+ Assertions.assertEquals("embedding", alteredTable.columns()[0].name());
+ Assertions.assertEquals("4",
alteredTable.properties().get(LANCE_TABLE_VERSION));
+ Assertions.assertEquals(1, alteredTable.index().length);
+ Assertions.assertEquals("embedding_idx", alteredTable.index()[0].name());
+ Assertions.assertEquals(1, storedTable.get().columns().size());
+ Assertions.assertEquals("4",
storedTable.get().properties().get(LANCE_TABLE_VERSION));
+
+ InOrder inOrder = Mockito.inOrder(schemaDataset, alterDataset);
+ inOrder.verify(schemaDataset).getSchema();
+ inOrder.verify(alterDataset).createIndex(any(IndexOptions.class));
+ inOrder.verify(alterDataset).getVersion();
+ verify(store, Mockito.times(2))
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ @Test
+ public void testAlterTableStopsWhenRegisteredSchemaCannotBeHydrated() throws
Exception {
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location =
tempDir.resolve("unavailable-registered-table").toString();
+ AtomicReference<TableEntity> storedTable =
+ new AtomicReference<>(
+ tableEntity(
+ ident,
+ List.of(),
+ Map.of(
+ Table.PROPERTY_LOCATION,
+ location,
+ LANCE_TABLE_REGISTER,
+ Boolean.TRUE.toString(),
+ LANCE_TABLE_VERSION,
+ "3")));
+ stubMutableTable(ident, storedTable);
+
+ Dataset alterDataset = mock(Dataset.class);
+ Version alteredVersion = mock(Version.class);
+ when(alterDataset.getVersion()).thenReturn(alteredVersion);
+ when(alteredVersion.getId()).thenReturn(4L);
+ Mockito.doThrow(new RuntimeException("storage unavailable"))
+ .doReturn(alterDataset)
+ .when(lanceTableOps)
+ .openDataset(location, Map.of());
+
+ IllegalStateException failure =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ lanceTableOps.alterTable(
+ ident,
+ TableChange.addIndex(
+ Index.IndexType.SCALAR, "embedding_idx", new
String[][] {{"embedding"}})));
+
+ Assertions.assertTrue(failure.getMessage().contains("schema"));
+ Assertions.assertEquals("storage unavailable",
failure.getCause().getMessage());
+ Assertions.assertTrue(storedTable.get().columns().isEmpty());
+ Assertions.assertEquals("3",
storedTable.get().properties().get(LANCE_TABLE_VERSION));
+ verify(lanceTableOps).openDataset(location, Map.of());
+ verify(alterDataset, never()).createIndex(any(IndexOptions.class));
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
@Test
public void testHandleLanceTableChangeRespectsOrder() {
Table table = mock(Table.class);
@@ -954,4 +1062,19 @@ public class TestLanceTableOperations {
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.EPOCH).build())
.build();
}
+
+ private void stubMutableTable(NameIdentifier ident,
AtomicReference<TableEntity> storedTable)
+ throws IOException {
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenAnswer(invocation -> storedTable.get());
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ TableEntity updated = updater.apply(storedTable.get());
+ storedTable.set(updated);
+ return updated;
+ });
+ }
}
diff --git a/docs/lakehouse-generic-lance-table.md
b/docs/lakehouse-generic-lance-table.md
index 945752ca05..2961527486 100644
--- a/docs/lakehouse-generic-lance-table.md
+++ b/docs/lakehouse-generic-lance-table.md
@@ -135,8 +135,9 @@ Gravitino. It adds a dataset version check to every
`loadTable` call.
:::note Zero-column Lance dataset
If a Lance dataset genuinely has no columns, `DECLARED_AND_EMPTY` mode records
the checked dataset
version (`lance.version`) on the first `loadTable` call. Subsequent loads skip
opening the dataset
-as long as the stored version is unchanged. Once columns are written to the
dataset, the next
-`VERSION_CHECK` load or an explicit `alterTable` will detect the change and
repair the schema.
+as long as the stored version is unchanged. Before recording a new version,
Lance table
+alterations recheck an empty stored schema and abort if it cannot be loaded,
so incomplete column
+metadata is not associated with the latest dataset version.
:::
### Table Operations
