This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 44846562db [#13309][#13310] fix(core): Keep column ids stable and pass
the new schema on cross-schema table rename (#13316)
44846562db is described below
commit 44846562dbd28319ccdfe19ed32b1d549d901b4f
Author: Jerry Shao <[email protected]>
AuthorDate: Fri Sep 18 22:15:44 2026 +0800
[#13309][#13310] fix(core): Keep column ids stable and pass the new schema
on cross-schema table rename (#13316)
### What changes were proposed in this pull request?
Tags, owners and privileges are attached to a column by its id. This PR
fixes three paths that gave existing columns new ids or left column rows
behind, so column-level metadata was lost. It also passes the right
schema to authorization plugins on a cross-schema table rename.
**Cross-schema table move with a column change (#13309)**
- `TableColumnMetaService.updateColumnPOsFromTableDiff` now moves all of
the table's column rows to the new `schema_id` whenever the table moves.
Before, it did this only when no column changed, so a move plus a column
change left the unchanged columns under the old schema, and a later
cascade drop of that schema deleted them.
**Re-import (#13309)**
- `TableMetaService.insertTable(overwrite = true)` reuses the stored
column ids by name, but only when the upsert kept the same table id and
that table already existed. This is the case when a table is
re-imported, e.g. loaded after it was renamed outside Gravitino.
- A stored id is never given to a second column. If legacy data holds
two live columns with one name, the most recently written one keeps its
id.
- The tag and owner relations of stored columns that are gone are
removed in the same transaction.
- When the upsert resolves to another table's row, which MySQL/H2 can do
through the `(schema_id, table_name)` key, no ids are reused, so that
table's columns are not inherited.
**Lance `VERSION_CHECK` schema refresh (#13309)**
- `LanceTableOperations.replaceColumnsFromDataset` reuses the id,
comment and audit info of each column that is still in the dataset.
Type, nullability and position come from the dataset.
**Cross-schema table rename and authorization (#13310)**
- `TableHookDispatcher.alterTable` builds the new identifier from
`RenameTable.getNewSchemaName()`. If the request has several renames, it
takes the last one that sets a schema.
- `AuthorizationUtils` gets a new `authorizationPluginRenamePrivileges`
overload that takes the new `NameIdentifier`. The existing `String`
overload delegates to it and behaves as before.
- `RangerAuthorizationHadoopSQLPlugin` renamed a table level by level,
starting with the schema. Once the plugin gets the new schema, that
schema step would match every policy of the old schema and move them
all. It now skips the schema step for a table rename and moves only the
table's own policies (table and column levels) to the new schema and
name.
### Why are the changes needed?
- Moving a table to another schema and changing a column in one
`alterTable`, then dropping the old schema, deleted the unchanged
columns and their tags, owners and privileges.
- Re-importing a table, e.g. after an out-of-band rename, gave every
column a new id, so all column tags were lost.
- On a Lance catalog with `lance.schema-refresh-mode=VERSION_CHECK`,
every dataset version change gave all columns new ids and cleared their
comments.
- On a cross-schema rename, authorization plugins such as Ranger were
told the table moved to `<old schema>.<new name>`, which doesn't exist.
Fix: #13309, #13310
Part of #13303
### Does this PR introduce _any_ user-facing change?
No API or configuration change. Behaviour changes:
- Column tags, owners and privileges survive a table move with a column
change, a re-import and a Lance schema refresh.
- Lance columns keep their comments across a schema refresh.
- Authorization plugins get the correct schema on a cross-schema table
rename.
Not changed here (tracked in #13303): when the MySQL/H2 upsert resolves
to a stale registration with the same name, that registration's column
relations are still orphaned.
### How was this patch tested?
- **Unit tests** (H2 locally; each one fails without its fix):
- `TestTableColumnMetaService`:
- `testMoveTableWithColumnChangeMovesAllColumns`
- `testOverwriteSameTableKeepsColumnIds`
- `testOverwriteNeverGivesOneStoredIdToTwoColumns`
- `testOverwriteWithoutColumnsRemovesAllColumnRelations`
- `testOverwriteReusesNewestIdOfDuplicateLegacyColumns`
- `testOverwriteOfAnotherTableDoesNotInheritColumnIds` (MySQL/H2 only)
-
`TestLanceTableOperations.testVersionCheckRefreshKeepsExistingColumnIdsAndComments`
- `TestTableHookDispatcher`:
`testRenameAcrossSchemasPassesNewSchemaToAuthorization`,
`testRenameTwiceKeepsSchemaFromEarlierRename`
-
`TestAuthorizationUtils.testRenameTableAcrossSchemasNotifiesAuthorizationPluginWithNewSchema`
- `TestRangerAuthorizationHadoopSQLPlugin`: a table rename across
schemas and within one schema, against an in-memory Ranger
- **Integration tests**:
-
`CatalogGenericCatalogLanceIT.testVersionCheckRefreshKeepsColumnTagsAndComments`:
passes locally and fails without the fix.
- `CatalogHive2IT.testOutOfBandRenameKeepsColumnTags` (also runs in
`CatalogHive3IT`): renames the table directly in HMS, then checks that
the column keeps its tag. Needs Docker.
- `TagIT.testMovedTableKeepsColumnTagsAfterOldSchemaIsDropped`: needs
Docker.
- `./gradlew :core:test -PskipITs` and the `catalog-lakehouse-generic`
unit tests pass locally. The MySQL/PostgreSQL backends and the
Docker-based ITs still need to run in CI.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../ranger/RangerAuthorizationHadoopSQLPlugin.java | 27 +-
.../TestRangerAuthorizationHadoopSQLPlugin.java | 147 +++++++++++
.../hive/integration/test/CatalogHive2IT.java | 61 +++++
.../lakehouse/lance/LanceTableOperations.java | 30 ++-
.../lakehouse/lance/TestLanceTableOperations.java | 72 +++++
.../test/CatalogGenericCatalogLanceIT.java | 90 +++++++
.../gravitino/client/integration/test/TagIT.java | 48 ++++
.../authorization/AuthorizationUtils.java | 24 +-
.../apache/gravitino/hook/TableHookDispatcher.java | 15 +-
.../relational/service/TableColumnMetaService.java | 91 ++++++-
.../relational/service/TableMetaService.java | 26 +-
.../authorization/TestAuthorizationUtils.java | 36 +++
.../gravitino/hook/TestTableHookDispatcher.java | 52 +++-
.../service/TestTableColumnMetaService.java | 292 +++++++++++++++++++++
14 files changed, 980 insertions(+), 31 deletions(-)
diff --git
a/authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationHadoopSQLPlugin.java
b/authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationHadoopSQLPlugin.java
index 3f1ca387f5..35296e4611 100644
---
a/authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationHadoopSQLPlugin.java
+++
b/authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationHadoopSQLPlugin.java
@@ -170,6 +170,12 @@ public class RangerAuthorizationHadoopSQLPlugin extends
RangerAuthorizationPlugi
for (int index = 0; index < mappingOldAndNewMetadata.size(); index++) {
oldMetadataNames.add(mappingOldAndNewMetadata.get(index).getKey());
newMetadataNames.add(mappingOldAndNewMetadata.get(index).getValue());
+ // A table rename never changes the schema-level policies. When the
table also moves to
+ // another schema, updating them would move every policy of the old
schema, so the table's
+ // own policies are updated from the table level down instead.
+ if (index == 0 && newAuthzMetadataObject.type().equals(TABLE)) {
+ continue;
+ }
AuthorizationMetadataObject.Type type;
if (index == 0) {
@@ -235,16 +241,21 @@ public class RangerAuthorizationHadoopSQLPlugin extends
RangerAuthorizationPlugi
// Doesn't need to rename the policy `*`
return;
}
- policyNames.set(index,
newAuthzMetaObject.names().get(index));
+ // Every level up to the renamed one takes its new name,
because a renamed
+ // table may also have moved to another schema.
+ for (int i = 0; i <= index; i++) {
+ policyNames.set(i, newAuthzMetaObject.names().get(i));
+ }
policy.setName(AuthorizationSecurableObject.DOT_JOINER.join(policyNames));
}
- // Update the policy resource name to new name
- policy
- .getResources()
- .put(
- policyResourceDefinesRule().get(index),
- new RangerPolicy.RangerPolicyResource(
- newAuthzMetaObject.names().get(index)));
+ // Update the policy resource names, up to the renamed level,
to the new names
+ for (int i = 0; i <= index; i++) {
+ policy
+ .getResources()
+ .put(
+ policyResourceDefinesRule().get(i),
+ new
RangerPolicy.RangerPolicyResource(newAuthzMetaObject.names().get(i)));
+ }
boolean alreadyExist =
existNewPolicies.stream()
diff --git
a/authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/TestRangerAuthorizationHadoopSQLPlugin.java
b/authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/TestRangerAuthorizationHadoopSQLPlugin.java
new file mode 100644
index 0000000000..5b268bd12d
--- /dev/null
+++
b/authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/TestRangerAuthorizationHadoopSQLPlugin.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.authorization.ranger;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.authorization.MetadataObjectChange;
+import
org.apache.gravitino.authorization.ranger.reference.RangerDefines.PolicyResource;
+import org.apache.ranger.plugin.model.RangerPolicy;
+import org.apache.ranger.plugin.util.SearchFilter;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class TestRangerAuthorizationHadoopSQLPlugin {
+
+ private final List<RangerPolicy> policies = new ArrayList<>();
+ private RangerAuthorizationHadoopSQLPlugin plugin;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ policies.clear();
+ // The constructor connects to Ranger, so only the tested methods run for
real.
+ plugin = Mockito.mock(RangerAuthorizationHadoopSQLPlugin.class,
Mockito.CALLS_REAL_METHODS);
+ RangerClientExtension rangerClient =
Mockito.mock(RangerClientExtension.class);
+ // An in-memory Ranger: a search returns the policies whose resources
match every filter.
+ Mockito.when(rangerClient.findPolicies(any()))
+ .thenAnswer(invocation -> findPolicies(invocation.getArgument(0)));
+ Mockito.when(rangerClient.updatePolicy(anyLong(), any()))
+ .thenAnswer(invocation -> invocation.getArgument(1));
+ plugin.setRangerClient(rangerClient);
+ }
+
+ @Test
+ public void testRenameTableAcrossSchemasMovesOnlyThatTablesPolicies() {
+ RangerPolicy schemaPolicy = addPolicy(1, "s1", "s1", null, null);
+ RangerPolicy tablePolicy = addPolicy(2, "s1.t1", "s1", "t1", null);
+ RangerPolicy columnPolicy = addPolicy(3, "s1.t1.*", "s1", "t1", "*");
+ RangerPolicy otherTablePolicy = addPolicy(4, "s1.t9", "s1", "t9", null);
+
+ plugin.onMetadataUpdated(
+ MetadataObjectChange.rename(
+ MetadataObjects.parse("catalog.s1.t1", MetadataObject.Type.TABLE),
+ MetadataObjects.parse("catalog.s2.t2", MetadataObject.Type.TABLE),
+ null));
+
+ // The renamed table's policies follow it to the new schema and name.
+ Assertions.assertEquals("s2.t2", tablePolicy.getName());
+ Assertions.assertEquals(resources("s2", "t2", null),
resourcesOf(tablePolicy));
+ Assertions.assertEquals(resources("s2", "t2", "*"),
resourcesOf(columnPolicy));
+ // The old schema and its other tables keep their policies.
+ Assertions.assertEquals("s1", schemaPolicy.getName());
+ Assertions.assertEquals(resources("s1", null, null),
resourcesOf(schemaPolicy));
+ Assertions.assertEquals("s1.t9", otherTablePolicy.getName());
+ Assertions.assertEquals(resources("s1", "t9", null),
resourcesOf(otherTablePolicy));
+ }
+
+ @Test
+ public void testRenameTableInSameSchemaRenamesOnlyThatTablesPolicies() {
+ RangerPolicy schemaPolicy = addPolicy(1, "s1", "s1", null, null);
+ RangerPolicy tablePolicy = addPolicy(2, "s1.t1", "s1", "t1", null);
+ RangerPolicy otherTablePolicy = addPolicy(3, "s1.t9", "s1", "t9", null);
+
+ plugin.onMetadataUpdated(
+ MetadataObjectChange.rename(
+ MetadataObjects.parse("catalog.s1.t1", MetadataObject.Type.TABLE),
+ MetadataObjects.parse("catalog.s1.t2", MetadataObject.Type.TABLE),
+ null));
+
+ Assertions.assertEquals("s1.t2", tablePolicy.getName());
+ Assertions.assertEquals(resources("s1", "t2", null),
resourcesOf(tablePolicy));
+ Assertions.assertEquals(resources("s1", null, null),
resourcesOf(schemaPolicy));
+ Assertions.assertEquals(resources("s1", "t9", null),
resourcesOf(otherTablePolicy));
+ }
+
+ private RangerPolicy addPolicy(long id, String name, String db, String
table, String column) {
+ RangerPolicy policy = new RangerPolicy();
+ policy.setId(id);
+ policy.setName(name);
+ Map<String, RangerPolicy.RangerPolicyResource> policyResources = new
HashMap<>();
+ resources(db, table, column)
+ .forEach((k, v) -> policyResources.put(k, new
RangerPolicy.RangerPolicyResource(v)));
+ policy.setResources(policyResources);
+ policies.add(policy);
+ return policy;
+ }
+
+ private List<RangerPolicy> findPolicies(Map<String, String> filters) {
+ return policies.stream()
+ .filter(
+ policy ->
+ filters.entrySet().stream()
+ .filter(e ->
e.getKey().startsWith(SearchFilter.RESOURCE_PREFIX))
+ .allMatch(
+ e ->
+ Objects.equals(
+ resourcesOf(policy)
+ .get(
+ e.getKey()
+
.substring(SearchFilter.RESOURCE_PREFIX.length())),
+ e.getValue())))
+ .collect(Collectors.toList());
+ }
+
+ private static Map<String, String> resources(String db, String table, String
column) {
+ Map<String, String> resources = new HashMap<>();
+ resources.put(PolicyResource.DATABASE.getName(), db);
+ if (table != null) {
+ resources.put(PolicyResource.TABLE.getName(), table);
+ }
+ if (column != null) {
+ resources.put(PolicyResource.COLUMN.getName(), column);
+ }
+ return resources;
+ }
+
+ private static Map<String, String> resourcesOf(RangerPolicy policy) {
+ return policy.getResources().entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e ->
e.getValue().getValues().get(0)));
+ }
+}
diff --git
a/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/CatalogHive2IT.java
b/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/CatalogHive2IT.java
index d296b19878..6ec869c7f9 100644
---
a/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/CatalogHive2IT.java
+++
b/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/CatalogHive2IT.java
@@ -52,6 +52,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.CatalogChange;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.MetalakeChange;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
@@ -1401,6 +1402,59 @@ public class CatalogHive2IT extends BaseIT {
"please ensure that the type of the new column position is
compatible with the old one"));
}
+ @Test
+ public void testOutOfBandRenameKeepsColumnTags() throws InterruptedException
{
+ // Hive stores names in lower case, and tag operations resolve a column by
its stored name.
+ String schema = schemaName.toLowerCase(Locale.ROOT);
+ NameIdentifier ident =
+ NameIdentifier.of(schema,
GravitinoITUtils.genRandomName("hive_oob_rename_table"));
+ catalog
+ .asTableCatalog()
+ .createTable(
+ ident, createColumns(), TABLE_COMMENT, createProperties(),
Transforms.EMPTY_TRANSFORM);
+ String tagName = GravitinoITUtils.genRandomName("hive_oob_rename_tag");
+ metalake.createTag(tagName, "comment", Collections.emptyMap());
+ try {
+ loadColumn(ident, HIVE_COL_NAME1).supportsTags().associateTags(new
String[] {tagName}, null);
+
+ // Rename the table directly in the Hive Metastore. It keeps its
Gravitino id in its table
+ // parameters, so loading it under the new name imports it again over
the same id.
+ String newName =
GravitinoITUtils.genRandomName("hive_oob_renamed_table");
+ HiveTable hiveTable = loadHiveTable(schema, ident.name());
+ HiveTable.Builder renamed =
+ HiveTable.builder()
+ .withName(newName)
+ .withColumns(hiveTable.columns())
+ .withProperties(hiveTable.properties())
+ .withAuditInfo(hiveTable.auditInfo())
+ .withDistribution(hiveTable.distribution())
+ .withSortOrders(hiveTable.sortOrder())
+ .withPartitioning(hiveTable.partitioning())
+ .withCatalogName(hiveTable.catalogName())
+ .withDatabaseName(hiveTable.databaseName());
+ if (hiveTable.comment() != null) {
+ renamed.withComment(hiveTable.comment());
+ }
+ hiveClientPool.run(
+ client -> {
+ client.alterTable(hmsCatalog, schema, ident.name(),
renamed.build());
+ return null;
+ });
+
+ // The column keeps its id, so the tag follows the table to its new name.
+ NameIdentifier newIdent = NameIdentifier.of(schema, newName);
+ Column column = loadColumn(newIdent, HIVE_COL_NAME1);
+ Assertions.assertArrayEquals(new String[] {tagName},
column.supportsTags().listTags());
+ MetadataObject[] objects =
metalake.getTag(tagName).associatedObjects().objects();
+ Assertions.assertEquals(1, objects.length);
+ Assertions.assertEquals(
+ String.join(".", catalogName, schema, newName,
HIVE_COL_NAME1).toLowerCase(Locale.ROOT),
+ objects[0].fullName().toLowerCase(Locale.ROOT));
+ } finally {
+ metalake.deleteTag(tagName);
+ }
+ }
+
@Test
public void testCrossSchemaTableRename() throws TException,
InterruptedException {
// Create a second schema to serve as the rename destination.
@@ -1946,4 +2000,11 @@ public class CatalogHive2IT extends BaseIT {
schemaName.toLowerCase()));
return properties;
}
+
+ private Column loadColumn(NameIdentifier tableIdent, String columnName) {
+ return
Arrays.stream(catalog.asTableCatalog().loadTable(tableIdent).columns())
+ .filter(c -> c.name().equals(columnName))
+ .findFirst()
+ .get();
+ }
}
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 c4e31e196e..a4d2a697e2 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
@@ -722,12 +722,36 @@ public class LanceTableOperations extends
ManagedTableOperations {
.withCreator(PrincipalUtils.getCurrentPrincipal().getName())
.withCreateTime(Instant.now())
.build();
+ // Tags, owners and privileges are attached to a column by id, so a column
that is still in the
+ // dataset keeps its id. It also keeps its comment and audit info, which
the dataset doesn't
+ // carry.
+ Map<String, ColumnEntity> existingColumns =
+ tableEntity.columns().stream()
+ .collect(
+ Collectors.toMap(
+ ColumnEntity::name, Function.identity(), (first, second)
-> first));
List<ColumnEntity> columnEntities =
IntStream.range(0, columns.length)
.mapToObj(
- i ->
- ColumnEntity.toColumnEntity(
- columns[i], i, idGenerator.nextId(), columnAuditInfo))
+ i -> {
+ ColumnEntity existing =
existingColumns.get(columns[i].name());
+ if (existing == null) {
+ return ColumnEntity.toColumnEntity(
+ columns[i], i, idGenerator.nextId(), columnAuditInfo);
+ }
+ return ColumnEntity.builder()
+ .withId(existing.id())
+ .withName(columns[i].name())
+ .withPosition(i)
+ .withDataType(columns[i].dataType())
+ .withComment(
+ columns[i].comment() != null ? columns[i].comment()
: existing.comment())
+ .withNullable(columns[i].nullable())
+ .withAutoIncrement(columns[i].autoIncrement())
+ .withDefaultValue(columns[i].defaultValue())
+ .withAuditInfo((AuditInfo) existing.auditInfo())
+ .build();
+ })
.collect(Collectors.toList());
return TableEntity.builder()
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 53e5c2d919..3556337d15 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
@@ -476,6 +476,78 @@ public class TestLanceTableOperations {
Assertions.assertEquals("9",
loadedTable.properties().get(LANCE_TABLE_VERSION));
}
+ @Test
+ public void testVersionCheckRefreshKeepsExistingColumnIdsAndComments()
throws Exception {
+ lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
+ NameIdentifier ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("version-check-keep-ids").toString();
+ AuditInfo columnAudit =
+
AuditInfo.builder().withCreator("column_creator").withCreateTime(Instant.EPOCH).build();
+ TableEntity tableEntity =
+ tableEntity(
+ ident,
+ List.of(
+ ColumnEntity.builder()
+ .withId(10L)
+ .withName("id")
+ .withComment("the id")
+ .withDataType(Types.IntegerType.get())
+ .withPosition(0)
+ .withAuditInfo(columnAudit)
+ .build(),
+ ColumnEntity.builder()
+ .withId(11L)
+ .withName("dropped")
+ .withDataType(Types.StringType.get())
+ .withPosition(1)
+ .withAuditInfo(columnAudit)
+ .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(12L);
+ AtomicReference<TableEntity> updated = 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);
+ updated.set(updater.apply(tableEntity));
+ return updated.get();
+ });
+
+ // The dataset moved to a new version: "id" is still there, "dropped" is
gone and "name" is new.
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema())
+ .thenReturn(
+ new Schema(
+ List.of(
+ Field.nullable("name", new ArrowType.Utf8()),
+ Field.nullable("id", new ArrowType.Int(32, true)))));
+ 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("9",
loadedTable.properties().get(LANCE_TABLE_VERSION));
+ List<ColumnEntity> columns = updated.get().columns();
+ Assertions.assertEquals(2, columns.size());
+
+ ColumnEntity name = columns.get(0);
+ Assertions.assertEquals("name", name.name());
+ Assertions.assertEquals(12L, name.id());
+ Assertions.assertEquals(0, name.position());
+
+ ColumnEntity id = columns.get(1);
+ Assertions.assertEquals("id", id.name());
+ Assertions.assertEquals(10L, id.id());
+ Assertions.assertEquals(1, id.position());
+ Assertions.assertEquals("the id", id.comment());
+ Assertions.assertEquals(columnAudit, id.auditInfo());
+ Assertions.assertEquals("the id", loadedTable.columns()[1].comment());
+ }
+
@Test
public void testVersionCheckSkipsRefreshWhenVersionIsCurrent() throws
Exception {
lanceTableOps.setCatalogProperties(Map.of(LANCE_SCHEMA_REFRESH_MODE,
"version-check"));
diff --git
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/integration/test/CatalogGenericCatalogLanceIT.java
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/integration/test/CatalogGenericCatalogLanceIT.java
index 2654bec7a5..277ed1ec6b 100644
---
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/integration/test/CatalogGenericCatalogLanceIT.java
+++
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/integration/test/CatalogGenericCatalogLanceIT.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.catalog.lakehouse.lance.integration.test;
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_TABLE_DECLARED;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_FORMAT;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_REGISTER;
@@ -48,6 +49,7 @@ import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
import org.apache.commons.io.FileUtils;
import org.apache.gravitino.Catalog;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Schema;
import org.apache.gravitino.client.GravitinoMetalake;
@@ -56,6 +58,7 @@ import org.apache.gravitino.integration.test.util.BaseIT;
import org.apache.gravitino.integration.test.util.GravitinoITUtils;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.expressions.NamedReference;
import org.apache.gravitino.rel.expressions.distributions.Distribution;
@@ -419,6 +422,86 @@ public class CatalogGenericCatalogLanceIT extends BaseIT {
RuntimeException.class, () ->
catalog.asTableCatalog().loadTable(newNameIdentifier));
}
+ @Test
+ void testVersionCheckRefreshKeepsColumnTagsAndComments() {
+ String refreshCatalogName =
GravitinoITUtils.genRandomName("lance_version_check_catalog");
+ Catalog refreshCatalog =
+ metalake.createCatalog(
+ refreshCatalogName,
+ Catalog.Type.RELATIONAL,
+ provider,
+ "comment",
+ ImmutableMap.of(LANCE_SCHEMA_REFRESH_MODE, "VERSION_CHECK"));
+ String refreshSchemaName = GravitinoITUtils.genRandomName(SCHEMA_PREFIX);
+ String tagName = GravitinoITUtils.genRandomName("lance_version_check_tag");
+ try {
+ refreshCatalog
+ .asSchemas()
+ .createSchema(refreshSchemaName, "comment",
createSchemaProperties());
+ NameIdentifier ident =
+ NameIdentifier.of(refreshSchemaName,
GravitinoITUtils.genRandomName(TABLE_PREFIX));
+ Map<String, String> properties = createProperties();
+ properties.put(Table.PROPERTY_TABLE_FORMAT, LANCE_TABLE_FORMAT);
+ properties.put(Table.PROPERTY_LOCATION, tempDirectory + "/" +
ident.name());
+ TableCatalog tableCatalog = refreshCatalog.asTableCatalog();
+ Table created =
+ tableCatalog.createTable(
+ ident,
+ new Column[] {
+ Column.of(LANCE_COL_NAME1, Types.IntegerType.get(),
"col_1_comment"),
+ Column.of(LANCE_COL_NAME2, Types.StringType.get(),
"col_2_comment")
+ },
+ TABLE_COMMENT,
+ properties,
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ new SortOrder[0]);
+
+ metalake.createTag(tagName, "comment", Collections.emptyMap());
+ findColumn(tableCatalog.loadTable(ident), LANCE_COL_NAME1)
+ .supportsTags()
+ .associateTags(new String[] {tagName}, null);
+
+ // Change the dataset outside Gravitino: keep column 1, drop column 2
and add column 3. This
+ // writes a new dataset version, so the next load refreshes the columns
from the dataset.
+ org.apache.arrow.vector.types.pojo.Schema newSchema =
+ new org.apache.arrow.vector.types.pojo.Schema(
+ Arrays.asList(
+ Field.nullable(LANCE_COL_NAME1, new ArrowType.Int(32, true)),
+ Field.nullable(LANCE_COL_NAME3, new ArrowType.Utf8())));
+ try (RootAllocator allocator = new RootAllocator();
+ Dataset ignored =
+ Dataset.write()
+ .allocator(allocator)
+ .schema(newSchema)
+ .uri(created.properties().get(Table.PROPERTY_LOCATION))
+ .mode(WriteParams.WriteMode.OVERWRITE)
+ .execute()) {
+ // The new dataset version is written.
+ }
+
+ Table refreshed = tableCatalog.loadTable(ident);
+ Assertions.assertArrayEquals(
+ new String[] {LANCE_COL_NAME1, LANCE_COL_NAME3},
+
Arrays.stream(refreshed.columns()).map(Column::name).toArray(String[]::new));
+
+ // Column 1 keeps its id, so it keeps its tag and its comment.
+ Column column1 = findColumn(refreshed, LANCE_COL_NAME1);
+ Assertions.assertEquals("col_1_comment", column1.comment());
+ Assertions.assertArrayEquals(new String[] {tagName},
column1.supportsTags().listTags());
+ MetadataObject[] objects =
metalake.getTag(tagName).associatedObjects().objects();
+ Assertions.assertEquals(1, objects.length);
+ Assertions.assertEquals(
+ String.join(".", refreshCatalogName, refreshSchemaName,
ident.name(), LANCE_COL_NAME1),
+ objects[0].fullName());
+ Assertions.assertEquals(
+ 0, findColumn(refreshed,
LANCE_COL_NAME3).supportsTags().listTags().length);
+ } finally {
+ metalake.deleteTag(tagName);
+ metalake.dropCatalog(refreshCatalogName, true);
+ }
+ }
+
@Test
void testLanceTableFormat() {
String tableName = GravitinoITUtils.genRandomName(TABLE_PREFIX);
@@ -1193,4 +1276,11 @@ public class CatalogGenericCatalogLanceIT extends BaseIT
{
LOG.warn("Failed to delete external table directory: {}",
externalTableLocation, e);
}
}
+
+ private static Column findColumn(Table table, String columnName) {
+ return Arrays.stream(table.columns())
+ .filter(c -> c.name().equals(columnName))
+ .findFirst()
+ .get();
+ }
}
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/TagIT.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/TagIT.java
index c5e0689469..80a6018ab0 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/TagIT.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/TagIT.java
@@ -912,6 +912,54 @@ public class TagIT extends BaseIT {
}
}
+ @Test
+ public void testMovedTableKeepsColumnTagsAfterOldSchemaIsDropped() {
+ String sourceSchema =
GravitinoITUtils.genRandomName("tag_it_move_source_schema");
+ relationalCatalog.asSchemas().createSchema(sourceSchema, "comment",
Collections.emptyMap());
+ NameIdentifier sourceIdent =
+ NameIdentifier.of(sourceSchema,
GravitinoITUtils.genRandomName("tag_it_move_table"));
+ relationalCatalog
+ .asTableCatalog()
+ .createTable(
+ sourceIdent,
+ new Column[] {
+ Column.of("c1", Types.IntegerType.get()), Column.of("c2",
Types.StringType.get())
+ },
+ "comment",
+ Collections.emptyMap());
+ NameIdentifier movedIdent = NameIdentifier.of(schema.name(),
sourceIdent.name());
+ try {
+ Tag tag =
+ metalake.createTag(
+ GravitinoITUtils.genRandomName("tag_it_move_table_tag"),
+ "comment",
+ Collections.emptyMap());
+ loadColumn(sourceIdent, "c1").supportsTags().associateTags(new String[]
{tag.name()}, null);
+
+ // Move the table to another schema and change another column in the
same request.
+ relationalCatalog
+ .asTableCatalog()
+ .alterTable(
+ sourceIdent,
+ TableChange.rename(sourceIdent.name(), schema.name()),
+ TableChange.updateColumnComment(new String[] {"c2"}, "new
comment"));
+
+ // Dropping the old schema must not take the moved table's unchanged
column with it.
+
Assertions.assertTrue(relationalCatalog.asSchemas().dropSchema(sourceSchema,
true));
+
+ Column c1 = loadColumn(movedIdent, "c1");
+ Assertions.assertArrayEquals(new String[] {tag.name()},
c1.supportsTags().listTags());
+ MetadataObject[] objects =
metalake.getTag(tag.name()).associatedObjects().objects();
+ Assertions.assertEquals(1, objects.length);
+ Assertions.assertEquals(
+ String.join(".", relationalCatalog.name(), schema.name(),
movedIdent.name(), "c1"),
+ objects[0].fullName());
+ } finally {
+ relationalCatalog.asTableCatalog().dropTable(movedIdent);
+ relationalCatalog.asSchemas().dropSchema(sourceSchema, true);
+ }
+ }
+
private NameIdentifier createColumnTestTable(String prefix) {
NameIdentifier tableIdent =
NameIdentifier.of(schema.name(),
GravitinoITUtils.genRandomName(prefix));
diff --git
a/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
b/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
index 43f99539ad..42eb56ba0b 100644
---
a/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
+++
b/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
@@ -370,18 +370,36 @@ public class AuthorizationUtils {
public static void authorizationPluginRenamePrivileges(
NameIdentifier ident, Entity.EntityType type, String newName,
List<String> locations) {
+ authorizationPluginRenamePrivileges(
+ ident, type, NameIdentifier.of(ident.namespace(), newName), locations);
+ }
+
+ /**
+ * Renames the privileges of an entity in the authorization plugins, when
the new identifier may
+ * be under a different parent, e.g. a table moved to another schema.
+ *
+ * @param ident the identifier of the entity before the rename
+ * @param type the entity type
+ * @param newIdent the identifier of the entity after the rename
+ * @param locations the storage locations of the entity, or {@code null}
+ */
+ public static void authorizationPluginRenamePrivileges(
+ NameIdentifier ident,
+ Entity.EntityType type,
+ NameIdentifier newIdent,
+ List<String> locations) {
// If we enable authorization, we should rename the privileges about the
entity in the
// authorization plugin.
if (GravitinoEnv.getInstance().internalAccessControlDispatcher() != null) {
notifyEntityNameIdMappingChange(ident, type);
MetadataObject oldMetadataObject =
NameIdentifierUtil.toMetadataObject(ident, type);
- MetadataObject newMetadataObject =
-
NameIdentifierUtil.toMetadataObject(NameIdentifier.of(ident.namespace(),
newName), type);
+ MetadataObject newMetadataObject =
NameIdentifierUtil.toMetadataObject(newIdent, type);
MetadataObjectChange renameChange =
MetadataObjectChange.rename(oldMetadataObject, newMetadataObject,
locations);
- String metalake = type == Entity.EntityType.METALAKE ? newName :
ident.namespace().level(0);
+ String metalake =
+ type == Entity.EntityType.METALAKE ? newIdent.name() :
ident.namespace().level(0);
// For a renamed catalog, we should pass the new name catalog, otherwise
we can't find the
// catalog in the entity store
diff --git
a/core/src/main/java/org/apache/gravitino/hook/TableHookDispatcher.java
b/core/src/main/java/org/apache/gravitino/hook/TableHookDispatcher.java
index 768c1e5410..76dc96d2ed 100644
--- a/core/src/main/java/org/apache/gravitino/hook/TableHookDispatcher.java
+++ b/core/src/main/java/org/apache/gravitino/hook/TableHookDispatcher.java
@@ -104,10 +104,15 @@ public class TableHookDispatcher implements
TableDispatcher {
public Table alterTable(NameIdentifier ident, TableChange... changes)
throws NoSuchTableException, IllegalArgumentException {
TableChange.RenameTable lastRenameChange = null;
+ // Renames apply in order, so the last one that sets a schema decides the
table's schema.
+ String newSchemaName = ident.namespace().level(2);
List<String> locations = null;
for (TableChange change : changes) {
if (change instanceof TableChange.RenameTable) {
lastRenameChange = (TableChange.RenameTable) change;
+ if (lastRenameChange.getNewSchemaName().isPresent()) {
+ newSchemaName = lastRenameChange.getNewSchemaName().get();
+ }
}
}
if (lastRenameChange != null) {
@@ -116,9 +121,15 @@ public class TableHookDispatcher implements
TableDispatcher {
Table alteredTable = dispatcher.alterTable(ident, changes);
if (lastRenameChange != null) {
- // todo: support rename across different schemas
+ // The rename may also move the table to another schema.
+ NameIdentifier newIdent =
+ NameIdentifierUtil.ofTable(
+ ident.namespace().level(0),
+ ident.namespace().level(1),
+ newSchemaName,
+ lastRenameChange.getNewName());
AuthorizationUtils.authorizationPluginRenamePrivileges(
- ident, Entity.EntityType.TABLE, lastRenameChange.getNewName(),
locations);
+ ident, Entity.EntityType.TABLE, newIdent, locations);
}
return alteredTable;
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableColumnMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableColumnMetaService.java
index 62653e23f7..fd64a42b80 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableColumnMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableColumnMetaService.java
@@ -22,14 +22,18 @@ import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATI
import com.google.common.collect.Lists;
import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.ColumnEntity;
import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.metrics.Monitored;
@@ -115,6 +119,62 @@ public class TableColumnMetaService {
insertColumnPOsInBatches(columnPOs);
}
+ /**
+ * Gives each column the id of the stored live column with the same name, if
there is one, and
+ * removes the relations of the stored columns that are gone.
+ *
+ * <p>Tags, owners and privileges are attached to a column by id, so a table
that is written again
+ * as a whole, e.g. re-imported, must keep the ids of the columns it still
has. Columns with no
+ * stored counterpart keep the id they were given, and a stored id is never
given to a second
+ * column. Names are matched exactly, as the catalog reports them. A stored
column whose id is not
+ * kept is dropped, so its relations are removed in the same transaction, as
for a column dropped
+ * by an alter.
+ *
+ * @param tableId the id of the stored table
+ * @param tableVersion the table version whose columns are reused
+ * @param columns the columns to write
+ * @return the columns, with the stored ids where the names match
+ */
+ List<ColumnEntity> reuseStoredColumnIds(
+ Long tableId, Long tableVersion, List<ColumnEntity> columns) {
+ List<ColumnPO> storedColumns = getColumnsByTableIdAndVersion(tableId,
tableVersion);
+ if (storedColumns.isEmpty()) {
+ return columns;
+ }
+
+ // A stored id that one of the columns already carries stays with that
column, so it is never
+ // handed to a second column.
+ Set<Long> carriedIds =
columns.stream().map(ColumnEntity::id).collect(Collectors.toSet());
+ Map<String, Long> reusableIdsByName = new HashMap<>();
+ // Legacy data may hold two live columns with one name. Visit the most
recently written one
+ // first, so the same id is reused whatever order the database returns the
rows in.
+ List<ColumnPO> newestFirst = Lists.newArrayList(storedColumns);
+ newestFirst.sort(
+ Comparator.comparing(ColumnPO::getTableVersion)
+ .thenComparing(ColumnPO::getColumnId)
+ .reversed());
+ for (ColumnPO storedColumn : newestFirst) {
+ if (!carriedIds.contains(storedColumn.getColumnId())) {
+ reusableIdsByName.putIfAbsent(storedColumn.getColumnName(),
storedColumn.getColumnId());
+ }
+ }
+
+ List<ColumnEntity> result = Lists.newArrayListWithCapacity(columns.size());
+ for (ColumnEntity column : columns) {
+ Long storedId = reusableIdsByName.remove(column.name());
+ result.add(storedId == null ? column : withId(column, storedId));
+ }
+
+ Set<Long> keptIds =
result.stream().map(ColumnEntity::id).collect(Collectors.toSet());
+ deleteColumnRelations(
+ storedColumns.stream()
+ .map(ColumnPO::getColumnId)
+ .filter(id -> !keptIds.contains(id))
+ .distinct()
+ .collect(Collectors.toList()));
+ return result;
+ }
+
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteColumnsByTableId")
@@ -187,15 +247,18 @@ public class TableColumnMetaService {
}
}
+ // If the table moved to another schema, move all of its existing column
rows as well, whether
+ // or not any column changed. Only the changed columns get new rows below,
so otherwise the
+ // unchanged ones would stay under the old schema and be dropped with it.
+ if (!newTable.namespace().equals(oldTable.namespace())) {
+ SessionUtils.doWithoutCommit(
+ TableColumnMapper.class,
+ mapper ->
+ mapper.updateSchemaIdByTableId(newTablePO.getTableId(),
newTablePO.getSchemaId()));
+ }
+
// If there is no change, directly return
if (columnPOsToInsert.isEmpty()) {
- // If namespace is changed, just update the schema_id of the columns.
- if (!newTable.namespace().equals(oldTable.namespace())) {
- SessionUtils.doWithoutCommit(
- TableColumnMapper.class,
- mapper ->
- mapper.updateSchemaIdByTableId(newTablePO.getTableId(),
newTablePO.getSchemaId()));
- }
return;
}
@@ -227,6 +290,20 @@ public class TableColumnMetaService {
});
}
+ private static ColumnEntity withId(ColumnEntity column, Long id) {
+ return ColumnEntity.builder()
+ .withId(id)
+ .withName(column.name())
+ .withPosition(column.position())
+ .withDataType(column.dataType())
+ .withComment(column.comment())
+ .withNullable(column.nullable())
+ .withAutoIncrement(column.autoIncrement())
+ .withDefaultValue(column.defaultValue())
+ .withAuditInfo((AuditInfo) column.auditInfo())
+ .build();
+ }
+
private void insertColumnPOsInBatches(List<ColumnPO> columnPOs) {
// Column inserts run inside the table transaction, so no batch commits
independently.
Lists.partition(columnPOs, COLUMN_INSERT_BATCH_SIZE)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
index 5ad3569fd0..7ace373f1d 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
@@ -33,6 +33,7 @@ import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.ColumnEntity;
import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.metrics.Monitored;
@@ -168,16 +169,27 @@ public class TableMetaService {
}
}),
() -> {
+ List<ColumnEntity> columns = tableEntity.columns();
// We need to delete the columns first if we want to overwrite the
table.
if (overwrite) {
- TableColumnMetaService.getInstance()
- .deleteColumnsByTableId(persistedPO.get().getTableId());
+ TablePO storedPO = persistedPO.get();
+ // Overwriting the same table, e.g. re-importing it after an
out-of-band rename, keeps
+ // the stored ids of the columns that still exist, so their
tags, owners and
+ // privileges stay attached. When the upsert resolved to another
table's row, the
+ // ids differ and that table's columns are not inherited.
+ if (columns != null
+ && po.getTableId().equals(storedPO.getTableId())
+ && storedPO.getCurrentVersion() > POConverters.INIT_VERSION)
{
+ columns =
+ TableColumnMetaService.getInstance()
+ .reuseStoredColumnIds(
+ storedPO.getTableId(),
storedPO.getCurrentVersion(), columns);
+ }
+
TableColumnMetaService.getInstance().deleteColumnsByTableId(storedPO.getTableId());
}
- },
- () -> {
- if (tableEntity.columns() != null &&
!tableEntity.columns().isEmpty()) {
- TableColumnMetaService.getInstance()
- .insertColumnPOs(persistedPO.get(), tableEntity.columns());
+
+ if (columns != null && !columns.isEmpty()) {
+
TableColumnMetaService.getInstance().insertColumnPOs(persistedPO.get(),
columns);
}
});
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationUtils.java
b/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationUtils.java
index bef17095d0..0049a36ec2 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationUtils.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationUtils.java
@@ -425,6 +425,42 @@ class TestAuthorizationUtils {
Assertions.assertEquals(locations, renameChange.locations());
}
+ @Test
+ void testRenameTableAcrossSchemasNotifiesAuthorizationPluginWithNewSchema() {
+ NameIdentifier ident = NameIdentifier.of("metalake", "catalog", "schema",
"table");
+ NameIdentifier newIdent = NameIdentifier.of("metalake", "catalog",
"new_schema", "new_table");
+ List<String> locations = Lists.newArrayList("/warehouse/schema/table");
+
+ AccessControlDispatcher accessControlDispatcher =
Mockito.mock(AccessControlDispatcher.class);
+ CatalogManager catalogManager = Mockito.mock(CatalogManager.class);
+ BaseCatalog<?> baseCatalog = Mockito.mock(BaseCatalog.class);
+ AuthorizationPlugin authorizationPlugin =
Mockito.mock(AuthorizationPlugin.class);
+ CatalogTestUtils.mockDoWithCatalog(catalogManager, baseCatalog);
+
Mockito.when(baseCatalog.getAuthorizationPlugin()).thenReturn(authorizationPlugin);
+
+ GravitinoEnv envMock = Mockito.mock(GravitinoEnv.class);
+
Mockito.when(envMock.internalAccessControlDispatcher()).thenReturn(accessControlDispatcher);
+ Mockito.when(envMock.catalogManager()).thenReturn(catalogManager);
+
+ try (MockedStatic<GravitinoEnv> envStatic =
Mockito.mockStatic(GravitinoEnv.class)) {
+ envStatic.when(GravitinoEnv::getInstance).thenReturn(envMock);
+
+ AuthorizationUtils.authorizationPluginRenamePrivileges(
+ ident, Entity.EntityType.TABLE, newIdent, locations);
+ }
+
+ ArgumentCaptor<MetadataObjectChange[]> changesCaptor =
+ ArgumentCaptor.forClass(MetadataObjectChange[].class);
+
Mockito.verify(authorizationPlugin).onMetadataUpdated(changesCaptor.capture());
+ MetadataObjectChange.RenameMetadataObject renameChange =
+ Assertions.assertInstanceOf(
+ MetadataObjectChange.RenameMetadataObject.class,
changesCaptor.getValue()[0]);
+ Assertions.assertEquals("catalog.schema.table",
renameChange.metadataObject().fullName());
+ Assertions.assertEquals(
+ "catalog.new_schema.new_table",
renameChange.newMetadataObject().fullName());
+ Assertions.assertEquals(locations, renameChange.locations());
+ }
+
@Test
void
testRemoveTablePrivilegesNotifiesAuthorizationPluginWithExpectedChange() {
NameIdentifier ident = NameIdentifier.of("metalake", "catalog", "schema",
"table");
diff --git
a/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
b/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
index befd3a9dd4..6aff59c0e7 100644
--- a/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
+++ b/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
@@ -222,10 +222,60 @@ public class TestTableHookDispatcher {
.when(() -> AuthorizationUtils.getMetadataObjectLocation(ident,
Entity.EntityType.TABLE))
.thenReturn(locations);
assertSame(alteredTable, hook.alterTable(ident, renameChange));
+ NameIdentifier newIdent = NameIdentifier.of(METALAKE, CATALOG, "schema",
"newName");
authorizationUtils.verify(
() ->
AuthorizationUtils.authorizationPluginRenamePrivileges(
- ident, Entity.EntityType.TABLE, "newName", locations));
+ ident, Entity.EntityType.TABLE, newIdent, locations));
+ }
+ }
+
+ @Test
+ public void testRenameAcrossSchemasPassesNewSchemaToAuthorization() {
+ TableDispatcher dispatcher = Mockito.mock(TableDispatcher.class);
+ TableHookDispatcher hook = new TableHookDispatcher(dispatcher, () -> null);
+ NameIdentifier ident = NameIdentifier.of(METALAKE, CATALOG, "schema",
"table");
+ Table alteredTable = Mockito.mock(Table.class);
+ TableChange renameChange = TableChange.rename("newName", "newSchema");
+ List<String> locations = ImmutableList.of("/test");
+ Mockito.when(dispatcher.alterTable(ident,
renameChange)).thenReturn(alteredTable);
+
+ try (MockedStatic<AuthorizationUtils> authorizationUtils =
+ Mockito.mockStatic(AuthorizationUtils.class)) {
+ authorizationUtils
+ .when(() -> AuthorizationUtils.getMetadataObjectLocation(ident,
Entity.EntityType.TABLE))
+ .thenReturn(locations);
+ assertSame(alteredTable, hook.alterTable(ident, renameChange));
+ NameIdentifier newIdent = NameIdentifier.of(METALAKE, CATALOG,
"newSchema", "newName");
+ authorizationUtils.verify(
+ () ->
+ AuthorizationUtils.authorizationPluginRenamePrivileges(
+ ident, Entity.EntityType.TABLE, newIdent, locations));
+ }
+ }
+
+ @Test
+ public void testRenameTwiceKeepsSchemaFromEarlierRename() {
+ TableDispatcher dispatcher = Mockito.mock(TableDispatcher.class);
+ TableHookDispatcher hook = new TableHookDispatcher(dispatcher, () -> null);
+ NameIdentifier ident = NameIdentifier.of(METALAKE, CATALOG, "schema",
"table");
+ Table alteredTable = Mockito.mock(Table.class);
+ // The table dispatcher moves the table to the last schema set by any
rename.
+ TableChange moveChange = TableChange.rename("t2", "newSchema");
+ TableChange renameChange = TableChange.rename("t3");
+ Mockito.when(dispatcher.alterTable(ident, moveChange,
renameChange)).thenReturn(alteredTable);
+
+ try (MockedStatic<AuthorizationUtils> authorizationUtils =
+ Mockito.mockStatic(AuthorizationUtils.class)) {
+ authorizationUtils
+ .when(() -> AuthorizationUtils.getMetadataObjectLocation(ident,
Entity.EntityType.TABLE))
+ .thenReturn(ImmutableList.of());
+ assertSame(alteredTable, hook.alterTable(ident, moveChange,
renameChange));
+ NameIdentifier newIdent = NameIdentifier.of(METALAKE, CATALOG,
"newSchema", "t3");
+ authorizationUtils.verify(
+ () ->
+ AuthorizationUtils.authorizationPluginRenamePrivileges(
+ ident, Entity.EntityType.TABLE, newIdent,
ImmutableList.of()));
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableColumnMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableColumnMetaService.java
index 377cbf8801..f53defa175 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableColumnMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableColumnMetaService.java
@@ -45,6 +45,7 @@ import
org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.session.SqlSessions;
import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.TestTemplate;
public class TestTableColumnMetaService extends TestJDBCBackend {
@@ -243,6 +244,297 @@ public class TestTableColumnMetaService extends
TestJDBCBackend {
Assertions.assertEquals(2, countActiveColumnRelations(kept));
}
+ @TestTemplate
+ public void testMoveTableWithColumnChangeMovesAllColumns() throws Exception {
+ String catalogName = "catalog1";
+ String oldSchemaName = "schema1";
+ String newSchemaName = "schema2";
+ createParentEntities(METALAKE_NAME, catalogName, oldSchemaName,
AUDIT_INFO);
+ createAndInsertSchema(METALAKE_NAME, catalogName, newSchemaName);
+
+ ColumnEntity unchanged = newIntColumn("unchanged", 0, "comment");
+ ColumnEntity changed = newIntColumn("changed", 1, "comment");
+ TableEntity table =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("table_move")
+ .withNamespace(Namespace.of(METALAKE_NAME, catalogName,
oldSchemaName))
+ .withColumns(Lists.newArrayList(unchanged, changed))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(table, false);
+
+ // Move the table to the other schema and change one column in the same
update.
+ ColumnEntity changedColumn = newIntColumn("changed", 1, "new comment",
changed.id());
+ TableEntity moved =
+ TableEntity.builder()
+ .withId(table.id())
+ .withName(table.name())
+ .withNamespace(Namespace.of(METALAKE_NAME, catalogName,
newSchemaName))
+ .withColumns(Lists.newArrayList(unchanged, changedColumn))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().updateTable(table.nameIdentifier(),
(TableEntity old) -> moved);
+
+ // Dropping the old schema must not take any of the moved table's columns
with it.
+ Assertions.assertTrue(
+ SchemaMetaService.getInstance()
+ .deleteSchema(NameIdentifier.of(METALAKE_NAME, catalogName,
oldSchemaName), true));
+
+ TableEntity retrieved =
+
TableMetaService.getInstance().getTableByIdentifier(moved.nameIdentifier());
+ compareTwoColumns(moved.columns(), retrieved.columns());
+ }
+
+ @TestTemplate
+ public void testOverwriteSameTableKeepsColumnIds() throws Exception {
+ String catalogName = "catalog1";
+ String schemaName = "schema1";
+ createParentEntities(METALAKE_NAME, catalogName, schemaName, AUDIT_INFO);
+
+ ColumnEntity kept = newIntColumn("kept", 0, "comment");
+ ColumnEntity dropped = newIntColumn("dropped", 1, "comment");
+ TableEntity table =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("table_before_rename")
+ .withNamespace(Namespace.of(METALAKE_NAME, catalogName,
schemaName))
+ .withColumns(Lists.newArrayList(kept, dropped))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(table, false);
+ insertColumnRelations(kept.id());
+ insertColumnRelations(dropped.id());
+
+ // Re-import the same table under a new name, e.g. after an out-of-band
rename. The importer
+ // gives every column a fresh id and doesn't know the stored ones.
+ ColumnEntity reimportedKept = newIntColumn("kept", 0, "comment");
+ ColumnEntity added = newIntColumn("added", 1, "comment");
+ TableEntity reimported =
+ TableEntity.builder()
+ .withId(table.id())
+ .withName("table_after_rename")
+ .withNamespace(table.namespace())
+ .withColumns(Lists.newArrayList(reimportedKept, added))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(reimported, true);
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()));
+ Map<String, Long> columnIds =
+ TableMetaService.getInstance()
+ .getTableByIdentifier(reimported.nameIdentifier())
+ .columns()
+ .stream()
+ .collect(Collectors.toMap(ColumnEntity::name, ColumnEntity::id));
+ Assertions.assertEquals(2, columnIds.size());
+ Assertions.assertEquals(kept.id(), columnIds.get("kept"));
+ Assertions.assertEquals(added.id(), columnIds.get("added"));
+ Assertions.assertEquals(2, countActiveColumnRelations(kept.id()));
+ // The column that is gone from the table loses its relations with it.
+ Assertions.assertEquals(0, countActiveColumnRelations(dropped.id()));
+
+ // Overwriting again works too: the rows written by the first overwrite
are retired first.
+ TableMetaService.getInstance().insertTable(reimported, true);
+ TableEntity retrieved =
+
TableMetaService.getInstance().getTableByIdentifier(reimported.nameIdentifier());
+ Assertions.assertEquals(
+ kept.id(),
+ retrieved.columns().stream().filter(c ->
c.name().equals("kept")).findFirst().get().id());
+ Assertions.assertEquals(2, retrieved.columns().size());
+ }
+
+ @TestTemplate
+ public void testOverwriteNeverGivesOneStoredIdToTwoColumns() throws
Exception {
+ String catalogName = "catalog1";
+ String schemaName = "schema1";
+ createParentEntities(METALAKE_NAME, catalogName, schemaName, AUDIT_INFO);
+
+ ColumnEntity original = newIntColumn("b", 0, "comment");
+ TableEntity table =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("table_carried_id")
+ .withNamespace(Namespace.of(METALAKE_NAME, catalogName,
schemaName))
+ .withColumns(Lists.newArrayList(original))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(table, false);
+ insertColumnRelations(original.id());
+
+ // The stored column already travels under a new name, and a new column
takes its old name. The
+ // stored id stays with the column that carries it.
+ ColumnEntity renamed = newIntColumn("c", 0, "comment", original.id());
+ ColumnEntity newColumn = newIntColumn("b", 1, "comment");
+ TableEntity overwritten =
+ TableEntity.builder()
+ .withId(table.id())
+ .withName(table.name())
+ .withNamespace(table.namespace())
+ .withColumns(Lists.newArrayList(renamed, newColumn))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(overwritten, true);
+
+ compareTwoColumns(
+ overwritten.columns(),
+
TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()).columns());
+ Assertions.assertEquals(2, countActiveColumnRelations(original.id()));
+ }
+
+ @TestTemplate
+ public void testOverwriteReusesNewestIdOfDuplicateLegacyColumns() throws
Exception {
+ String catalogName = "catalog1";
+ String schemaName = "schema1";
+ createParentEntities(METALAKE_NAME, catalogName, schemaName, AUDIT_INFO);
+
+ ColumnEntity older = newIntColumn("a", 0, "comment");
+ TableEntity table =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("table_duplicate_names")
+ .withNamespace(Namespace.of(METALAKE_NAME, catalogName,
schemaName))
+ .withColumns(Lists.newArrayList(older))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(table, false);
+ ColumnEntity newer = newIntColumn("b", 1, "comment");
+ TableEntity withNewer =
+ TableEntity.builder()
+ .withId(table.id())
+ .withName(table.name())
+ .withNamespace(table.namespace())
+ .withColumns(Lists.newArrayList(older, newer))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance()
+ .updateTable(table.nameIdentifier(), (TableEntity old) -> withNewer);
+ insertColumnRelations(older.id());
+ insertColumnRelations(newer.id());
+
+ // Legacy data: two live columns share one name. The newer one was written
at a later version.
+ try (SqlSession session =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = session.getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.executeUpdate(
+ "UPDATE "
+ + TableColumnMapper.COLUMN_TABLE_NAME
+ + " SET column_name = 'a' WHERE column_id = "
+ + newer.id());
+ }
+
+ ColumnEntity reimported = newIntColumn("a", 0, "comment");
+ TableEntity overwritten =
+ TableEntity.builder()
+ .withId(table.id())
+ .withName(table.name())
+ .withNamespace(table.namespace())
+ .withColumns(Lists.newArrayList(reimported))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(overwritten, true);
+
+ List<ColumnEntity> columns =
+
TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()).columns();
+ Assertions.assertEquals(1, columns.size());
+ Assertions.assertEquals(newer.id(), columns.get(0).id());
+ Assertions.assertEquals(2, countActiveColumnRelations(newer.id()));
+ Assertions.assertEquals(0, countActiveColumnRelations(older.id()));
+ }
+
+ @TestTemplate
+ public void testOverwriteWithoutColumnsRemovesAllColumnRelations() throws
Exception {
+ String catalogName = "catalog1";
+ String schemaName = "schema1";
+ createParentEntities(METALAKE_NAME, catalogName, schemaName, AUDIT_INFO);
+
+ ColumnEntity column = newIntColumn("column", 0, "comment");
+ TableEntity table =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("table_without_columns")
+ .withNamespace(Namespace.of(METALAKE_NAME, catalogName,
schemaName))
+ .withColumns(Lists.newArrayList(column))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(table, false);
+ insertColumnRelations(column.id());
+
+ TableEntity overwritten =
+ TableEntity.builder()
+ .withId(table.id())
+ .withName(table.name())
+ .withNamespace(table.namespace())
+ .withColumns(Lists.newArrayList())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(overwritten, true);
+
+ Assertions.assertTrue(
+ TableMetaService.getInstance()
+ .getTableByIdentifier(table.nameIdentifier())
+ .columns()
+ .isEmpty());
+ Assertions.assertEquals(0, countActiveColumnRelations(column.id()));
+ }
+
+ @TestTemplate
+ public void testOverwriteOfAnotherTableDoesNotInheritColumnIds() throws
Exception {
+ // Only MySQL/H2 can resolve the upsert to another table's row through the
natural key.
+ // PostgreSQL's upsert targets table_id and rejects the insert instead.
+ Assumptions.assumeFalse("postgresql".equalsIgnoreCase(backendType));
+ String catalogName = "catalog1";
+ String schemaName = "schema1";
+ createParentEntities(METALAKE_NAME, catalogName, schemaName, AUDIT_INFO);
+
+ ColumnEntity storedColumn = newIntColumn("column", 0, "comment");
+ TableEntity stale =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("table_same_name")
+ .withNamespace(Namespace.of(METALAKE_NAME, catalogName,
schemaName))
+ .withColumns(Lists.newArrayList(storedColumn))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(stale, false);
+
+ ColumnEntity newColumn = newIntColumn("column", 0, "comment");
+ TableEntity other =
+ TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(stale.name())
+ .withNamespace(stale.namespace())
+ .withColumns(Lists.newArrayList(newColumn))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TableMetaService.getInstance().insertTable(other, true);
+
+ TableEntity stored =
+
TableMetaService.getInstance().getTableByIdentifier(other.nameIdentifier());
+ Assertions.assertEquals(1, stored.columns().size());
+ Assertions.assertEquals(newColumn.id(), stored.columns().get(0).id());
+ }
+
+ private ColumnEntity newIntColumn(String name, int position, String comment)
{
+ return newIntColumn(name, position, comment,
RandomIdGenerator.INSTANCE.nextId());
+ }
+
+ private ColumnEntity newIntColumn(String name, int position, String comment,
long id) {
+ return ColumnEntity.builder()
+ .withId(id)
+ .withName(name)
+ .withPosition(position)
+ .withComment(comment)
+ .withDataType(Types.IntegerType.get())
+ .withNullable(true)
+ .withAutoIncrement(false)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
private void insertColumnRelations(long columnId) throws SQLException {
try (SqlSession session =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
