jerryshao commented on code in PR #13197:
URL: https://github.com/apache/gravitino/pull/13197#discussion_r4080131747
##########
core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java:
##########
@@ -573,18 +593,165 @@ private EntityCombinedTable importTable(NameIdentifier
identifier) {
.withAuditInfo(audit)
.build();
try {
- store.put(tableEntity, true);
- } catch (EntityAlreadyExistsException e) {
+ if (caseAlias) {
+ TableEntity registered = store.get(observedOwner, TABLE,
TableEntity.class);
+ if (registered.id() != uid) {
+ throw new OptimisticLockException(
+ "The registered owner of table ID %d changed during import;
retry the load", uid);
+ }
+ return EntityCombinedTable.of(table.tableFromCatalog(), registered)
+ .withHiddenProperties(table.hiddenProperties());
+ }
+ if (observedOwner != null && !observedOwner.equals(identifier)) {
+ // Updating the observed row uses the store's version check. A second
server that observed
+ // the same old name cannot move the row again after the first import
commits.
+ store.update(
+ observedOwner,
+ TableEntity.class,
+ TABLE,
+ current -> withReusedColumnIds(tableEntity, current));
+ } else {
+ // Import a new name without overwrite so a concurrent import of the
same ID cannot move
+ // its row. Preserve overwrite when repairing a registration already
stored at this name.
+ boolean overwriteByName = stringId == null || store.exists(identifier,
TABLE);
+ store.put(tableEntity, overwriteByName);
Review Comment:
[Important] Inserting without overwrite also collides with **soft-deleted**
rows, so a re-registered catalog becomes unloadable for up to a week.
`table_meta`'s primary key is `table_id` alone
(scripts/mysql/schema-0.5.0-mysql.sql:76), and dropping a catalog only sets
`deleted_at` (`softDeleteTableMetasByCatalogId`,
TableMetaBaseSQLProvider.java:313). The dead row keeps the ID until the GC
removes it after `gravitino.entity.store.deleteAfterTimeMs` (default 7 days,
Configs.java:64).
Drop a catalog in Gravitino and re-create it over the same Hive metastore -
the usual way to change catalog properties - and every external table still
carries its old `gravitino.identifier`. `findRegisteredTableById` finds no
*live* owner in the new schema (`listTablePOsBySchemaId` filters `deleted_at =
0`), so this line inserts without overwrite, the plain INSERT
(TableMetaBaseSQLProvider.insertTableMeta:205) hits the dead row's primary key,
the converter maps that to `EntityAlreadyExistsException`
(H2ExceptionConverter.java:40-44), and loadTable:188 tells the operator to
remove a copied identifier from a table nobody copied. Nothing in that catalog
loads until the GC runs. Before this PR the overwrite revived the row.
The store guard was written to permit exactly this -
`selectTableMetaByIdForUpdate` filters `deleted_at = 0`
(TableMetaBaseSQLProvider.java:202) so a soft-deleted owner never blocks a
re-import - but that guard is no longer on this path. Falling back to an
overwrite when the conflicting ID has no live owner would keep both properties.
`SchemaOperationDispatcher.java:696-697` has the identical shape (`schema_meta`
PK is `schema_id`), and a re-registered catalog hits the schema first, so the
failure arrives before any table is touched.
Verified by: read the MySQL DDL,
`insertTableMeta`/`selectTableMetaByIdForUpdate`/`softDeleteTableMetasByCatalogId`
in TableMetaBaseSQLProvider on this branch,
H2ExceptionConverter.toGravitinoException, Configs.java:64, and both dispatcher
import paths. The PR's own TestTableMetaService.java:141 asserts that
`backend.insert(copiedId, false)` throws `EntityAlreadyExistsException` on a
duplicate ID, which is the same collision - only there the owner row is live.
##########
core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java:
##########
@@ -573,18 +593,165 @@ private EntityCombinedTable importTable(NameIdentifier
identifier) {
.withAuditInfo(audit)
.build();
try {
- store.put(tableEntity, true);
- } catch (EntityAlreadyExistsException e) {
+ if (caseAlias) {
+ TableEntity registered = store.get(observedOwner, TABLE,
TableEntity.class);
+ if (registered.id() != uid) {
+ throw new OptimisticLockException(
+ "The registered owner of table ID %d changed during import;
retry the load", uid);
+ }
+ return EntityCombinedTable.of(table.tableFromCatalog(), registered)
+ .withHiddenProperties(table.hiddenProperties());
+ }
+ if (observedOwner != null && !observedOwner.equals(identifier)) {
+ // Updating the observed row uses the store's version check. A second
server that observed
+ // the same old name cannot move the row again after the first import
commits.
+ store.update(
+ observedOwner,
+ TableEntity.class,
+ TABLE,
+ current -> withReusedColumnIds(tableEntity, current));
+ } else {
+ // Import a new name without overwrite so a concurrent import of the
same ID cannot move
+ // its row. Preserve overwrite when repairing a registration already
stored at this name.
+ boolean overwriteByName = stringId == null || store.exists(identifier,
TABLE);
+ store.put(tableEntity, overwriteByName);
+ }
+ } catch (EntityAlreadyExistsException | OptimisticLockException e) {
throw e;
+ } catch (NoSuchEntityException e) {
+ throw new OptimisticLockException(
+ e, "The registered owner of table ID %d changed during import; retry
the load", uid);
} catch (Exception e) {
- LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", identifier, e);
+ LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "import", identifier,
e);
throw new RuntimeException("Failed to import the table entity to the
store", e);
}
return EntityCombinedTable.of(table.tableFromCatalog(), tableEntity)
.withHiddenProperties(table.hiddenProperties());
}
+ /**
+ * Tells an external rename or case alias apart from a copied id before
import.
+ *
+ * <p>An import that finds a {@link StringIdentifier} but no row under this
name overwrites the
+ * row that owns the id. That is right after an external rename: the old
name is gone and the row
+ * should follow the table. It is wrong when the id was copied ({@code
CREATE TABLE t2 LIKE t1}
+ * carries {@code TBLPROPERTIES}, so does a copy tool or a restored backup):
the source table is
+ * still there, and re-binding would move its row and every attachment keyed
by that id (owner,
+ * tags, policies, role grants) to the copy. The store cannot tell the two
apart; only the
+ * external catalog can, so this asks it whether the id's current owner
still exists.
+ */
+ private Pair<NameIdentifier, Boolean> checkImportedIdNotCopied(
+ NameIdentifier identifier, long id) {
+ NameIdentifier currentOwner =
findRegisteredTableById(identifier.namespace(), id);
Review Comment:
[Important] The owner lookup is namespace-scoped, so an external
**cross-schema rename** is now reported as a copied identifier and the table
cannot be loaded at all.
Gravitino moves tables between schemas itself (`RenameTable` with a new
schema name, handled by `TableMetaService.updateTable`, which branches on
`isSchemaChanged` at TableMetaService.java:221), so an out-of-band `ALTER TABLE
db1.orders RENAME TO db2.orders` - Hive and Iceberg both support it - is a real
case rather than a hypothetical.
After such a rename, loading `db2.orders`: this line lists only `db2`
(findRegisteredTableById, TableOperationDispatcher.java:742), returns null, so
the external `tableExists(currentOwner)` check below never runs; the import
falls to the insert without overwrite, which collides with the ID's row in
`db1`, and loadTable:188 asks the operator to remove `gravitino.identifier`
from a table nobody copied. Removing it mints a fresh ID, so the owner, tags,
policies, role grants and statistics stay behind on the stale `db1.orders` row
- the same loss this PR set out to prevent, reached from the other side. Before
this PR the upsert followed the table.
The earlier review's other suggestion covers this: widen the lookup to the
catalog, so `tableExists(currentOwner)` gets to run and separates a
cross-schema rename (source gone -> re-bind) from a cross-schema copy (source
alive -> reject with the good message). The cross-parent store guard still
backstops a genuine copy. Failing that, the message at loadTable:188 should not
assert that the identifier was copied when nothing here has checked whether the
other object still exists.
Verified by: read `checkImportedIdNotCopied` (:644-690),
`findRegisteredTableById` (:742-753), `importTable` (:547-640) and `loadTable`
(:163-215) on this branch; `TableMetaService.updateTable` (:204-313) for the
cross-schema move it supports; the PR's tests cover the cross-schema *copy*
(TestTableOperationDispatcher.java:356) but no cross-schema rename.
##########
core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java:
##########
@@ -573,18 +593,165 @@ private EntityCombinedTable importTable(NameIdentifier
identifier) {
.withAuditInfo(audit)
.build();
try {
- store.put(tableEntity, true);
- } catch (EntityAlreadyExistsException e) {
+ if (caseAlias) {
+ TableEntity registered = store.get(observedOwner, TABLE,
TableEntity.class);
+ if (registered.id() != uid) {
+ throw new OptimisticLockException(
+ "The registered owner of table ID %d changed during import;
retry the load", uid);
+ }
+ return EntityCombinedTable.of(table.tableFromCatalog(), registered)
+ .withHiddenProperties(table.hiddenProperties());
+ }
Review Comment:
[Question] The alias no longer writes, but every load of the alias spelling
still takes the schema WRITE lock and two catalog round trips.
`internalLoadTable` decides `imported` from `store.get(ident)`
(TableOperationDispatcher.java:805-814), and this branch deliberately leaves no
row under the alias spelling. So each load of `T` while `t` is stored is still
`!imported`: `loadTable` re-enters `importTable` under `LockType.WRITE` on the
schema (:175-176), and `checkImportedIdNotCopied` calls `tableExists` and then
`listTables` on the external catalog (:657-669) before arriving here and
returning the stored entity.
That is a clear improvement over the rename-per-load this replaced - no
version bump, no column rewrite, and `testLoadTableAcceptsCaseAlias` pins it.
What remains is that reads of the alias serialize against every import in that
schema and pay two extra RPCs each time, on an Iceberg REST catalog where
aliases are the normal case. Would resolving the requested name to the stored
one before the import decision (rather than after) be worth it, or is the alias
spelling expected to be rare enough in practice?
Verified by: read `importTable` (:547-640), `checkImportedIdNotCopied`
(:644-690), `loadTable` (:163-215) and `internalLoadTable` (:764-815) on this
branch, and `testLoadTableAcceptsCaseAlias`
(TestTableOperationDispatcher.java:321-354), which asserts no `put`/`update`
but does not look at the lock or the catalog calls.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java:
##########
@@ -81,6 +81,14 @@ public void insertTopic(TopicEntity topicEntity, boolean
overwrite) throws IOExc
TopicMetaMapper.class,
mapper -> {
if (overwrite) {
+ // A copied id must not move another schema's topic
here.
+
OccWriteSupport.checkOverwriteIdNotOwnedByOtherParent(
+ () ->
mapper.selectTopicMetaByIdForUpdate(po.getTopicId()),
+ owner -> Objects.equals(owner.getSchemaId(),
po.getSchemaId()),
+ owner ->
+ String.format(
+ "Topic ID %d belongs to schema ID %d,
not schema ID %d",
+ po.getTopicId(), owner.getSchemaId(),
po.getSchemaId()));
Review Comment:
[Nit] For topics this guard's message never reaches the caller.
`TopicOperationDispatcher.importTopic` still does `store.put(topicEntity,
true)` inside a `try { ... } catch (Exception e)` that wraps everything into
`RuntimeException("Failed to import topic entity to the store")`
(TopicOperationDispatcher.java:311-316) - it has no
`EntityAlreadyExistsException` branch, and no dispatcher-side
`checkImportedIdNotCopied`. So the careful "Topic ID %d belongs to schema ID
%d, not schema ID %d" you build here is swallowed into a 500 with a generic
message, and the same-parent copy that the table and schema dispatchers now
reject is still accepted for topics.
Not necessarily wrong to leave as is - a Kafka topic's ID comes from the
topic UUID rather than copied properties
(TopicOperationDispatcher.java:283-290), so the copy vector is much weaker
here. But it is worth a line in the PR description saying topics get the store
guard only, or at least letting `importTopic` rethrow
`EntityAlreadyExistsException` so the message survives.
Verified by: read `TopicOperationDispatcher.importTopic` (:267-317) and
`internalLoadTopic` (:319-340) on this branch, and grepped for
`checkImportedIdNotCopied` - two definitions, in the table and schema
dispatchers only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
