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 a9db07c53e [#14335] fix(core): Clean owner and privilege rows of
deleted jobs and job templates, and keep columns when the catalog reports none
(#14336)
a9db07c53e is described below
commit a9db07c53e66df31d18b2ded58d00e1edb5da6e4
Author: Qi Yu <[email protected]>
AuthorDate: Thu Oct 29 19:30:34 2026 +0800
[#14335] fix(core): Clean owner and privilege rows of deleted jobs and job
templates, and keep columns when the catalog reports none (#14336)
### What changes were proposed in this pull request?
- `JobMetaService.deleteJob` soft-deletes the job's owner relation in
the same transaction.
- `JobTemplateMetaService.deleteJobTemplate` soft-deletes the template's
owner relation and securable objects, and the owner relations of the
jobs deleted with it (new
`OwnerMetaMapper.softDeleteOwnerRelOfJobsByTemplateId`). Jobs can't be
securable objects, so they have no securable-object rows to clean.
- A table whose catalog reports no column list (`columns() == null`) is
handled by `TableOperationDispatcher` as follows. Where it fails, it
throws an `IllegalStateException` that names the catalog and the table:
- `loadTable` fails before importing the table or updating its stored
columns.
- `createTable` fails before storing the table, where the snapshot used
to fail with an NPE. The table stays created in the catalog, as before.
- `alterTable` on a table stored in Gravitino completes.
`updateColumnsIfNecessary` keeps the stored columns as they were before
the alter, instead of treating a missing list as "all columns dropped",
and the result reports those stored columns, with a warning logged. So
the hook's authorization rename and the success event run as usual. A
column drop or rename in that alter is not reflected in the stored
columns.
- `alterTable` on a table not stored in Gravitino, or on a managed
table, fails after the catalog has applied the change, since there are
no stored columns to report. The hook's authorization rename and the
success event are then skipped, and a retried rename fails because the
table already has its new name.
- `DTOConverters.toDTO(Table)` keeps `null` columns instead of throwing
an NPE. Connector results are snapshotted through it before the
dispatcher check runs, so the check could not report its error
otherwise. `TableDTO.columns` is marked `@Nullable` for that internal
snapshot only. A `null` column list still never reaches the REST
response, so the clients are unchanged.
### Why are the changes needed?
- Deleting a job or job template left its owner and privilege rows live
until the orphan relation GC collected them, so roles kept listing
privileges on deleted templates.
- A catalog returning `columns() == null` would have made
`updateColumnsIfNecessary` delete all of a table's stored columns and
their tags, policies and owners. Today an NPE in the snapshot fails the
call first, so this is a defensive fix that also replaces the NPE with a
clear error.
Fixes #14335
### Does this PR introduce _any_ user-facing change?
No API change. Creating or loading a table whose catalog reports no
column list fails with a clear error instead of an NPE. Altering a
stored one completes and reports the stored columns.
### How was this patch tested?
- New tests:
- `TestJobMetaService.testDeleteJobCleansOwner` and
`TestJobTemplateMetaService.testDeleteJobTemplateCleansOwnersAndPrivileges`.
Both check that rows of another job or template survive.
- `TestTableOperationDispatcher`:
`testLoadFailsWhenCatalogReportsNoColumns`,
`testCreateFailsWhenCatalogReportsNoColumns` (nothing is stored),
`testAlterKeepsStoredColumnsWhenCatalogReportsNoColumns`,
`testRenameCompletesWhenCatalogReportsNoColumns` (the stored column ids
move with the table) and
`testAlterUnstoredTableFailsWhenCatalogReportsNoColumns`. The alter
tests also check that the catalog applied the change. The alter check
for managed tables has no test.
- `TestJobMetaService`, `TestJobTemplateMetaService`,
`TestPolicyMetaService` and `TestOwnerMetaService` passed on H2, MySQL
and PostgreSQL for the first commit. `TestTableOperationDispatcher`
(49), `TestTableEvent` and `TestDTOConverters` pass.
- Mutation checks: removing any of the following makes its test fail:
- the template relation cleanup;
- the owner cleanup for jobs deleted with their template;
- the job owner cleanup;
- the load check, the create check, the alter check for unstored tables,
or reporting the stored columns on alter;
- the column guard (restoring the old "all columns dropped" handling);
- the `toDTO` null check.
---
.../org/apache/gravitino/dto/rel/TableDTO.java | 17 ++-
.../apache/gravitino/dto/util/DTOConverters.java | 8 +-
.../gravitino/catalog/EntityCombinedTable.java | 17 ++-
.../catalog/TableOperationDispatcher.java | 88 +++++++++++--
.../storage/relational/mapper/OwnerMetaMapper.java | 5 +
.../mapper/OwnerMetaSQLProviderFactory.java | 5 +
.../provider/base/OwnerMetaBaseSQLProvider.java | 17 +++
.../storage/relational/service/JobMetaService.java | 12 +-
.../relational/service/JobTemplateMetaService.java | 21 ++++
.../catalog/TestTableOperationDispatcher.java | 137 +++++++++++++++++++++
.../gravitino/connector/TestCatalogOperations.java | 17 ++-
.../storage/relational/TestJDBCBackend.java | 18 +++
.../relational/service/TestJobMetaService.java | 48 ++++++++
.../service/TestJobTemplateMetaService.java | 79 ++++++++++++
14 files changed, 468 insertions(+), 21 deletions(-)
diff --git a/common/src/main/java/org/apache/gravitino/dto/rel/TableDTO.java
b/common/src/main/java/org/apache/gravitino/dto/rel/TableDTO.java
index a23a1cd37c..0140d68a3b 100644
--- a/common/src/main/java/org/apache/gravitino/dto/rel/TableDTO.java
+++ b/common/src/main/java/org/apache/gravitino/dto/rel/TableDTO.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.dto.rel;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import java.util.Map;
+import javax.annotation.Nullable;
import lombok.EqualsAndHashCode;
import org.apache.gravitino.dto.AuditDTO;
import org.apache.gravitino.dto.rel.indexes.IndexDTO;
@@ -42,6 +43,8 @@ public class TableDTO implements Table {
@JsonProperty("comment")
private String comment;
+ // Null only for the internal connector snapshot; responses always carry a
column list.
+ @Nullable
@JsonProperty("columns")
private ColumnDTO[] columns;
@@ -106,8 +109,11 @@ public class TableDTO implements Table {
}
/**
- * @return The columns of the table.
+ * @return The columns of the table. Null only for the internal connector
snapshot of a table
+ * whose catalog reports no column list; such a table is rejected before
it is returned to a
+ * client.
*/
+ @Nullable
@Override
public Column[] columns() {
return columns;
@@ -189,7 +195,7 @@ public class TableDTO implements Table {
/** The comment associated with the table. */
protected String comment;
/** The columns of the table. */
- protected ColumnDTO[] columns;
+ @Nullable protected ColumnDTO[] columns;
/** The properties associated with the table. */
protected Map<String, String> properties;
/** The audit information for the table. */
@@ -231,10 +237,11 @@ public class TableDTO implements Table {
/**
* Sets the columns of the table.
*
- * @param columns The columns of the table.
+ * @param columns The columns of the table. Null is only for the internal
connector snapshot of
+ * a table whose catalog reports no column list; a DTO sent to clients
always has columns.
* @return The Builder instance.
*/
- public S withColumns(ColumnDTO[] columns) {
+ public S withColumns(@Nullable ColumnDTO[] columns) {
this.columns = columns;
return (S) this;
}
@@ -309,7 +316,7 @@ public class TableDTO implements Table {
* Builds a Table DTO based on the provided builder parameters.
*
* @return A new TableDTO instance.
- * @throws IllegalArgumentException If required fields name, columns and
audit are not set.
+ * @throws IllegalArgumentException If required fields name and audit are
not set.
*/
public TableDTO build() {
Preconditions.checkArgument(name != null && !name.isEmpty(), "name
cannot be null or empty");
diff --git
a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
index 5fdf3c51f4..406d52bee2 100644
--- a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
+++ b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
@@ -298,8 +298,14 @@ public class DTOConverters {
return TableDTO.builder()
.withName(table.name())
.withComment(table.comment())
+ // Connector results are snapshotted through here before the table
dispatcher rejects a
+ // missing column list with a clear error, so a null list must not
throw here.
.withColumns(
-
Arrays.stream(table.columns()).map(DTOConverters::toDTO).toArray(ColumnDTO[]::new))
+ table.columns() == null
+ ? null
+ : Arrays.stream(table.columns())
+ .map(DTOConverters::toDTO)
+ .toArray(ColumnDTO[]::new))
.withProperties(table.properties())
.withSortOrders(toDTOs(table.sortOrder()))
.withDistribution(toDTO(table.distribution()))
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/EntityCombinedTable.java
b/core/src/main/java/org/apache/gravitino/catalog/EntityCombinedTable.java
index a9ee8513cb..44df60eabb 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/EntityCombinedTable.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/EntityCombinedTable.java
@@ -57,6 +57,9 @@ public final class EntityCombinedTable implements Table {
// unique id, and based on this id, we can label and control the access to
the entities.
private boolean imported;
+ // The columns to report instead of the catalog's, set only when the catalog
reports none.
+ private Column[] columnsOverride;
+
private EntityCombinedTable(Table table, TableEntity tableEntity) {
this.table = table;
this.tableEntity = tableEntity;
@@ -78,6 +81,18 @@ public final class EntityCombinedTable implements Table {
return this;
}
+ /**
+ * Reports the given columns instead of the catalog table's. Used when the
catalog returns a table
+ * without a column list and the stored columns stand in for it.
+ *
+ * @param columns The columns to report.
+ * @return This EntityCombinedTable.
+ */
+ EntityCombinedTable withColumns(Column[] columns) {
+ this.columnsOverride = columns;
+ return this;
+ }
+
public EntityCombinedTable withImported(boolean imported) {
this.imported = imported;
return this;
@@ -99,7 +114,7 @@ public final class EntityCombinedTable implements Table {
@Override
public Column[] columns() {
- return table.columns();
+ return columnsOverride != null ? columnsOverride : table.columns();
}
@Override
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 4e50f44aa4..9eb60ee9f0 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
@@ -331,9 +331,15 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
IllegalArgumentException.class);
Table alteredTable = catalogResult.table;
+ // The change is already applied in the catalog. If the result has
no column list and
+ // Gravitino has no stored columns to report instead, the alter
fails, because clients
+ // require one. The hook's authorization rename and the success
event are then skipped,
+ // and a retried rename fails because the table already has its new
name.
if (catalogResult.managed) {
- return EntityCombinedTable.of(alteredTable)
- .withHiddenProperties(catalogResult.hiddenProperties);
+ return checkColumnsReported(
+ ident,
+ EntityCombinedTable.of(alteredTable)
+ .withHiddenProperties(catalogResult.hiddenProperties));
}
StringIdentifier stringId =
getStringIdFromProperties(alteredTable.properties());
@@ -344,8 +350,10 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
te = getEntity(ident, TABLE, TableEntity.class);
}
if (te == null) {
- return EntityCombinedTable.of(alteredTable)
- .withHiddenProperties(catalogResult.hiddenProperties);
+ return checkColumnsReported(
+ ident,
+ EntityCombinedTable.of(alteredTable)
+ .withHiddenProperties(catalogResult.hiddenProperties));
}
}
@@ -404,8 +412,22 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
ident, newIdent);
}
- return EntityCombinedTable.of(alteredTable, updatedTableEntity)
- .withHiddenProperties(catalogResult.hiddenProperties);
+ EntityCombinedTable result =
+ EntityCombinedTable.of(alteredTable, updatedTableEntity)
+ .withHiddenProperties(catalogResult.hiddenProperties);
+ // The store kept the columns it had before this alter (see
updateColumnsIfNecessary).
+ // Report them, so the alter completes, including the hook's
authorization rename,
+ // instead of failing after the catalog has already applied it. A
column drop or rename
+ // in this alter is not reflected in them.
+ if (alteredTable.columns() == null && updatedTableEntity != null) {
+ LOG.warn(
+ "Catalog {} returned altered table {} without a column list,
reporting the stored "
+ + "columns",
+ getCatalogIdentifier(ident),
+ ident);
+ result.withColumns(toColumns(updatedTableEntity.columns()));
+ }
+ return checkColumnsReported(ident, result);
});
}
@@ -629,6 +651,8 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
},
NoSuchTableException.class);
Table table = catalogResult.table;
+ // Fail before importing the table or updating its stored columns.
+ checkColumnsReported(ident, table);
if (catalogResult.managed) {
return EntityCombinedTable.of(table)
@@ -718,6 +742,8 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
NoSuchSchemaException.class,
TableAlreadyExistsException.class);
Table table = catalogResult.table;
+ // Fail before storing the table, as a missing column list failed the
snapshot before.
+ checkColumnsReported(ident, table);
// If the table is managed by Gravitino, we don't need to create
TableEntity and store it again.
if (catalogResult.managed) {
@@ -769,6 +795,34 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
return snapshotTable(catalog, table, isManagedEntity(catalog,
Capability.Scope.TABLE));
}
+ private <T extends Table> T checkColumnsReported(NameIdentifier ident, T
table) {
+ if (table.columns() == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Catalog %s returned table %s without a column list",
+ getCatalogIdentifier(ident), ident));
+ }
+ return table;
+ }
+
+ private static Column[] toColumns(@Nullable List<ColumnEntity>
columnEntities) {
+ if (columnEntities == null) {
+ return new Column[0];
+ }
+ return columnEntities.stream()
+ .sorted(Comparator.comparing(ColumnEntity::position))
+ .map(
+ c ->
+ Column.of(
+ c.name(),
+ c.dataType(),
+ c.comment(),
+ c.nullable(),
+ c.autoIncrement(),
+ c.defaultValue()))
+ .toArray(Column[]::new);
+ }
+
private List<ColumnEntity> toColumnEntities(Column[] columns, AuditInfo
audit) {
return columns == null
? Collections.emptyList()
@@ -977,12 +1031,24 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
return Pair.of(false, Collections.emptyList());
}
+ // A catalog that reports no column list says nothing about the columns,
so keep the stored
+ // ones. Treating it as "all columns dropped" would delete their tags,
policies and owners. Load
+ // rejects such a table before getting here, and alter keeps the columns
as they were before it.
+ if (tableFromCatalog.columns() == null) {
+ LOG.warn(
+ "Table {} from the underlying source has no column list, keeping the
stored columns",
+ tableFromCatalog.name());
+ return Pair.of(
+ false,
+ tableFromGravitino.columns() == null
+ ? Collections.emptyList()
+ : tableFromGravitino.columns());
+ }
+
Map<String, Pair<Integer, Column>> columnsFromCatalogTable =
- tableFromCatalog.columns() == null
- ? Collections.emptyMap()
- : IntStream.range(0, tableFromCatalog.columns().length)
- .mapToObj(i -> Pair.of(i, tableFromCatalog.columns()[i]))
- .collect(Collectors.toMap(p -> p.getRight().name(),
Function.identity()));
+ IntStream.range(0, tableFromCatalog.columns().length)
+ .mapToObj(i -> Pair.of(i, tableFromCatalog.columns()[i]))
+ .collect(Collectors.toMap(p -> p.getRight().name(),
Function.identity()));
Map<String, ColumnEntity> columnsFromTableEntity =
tableFromGravitino.columns() == null
? Collections.emptyMap()
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
index fc456f10cb..12dab8d833 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
@@ -135,6 +135,11 @@ public interface OwnerMetaMapper {
method = "softDeleteOwnerRelBySchemaIds")
void softDeleteOwnerRelBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
+ @UpdateProvider(
+ type = OwnerMetaSQLProviderFactory.class,
+ method = "softDeleteOwnerRelOfJobsByTemplateId")
+ void softDeleteOwnerRelOfJobsByTemplateId(@Param("jobTemplateId") Long
jobTemplateId);
+
@UpdateProvider(
type = OwnerMetaSQLProviderFactory.class,
method = "deleteOwnerMetasByLegacyTimeline")
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
index eea8ddb9d3..bcdd7da2a8 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
@@ -189,6 +189,11 @@ public class OwnerMetaSQLProviderFactory {
return getProvider().softDeleteOwnerRelBySchemaIds(schemaIds);
}
+ public static String softDeleteOwnerRelOfJobsByTemplateId(
+ @Param("jobTemplateId") Long jobTemplateId) {
+ return getProvider().softDeleteOwnerRelOfJobsByTemplateId(jobTemplateId);
+ }
+
public static String deleteOwnerMetasByLegacyTimeline(
@Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
return getProvider().deleteOwnerMetasByLegacyTimeline(legacyTimeline,
limit);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
index 46806b59e3..8ce23075c1 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
@@ -25,6 +25,7 @@ import
org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper;
import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.JobMetaMapper;
import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
@@ -284,6 +285,22 @@ public class OwnerMetaBaseSQLProvider {
+ ")";
}
+ public String softDeleteOwnerRelOfJobsByTemplateId(@Param("jobTemplateId")
Long jobTemplateId) {
+ String now = currentTimestampMillisExpression();
+ // No deleted_at filter on the job rows: the template delete soft-deletes
its jobs earlier in
+ // the same transaction, and filtering them out would make this a no-op.
+ return "UPDATE "
+ + OWNER_TABLE_NAME
+ + " SET deleted_at = "
+ + now
+ + ", updated_at = "
+ + now
+ + " WHERE deleted_at = 0 AND metadata_object_type = 'JOB' AND
metadata_object_id IN ("
+ + " SELECT job_run_id FROM "
+ + JobMetaMapper.TABLE_NAME
+ + " WHERE job_template_id = #{jobTemplateId})";
+ }
+
public String softDeleteOwnerRelBySchemaIds(@Param("schemaIds") List<Long>
schemaIds) {
String now = currentTimestampMillisExpression();
return "<script>"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java
index a07a13ce0f..fab897cce8 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java
@@ -29,6 +29,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
import org.apache.gravitino.HasIdentifier;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.IllegalNamespaceException;
@@ -37,6 +38,7 @@ import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.meta.JobEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.JobMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.po.JobPO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -245,7 +247,7 @@ public class JobMetaService {
});
}
- /** Deletes a job only if the observed version is still active. */
+ /** Deletes a job and its owner relation only if the observed version is
still active. */
void deleteJobWithVersion(NameIdentifier ident, JobPO observed) {
SessionUtils.doMultipleWithCommit(
() ->
@@ -256,7 +258,13 @@ public class JobMetaService {
mapper ->
mapper.softDeleteJobByRunIdWithVersion(
observed.jobRunId(),
observed.currentVersion())),
- () -> writeFailure(ident, observed)));
+ () -> writeFailure(ident, observed)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ observed.jobRunId(), MetadataObject.Type.JOB.name())));
}
private RuntimeException writeFailure(NameIdentifier ident, JobPO observed) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java
index 9628fa7f7e..bd69a6230d 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java
@@ -29,6 +29,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
import org.apache.gravitino.HasIdentifier;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.NoSuchEntityException;
@@ -38,6 +39,8 @@ import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.EntityChangeLogWriter;
import org.apache.gravitino.storage.relational.mapper.JobMetaMapper;
import org.apache.gravitino.storage.relational.mapper.JobTemplateMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.po.JobTemplatePO;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -259,10 +262,28 @@ public class JobTemplateMetaService {
SessionUtils.doWithoutCommit(
JobMetaMapper.class,
mapper ->
mapper.softDeleteJobsByTemplateId(observed.jobTemplateId())),
+ () -> deleteJobTemplateRelations(observed.jobTemplateId()),
() ->
EntityChangeLogWriter.append(ident,
Entity.EntityType.JOB_TEMPLATE, OperateType.DROP));
}
+ private static void deleteJobTemplateRelations(Long jobTemplateId) {
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ jobTemplateId, MetadataObject.Type.JOB_TEMPLATE.name()));
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(
+ jobTemplateId, MetadataObject.Type.JOB_TEMPLATE.name()));
+ // The template's jobs were deleted with it. Jobs have owners but can't be
securable objects.
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper -> mapper.softDeleteOwnerRelOfJobsByTemplateId(jobTemplateId));
+ }
+
/** Locks the observed template while a job is inserted in the same
transaction. */
void lockTemplateForJobWrite(String name, Long templateId, Long metalakeId) {
OccWriteSupport.lockParentForChildWrite(
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
index b9ac73e877..5d58fe6900 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
@@ -63,6 +63,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.TestCatalog;
import org.apache.gravitino.TestColumn;
+import org.apache.gravitino.TestTable;
import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.connector.HiddenPropertyMaskUtils;
import org.apache.gravitino.connector.TestCatalogOperations;
@@ -1492,6 +1493,142 @@ public class TestTableOperationDispatcher extends
TestOperationDispatcher {
.build();
}
+ @Test
+ public void testLoadFailsWhenCatalogReportsNoColumns() throws IOException {
+ NameIdentifier tableIdent =
createTableReportingNoColumns("schema_null_columns_load");
+ Map<String, Long> ids = columnIds(tableIdent);
+
+ IllegalStateException e =
+ Assertions.assertThrows(
+ IllegalStateException.class, () ->
tableOperationDispatcher.loadTable(tableIdent));
+
+ Assertions.assertTrue(e.getMessage().contains("without a column list"),
e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains(catalog), e.getMessage());
+ Assertions.assertEquals(ids, columnIds(tableIdent));
+ }
+
+ @Test
+ public void testAlterKeepsStoredColumnsWhenCatalogReportsNoColumns() throws
IOException {
+ NameIdentifier tableIdent =
createTableReportingNoColumns("schema_null_columns_alter");
+ Map<String, Long> ids = columnIds(tableIdent);
+
+ Table altered =
+ tableOperationDispatcher.alterTable(tableIdent,
TableChange.setProperty("k3", "v3"));
+
+ // The catalog applied the change, the stored columns are kept and
reported instead.
+ Assertions.assertEquals(
+ "v3",
testCatalogOperations().loadTable(tableIdent).properties().get("k3"));
+ Assertions.assertEquals(ids, columnIds(tableIdent));
+ Assertions.assertArrayEquals(
+ new String[] {"c1"},
Arrays.stream(altered.columns()).map(Column::name).toArray());
+ }
+
+ @Test
+ public void testRenameCompletesWhenCatalogReportsNoColumns() throws
IOException {
+ NameIdentifier tableIdent =
createTableReportingNoColumns("schema_null_columns_rename");
+ Map<String, Long> ids = columnIds(tableIdent);
+ NameIdentifier newIdent = NameIdentifier.of(tableIdent.namespace(),
"table_null_columns_new");
+
+ Table altered =
+ tableOperationDispatcher.alterTable(tableIdent,
TableChange.rename(newIdent.name()));
+
+ // The rename completes, so the hook's authorization rename that follows
it runs too.
+ Assertions.assertEquals(newIdent.name(), altered.name());
+ Assertions.assertArrayEquals(
+ new String[] {"c1"},
Arrays.stream(altered.columns()).map(Column::name).toArray());
+ Assertions.assertFalse(entityStore.exists(tableIdent, TABLE));
+ Assertions.assertEquals(ids, columnIds(newIdent));
+ }
+
+ @Test
+ public void testAlterUnstoredTableFailsWhenCatalogReportsNoColumns() {
+ Namespace tableNs = Namespace.of(metalake, catalog,
"schema_null_columns_unstored");
+ Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2");
+
schemaOperationDispatcher.createSchema(NameIdentifier.of(tableNs.levels()),
"comment", props);
+ NameIdentifier tableIdent = NameIdentifier.of(tableNs,
"table_null_columns");
+ // Created in the catalog only, so Gravitino has no stored table and no id
for it.
+ testCatalogOperations()
+ .replaceTable(
+ tableIdent,
+ TestTable.builder()
+ .withName(tableIdent.name())
+ .withProperties(props)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build());
+
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> tableOperationDispatcher.alterTable(tableIdent,
TableChange.setProperty("k3", "v3")));
+ Assertions.assertEquals(
+ "v3",
testCatalogOperations().loadTable(tableIdent).properties().get("k3"));
+ }
+
+ @Test
+ public void testCreateFailsWhenCatalogReportsNoColumns() throws IOException {
+ Namespace tableNs = Namespace.of(metalake, catalog,
"schema_null_columns_create");
+ Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2");
+
schemaOperationDispatcher.createSchema(NameIdentifier.of(tableNs.levels()),
"comment", props);
+ NameIdentifier tableIdent = NameIdentifier.of(tableNs,
"table_null_columns");
+ Column[] columns =
+ new Column[] {
+ TestColumn.builder()
+ .withName("c1")
+ .withPosition(0)
+ .withType(Types.IntegerType.get())
+ .build()
+ };
+
+ TestCatalogOperations ops = testCatalogOperations();
+ ops.setReportNoColumnsOnCreate(true);
+ try {
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ tableOperationDispatcher.createTable(
+ tableIdent, columns, "comment", props, new Transform[0]));
+ } finally {
+ ops.setReportNoColumnsOnCreate(false);
+ }
+
+ Assertions.assertFalse(entityStore.exists(tableIdent, TABLE));
+ }
+
+ private NameIdentifier createTableReportingNoColumns(String schemaName) {
+ Namespace tableNs = Namespace.of(metalake, catalog, schemaName);
+ Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2");
+
schemaOperationDispatcher.createSchema(NameIdentifier.of(tableNs.levels()),
"comment", props);
+ NameIdentifier tableIdent = NameIdentifier.of(tableNs,
"table_null_columns");
+ Column[] columns =
+ new Column[] {
+ TestColumn.builder()
+ .withName("c1")
+ .withPosition(0)
+ .withType(Types.IntegerType.get())
+ .build()
+ };
+ tableOperationDispatcher.createTable(tableIdent, columns, "comment",
props, new Transform[0]);
+
+ testCatalogOperations()
+ .replaceTable(
+ tableIdent,
+ TestTable.builder()
+ .withName(tableIdent.name())
+ .withComment("comment")
+ .withProperties(props)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build());
+ return tableIdent;
+ }
+
+ private TestCatalogOperations testCatalogOperations() {
+ TestCatalog testCatalog =
+ (TestCatalog)
+ catalogManager.loadCatalogAndWrap(NameIdentifier.of(metalake,
catalog)).catalog();
+ return (TestCatalogOperations) testCatalog.ops();
+ }
+
@Test
public void testRenameIntoNameOfStaleStoredColumn() throws IOException {
Namespace tableNs = Namespace.of(metalake, catalog, "schema_stale_column");
diff --git
a/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
b/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
index 3ac659243c..b56272877e 100644
---
a/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
+++
b/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
@@ -132,6 +132,8 @@ public class TestCatalogOperations
private final Map<String, String> config;
+ private boolean reportNoColumnsOnCreate;
+
public static final String FAIL_CREATE = "fail-create";
public static final String FAIL_TEST = "need-fail";
@@ -173,6 +175,16 @@ public class TestCatalogOperations
.toArray(NameIdentifier[]::new);
}
+ /** Makes {@link #createTable} return a table without a column list, as a
broken source would. */
+ public void setReportNoColumnsOnCreate(boolean reportNoColumnsOnCreate) {
+ this.reportNoColumnsOnCreate = reportNoColumnsOnCreate;
+ }
+
+ /** Replaces a table, e.g. with one whose source reports no column list. */
+ public void replaceTable(NameIdentifier ident, TestTable table) {
+ tables.put(ident, table);
+ }
+
@Override
public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
if (tables.containsKey(ident)) {
@@ -236,7 +248,7 @@ public class TestCatalogOperations
.withComment(comment)
.withProperties(new HashMap<>(properties))
.withAuditInfo(auditInfo)
- .withColumns(sortedColumns)
+ .withColumns(reportNoColumnsOnCreate ? null : sortedColumns)
.withDistribution(distribution)
.withSortOrders(sortOrders)
.withPartitioning(partitions)
@@ -1454,6 +1466,9 @@ public class TestCatalogOperations
}
private Column[] updateColumns(Column[] columns, TableChange.ColumnChange[]
columnChanges) {
+ if (columns == null) {
+ return null;
+ }
Map<String, Column> columnMap =
Arrays.stream(columns).collect(Collectors.toMap(Column::name,
Function.identity()));
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java
index 894937a1e6..bce9a024e3 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java
@@ -342,6 +342,24 @@ public abstract class TestJDBCBackend {
}
}
+ protected int countActiveRowsWhere(String table, String whereClause) {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet rs =
+ statement.executeQuery(
+ String.format(
+ "SELECT count(*) FROM %s WHERE %s AND deleted_at = 0",
table, whereClause))) {
+ if (rs.next()) {
+ return rs.getInt(1);
+ }
+ throw new RuntimeException("Doesn't contain data");
+ } catch (SQLException se) {
+ throw new RuntimeException("SQL execution failed", se);
+ }
+ }
+
protected BaseMetalake createBaseMakeLake(Long id, String name, AuditInfo
auditInfo) {
return BaseMetalake.builder()
.withId(id)
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
index abd5802a53..20d33937d6 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
@@ -22,14 +22,17 @@ import java.io.IOException;
import java.time.Instant;
import java.util.List;
import org.apache.commons.lang3.ArrayUtils;
+import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.JobEntity;
import org.apache.gravitino.meta.JobTemplateEntity;
+import org.apache.gravitino.meta.UserEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.gravitino.utils.NameIdentifierUtil;
@@ -231,6 +234,45 @@ public class TestJobMetaService extends TestJDBCBackend {
.deleteJob(NameIdentifierUtil.ofJob(METALAKE_NAME, job.name())));
}
+ @TestTemplate
+ public void testDeleteJobCleansOwner() throws IOException {
+ BaseMetalake metalake =
+ createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), METALAKE_NAME,
AUDIT_INFO);
+ backend.insert(metalake, false);
+ backend.insert(newShellJobTemplateEntity("template", "", METALAKE_NAME),
false);
+ UserEntity user =
+ createUserEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofUserNamespace(METALAKE_NAME),
+ "job_owner",
+ AUDIT_INFO);
+ backend.insert(user, false);
+
+ JobEntity dropped =
+ TestJobTemplateMetaService.newJobEntity(
+ "template", JobHandle.Status.SUCCEEDED, METALAKE_NAME);
+ JobEntity kept =
+ TestJobTemplateMetaService.newJobEntity(
+ "template", JobHandle.Status.SUCCEEDED, METALAKE_NAME);
+ backend.insert(dropped, false);
+ backend.insert(kept, false);
+ for (JobEntity job : new JobEntity[] {dropped, kept}) {
+ OwnerMetaService.getInstance()
+ .setOwner(
+ NameIdentifierUtil.ofJob(METALAKE_NAME, job.name()),
+ Entity.EntityType.JOB,
+ user.nameIdentifier(),
+ user.type());
+ }
+
+ Assertions.assertTrue(
+ JobMetaService.getInstance()
+ .deleteJob(NameIdentifierUtil.ofJob(METALAKE_NAME,
dropped.name())));
+
+ Assertions.assertEquals(0, countJobOwnerRows(dropped.id()));
+ Assertions.assertEquals(1, countJobOwnerRows(kept.id()));
+ }
+
@TestTemplate
public void testUpdateJob() throws IOException {
BaseMetalake metalake =
@@ -491,4 +533,10 @@ public class TestJobMetaService extends TestJDBCBackend {
.withFinishedAt(oldJob.finishedAt())
.build()));
}
+
+ private int countJobOwnerRows(long jobId) {
+ return countActiveRowsWhere(
+ "owner_meta",
+ String.format("metadata_object_id = %d AND metadata_object_type =
'JOB'", jobId));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
index 31ed1f9100..aca0ea972a 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
@@ -18,12 +18,17 @@
*/
package org.apache.gravitino.storage.relational.service;
+import com.google.common.collect.Lists;
import java.io.IOException;
import java.time.Instant;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.authorization.AuthorizationUtils;
+import org.apache.gravitino.authorization.Privileges;
+import org.apache.gravitino.authorization.SecurableObjects;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.job.JobHandle;
@@ -31,6 +36,7 @@ import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.JobEntity;
import org.apache.gravitino.meta.JobTemplateEntity;
+import org.apache.gravitino.meta.UserEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
import
org.apache.gravitino.storage.relational.EntityChangeLogNameIdentifierCodec;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
@@ -232,6 +238,66 @@ public class TestJobTemplateMetaService extends
TestJDBCBackend {
Assertions.assertEquals(0, jobs.size());
}
+ @TestTemplate
+ public void testDeleteJobTemplateCleansOwnersAndPrivileges() throws
IOException {
+ BaseMetalake metalake =
+ createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), METALAKE_NAME,
AUDIT_INFO);
+ backend.insert(metalake, false);
+ UserEntity user =
+ createUserEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofUserNamespace(METALAKE_NAME),
+ "job_template_owner",
+ AUDIT_INFO);
+ backend.insert(user, false);
+
+ JobTemplateEntity dropped = newShellJobTemplateEntity("dropped", "",
METALAKE_NAME);
+ JobTemplateEntity kept = newShellJobTemplateEntity("kept", "",
METALAKE_NAME);
+ backend.insert(dropped, false);
+ backend.insert(kept, false);
+ JobEntity droppedJob = newJobEntity("dropped", JobHandle.Status.SUCCEEDED,
METALAKE_NAME);
+ JobEntity keptJob = newJobEntity("kept", JobHandle.Status.SUCCEEDED,
METALAKE_NAME);
+ backend.insert(droppedJob, false);
+ backend.insert(keptJob, false);
+
+ OwnerMetaService ownerMetaService = OwnerMetaService.getInstance();
+ for (NameIdentifier ident :
+ new NameIdentifier[] {dropped.nameIdentifier(),
kept.nameIdentifier()}) {
+ ownerMetaService.setOwner(
+ ident, Entity.EntityType.JOB_TEMPLATE, user.nameIdentifier(),
user.type());
+ }
+ for (JobEntity job : new JobEntity[] {droppedJob, keptJob}) {
+ ownerMetaService.setOwner(
+ NameIdentifierUtil.ofJob(METALAKE_NAME, job.name()),
+ Entity.EntityType.JOB,
+ user.nameIdentifier(),
+ user.type());
+ }
+ backend.insert(
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "job_template_role",
+ AUDIT_INFO,
+ Lists.newArrayList(
+ SecurableObjects.ofJobTemplate(
+ "dropped",
Lists.newArrayList(Privileges.UseJobTemplate.allow())),
+ SecurableObjects.ofJobTemplate(
+ "kept",
Lists.newArrayList(Privileges.UseJobTemplate.allow()))),
+ null),
+ false);
+
+ Assertions.assertTrue(
+
JobTemplateMetaService.getInstance().deleteJobTemplate(dropped.nameIdentifier()));
+
+ Assertions.assertEquals(0, countOwnerRows(dropped.id(), "JOB_TEMPLATE"));
+ Assertions.assertEquals(0, countSecurableObjectRows(dropped.id(),
"JOB_TEMPLATE"));
+ Assertions.assertEquals(0, countOwnerRows(droppedJob.id(), "JOB"));
+ Assertions.assertEquals(1, countOwnerRows(kept.id(), "JOB_TEMPLATE"));
+ Assertions.assertEquals(1, countSecurableObjectRows(kept.id(),
"JOB_TEMPLATE"));
+ Assertions.assertEquals(1, countOwnerRows(keptJob.id(), "JOB"));
+ }
+
@TestTemplate
public void testUpdateJobTemplate() throws IOException {
BaseMetalake metalake =
@@ -376,6 +442,19 @@ public class TestJobTemplateMetaService extends
TestJDBCBackend {
Assertions.assertEquals(operateType, change.getOperateType());
}
+ private int countOwnerRows(long objectId, String objectType) {
+ return countActiveRowsWhere(
+ "owner_meta",
+ String.format(
+ "metadata_object_id = %d AND metadata_object_type = '%s'",
objectId, objectType));
+ }
+
+ private int countSecurableObjectRows(long objectId, String objectType) {
+ return countActiveRowsWhere(
+ "role_meta_securable_object",
+ String.format("metadata_object_id = %d AND type = '%s'", objectId,
objectType));
+ }
+
static JobEntity newJobEntity(String templateName, JobHandle.Status status,
String metalake) {
// Any status other than QUEUED implies the job has at least started.
boolean isStarted = status != JobHandle.Status.QUEUED;
