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 21c348be20 [#11027][#11178] fix(lance): Refresh Lance table schema
from dataset on load (#11176)
21c348be20 is described below
commit 21c348be20912ce726bf8c83eef60ccacc566ab2
Author: Qi Yu <[email protected]>
AuthorDate: Mon Jun 8 15:43:27 2026 +0800
[#11027][#11178] fix(lance): Refresh Lance table schema from dataset on
load (#11176)
### What changes were proposed in this pull request?
This PR updates Lance table loading in the Generic Lakehouse catalog to
repair Gravitino table metadata from the underlying Lance dataset when
needed.
The refresh behavior is:
- `DECLARED_AND_EMPTY` by default: refresh tables marked with
`lance.declared=true` or tables with empty stored columns.
- `VERSION_CHECK`: open the Lance dataset during `loadTable`, compare
its version with the stored `lance.version`, and refresh columns when
the version changes.
When a declared table has been materialized, the refresh persists its
actual columns and dataset version and removes `lance.declared`. As a
result, Lance REST `describeTable` reports `is_only_declared=false`
after the first data write.
This PR also supports the Lance `loadDetailedMetadata` option and
documents the new catalog property `lance.schema-refresh-mode`.
### Why are the changes needed?
Spark staged create/declare flows register a table in Gravitino before
writing the actual Lance dataset. After the write, Gravitino can still
contain empty columns and the immutable `lance.declared=true` marker.
This causes the Web UI and client APIs to return an empty schema and
causes `describeTable` to incorrectly report that the materialized table
is still declaration-only.
Refreshing the stored metadata from the Lance dataset during `loadTable`
repairs both states.
Fix: #11027
Fix: #11178
Fix: #9501
### Does this PR introduce _any_ user-facing change?
Yes.
A new Generic Lakehouse catalog property is added:
- `lance.schema-refresh-mode`
- `DECLARED_AND_EMPTY`: the default; repairs declared tables and tables
with empty stored columns.
- `VERSION_CHECK`: refreshes table columns when the Lance dataset
version changes.
Materialized declared tables now return their actual schema and report
`is_only_declared=false`. No generic Table API parameter is added.
### How was this patch tested?
- Added unit tests for declared-table repair, empty stored schemas,
dataset version changes, genuinely empty datasets, missing locations,
dataset open failures, and concurrent metadata repair.
- Added Lance REST and Spark integration coverage for schema refresh and
declared-table state.
- `./gradlew :catalogs:catalog-lakehouse-generic:spotlessApply
:lance:lance-rest-server:spotlessApply
:catalogs:catalog-lakehouse-generic:test --tests
org.apache.gravitino.catalog.lakehouse.lance.TestLanceTableOperations
--tests
org.apache.gravitino.catalog.lakehouse.generic.TestPropertiesMetadata
:lance:lance-rest-server:compileTestJava -PskipWeb=true
-PskipDockerTests=true`
- `./gradlew :docs:build`
---
.../generic/GenericCatalogPropertiesMetadata.java | 12 +
.../lakehouse/lance/LanceTableOperations.java | 335 +++++++++-
.../lakehouse/generic/TestPropertiesMetadata.java | 8 +-
.../lakehouse/lance/TestLanceTableOperations.java | 701 +++++++++++++++++++++
.../catalog/TableOperationDispatcher.java | 6 +
docs/lakehouse-generic-catalog.md | 11 +-
docs/lakehouse-generic-lance-table.md | 37 +-
.../lance/common/ops/LanceTableOperations.java | 8 +-
.../gravitino/GravitinoLanceTableOperations.java | 10 +-
.../lance/common/utils/LanceConstants.java | 1 +
.../lance/service/rest/LanceTableOperations.java | 7 +-
.../gravitino/TestGravitinoLanceModeParsing.java | 2 +-
.../integration/test/LanceSparkRESTServiceIT.java | 4 +
.../service/rest/TestLanceNamespaceOperations.java | 9 +-
14 files changed, 1107 insertions(+), 44 deletions(-)
diff --git
a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogPropertiesMetadata.java
b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogPropertiesMetadata.java
index dd805181f8..7bd6f58605 100644
---
a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogPropertiesMetadata.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.catalog.lakehouse.generic;
+import static org.apache.gravitino.connector.PropertyEntry.enumPropertyEntry;
import static
org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyEntry;
import static
org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyPrefixEntry;
+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 com.google.common.collect.ImmutableList;
@@ -28,6 +30,7 @@ import com.google.common.collect.Maps;
import java.util.List;
import java.util.Map;
import org.apache.gravitino.Catalog;
+import org.apache.gravitino.catalog.lakehouse.lance.LanceTableOperations;
import org.apache.gravitino.connector.BaseCatalogPropertiesMetadata;
import org.apache.gravitino.connector.PropertyEntry;
@@ -50,6 +53,15 @@ public class GenericCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetad
false /* immutable */,
null, /* defaultValue */
false /* hidden */,
+ false /* reserved */),
+ enumPropertyEntry(
+ LANCE_SCHEMA_REFRESH_MODE,
+ "Controls when Lance table schemas are refreshed from the
underlying dataset.",
+ false /* required */,
+ false /* immutable */,
+ LanceTableOperations.SchemaRefreshMode.class,
+ LanceTableOperations.SchemaRefreshMode.DECLARED_AND_EMPTY,
+ false /* hidden */,
false /* reserved */));
PROPERTIES_METADATA = Maps.uniqueIndex(propertyEntries,
PropertyEntry::getName);
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 89c99d5365..38dbf0e90d 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
@@ -20,31 +20,41 @@ package org.apache.gravitino.catalog.lakehouse.lance;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_CREATION_MODE;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_REGISTER;
+import static org.apache.gravitino.rel.Column.DEFAULT_VALUE_NOT_SET;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import java.io.IOException;
+import java.time.Instant;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
-import org.apache.arrow.memory.RootAllocator;
+import java.util.stream.IntStream;
import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.ManagedSchemaOperations;
import org.apache.gravitino.catalog.ManagedTableOperations;
+import org.apache.gravitino.connector.GenericColumn;
import org.apache.gravitino.connector.GenericTable;
import org.apache.gravitino.connector.SupportsSchemas;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NoSuchTableException;
import org.apache.gravitino.exceptions.TableAlreadyExistsException;
import org.apache.gravitino.lance.common.ops.gravitino.LanceDataTypeConverter;
import org.apache.gravitino.lance.common.utils.LanceConstants;
import org.apache.gravitino.lance.common.utils.LancePropertiesUtils;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.ColumnEntity;
+import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.TableChange;
@@ -53,6 +63,7 @@ import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.utils.PrincipalUtils;
import org.lance.Dataset;
import org.lance.ReadOptions;
import org.lance.WriteParams;
@@ -74,6 +85,41 @@ public class LanceTableOperations extends
ManagedTableOperations {
OVERWRITE
}
+ /**
+ * Controls when Gravitino refreshes a Lance table's stored columns from the
underlying dataset.
+ *
+ * <p>{@link #DECLARED_AND_EMPTY} is the default. It covers two
complementary repair cases:
+ *
+ * <ul>
+ * <li><b>Declared tables</b> ({@code lance.declared=true}): metadata-only
tables whose schema
+ * has not yet been written to Gravitino (e.g. Spark staged-create
flow).
+ * <li><b>Empty stored columns</b>: tables where Gravitino has no column
metadata but the Lance
+ * dataset at the table's location already carries a real schema (e.g.
tables registered via
+ * the register-table path before schema was captured).
+ * </ul>
+ *
+ * <p><b>Zero-column Lance dataset:</b> when the dataset has no columns,
Gravitino records the
+ * checked dataset version ({@code lance.version}) without modifying stored
columns. Subsequent
+ * {@code loadTable} calls skip the dataset open because the stored version
acts as a "confirmed
+ * empty" marker. The dataset is re-examined only after the stored version
changes (for example
+ * via an explicit {@code alterTable}) or when the mode is switched to
{@link #VERSION_CHECK}.
+ */
+ public enum SchemaRefreshMode {
+ /**
+ * Default mode. Refreshes stored columns from the Lance dataset for
declared tables ({@code
+ * lance.declared=true}) and for tables whose Gravitino column list is
empty. Also repairs
+ * empty-column tables that were registered before their schema was
captured.
+ */
+ DECLARED_AND_EMPTY,
+ /**
+ * Opens the Lance dataset on every {@code loadTable}, compares the
dataset version with the
+ * stored {@code lance.version}, and refreshes columns when the version
has changed. The version
+ * is the sole gating factor: if the version is unchanged the schema read
is skipped even when
+ * stored columns are empty.
+ */
+ VERSION_CHECK
+ }
+
private final EntityStore store;
private final ManagedSchemaOperations schemaOps;
@@ -114,6 +160,68 @@ public class LanceTableOperations extends
ManagedTableOperations {
catalogProperties == null ? Map.of() :
ImmutableMap.copyOf(catalogProperties);
}
+ @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);
+ }
+
@Override
public Table createTable(
NameIdentifier ident,
@@ -184,26 +292,18 @@ public class LanceTableOperations extends
ManagedTableOperations {
// 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.
- GenericTable table = (GenericTable) super.alterTable(ident, changes);
- Map<String, String> updatedProperties = new HashMap<>(table.properties());
- updatedProperties.put(LanceConstants.LANCE_TABLE_VERSION,
String.valueOf(version));
- return GenericTable.builder()
- .withName(table.name())
- .withColumns(table.columns())
- .withComment(table.comment())
- .withProperties(updatedProperties)
- .withAuditInfo(table.auditInfo())
- .withPartitioning(table.partitioning())
- .withSortOrders(table.sortOrder())
- .withDistribution(table.distribution())
- .withIndexes(table.index())
- .build();
+ TableChange[] metadataChanges = Arrays.copyOf(changes, changes.length + 1);
+ metadataChanges[changes.length] =
+ TableChange.setProperty(LanceConstants.LANCE_TABLE_VERSION,
String.valueOf(version));
+ return super.alterTable(ident, metadataChanges);
}
@Override
public boolean purgeTable(NameIdentifier ident) {
try {
- Table table = loadTable(ident);
+ // Use super.loadTable to avoid triggering an unnecessary schema-refresh
(which may open the
+ // dataset) for a table that is about to be deleted anyway.
+ Table table = super.loadTable(ident);
boolean external =
Optional.ofNullable(table.properties().get(Table.PROPERTY_EXTERNAL))
.map(Boolean::parseBoolean)
@@ -238,7 +338,8 @@ public class LanceTableOperations extends
ManagedTableOperations {
@Override
public boolean dropTable(NameIdentifier ident) {
try {
- Table table = loadTable(ident);
+ // Use super.loadTable to skip schema-refresh overhead when dropping.
+ Table table = super.loadTable(ident);
boolean external =
Optional.ofNullable(table.properties().get(Table.PROPERTY_EXTERNAL))
.map(Boolean::parseBoolean)
@@ -324,7 +425,6 @@ public class LanceTableOperations extends
ManagedTableOperations {
LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties,
properties);
try (Dataset ignored =
Dataset.write()
- .allocator(new RootAllocator())
.schema(convertColumnsToArrowSchema(columns))
.uri(location)
.mode(WriteParams.WriteMode.CREATE)
@@ -365,7 +465,7 @@ public class LanceTableOperations extends
ManagedTableOperations {
}
}
- private org.apache.arrow.vector.types.pojo.Schema
convertColumnsToArrowSchema(Column[] columns) {
+ private Schema convertColumnsToArrowSchema(Column[] columns) {
List<Field> fields =
Arrays.stream(columns)
.map(
@@ -373,7 +473,198 @@ public class LanceTableOperations extends
ManagedTableOperations {
LanceDataTypeConverter.CONVERTER.toArrowField(
col.name(), col.dataType(), col.nullable()))
.collect(Collectors.toList());
- return new org.apache.arrow.vector.types.pojo.Schema(fields);
+ return new Schema(fields);
+ }
+
+ private SchemaRefreshMode schemaRefreshMode() {
+ return
Optional.ofNullable(catalogProperties.get(LanceConstants.LANCE_SCHEMA_REFRESH_MODE))
+ .map(mode -> mode.trim().replace('-', '_').toUpperCase())
+ .map(SchemaRefreshMode::valueOf)
+ .orElse(SchemaRefreshMode.DECLARED_AND_EMPTY);
+ }
+
+ private boolean isDeclaredOnly(Table table) {
+ return isDeclaredOnly(table.properties());
+ }
+
+ private boolean isDeclaredOnly(Map<String, String> properties) {
+ return
Optional.ofNullable(properties.get(LanceConstants.LANCE_TABLE_DECLARED))
+ .map(Boolean::parseBoolean)
+ .orElse(false);
+ }
+
+ private boolean isDatasetVersionChanged(Table table, long datasetVersion) {
+ return isDatasetVersionChanged(table.properties(), datasetVersion);
+ }
+
+ private boolean isDatasetVersionChanged(Map<String, String> properties, long
datasetVersion) {
+ String version = properties.get(LanceConstants.LANCE_TABLE_VERSION);
+ if (StringUtils.isBlank(version)) {
+ return true;
+ }
+
+ try {
+ return Long.parseLong(version) != datasetVersion;
+ } catch (NumberFormatException e) {
+ return true;
+ }
+ }
+
+ private Column[] extractColumns(Schema arrowSchema) {
+ return arrowSchema.getFields().stream()
+ .map(
+ field ->
+ Column.of(
+ field.getName(),
+ LanceDataTypeConverter.CONVERTER.toGravitino(field),
+ null,
+ field.isNullable(),
+ false,
+ DEFAULT_VALUE_NOT_SET))
+ .toArray(Column[]::new);
+ }
+
+ private Table repairTableMetadata(NameIdentifier ident, Column[] columns,
long datasetVersion) {
+ try {
+ TableEntity tableEntity =
+ store.update(
+ ident,
+ TableEntity.class,
+ Entity.EntityType.TABLE,
+ current -> {
+ if (!needsSchemaRefresh(current, datasetVersion)) {
+ return current;
+ }
+ return replaceColumnsFromDataset(current, columns,
datasetVersion);
+ });
+ return toGenericTable(tableEntity);
+ } catch (NoSuchEntityException e) {
+ throw new NoSuchTableException(e, "Table %s does not exist", ident);
+ } catch (EntityAlreadyExistsException e) {
+ throw new IllegalArgumentException("Failed to repair table " + ident, e);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to repair table " + ident, e);
+ }
+ }
+
+ private Table recordCheckedEmptyVersion(NameIdentifier ident, long
datasetVersion) {
+ try {
+ TableEntity tableEntity =
+ store.update(
+ ident,
+ TableEntity.class,
+ Entity.EntityType.TABLE,
+ current -> {
+ if (!isDatasetVersionChanged(current.properties(),
datasetVersion)) {
+ return current;
+ }
+ Map<String, String> updatedProperties = new
HashMap<>(current.properties());
+ updatedProperties.put(
+ LanceConstants.LANCE_TABLE_VERSION,
String.valueOf(datasetVersion));
+ // Always use an empty column list: the dataset is confirmed
empty at this version.
+ // Using current.columns() would preserve stale columns when
the dataset schema was
+ // cleared externally, causing future VERSION_CHECK loads to
return stale metadata
+ // permanently (the version sentinel would match but columns
would be wrong).
+ return TableEntity.builder()
+ .withId(current.id())
+ .withName(current.name())
+ .withNamespace(current.namespace())
+ .withComment(current.comment())
+ .withColumns(List.of())
+ .withProperties(updatedProperties)
+ .withPartitioning(current.partitioning())
+ .withDistribution(current.distribution())
+ .withSortOrders(current.sortOrders())
+ .withIndexes(current.indexes())
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(current.auditInfo().creator())
+ .withCreateTime(current.auditInfo().createTime())
+
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
+ .withLastModifiedTime(Instant.now())
+ .build())
+ .build();
+ });
+ return toGenericTable(tableEntity);
+ } catch (NoSuchEntityException e) {
+ throw new NoSuchTableException(e, "Table %s does not exist", ident);
+ } catch (EntityAlreadyExistsException e) {
+ throw new IllegalArgumentException("Failed to record empty version for
table " + ident, e);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to record empty version for table " +
ident, e);
+ }
+ }
+
+ private boolean needsSchemaRefresh(TableEntity tableEntity, long
datasetVersion) {
+ return isDeclaredOnly(tableEntity.properties())
+ || tableEntity.columns().isEmpty()
+ || isDatasetVersionChanged(tableEntity.properties(), datasetVersion);
+ }
+
+ private TableEntity replaceColumnsFromDataset(
+ TableEntity tableEntity, Column[] columns, long datasetVersion) {
+ Map<String, String> updatedProperties = new
HashMap<>(tableEntity.properties());
+ updatedProperties.put(LanceConstants.LANCE_TABLE_VERSION,
String.valueOf(datasetVersion));
+ updatedProperties.remove(LanceConstants.LANCE_TABLE_DECLARED);
+
+ AuditInfo columnAuditInfo =
+ AuditInfo.builder()
+ .withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+ .withCreateTime(Instant.now())
+ .build();
+ List<ColumnEntity> columnEntities =
+ IntStream.range(0, columns.length)
+ .mapToObj(
+ i ->
+ ColumnEntity.toColumnEntity(
+ columns[i], i, idGenerator.nextId(), columnAuditInfo))
+ .collect(Collectors.toList());
+
+ return TableEntity.builder()
+ .withId(tableEntity.id())
+ .withName(tableEntity.name())
+ .withNamespace(tableEntity.namespace())
+ .withComment(tableEntity.comment())
+ .withColumns(columnEntities)
+ .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(Instant.now())
+ .build())
+ .build();
+ }
+
+ private GenericTable toGenericTable(TableEntity tableEntity) {
+ return GenericTable.builder()
+ .withName(tableEntity.name())
+ .withComment(tableEntity.comment())
+ .withColumns(
+
tableEntity.columns().stream().map(this::toGenericColumn).toArray(Column[]::new))
+ .withProperties(tableEntity.properties())
+ .withAuditInfo(tableEntity.auditInfo())
+ .withSortOrders(tableEntity.sortOrders())
+ .withPartitioning(tableEntity.partitioning())
+ .withDistribution(tableEntity.distribution())
+ .withIndexes(tableEntity.indexes())
+ .build();
+ }
+
+ private GenericColumn toGenericColumn(ColumnEntity columnEntity) {
+ return GenericColumn.builder()
+ .withName(columnEntity.name())
+ .withComment(columnEntity.comment())
+ .withAutoIncrement(columnEntity.autoIncrement())
+ .withNullable(columnEntity.nullable())
+ .withType(columnEntity.dataType())
+ .withDefaultValue(columnEntity.defaultValue())
+ .build();
}
/**
@@ -435,8 +726,10 @@ public class LanceTableOperations extends
ManagedTableOperations {
}
Dataset openDataset(String location, Map<String, String> storageOptions) {
+ // Do not pass an explicit allocator: OpenDatasetBuilder then sets
selfManagedAllocator=true
+ // so Dataset.close() will release the off-heap memory. Passing new
RootAllocator() sets
+ // selfManagedAllocator=false and leaks the allocator on every call.
return Dataset.open()
- .allocator(new RootAllocator())
.uri(location)
.readOptions(new
ReadOptions.Builder().setStorageOptions(storageOptions).build())
.build();
diff --git
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestPropertiesMetadata.java
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestPropertiesMetadata.java
index 5371a63f5a..a57474be2e 100644
---
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestPropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestPropertiesMetadata.java
@@ -22,6 +22,7 @@ package org.apache.gravitino.catalog.lakehouse.generic;
import com.google.common.collect.ImmutableMap;
import java.util.Map;
import org.apache.gravitino.Schema;
+import org.apache.gravitino.catalog.lakehouse.lance.LanceTableOperations;
import org.apache.gravitino.connector.PropertiesMetadata;
import org.apache.gravitino.rel.Table;
import org.junit.jupiter.api.Assertions;
@@ -44,7 +45,8 @@ public class TestPropertiesMetadata {
Map<String, String> catalogProperties =
ImmutableMap.of(
"location", "/tmp/test1",
- "lance.storage.endpoint", "http://minio:9000";);
+ "lance.storage.endpoint", "http://minio:9000";,
+ "lance.schema-refresh-mode", "VERSION_CHECK");
String catalogLocation =
(String)
@@ -55,6 +57,10 @@ public class TestPropertiesMetadata {
Assertions.assertEquals(
"http://minio:9000";,
catalogPropertiesMetadata.getOrDefault(catalogProperties,
"lance.storage.endpoint"));
+
Assertions.assertTrue(catalogPropertiesMetadata.containsProperty("lance.schema-refresh-mode"));
+ Assertions.assertEquals(
+ LanceTableOperations.SchemaRefreshMode.VERSION_CHECK,
+ catalogPropertiesMetadata.getOrDefault(catalogProperties,
"lance.schema-refresh-mode"));
}
@Test
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 7362f903cb..59b2a41ff9 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
@@ -19,18 +19,37 @@
package org.apache.gravitino.catalog.lakehouse.lance;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_CREATION_MODE;
+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_VERSION;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.collect.Maps;
+import java.time.Instant;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.UserPrincipal;
import org.apache.gravitino.catalog.ManagedSchemaOperations;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.ColumnEntity;
+import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.TableChange;
@@ -39,6 +58,7 @@ import
org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.rel.types.Types;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.utils.PrincipalUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -100,6 +120,274 @@ public class TestLanceTableOperations {
new Index[0]));
}
+ @Test
+ public void testLoadDeclaredTableSchemaFromLocation() throws Exception {
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("declared-table").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(),
+ Map.of(
+ Table.PROPERTY_LOCATION,
+ location,
+ LANCE_TABLE_DECLARED,
+ "true",
+ LANCE_STORAGE_OPTIONS_PREFIX + "endpoint",
+ "http://endpoint";));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(idGenerator.nextId()).thenReturn(10L, 11L);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema())
+ .thenReturn(
+ new Schema(
+ List.of(
+ Field.nullable("id", new ArrowType.Int(32, true)),
+ Field.nullable("name", new ArrowType.Utf8()))));
+ when(dataset.version()).thenReturn(8L);
+ Mockito.doReturn(dataset)
+ .when(lanceTableOps)
+ .openDataset(location, Map.of("endpoint", "http://endpoint";));
+
+ Table loadedTable =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ Assertions.assertEquals(2, loadedTable.columns().length);
+ Assertions.assertEquals("id", loadedTable.columns()[0].name());
+ Assertions.assertEquals(Types.IntegerType.get(),
loadedTable.columns()[0].dataType());
+ Assertions.assertEquals("name", loadedTable.columns()[1].name());
+ Assertions.assertEquals(Types.StringType.get(),
loadedTable.columns()[1].dataType());
+ Assertions.assertEquals("8",
loadedTable.properties().get(LANCE_TABLE_VERSION));
+
Assertions.assertFalse(loadedTable.properties().containsKey(LANCE_TABLE_DECLARED));
+ }
+
+ @Test
+ public void testLoadTableWithStoredColumnsDoesNotReadLocation() throws
Exception {
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("normal-table").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(10L)
+ .withName("id")
+ .withDataType(Types.IntegerType.get())
+ .withPosition(0)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build()),
+ Map.of(Table.PROPERTY_LOCATION, location));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+
+ Table loadedTable = lanceTableOps.loadTable(ident);
+
+ Assertions.assertEquals(1, loadedTable.columns().length);
+ Assertions.assertEquals("id", loadedTable.columns()[0].name());
+ verify(lanceTableOps, never()).openDataset(anyString(), any());
+ }
+
+ @Test
+ public void testVersionCheckRefreshesSchemaFromLocation() throws Exception {
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("version-check-table").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(10L)
+ .withName("old_col")
+ .withDataType(Types.StringType.get())
+ .withPosition(0)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build()),
+ Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_VERSION,
"8"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(idGenerator.nextId()).thenReturn(11L, 12L);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema())
+ .thenReturn(
+ new Schema(
+ List.of(
+ Field.nullable("id", new ArrowType.Int(32, true)),
+ Field.nullable("name", new ArrowType.Utf8()))));
+ when(dataset.version()).thenReturn(9L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loadedTable =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ Assertions.assertEquals(2, loadedTable.columns().length);
+ Assertions.assertEquals("id", loadedTable.columns()[0].name());
+ Assertions.assertEquals("name", loadedTable.columns()[1].name());
+ Assertions.assertEquals("9",
loadedTable.properties().get(LANCE_TABLE_VERSION));
+ }
+
+ @Test
+ public void testVersionCheckSkipsRefreshWhenVersionIsCurrent() throws
Exception {
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("current-version-table").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(10L)
+ .withName("id")
+ .withDataType(Types.IntegerType.get())
+ .withPosition(0)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build()),
+ Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_VERSION,
"9"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.version()).thenReturn(9L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loadedTable = lanceTableOps.loadTable(ident);
+
+ Assertions.assertEquals(1, loadedTable.columns().length);
+ Assertions.assertEquals("id", loadedTable.columns()[0].name());
+ verify(dataset, never()).getSchema();
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ @Test
+ public void
testVersionCheckRefreshIsIdempotentWhenCurrentEntityWasAlreadyRepaired()
+ throws Exception {
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location =
tempDir.resolve("concurrent-version-check-table").toString();
+ TableEntity staleTableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(10L)
+ .withName("old_col")
+ .withDataType(Types.StringType.get())
+ .withPosition(0)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build()),
+ Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_VERSION,
"8"));
+ TableEntity alreadyRepairedTableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(11L)
+ .withName("id")
+ .withDataType(Types.IntegerType.get())
+ .withPosition(0)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build(),
+ ColumnEntity.builder()
+ .withId(12L)
+ .withName("name")
+ .withDataType(Types.StringType.get())
+ .withPosition(1)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build()),
+ Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_VERSION,
"9"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(staleTableEntity);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(alreadyRepairedTableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema())
+ .thenReturn(
+ new Schema(
+ List.of(
+ Field.nullable("id", new ArrowType.Int(32, true)),
+ Field.nullable("name", new ArrowType.Utf8()))));
+ when(dataset.version()).thenReturn(9L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loadedTable =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ Assertions.assertEquals(2, loadedTable.columns().length);
+ Assertions.assertEquals("id", loadedTable.columns()[0].name());
+ Assertions.assertEquals("name", loadedTable.columns()[1].name());
+ Assertions.assertEquals("9",
loadedTable.properties().get(LANCE_TABLE_VERSION));
+ }
+
+ @Test
+ public void testAlterTablePersistsUpdatedLanceVersion() throws Exception {
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("alter-table").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(10L)
+ .withName("id")
+ .withDataType(Types.IntegerType.get())
+ .withPosition(0)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build()),
+ Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_VERSION,
"8"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ AtomicReference<TableEntity> storedTable = new AtomicReference<>();
+ 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(tableEntity);
+ storedTable.set(updated);
+ return updated;
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ Version version = mock(Version.class);
+ when(dataset.getVersion()).thenReturn(version);
+ when(version.getId()).thenReturn(9L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table alteredTable =
+ PrincipalUtils.doAs(
+ new UserPrincipal("tester"),
+ () ->
+ lanceTableOps.alterTable(
+ ident, TableChange.deleteColumn(new String[] {"id"},
false)));
+
+ Assertions.assertEquals("9",
alteredTable.properties().get(LANCE_TABLE_VERSION));
+ Assertions.assertEquals("9",
storedTable.get().properties().get(LANCE_TABLE_VERSION));
+ }
+
@Test
public void testHandleLanceTableChangeRespectsOrder() {
Table table = mock(Table.class);
@@ -168,4 +456,417 @@ public class TestLanceTableOperations {
Mockito.verify(dataset).dropColumns(anyList());
Mockito.verify(dataset).getVersion();
}
+
+ @Test
+ public void
testVersionCheckSkipsSchemaReadForEmptySchemaWhenVersionUnchanged() throws
Exception {
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("empty-version-check").toString();
+ // Table has empty columns but a known stored version (confirmed-empty
state)
+ TableEntity tableEntity =
+ tableEntity(
+ ident, List.of(), Map.of(Table.PROPERTY_LOCATION, location,
LANCE_TABLE_VERSION, "5"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.version()).thenReturn(5L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded = lanceTableOps.loadTable(ident);
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ // version unchanged — schema read must be skipped even though columns are
empty
+ verify(dataset, never()).getSchema();
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ @Test
+ public void testEmptyDatasetRecordsVersionOnFirstLoad() throws Exception {
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("empty-dataset-first").toString();
+ TableEntity tableEntity =
+ tableEntity(ident, List.of(), Map.of(Table.PROPERTY_LOCATION,
location));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema()).thenReturn(new Schema(List.of()));
+ when(dataset.version()).thenReturn(3L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ Assertions.assertEquals("3", loaded.properties().get(LANCE_TABLE_VERSION));
+ verify(store).update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any());
+ }
+
+ @Test
+ public void testEmptyDatasetSkipsOpenWhenVersionAlreadyRecorded() throws
Exception {
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("empty-dataset-second").toString();
+ // Simulate the table after first-load recorded lance.version=3 but
columns still empty
+ TableEntity tableEntity =
+ tableEntity(
+ ident, List.of(), Map.of(Table.PROPERTY_LOCATION, location,
LANCE_TABLE_VERSION, "3"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+
+ Table loaded = lanceTableOps.loadTable(ident);
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ verify(lanceTableOps, never()).openDataset(anyString(), any());
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ //
---------------------------------------------------------------------------
+ // VERSION_CHECK: dataset becomes empty while stored columns are non-empty
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void testVersionCheckClearsStaleColumnsWhenDatasetBecomesEmpty()
throws Exception {
+ // Regression test for: VERSION_CHECK + stored columns non-empty + dataset
schema becomes empty.
+ // recordCheckedEmptyVersion must clear the stale stored columns (not
preserve them via
+ // current.columns()), otherwise the version sentinel locks in permanently
stale metadata.
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location =
tempDir.resolve("stale-columns-empty-dataset").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(10L)
+ .withName("id")
+ .withDataType(Types.IntegerType.get())
+ .withPosition(0)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build(),
+ ColumnEntity.builder()
+ .withId(11L)
+ .withName("name")
+ .withDataType(Types.StringType.get())
+ .withPosition(1)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build()),
+ Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_VERSION,
"8"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema()).thenReturn(new Schema(List.of()));
+ when(dataset.version()).thenReturn(9L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ // Stale columns must be cleared; storing [id, name] here would
permanently lock them in.
+ Assertions.assertEquals(0, loaded.columns().length);
+ // The new dataset version must be persisted so future VERSION_CHECK loads
see the match.
+ Assertions.assertEquals("9", loaded.properties().get(LANCE_TABLE_VERSION));
+ }
+
+ @Test
+ public void testVersionCheckStaleColumnsAreNotReturnedOnSubsequentLoad()
throws Exception {
+ // After the fix: once stale columns are cleared and version=9 is
recorded, the next loadTable
+ // must early-return with empty columns (not re-open the dataset and not
return stale data).
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("subsequent-empty-load").toString();
+ // Simulate the store state after the first load cleared stale columns.
+ TableEntity tableEntity =
+ tableEntity(
+ ident, List.of(), Map.of(Table.PROPERTY_LOCATION, location,
LANCE_TABLE_VERSION, "9"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.version()).thenReturn(9L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded = lanceTableOps.loadTable(ident);
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ // Version matched: schema read must be skipped entirely.
+ verify(dataset, never()).getSchema();
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ //
---------------------------------------------------------------------------
+ // VERSION_CHECK: empty stored columns
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void testVersionCheckFirstLoadEmptyStoredColumnsEmptyDataset() throws
Exception {
+ // VERSION_CHECK + no stored version + empty stored columns + dataset is
also empty.
+ // Should open dataset, read schema (empty), and record version without
creating any columns.
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("vc-empty-first-load").toString();
+ TableEntity tableEntity =
+ tableEntity(ident, List.of(), Map.of(Table.PROPERTY_LOCATION,
location));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema()).thenReturn(new Schema(List.of()));
+ when(dataset.version()).thenReturn(5L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ Assertions.assertEquals("5", loaded.properties().get(LANCE_TABLE_VERSION));
+ verify(store).update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any());
+ }
+
+ @Test
+ public void
testVersionCheckEmptyStoredColumnsVersionBumpedDatasetStillEmpty() throws
Exception {
+ // VERSION_CHECK + stored empty columns with version=5 + dataset bumped to
version=6 (still
+ // empty). Should detect version change, read schema (empty), and update
version to 6.
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("vc-empty-version-bump").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident, List.of(), Map.of(Table.PROPERTY_LOCATION, location,
LANCE_TABLE_VERSION, "5"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema()).thenReturn(new Schema(List.of()));
+ when(dataset.version()).thenReturn(6L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ Assertions.assertEquals("6", loaded.properties().get(LANCE_TABLE_VERSION));
+ }
+
+ //
---------------------------------------------------------------------------
+ // DECLARED_AND_EMPTY: non-declared empty stored columns with real dataset
schema
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void
testDeclaredAndEmptyRepairsNonDeclaredEmptyStoredColumnsFromRealDataset()
+ throws Exception {
+ // DECLARED_AND_EMPTY + non-declared table + empty stored columns (no
version) + dataset has a
+ // real schema. loadTable should open the dataset, read the schema, and
persist the columns.
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location =
tempDir.resolve("dae-empty-stored-real-schema").toString();
+ TableEntity tableEntity =
+ tableEntity(ident, List.of(), Map.of(Table.PROPERTY_LOCATION,
location));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(idGenerator.nextId()).thenReturn(20L, 21L);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema())
+ .thenReturn(
+ new Schema(
+ List.of(
+ Field.nullable("col_a", new ArrowType.Int(64, true)),
+ Field.nullable("col_b", new ArrowType.Bool()))));
+ when(dataset.version()).thenReturn(7L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ Assertions.assertEquals(2, loaded.columns().length);
+ Assertions.assertEquals("col_a", loaded.columns()[0].name());
+ Assertions.assertEquals("col_b", loaded.columns()[1].name());
+ Assertions.assertEquals("7", loaded.properties().get(LANCE_TABLE_VERSION));
+ }
+
+ //
---------------------------------------------------------------------------
+ // Edge cases: no location, dataset open failure
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void testLoadTableWithNoLocationReturnsStoredMetadata() throws
Exception {
+ // A table without a PROPERTY_LOCATION must return stored metadata
immediately without
+ // attempting to open any dataset, even in VERSION_CHECK mode.
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ // Declared table with empty columns and no location triggers the
schema-refresh branch.
+ TableEntity tableEntity = tableEntity(ident, List.of(),
Map.of(LANCE_TABLE_DECLARED, "true"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+
+ Table loaded = lanceTableOps.loadTable(ident);
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ verify(lanceTableOps, never()).openDataset(anyString(), any());
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ @Test
+ public void testLoadTableFallsBackToStoredMetadataWhenDatasetOpenFails()
throws Exception {
+ // If the Lance dataset cannot be opened (e.g. storage not accessible),
loadTable must return
+ // the stored metadata rather than propagating the exception.
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("broken-dataset").toString();
+ TableEntity tableEntity =
+ tableEntity(ident, List.of(), Map.of(Table.PROPERTY_LOCATION,
location));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ Mockito.doThrow(new RuntimeException("storage unavailable"))
+ .when(lanceTableOps)
+ .openDataset(eq(location), any());
+
+ Table loaded = lanceTableOps.loadTable(ident);
+
+ Assertions.assertEquals(0, loaded.columns().length);
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ //
---------------------------------------------------------------------------
+ // Declared table: dataset open but no recordCheckedEmptyVersion
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void testDeclaredTableWithEmptyDatasetDoesNotRecordVersion() throws
Exception {
+ // Declared tables use lance.declared=true as the "not yet written"
signal. When the dataset is
+ // empty, the caller should NOT record the checked version (that would
advance the version
+ // sentinel while the declared flag is still present). The returned table
must stay unchanged.
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("declared-empty-dataset").toString();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(),
+ Map.of(Table.PROPERTY_LOCATION, location, LANCE_TABLE_DECLARED,
"true"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema()).thenReturn(new Schema(List.of()));
+ when(dataset.version()).thenReturn(3L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ // lance.declared is still present (no schema written yet), no version
recorded.
+ Assertions.assertTrue(
+ Boolean.parseBoolean(loaded.properties().get(LANCE_TABLE_DECLARED)),
+ "lance.declared must still be set");
+ Assertions.assertNull(
+ loaded.properties().get(LANCE_TABLE_VERSION), "lance.version must not
be recorded");
+ verify(store, never())
+ .update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
+ }
+
+ //
---------------------------------------------------------------------------
+ // VERSION_CHECK: declared table always refreshes even when version matches
+ //
---------------------------------------------------------------------------
+
+ @Test
+ public void
testVersionCheckDeclaredTableAlwaysRefreshesDespiteVersionMatch() throws
Exception {
+ // In VERSION_CHECK mode the early-return is gated on !declaredOnly, so
declared tables must
+ // always open the dataset and repair their schema regardless of the
stored lance.version.
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("vc-declared-version-match").toString();
+ // Stored version already matches the dataset version.
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(),
+ Map.of(
+ Table.PROPERTY_LOCATION,
+ location,
+ LANCE_TABLE_DECLARED,
+ "true",
+ LANCE_TABLE_VERSION,
+ "9"));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(tableEntity);
+ when(idGenerator.nextId()).thenReturn(30L);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<TableEntity, TableEntity> updater =
invocation.getArgument(3);
+ return updater.apply(tableEntity);
+ });
+
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema())
+ .thenReturn(new Schema(List.of(Field.nullable("id", new
ArrowType.Int(32, true)))));
+ when(dataset.version()).thenReturn(9L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+
+ Table loaded =
+ PrincipalUtils.doAs(new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident));
+
+ // Schema must be read and persisted — declared tables bypass the version
early-return.
+ Assertions.assertEquals(1, loaded.columns().length);
+ Assertions.assertEquals("id", loaded.columns()[0].name());
+ Assertions.assertFalse(
+ loaded.properties().containsKey(LANCE_TABLE_DECLARED),
+ "lance.declared must be removed after schema is written");
+ verify(dataset).getSchema();
+ }
+
+ private static TableEntity tableEntity(
+ NameIdentifier ident, List<ColumnEntity> columns, Map<String, String>
properties) {
+ return TableEntity.builder()
+ .withId(1L)
+ .withName(ident.name())
+ .withNamespace(ident.namespace())
+ .withComment("comment")
+ .withColumns(columns)
+ .withProperties(properties)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.EPOCH).build())
+ .build();
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
index c2646f4b40..f6ef8d9c3b 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
@@ -886,7 +886,13 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
.withId(entity.id())
.withName(entity.name())
.withNamespace(entity.namespace())
+ .withComment(entity.comment())
+ .withProperties(entity.properties())
.withColumns(columnsUpdateResult.getRight())
+ .withPartitioning(entity.partitioning())
+ .withDistribution(entity.distribution())
+ .withSortOrders(entity.sortOrders())
+ .withIndexes(entity.indexes())
.withAuditInfo(
AuditInfo.builder()
.withCreator(entity.auditInfo().creator())
diff --git a/docs/lakehouse-generic-catalog.md
b/docs/lakehouse-generic-catalog.md
index 0f24628c55..263b552c88 100644
--- a/docs/lakehouse-generic-catalog.md
+++ b/docs/lakehouse-generic-catalog.md
@@ -44,10 +44,11 @@ For detailed information on available operations, see
[Manage Relational Metadat
### Catalog Properties
-| Property | Description | Example
| Required | Since Version |
-|------------|----------------------------------------------|-------------------------|----------|---------------|
-| `provider` | Catalog provider type |
`lakehouse-generic` | Yes | 1.1.0 |
-| `location` | Root storage path for all schemas and tables |
`s3://bucket/lakehouse` | No | 1.1.0 |
+| Property | Description
| Example | Required | Since Version |
+|-----------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------|----------|---------------|
+| `provider` | Catalog provider type
| `lakehouse-generic` | Yes | 1.1.0 |
+| `location` | Root storage path for all schemas and tables
| `s3://bucket/lakehouse` | No | 1.1.0 |
+| `lance.schema-refresh-mode` | Lance table schema refresh mode.
`DECLARED_AND_EMPTY` (default) refreshes declared tables and tables with empty
stored columns. `VERSION_CHECK` additionally refreshes when the Lance dataset
version changes. | `DECLARED_AND_EMPTY` | No | 1.3.0 |
#### Key Property: `location`
@@ -199,4 +200,4 @@ For additional operations, refer to [Schema Operations
documentation](./manage-r
Since different lakehouse table formats have varying capabilities, table
operation support may differ. The following are table operations for different
lakehouse formats:
-- [Lance Format Support](./lakehouse-generic-lance-table.md)
\ No newline at end of file
+- [Lance Format Support](./lakehouse-generic-lance-table.md)
diff --git a/docs/lakehouse-generic-lance-table.md
b/docs/lakehouse-generic-lance-table.md
index 6e88c301a8..cdbd0bb79a 100644
--- a/docs/lakehouse-generic-lance-table.md
+++ b/docs/lakehouse-generic-lance-table.md
@@ -101,14 +101,14 @@ For Arrow types not natively mapped in Gravitino, use the
`External(arrow_field_
Required and optional properties for tables in a Generic Lakehouse Catalog:
-| Property | Description
| Default | Required | Since
Version |
-|-----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------------|---------------|
-| `format` | Table format: `lance`, only `lance` is fully
supported.
| (none) | Yes |
1.1.0 |
-| `location` | Storage path for table metadata and data, Lance
supports: S3, GCS, OSS, AZ, File, Memory and file-object-store.
| (none) | Conditional* |
1.1.0 |
-| `external` | Whether the data directory is an external location.
If it's `true`, dropping a table will only remove metadata in Gravitino and
will not delete the data directory, and purge table will delete both. For a
non-external table, dropping will drop both.
| false | No |
1.1.0 |
-| `lance.creation-mode` | Create mode: for create table, it can be `CREATE`,
`EXIST_OK` or `OVERWRITE`. and it should be `CREATE` or `OVERWRITE` for
registering tables
| `CREATE` | No
| 1.1.0 |
-| `lance.register` | Whether it is a register table operation. If it's
`true`, This API will not create data directory actually and it's the user's
responsibility to create and manage the data directory. `false` it will
actually create a table.
| false | No
| 1.1.0 |
-| `lance.storage.xxxx` | Any additional storage-specific properties required
by Lance format (e.g., S3 credentials, HDFS configs). Replace `xxxx` with
actual property names. For example, we can use
`lance.storage.aws_access_key_id` to set S3 aws_access_key_id when using a S3
location, for detail, refer to https://lancedb.com/docs/storage/integrations/
| (none) | No | 1.1.0 |
+| Property | Description
| Default | Required | Since Version |
+|-----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------------|---------------|
+| `format` | Table format: `lance`, only `lance` is fully
supported.
| (none) | Yes | 1.1.0
|
+| `location` | Storage path for table metadata and data, Lance
supports: S3, GCS, OSS, AZ, File, Memory and file-object-store.
| (none) | Conditional* | 1.1.0
|
+| `external` | Whether the data directory is an external location.
If it's `true`, dropping a table will only remove metadata in Gravitino and
will not delete the data directory, and purge table will delete both. For a
non-external table, dropping will drop both.
| false | No | 1.1.0
|
+| `lance.creation-mode` | Create mode: for create table, it can be `CREATE`,
`EXIST_OK` or `OVERWRITE`. and it should be `CREATE` or `OVERWRITE` for
registering tables
| `CREATE` | No |
1.1.0 |
+| `lance.register` | Whether it is a register table operation. If it's
`true`, This API will not create data directory actually and it's the user's
responsibility to create and manage the data directory. `false` it will
actually create a table.
| false | No |
1.1.0 |
+| `lance.storage.xxxx` | Any additional storage-specific properties required
by Lance format (e.g., S3 credentials, HDFS configs). Replace `xxxx` with
actual property names. For example, we can use
`lance.storage.aws_access_key_id` to set S3 aws_access_key_id when using a S3
location, for detail, refer to https://lancedb.com/docs/storage/integrations/
| (none) | No | 1.1.0 |
- `CREATE`: Create a new table, fail if the table already exists.
- `EXIST_OK`: Create a new table if it does not exist, otherwise do nothing.
@@ -118,6 +118,27 @@ Required and optional properties for tables in a Generic
Lakehouse Catalog:
Also set additional properties specific to your lakehouse format or custom
requirements.
+### Schema Refresh
+
+For Lance tables, Gravitino stores table columns in its metadata store. Some
Lance writers can also
+update the dataset directly at the Lance location. To keep Gravitino metadata
in sync, the Generic
+Lakehouse catalog supports catalog-level schema refresh modes:
+
+| Mode | Behavior
|
+|-----------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `DECLARED_AND_EMPTY` | Default. Refreshes schema from the Lance dataset for
two cases: (1) declared tables (`lance.declared=true`) whose schema has not yet
been written to Gravitino; (2) tables whose Gravitino column list is empty, for
example tables registered before their schema was captured. |
+| `VERSION_CHECK` | Opens the Lance dataset on every `loadTable`,
compares the dataset version with `lance.version`, and refreshes columns when
the version has changed.
|
+
+Use `VERSION_CHECK` only when tables may be modified directly through the
Lance path outside
+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.
+:::
+
### Table Operations
Table operations follow standard relational catalog patterns. See [Table
Operations](./manage-relational-metadata-using-gravitino.md#table-operations)
for comprehensive documentation.
diff --git
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/LanceTableOperations.java
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/LanceTableOperations.java
index 6bdfcb5054..41774fdc96 100644
---
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/LanceTableOperations.java
+++
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/LanceTableOperations.java
@@ -36,10 +36,16 @@ public interface LanceTableOperations {
* @param delimiter the delimiter used in the namespace
* @param version the version of the table to describe, if null, describe
the latest version
* @param checkDeclared whether to populate the is_only_declared response
field
+ * @param loadDetailedMetadata whether to include column schema in the
response; when false the
+ * schema field is omitted for a lightweight existence/property check
* @return the table description
*/
DescribeTableResponse describeTable(
- String tableId, String delimiter, Optional<Long> version, boolean
checkDeclared);
+ String tableId,
+ String delimiter,
+ Optional<Long> version,
+ boolean checkDeclared,
+ boolean loadDetailedMetadata);
/**
* Create a new table.
diff --git
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
index 21a53a8600..5321b6bb74 100644
---
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
+++
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
@@ -99,7 +99,11 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
@Override
public DescribeTableResponse describeTable(
- String tableId, String delimiter, Optional<Long> version, boolean
checkDeclared) {
+ String tableId,
+ String delimiter,
+ Optional<Long> version,
+ boolean checkDeclared,
+ boolean loadDetailedMetadata) {
if (!version.isEmpty()) {
throw new UnsupportedOperationException(
"Describing specific table version is not supported. It should be
null to indicate the"
@@ -126,7 +130,9 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
response.setMetadata(table.properties());
response.setProperties(table.properties());
response.setLocation(table.properties().get(LANCE_LOCATION));
- response.setSchema(toJsonArrowSchema(table.columns()));
+ if (loadDetailedMetadata) {
+ response.setSchema(toJsonArrowSchema(table.columns()));
+ }
response.setVersion(
Optional.ofNullable(table.properties().get(LANCE_TABLE_VERSION))
.map(Long::valueOf)
diff --git
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/LanceConstants.java
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/LanceConstants.java
index f5faaad1c2..197388e8d4 100644
---
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/LanceConstants.java
+++
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/LanceConstants.java
@@ -40,6 +40,7 @@ public class LanceConstants {
public static final String LANCE_TABLE_VERSION = "lance.version";
// Mark whether the table is declared only in metadata without creating a
Lance dataset.
public static final String LANCE_TABLE_DECLARED = "lance.declared";
+ public static final String LANCE_SCHEMA_REFRESH_MODE =
"lance.schema-refresh-mode";
public static final String LANCE_TABLE_FORMAT = "lance";
}
diff --git
a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/rest/LanceTableOperations.java
b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/rest/LanceTableOperations.java
index 5114b475c3..5279f9cadd 100644
---
a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/rest/LanceTableOperations.java
+++
b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/rest/LanceTableOperations.java
@@ -96,10 +96,15 @@ public class LanceTableOperations {
boolean shouldCheckDeclared =
Optional.ofNullable(checkDeclared)
.orElse(request != null &&
Boolean.TRUE.equals(request.getCheckDeclared()));
+ // loadDetailedMetadata defaults to true when absent so callers that
omit the field
+ // still receive the full schema in the response.
+ boolean shouldLoadDetailedMetadata =
+ request == null ||
!Boolean.FALSE.equals(request.getLoadDetailedMetadata());
DescribeTableResponse response =
lanceNamespace
.asTableOps()
- .describeTable(tableId, delimiter, version, shouldCheckDeclared);
+ .describeTable(
+ tableId, delimiter, version, shouldCheckDeclared,
shouldLoadDetailedMetadata);
return Response.ok(response).build();
} catch (Exception e) {
return LanceExceptionMapper.toRESTResponse(tableId, e);
diff --git
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
index cadfccebdf..d27799c1f8 100644
---
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
+++
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
@@ -215,7 +215,7 @@ class TestGravitinoLanceModeParsing {
GravitinoLanceTableOperations operations =
newTableOperations(tableCatalog);
DescribeTableResponse response =
- operations.describeTable("catalog.schema.table", ".",
Optional.empty(), true);
+ operations.describeTable("catalog.schema.table", ".",
Optional.empty(), true, true);
Assertions.assertEquals(Boolean.TRUE, response.getIsOnlyDeclared());
Assertions.assertEquals(Boolean.FALSE, response.getManagedVersioning());
diff --git
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
index cebf03abc1..3704247919 100644
---
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
+++
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
@@ -182,6 +182,10 @@ public class LanceSparkRESTServiceIT extends BaseIT {
Table table = catalog.asTableCatalog().loadTable(tableIdentifier);
assertTableLocationAndFormat(table, schemaName, tableName);
+ Assertions.assertEquals(2, table.columns().length);
+ Assertions.assertEquals(
+ Set.of("id", "score"),
+ Arrays.stream(table.columns()).map(column ->
column.name()).collect(Collectors.toSet()));
}
@Test
diff --git
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java
index a7ab97120a..a80c70a06b 100644
---
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java
+++
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java
@@ -633,7 +633,8 @@ public class TestLanceNamespaceOperations extends
JerseyTest {
DescribeTableResponse createTableResponse = new DescribeTableResponse();
createTableResponse.setLocation("/path/to/describe_table");
createTableResponse.setMetadata(ImmutableMap.of("key", "value"));
- when(tableOps.describeTable(any(), any(), any(),
anyBoolean())).thenReturn(createTableResponse);
+ when(tableOps.describeTable(any(), any(), any(), anyBoolean(),
anyBoolean()))
+ .thenReturn(createTableResponse);
DescribeTableRequest tableRequest = new DescribeTableRequest();
Response resp =
@@ -649,11 +650,11 @@ public class TestLanceNamespaceOperations extends
JerseyTest {
Assertions.assertEquals(createTableResponse.getLocation(),
response.getLocation());
Assertions.assertEquals(createTableResponse.getMetadata(),
response.getMetadata());
Mockito.verify(tableOps)
- .describeTable(eq(tableIds), eq(delimiter), eq(Optional.empty()),
eq(true));
+ .describeTable(eq(tableIds), eq(delimiter), eq(Optional.empty()),
eq(true), eq(true));
// Test not found exception
Mockito.reset(tableOps);
- when(tableOps.describeTable(any(), any(), any(), anyBoolean()))
+ when(tableOps.describeTable(any(), any(), any(), anyBoolean(),
anyBoolean()))
.thenThrow(new TableNotFoundException("Table not found", "",
tableIds));
resp =
target(String.format("/v1/table/%s/describe", tableIds))
@@ -664,7 +665,7 @@ public class TestLanceNamespaceOperations extends
JerseyTest {
// Test runtime exception
Mockito.reset(tableOps);
- when(tableOps.describeTable(any(), any(), any(), anyBoolean()))
+ when(tableOps.describeTable(any(), any(), any(), anyBoolean(),
anyBoolean()))
.thenThrow(new RuntimeException("Runtime exception"));
resp =
target(String.format("/v1/table/%s/describe", tableIds))
