This is an automated email from the ASF dual-hosted git repository.
roryqi 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 2bcbd2014f [#12578] feat(core): add policy-to-tag relation storage
(#12579)
2bcbd2014f is described below
commit 2bcbd2014f2aca5f473e46b246eadf8b418d25b0
Author: roryqi <[email protected]>
AuthorDate: Fri Aug 28 00:23:13 2026 +0800
[#12578] feat(core): add policy-to-tag relation storage (#12579)
### What changes were proposed in this pull request?
This PR extracts the relational storage part of policy-on-tag from
#12536.
The changes include:
1. Add the `policy_tag_relation_meta` table and upgrade scripts for H2,
MySQL, and PostgreSQL.
2. Persist the optional selector JSON on each policy-to-tag relation.
3. Support atomic and idempotent relation updates.
4. Support bidirectional batch reads without resolving every anchor ID
separately.
5. Clean up relations when a policy, tag, metalake, or legacy record is
deleted.
REST APIs, clients, authorization, events, and policy resolution are
intentionally excluded from this PR.
### Why are the changes needed?
Policy-on-tag needs a persistent `Policy -> Tag` relation before
higher-level APIs and policy resolution can be implemented.
Splitting the storage layer from #12536 keeps the schema, transaction
behavior, and lifecycle cleanup independently reviewable.
Fix: #12578
Part of #12179 and #12176.
### Does this PR introduce _any_ user-facing change?
No. This PR only adds internal relational storage and schema support. It
does not add or change REST or client APIs.
### How was this patch tested?
- `./gradlew :core:spotlessApply`
- `./gradlew :core:test --tests
org.apache.gravitino.storage.relational.service.TestPolicyTagRelService
-PskipDockerTests=false`
- Covers H2, MySQL, and PostgreSQL.
- Covers atomic rollback, idempotent updates, bidirectional batch reads,
metalake isolation, and cascade cleanup.
- Related relational entity store, metalake, policy, and tag service
tests.
- `git diff --check`
---
.../org/apache/gravitino/RelationalEntity.java | 32 ++
.../gravitino/SupportsRelationOperations.java | 6 +
.../gravitino/storage/relational/JDBCBackend.java | 33 ++
.../storage/relational/RelationalEntityStore.java | 2 +
.../relational/mapper/PolicyTagRelMapper.java | 104 ++++
.../mapper/PolicyTagRelSQLProviderFactory.java | 94 ++++
.../storage/relational/mapper/TagMetaMapper.java | 9 +
.../mapper/TagMetaSQLProviderFactory.java | 5 +
.../provider/DefaultMapperPackageProvider.java | 2 +
.../provider/base/PolicyTagRelBaseSQLProvider.java | 150 +++++
.../provider/base/TagMetaBaseSQLProvider.java | 5 +
.../postgresql/PolicyTagRelPostgreSQLProvider.java | 55 ++
.../storage/relational/po/PolicyTagRelPO.java | 154 ++++++
.../relational/service/MetalakeMetaService.java | 7 +
.../relational/service/PolicyTagRelService.java | 408 ++++++++++++++
.../storage/relational/service/TagMetaService.java | 11 +-
.../apache/gravitino/storage/TestSQLScripts.java | 59 +-
.../service/TestPolicyTagRelService.java | 610 +++++++++++++++++++++
scripts/h2/schema-2.0.0-h2.sql | 14 +
scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql | 14 +
scripts/mysql/schema-2.0.0-mysql.sql | 14 +
scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql | 14 +
scripts/postgresql/schema-2.0.0-postgresql.sql | 24 +
.../upgrade-1.3.0-to-2.0.0-postgresql.sql | 24 +
24 files changed, 1834 insertions(+), 16 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/RelationalEntity.java
b/core/src/main/java/org/apache/gravitino/RelationalEntity.java
index 4c9cd20bcc..54cf1d41f1 100644
--- a/core/src/main/java/org/apache/gravitino/RelationalEntity.java
+++ b/core/src/main/java/org/apache/gravitino/RelationalEntity.java
@@ -17,6 +17,9 @@
package org.apache.gravitino;
+import java.util.Optional;
+import javax.annotation.Nullable;
+
/**
* Represents a directed relation between two entities. The source is
identified by a {@link
* NameIdentifier} and an {@link Entity.EntityType}; the target is the actual
resolved entity,
@@ -29,6 +32,7 @@ public class RelationalEntity<T extends Entity &
HasIdentifier> {
private final NameIdentifier source;
private final Entity.EntityType sourceType;
private final T targetEntity;
+ @Nullable private final String relationValue;
/**
* Constructs a RelationalEntity.
@@ -43,10 +47,29 @@ public class RelationalEntity<T extends Entity &
HasIdentifier> {
NameIdentifier source,
Entity.EntityType sourceType,
T targetEntity) {
+ this(type, source, sourceType, targetEntity, null);
+ }
+
+ /**
+ * Constructs a RelationalEntity with an optional value stored on the
relation edge.
+ *
+ * @param type the relation type
+ * @param source the source identifier
+ * @param sourceType the entity type of the source
+ * @param targetEntity the resolved target entity
+ * @param relationValue the optional value carried by the relation edge
+ */
+ public RelationalEntity(
+ SupportsRelationOperations.Type type,
+ NameIdentifier source,
+ Entity.EntityType sourceType,
+ T targetEntity,
+ @Nullable String relationValue) {
this.type = type;
this.source = source;
this.sourceType = sourceType;
this.targetEntity = targetEntity;
+ this.relationValue = relationValue;
}
/**
@@ -84,4 +107,13 @@ public class RelationalEntity<T extends Entity &
HasIdentifier> {
public T targetEntity() {
return targetEntity;
}
+
+ /**
+ * Gets the optional value stored on this relation edge.
+ *
+ * @return the relation value, or empty when the edge has no value
+ */
+ public Optional<String> relationValue() {
+ return Optional.ofNullable(relationValue);
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
b/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
index ca1ba1c230..6158c6c3bd 100644
--- a/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
+++ b/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
@@ -43,6 +43,8 @@ public interface SupportsRelationOperations {
POLICY_METADATA_OBJECT_REL,
/** Metadata object to tag relationship */
TAG_METADATA_OBJECT_REL,
+ /** Policy to tag relationship */
+ POLICY_TAG_REL,
}
/**
@@ -291,6 +293,10 @@ public interface SupportsRelationOperations {
return Entity.EntityType.POLICY;
case TAG_METADATA_OBJECT_REL:
return Entity.EntityType.TAG;
+ case POLICY_TAG_REL:
+ return srcEntityType == Entity.EntityType.TAG
+ ? Entity.EntityType.POLICY
+ : Entity.EntityType.TAG;
default:
return srcEntityType;
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
index 6661d0014d..a0f3701d2e 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
@@ -84,6 +84,7 @@ import
org.apache.gravitino.storage.relational.service.ModelVersionMetaService;
import
org.apache.gravitino.storage.relational.service.OrphanedMetadataObjectRelationService;
import org.apache.gravitino.storage.relational.service.OwnerMetaService;
import org.apache.gravitino.storage.relational.service.PolicyMetaService;
+import org.apache.gravitino.storage.relational.service.PolicyTagRelService;
import org.apache.gravitino.storage.relational.service.RoleMetaService;
import org.apache.gravitino.storage.relational.service.SchemaMetaService;
import org.apache.gravitino.storage.relational.service.StatisticMetaService;
@@ -737,6 +738,13 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
return (List<E>)
TagMetaService.getInstance().listTagsForMetadataObject(nameIdentifier,
identType);
}
+ case POLICY_TAG_REL:
+ return (List<E>)
+ PolicyTagRelService.getInstance()
+ .listRelations(List.of(nameIdentifier), identType)
+ .stream()
+ .map(RelationalEntity::targetEntity)
+ .collect(Collectors.toList());
default:
throw new IllegalArgumentException(
String.format("Doesn't support the relation type %s", relType));
@@ -750,6 +758,8 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
switch (relType) {
case OWNER_REL:
return OwnerMetaService.getInstance().batchGetOwner(nameIdentifiers,
identType);
+ case POLICY_TAG_REL:
+ return
PolicyTagRelService.getInstance().listRelations(nameIdentifiers, identType);
default:
throw new IllegalArgumentException(
String.format("Doesn't support the relation type %s", relType));
@@ -862,6 +872,14 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
update.sourceEntityType(),
toTagValues(update.targetsToAdd()),
toTagValues(update.targetsToRemove()));
+ case POLICY_TAG_REL:
+ Preconditions.checkArgument(
+ update.sourceEntityType() == Entity.EntityType.TAG,
+ "Policy-to-tag relation updates must use a tag as the source
entity");
+ return (List<E>)
+ PolicyTagRelService.getInstance()
+ .updateRelations(
+ update.sourceIdentifier(), update.targetsToAdd(),
update.targetsToRemove());
default:
Preconditions.checkArgument(
!update.hasRelationValues(),
@@ -921,6 +939,21 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
return (E)
TagMetaService.getInstance()
.getTagForMetadataObject(srcIdentifier, srcType,
destEntityIdent);
+ case POLICY_TAG_REL:
+ return (E)
+ PolicyTagRelService.getInstance()
+ .listRelations(List.of(srcIdentifier), srcType)
+ .stream()
+ .filter(
+ relation ->
relation.targetEntity().nameIdentifier().equals(destEntityIdent))
+ .map(RelationalEntity::targetEntity)
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ srcType == Entity.EntityType.TAG ? "policy" :
"tag",
+ destEntityIdent.name()));
default:
throw new IllegalArgumentException(
String.format("Doesn't support the relation type %s", relType));
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
index bdac4f92b8..bf6c50ee09 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -565,6 +565,8 @@ public class RelationalEntityStore
return Entity.EntityType.POLICY;
case TAG_METADATA_OBJECT_REL:
return Entity.EntityType.TAG;
+ case POLICY_TAG_REL:
+ return Entity.EntityType.POLICY;
default:
throw new IllegalArgumentException(
String.format("Doesn't support the relation type %s", relType));
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelMapper.java
new file mode 100644
index 0000000000..370c2b2f5e
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelMapper.java
@@ -0,0 +1,104 @@
+/*
+ * 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.storage.relational.mapper;
+
+import java.util.List;
+import org.apache.gravitino.storage.relational.po.PolicyTagRelPO;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+import org.apache.ibatis.annotations.UpdateProvider;
+
+/** MyBatis mapper for policy-to-tag relations. */
+public interface PolicyTagRelMapper {
+ /** The policy-to-tag relation table name. */
+ String POLICY_TAG_RELATION_TABLE_NAME = "policy_tag_relation_meta";
+
+ /**
+ * Lists active policy-to-tag relations for the given tag names in a
metalake.
+ *
+ * @param metalakeName The metalake name.
+ * @param tagNames The tag names.
+ * @return The active policy-to-tag relations.
+ */
+ @SelectProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"listByTagNames")
+ List<PolicyTagRelPO> listByTagNames(
+ @Param("metalakeName") String metalakeName, @Param("tagNames")
List<String> tagNames);
+
+ /**
+ * Lists active policy-to-tag relations for the given policy names in a
metalake.
+ *
+ * @param metalakeName The metalake name.
+ * @param policyNames The policy names.
+ * @return The active policy-to-tag relations.
+ */
+ @SelectProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"listByPolicyNames")
+ List<PolicyTagRelPO> listByPolicyNames(
+ @Param("metalakeName") String metalakeName, @Param("policyNames")
List<String> policyNames);
+
+ /**
+ * Gets one active relation by policy ID and tag ID.
+ *
+ * @param policyId The policy ID.
+ * @param tagId The tag ID.
+ * @return The active relation, or null if no relation exists.
+ */
+ @SelectProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"getByPolicyIdAndTagId")
+ PolicyTagRelPO getByPolicyIdAndTagId(
+ @Param("policyId") Long policyId, @Param("tagId") Long tagId);
+
+ /**
+ * Inserts a policy-to-tag relation if the active pair does not exist.
+ *
+ * @param relation The relation to insert.
+ * @return The number of inserted rows.
+ */
+ @InsertProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"insertIfAbsent")
+ int insertIfAbsent(@Param("relation") PolicyTagRelPO relation);
+
+ /**
+ * Soft-deletes one active relation and advances its optimistic-concurrency
version.
+ *
+ * @param relation The observed relation and version.
+ * @return The number of updated rows.
+ */
+ @UpdateProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"softDeleteByIdAndVersion")
+ int softDeleteByIdAndVersion(@Param("relation") PolicyTagRelPO relation);
+
+ /**
+ * Soft-deletes active relations in a metalake.
+ *
+ * @param metalakeId The metalake ID.
+ * @return The number of updated rows.
+ */
+ @UpdateProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"softDeleteByMetalakeId")
+ int softDeleteByMetalakeId(@Param("metalakeId") Long metalakeId);
+
+ /**
+ * Physically deletes expired relation rows.
+ *
+ * @param legacyTimeline The exclusive deletion timestamp upper bound.
+ * @param limit The maximum number of rows to delete.
+ * @return The number of deleted rows.
+ */
+ @DeleteProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"deleteByLegacyTimeline")
+ int deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelSQLProviderFactory.java
new file mode 100644
index 0000000000..d9cdb827ce
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelSQLProviderFactory.java
@@ -0,0 +1,94 @@
+/*
+ * 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.storage.relational.mapper;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import
org.apache.gravitino.storage.relational.mapper.provider.base.PolicyTagRelBaseSQLProvider;
+import
org.apache.gravitino.storage.relational.mapper.provider.postgresql.PolicyTagRelPostgreSQLProvider;
+import org.apache.gravitino.storage.relational.po.PolicyTagRelPO;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/** Selects the policy-to-tag relation SQL provider for the active JDBC
backend. */
+public class PolicyTagRelSQLProviderFactory {
+
+ private static final Map<JDBCBackendType, PolicyTagRelBaseSQLProvider>
PROVIDERS =
+ ImmutableMap.of(
+ JDBCBackendType.MYSQL, new PolicyTagRelMySQLProvider(),
+ JDBCBackendType.H2, new PolicyTagRelH2Provider(),
+ JDBCBackendType.POSTGRESQL, new PolicyTagRelPostgreSQLProvider());
+
+ /**
+ * @return The SQL provider for the active backend.
+ */
+ public static PolicyTagRelBaseSQLProvider getProvider() {
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+ return PROVIDERS.get(JDBCBackendType.fromString(databaseId));
+ }
+
+ /** Delegates a tag-anchored list query. */
+ public static String listByTagNames(
+ @Param("metalakeName") String metalakeName, @Param("tagNames")
List<String> tagNames) {
+ return getProvider().listByTagNames(metalakeName, tagNames);
+ }
+
+ /** Delegates a policy-anchored list query. */
+ public static String listByPolicyNames(
+ @Param("metalakeName") String metalakeName, @Param("policyNames")
List<String> policyNames) {
+ return getProvider().listByPolicyNames(metalakeName, policyNames);
+ }
+
+ /** Delegates a single relation query. */
+ public static String getByPolicyIdAndTagId(
+ @Param("policyId") Long policyId, @Param("tagId") Long tagId) {
+ return getProvider().getByPolicyIdAndTagId(policyId, tagId);
+ }
+
+ /** Delegates an insert-if-absent operation. */
+ public static String insertIfAbsent(@Param("relation") PolicyTagRelPO
relation) {
+ return getProvider().insertIfAbsent(relation);
+ }
+
+ /** Delegates a relation soft delete. */
+ public static String softDeleteByIdAndVersion(@Param("relation")
PolicyTagRelPO relation) {
+ return getProvider().softDeleteByIdAndVersion(relation);
+ }
+
+ /** Delegates metalake deletion cleanup. */
+ public static String softDeleteByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ return getProvider().softDeleteByMetalakeId(metalakeId);
+ }
+
+ /** Delegates expired relation cleanup. */
+ public static String deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return getProvider().deleteByLegacyTimeline(legacyTimeline, limit);
+ }
+
+ static class PolicyTagRelMySQLProvider extends PolicyTagRelBaseSQLProvider {}
+
+ static class PolicyTagRelH2Provider extends PolicyTagRelBaseSQLProvider {}
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
index 68cb3d721f..2d6095bf03 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
@@ -80,6 +80,15 @@ public interface TagMetaMapper {
@SelectProvider(type = TagMetaSQLProviderFactory.class, method =
"selectTagByTagId")
TagPO selectTagByTagId(@Param("tagId") Long tagId);
+ /**
+ * Selects and exclusively locks an active tag by ID.
+ *
+ * @param tagId The tag ID.
+ * @return The locked tag, or null if it is not active.
+ */
+ @SelectProvider(type = TagMetaSQLProviderFactory.class, method =
"selectTagByTagIdForUpdate")
+ TagPO selectTagByTagIdForUpdate(@Param("tagId") Long tagId);
+
@SelectProvider(type = TagMetaSQLProviderFactory.class, method =
"listTagPOsByTagIds")
List<TagPO> listTagPOsByTagIds(@Param("tagIds") List<Long> tagIds);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
index 1dde368e69..400bcc0e08 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
@@ -106,6 +106,11 @@ public class TagMetaSQLProviderFactory {
return getProvider().selectTagByTagId(tagId);
}
+ /** Delegates an exclusive-lock tag query. */
+ public static String selectTagByTagIdForUpdate(@Param("tagId") Long tagId) {
+ return getProvider().selectTagByTagIdForUpdate(tagId);
+ }
+
public static String listTagPOsByTagIds(@Param("tagIds") List<Long> tagIds) {
return getProvider().listTagPOsByTagIds(tagIds);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
index 8065ecfa3b..6fdf25d142 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
@@ -38,6 +38,7 @@ import
org.apache.gravitino.storage.relational.mapper.OrphanedMetadataObjectRela
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.PolicyMetadataObjectRelMapper;
+import org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper;
import org.apache.gravitino.storage.relational.mapper.PolicyVersionMapper;
import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
@@ -77,6 +78,7 @@ public class DefaultMapperPackageProvider implements
MapperPackageProvider {
OrphanedMetadataObjectRelationMapper.class,
OwnerMetaMapper.class,
PolicyMetadataObjectRelMapper.class,
+ PolicyTagRelMapper.class,
PolicyMetaMapper.class,
PolicyVersionMapper.class,
RoleMetaMapper.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyTagRelBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyTagRelBaseSQLProvider.java
new file mode 100644
index 0000000000..b81e6b9ec6
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyTagRelBaseSQLProvider.java
@@ -0,0 +1,150 @@
+/*
+ * 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.storage.relational.mapper.provider.base;
+
+import static
org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper.POLICY_TAG_RELATION_TABLE_NAME;
+
+import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
+import org.apache.gravitino.storage.relational.po.PolicyTagRelPO;
+import org.apache.ibatis.annotations.Param;
+
+/** Base SQL provider for policy-to-tag relations. */
+public class PolicyTagRelBaseSQLProvider {
+
+ /** Returns SQL for listing relations anchored by tag names. */
+ public String listByTagNames(
+ @Param("metalakeName") String metalakeName, @Param("tagNames")
List<String> tagNames) {
+ return listRelations("tm.metalake_id", "tm.tag_name", "tagNames",
"tagName");
+ }
+
+ /** Returns SQL for listing relations anchored by policy names. */
+ public String listByPolicyNames(
+ @Param("metalakeName") String metalakeName, @Param("policyNames")
List<String> policyNames) {
+ return listRelations("pm.metalake_id", "pm.policy_name", "policyNames",
"policyName");
+ }
+
+ /** Returns SQL for getting one active relation. */
+ public String getByPolicyIdAndTagId(
+ @Param("policyId") Long policyId, @Param("tagId") Long tagId) {
+ return selectColumns()
+ + joins()
+ + " WHERE ptr.policy_id = #{policyId} AND ptr.tag_id = #{tagId}"
+ + activePredicates();
+ }
+
+ /** Returns SQL for inserting one relation if the active pair does not
exist. */
+ public String insertIfAbsent(@Param("relation") PolicyTagRelPO relation) {
+ return "INSERT IGNORE INTO "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " (policy_id, tag_id, selector, audit_info, current_version,
last_version, deleted_at)"
+ + " VALUES (#{relation.policyId}, #{relation.tagId},
#{relation.selector},"
+ + " #{relation.auditInfo}, #{relation.currentVersion},
#{relation.lastVersion},"
+ + " #{relation.deletedAt})";
+ }
+
+ /** Returns SQL for soft-deleting one relation with a version CAS. */
+ public String softDeleteByIdAndVersion(@Param("relation") PolicyTagRelPO
relation) {
+ return "UPDATE "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ // Advance the OCC token in the same statement as the soft delete.
Assign last_version
+ // first so MySQL computes both columns from the version observed
before this update.
+ + " SET last_version = current_version + 1,"
+ + " current_version = current_version + 1,"
+ + " deleted_at = "
+ + deletedAtNowExpression()
+ + " WHERE id = #{relation.id} AND current_version =
#{relation.currentVersion}"
+ + " AND deleted_at = 0";
+ }
+
+ /** Returns SQL for soft-deleting relations when a metalake is deleted. */
+ public String softDeleteByMetalakeId(@Param("metalakeId") Long metalakeId) {
+ return "UPDATE "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " SET deleted_at = "
+ + deletedAtNowExpression()
+ + " WHERE EXISTS (SELECT * FROM "
+ + PolicyMetaMapper.POLICY_META_TABLE_NAME
+ + " pm WHERE pm.metalake_id = #{metalakeId} AND pm.policy_id = "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + ".policy_id)"
+ + " AND deleted_at = 0";
+ }
+
+ /** Returns SQL for physically deleting expired relation rows. */
+ public String deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return "DELETE FROM "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit}";
+ }
+
+ /** Returns the database expression for the current epoch-millisecond
timestamp. */
+ protected String deletedAtNowExpression() {
+ return "(UNIX_TIMESTAMP() * 1000.0)"
+ + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000";
+ }
+
+ private String listRelations(
+ String metalakeIdColumn, String nameColumn, String collection, String
item) {
+ return "<script>"
+ + selectColumns()
+ + joins()
+ + " WHERE "
+ + metalakeIdColumn
+ + " IN (SELECT mm.metalake_id FROM "
+ + MetalakeMetaMapper.TABLE_NAME
+ + " mm WHERE mm.metalake_name = #{metalakeName} AND mm.deleted_at = 0)"
+ + " AND "
+ + nameColumn
+ + " IN <foreach item='"
+ + item
+ + "' collection='"
+ + collection
+ + "' open='(' separator=',' close=')'>#{"
+ + item
+ + "}</foreach>"
+ + activePredicates()
+ + " ORDER BY tm.tag_name, pm.policy_name"
+ + "</script>";
+ }
+
+ private String selectColumns() {
+ return "SELECT ptr.id AS id, ptr.policy_id AS policyId, pm.policy_name AS
policyName,"
+ + " ptr.tag_id AS tagId, tm.tag_name AS tagName, ptr.selector,"
+ + " ptr.audit_info AS auditInfo, ptr.current_version AS
currentVersion,"
+ + " ptr.last_version AS lastVersion, ptr.deleted_at AS deletedAt";
+ }
+
+ private String joins() {
+ return " FROM "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " ptr JOIN "
+ + PolicyMetaMapper.POLICY_META_TABLE_NAME
+ + " pm ON ptr.policy_id = pm.policy_id JOIN "
+ + TagMetaMapper.TAG_TABLE_NAME
+ + " tm ON ptr.tag_id = tm.tag_id";
+ }
+
+ private String activePredicates() {
+ return " AND ptr.deleted_at = 0 AND pm.deleted_at = 0 AND tm.deleted_at =
0";
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
index e33f2080c9..8ab7a8fe06 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
@@ -233,6 +233,11 @@ public class TagMetaBaseSQLProvider {
+ " WHERE tag_id = #{tagId} and deleted_at = 0";
}
+ /** Returns SQL that selects and locks an active tag by ID. */
+ public String selectTagByTagIdForUpdate(@Param("tagId") Long tagId) {
+ return selectTagByTagId(tagId) + " FOR UPDATE";
+ }
+
public String listTagPOsByTagIds(@Param("tagIds") List<Long> tagIds) {
return "<script>"
+ "SELECT tag_id as tagId, tag_name as tagName,"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyTagRelPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyTagRelPostgreSQLProvider.java
new file mode 100644
index 0000000000..101a3014ad
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyTagRelPostgreSQLProvider.java
@@ -0,0 +1,55 @@
+/*
+ * 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.storage.relational.mapper.provider.postgresql;
+
+import static
org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper.POLICY_TAG_RELATION_TABLE_NAME;
+
+import
org.apache.gravitino.storage.relational.mapper.provider.base.PolicyTagRelBaseSQLProvider;
+import org.apache.gravitino.storage.relational.po.PolicyTagRelPO;
+import org.apache.ibatis.annotations.Param;
+
+/** PostgreSQL SQL provider for policy-to-tag relations. */
+public class PolicyTagRelPostgreSQLProvider extends
PolicyTagRelBaseSQLProvider {
+
+ @Override
+ public String insertIfAbsent(@Param("relation") PolicyTagRelPO relation) {
+ return "INSERT INTO "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " (policy_id, tag_id, selector, audit_info, current_version,
last_version, deleted_at)"
+ + " VALUES (#{relation.policyId}, #{relation.tagId},
#{relation.selector},"
+ + " #{relation.auditInfo}, #{relation.currentVersion},
#{relation.lastVersion},"
+ + " #{relation.deletedAt})"
+ + " ON CONFLICT (policy_id, tag_id, deleted_at) DO NOTHING";
+ }
+
+ @Override
+ public String deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return "DELETE FROM "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " WHERE id IN (SELECT id FROM "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit})";
+ }
+
+ @Override
+ protected String deletedAtNowExpression() {
+ return "CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)";
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/PolicyTagRelPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/PolicyTagRelPO.java
new file mode 100644
index 0000000000..29d8bd17e0
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/po/PolicyTagRelPO.java
@@ -0,0 +1,154 @@
+/*
+ * 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.storage.relational.po;
+
+import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
+import javax.annotation.Nullable;
+import lombok.Getter;
+
+/** Persistent object for a policy-to-tag relation row. */
+@Getter
+public class PolicyTagRelPO {
+ private Long id;
+ private Long policyId;
+ private String policyName;
+ private Long tagId;
+ private String tagName;
+ @Nullable private String selector;
+ private String auditInfo;
+ private Long currentVersion;
+ private Long lastVersion;
+ private Long deletedAt;
+
+ /**
+ * @return A builder for a policy-to-tag relation persistent object.
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof PolicyTagRelPO)) {
+ return false;
+ }
+ PolicyTagRelPO that = (PolicyTagRelPO) o;
+ return Objects.equal(id, that.id)
+ && Objects.equal(policyId, that.policyId)
+ && Objects.equal(policyName, that.policyName)
+ && Objects.equal(tagId, that.tagId)
+ && Objects.equal(tagName, that.tagName)
+ && Objects.equal(selector, that.selector)
+ && Objects.equal(auditInfo, that.auditInfo)
+ && Objects.equal(currentVersion, that.currentVersion)
+ && Objects.equal(lastVersion, that.lastVersion)
+ && Objects.equal(deletedAt, that.deletedAt);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(
+ id,
+ policyId,
+ policyName,
+ tagId,
+ tagName,
+ selector,
+ auditInfo,
+ currentVersion,
+ lastVersion,
+ deletedAt);
+ }
+
+ /** Builder for {@link PolicyTagRelPO}. */
+ public static class Builder {
+ private final PolicyTagRelPO relation;
+
+ private Builder() {
+ relation = new PolicyTagRelPO();
+ }
+
+ /** Sets the relation row ID. */
+ public Builder withId(Long id) {
+ relation.id = id;
+ return this;
+ }
+
+ /** Sets the policy ID. */
+ public Builder withPolicyId(Long policyId) {
+ relation.policyId = policyId;
+ return this;
+ }
+
+ /** Sets the tag ID. */
+ public Builder withTagId(Long tagId) {
+ relation.tagId = tagId;
+ return this;
+ }
+
+ /** Sets the selector JSON. */
+ public Builder withSelector(@Nullable String selector) {
+ relation.selector = selector;
+ return this;
+ }
+
+ /** Sets the audit information JSON. */
+ public Builder withAuditInfo(String auditInfo) {
+ relation.auditInfo = auditInfo;
+ return this;
+ }
+
+ /** Sets the current version. */
+ public Builder withCurrentVersion(Long currentVersion) {
+ relation.currentVersion = currentVersion;
+ return this;
+ }
+
+ /** Sets the last version. */
+ public Builder withLastVersion(Long lastVersion) {
+ relation.lastVersion = lastVersion;
+ return this;
+ }
+
+ /** Sets the deletion timestamp. */
+ public Builder withDeletedAt(Long deletedAt) {
+ relation.deletedAt = deletedAt;
+ return this;
+ }
+
+ /**
+ * Builds the persistent object.
+ *
+ * @return The persistent object.
+ */
+ public PolicyTagRelPO build() {
+ Preconditions.checkArgument(relation.policyId != null, "Policy id is
required");
+ Preconditions.checkArgument(relation.tagId != null, "Tag id is
required");
+ Preconditions.checkArgument(relation.auditInfo != null, "Audit info is
required");
+ Preconditions.checkArgument(relation.currentVersion != null, "Current
version is required");
+ Preconditions.checkArgument(relation.lastVersion != null, "Last version
is required");
+ Preconditions.checkArgument(relation.deletedAt != null, "Deleted at is
required");
+ return relation;
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java
index 58c8c054fe..fb63973758 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java
@@ -49,6 +49,7 @@ import
org.apache.gravitino.storage.relational.mapper.ModelVersionAliasRelMapper
import org.apache.gravitino.storage.relational.mapper.ModelVersionMetaMapper;
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper;
import org.apache.gravitino.storage.relational.mapper.PolicyVersionMapper;
import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
@@ -283,6 +284,9 @@ public class MetalakeMetaService {
SessionUtils.doWithoutCommit(
TagMetadataObjectRelMapper.class,
mapper ->
mapper.softDeleteTagMetadataObjectRelsByMetalakeId(metalakeId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByMetalakeId(metalakeId)),
() ->
SessionUtils.doWithoutCommit(
PolicyMetaMapper.class,
@@ -374,6 +378,9 @@ public class MetalakeMetaService {
SessionUtils.doWithoutCommit(
TagMetadataObjectRelMapper.class,
mapper ->
mapper.softDeleteTagMetadataObjectRelsByMetalakeId(metalakeId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByMetalakeId(metalakeId)),
() ->
SessionUtils.doWithoutCommit(
OwnerMetaMapper.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
new file mode 100644
index 0000000000..fc011456c2
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
@@ -0,0 +1,408 @@
+/*
+ * 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.storage.relational.service;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.RelationEdgeTarget;
+import org.apache.gravitino.RelationalEntity;
+import org.apache.gravitino.SupportsRelationOperations;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.json.JsonUtils;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.meta.TagEntity;
+import org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
+import org.apache.gravitino.storage.relational.po.PolicyPO;
+import org.apache.gravitino.storage.relational.po.PolicyTagRelPO;
+import org.apache.gravitino.storage.relational.po.TagPO;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.PrincipalUtils;
+
+/** JDBC metadata service for policy-to-tag relations. */
+public class PolicyTagRelService {
+
+ private static final PolicyTagRelService INSTANCE = new
PolicyTagRelService();
+
+ /**
+ * @return The singleton service instance.
+ */
+ public static PolicyTagRelService getInstance() {
+ return INSTANCE;
+ }
+
+ private PolicyTagRelService() {}
+
+ /**
+ * Lists policy-to-tag relation edges from policy or tag anchors.
+ *
+ * @param anchors The policy or tag identifiers to query.
+ * @param anchorType The entity type of every anchor.
+ * @return The relation edges, including selector JSON as the relation value.
+ */
+ public List<RelationalEntity<?>> listRelations(
+ List<NameIdentifier> anchors, Entity.EntityType anchorType) {
+ if (anchors == null || anchors.isEmpty()) {
+ return Collections.emptyList();
+ }
+ Preconditions.checkArgument(
+ anchorType == Entity.EntityType.TAG || anchorType ==
Entity.EntityType.POLICY,
+ "Policy-to-tag relations do not support anchor type %s",
+ anchorType);
+ validateSameMetalake(anchors);
+
+ String metalake = anchors.get(0).namespace().level(0);
+ List<String> anchorNames =
+
anchors.stream().map(NameIdentifier::name).distinct().collect(Collectors.toList());
+ List<PolicyTagRelPO> relations =
+ SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class,
+ mapper ->
+ anchorType == Entity.EntityType.TAG
+ ? mapper.listByTagNames(metalake, anchorNames)
+ : mapper.listByPolicyNames(metalake, anchorNames));
+ if (relations.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ return anchorType == Entity.EntityType.TAG
+ ? policyTargets(metalake, relations)
+ : tagTargets(metalake, relations);
+ }
+
+ /**
+ * Creates or removes policy-to-tag relations for one tag.
+ *
+ * <p>An add for an existing policy and tag pair conflicts regardless of its
selector. Removing a
+ * missing pair is an idempotent no-op. The same pair cannot be added and
removed in one update.
+ *
+ * @param tagIdentifier The source tag identifier.
+ * @param targetsToAdd Policy targets to create.
+ * @param targetsToRemove Policy targets to remove.
+ * @return All active policy targets for the tag after the update.
+ * @throws IOException If selector audit information cannot be serialized.
+ * @throws EntityAlreadyExistsException If a relation to add already exists.
+ * @throws IllegalArgumentException If the same relation is both added and
removed.
+ */
+ public List<PolicyEntity> updateRelations(
+ NameIdentifier tagIdentifier,
+ RelationEdgeTarget[] targetsToAdd,
+ RelationEdgeTarget[] targetsToRemove)
+ throws IOException {
+ NameIdentifierUtil.checkTag(tagIdentifier);
+ String metalake = tagIdentifier.namespace().level(0);
+ RelationEdgeTarget[] targetsToAddOrEmpty = nullToEmpty(targetsToAdd);
+ RelationEdgeTarget[] targetsToRemoveOrEmpty = nullToEmpty(targetsToRemove);
+ validatePolicyTargets(metalake, targetsToAddOrEmpty);
+ validatePolicyTargets(metalake, targetsToRemoveOrEmpty);
+ validateNoOverlappingTargets(targetsToAddOrEmpty, targetsToRemoveOrEmpty);
+
+ List<PolicyEntity> updatedPolicies = new ArrayList<>();
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ try {
+ updatedPolicies.addAll(
+ updateRelationsWithoutCommit(
+ tagIdentifier, targetsToAddOrEmpty,
targetsToRemoveOrEmpty));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
+ }
+ return updatedPolicies;
+ }
+
+ private List<PolicyEntity> updateRelationsWithoutCommit(
+ NameIdentifier tagIdentifier,
+ RelationEdgeTarget[] targetsToAdd,
+ RelationEdgeTarget[] targetsToRemove)
+ throws IOException {
+ String metalake = tagIdentifier.namespace().level(0);
+ TagPO tagPO = lockTag(tagIdentifier);
+ long tagId = tagPO.getTagId();
+ Map<String, Long> policyIds = resolvePolicyIds(metalake, targetsToAdd,
targetsToRemove);
+
+ for (RelationEdgeTarget target : targetsToRemove) {
+ long policyId = policyIds.get(target.nameIdentifier().name());
+ PolicyTagRelPO existing =
+ SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.getByPolicyIdAndTagId(policyId, tagId));
+ if (existing != null) {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByIdAndVersion(existing));
+ if (deleted != 1) {
+ throw relationConflict(tagIdentifier);
+ }
+ }
+ }
+
+ for (RelationEdgeTarget target : targetsToAdd) {
+ long policyId = policyIds.get(target.nameIdentifier().name());
+ insertIfAbsent(
+ tagIdentifier,
+ target.nameIdentifier(),
+ policyId,
+ tagId,
+ target.relationValue().orElse(null));
+ }
+
+ return listRelations(Collections.singletonList(tagIdentifier),
Entity.EntityType.TAG).stream()
+ .map(relation -> (PolicyEntity) relation.targetEntity())
+ .collect(Collectors.toList());
+ }
+
+ private static List<RelationalEntity<?>> policyTargets(
+ String metalake, List<PolicyTagRelPO> relations) {
+ Set<String> policyNames =
+ relations.stream()
+ .map(PolicyTagRelPO::getPolicyName)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ List<NameIdentifier> policyIdentifiers =
+ policyNames.stream()
+ .map(name -> NameIdentifierUtil.ofPolicy(metalake, name))
+ .collect(Collectors.toList());
+ Map<String, PolicyEntity> policies =
+
PolicyMetaService.getInstance().batchGetPolicyByIdentifier(policyIdentifiers).stream()
+ .collect(Collectors.toMap(PolicyEntity::name, policy -> policy));
+
+ List<RelationalEntity<?>> result = new ArrayList<>();
+ for (PolicyTagRelPO relation : relations) {
+ PolicyEntity policy = policies.get(relation.getPolicyName());
+ if (policy != null) {
+ result.add(
+ new RelationalEntity<>(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ NameIdentifierUtil.ofTag(metalake, relation.getTagName()),
+ Entity.EntityType.TAG,
+ policy,
+ relation.getSelector()));
+ }
+ }
+ return result;
+ }
+
+ private static List<RelationalEntity<?>> tagTargets(
+ String metalake, List<PolicyTagRelPO> relations) {
+ Set<String> tagNames =
+ relations.stream()
+ .map(PolicyTagRelPO::getTagName)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ List<NameIdentifier> tagIdentifiers =
+ tagNames.stream()
+ .map(name -> NameIdentifierUtil.ofTag(metalake, name))
+ .collect(Collectors.toList());
+ Map<String, TagEntity> tags =
+
TagMetaService.getInstance().batchGetTagByIdentifier(tagIdentifiers).stream()
+ .collect(Collectors.toMap(TagEntity::name, tag -> tag));
+
+ List<RelationalEntity<?>> result = new ArrayList<>();
+ for (PolicyTagRelPO relation : relations) {
+ TagEntity tag = tags.get(relation.getTagName());
+ if (tag != null) {
+ result.add(
+ new RelationalEntity<>(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ NameIdentifierUtil.ofPolicy(metalake,
relation.getPolicyName()),
+ Entity.EntityType.POLICY,
+ tag,
+ relation.getSelector()));
+ }
+ }
+ return result;
+ }
+
+ private static void insertIfAbsent(
+ NameIdentifier tagIdentifier,
+ NameIdentifier policyIdentifier,
+ long policyId,
+ long tagId,
+ String selector)
+ throws IOException {
+ PolicyTagRelPO relation =
+ PolicyTagRelPO.builder()
+ .withPolicyId(policyId)
+ .withTagId(tagId)
+ .withSelector(selector)
+ .withAuditInfo(auditInfo())
+ .withCurrentVersion(1L)
+ .withLastVersion(1L)
+ .withDeletedAt(0L)
+ .build();
+ int inserted =
+ SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.insertIfAbsent(relation));
+ if (inserted == 1) {
+ return;
+ }
+
+ // A zero-row insert means another writer won the active-pair uniqueness
race. Re-read the
+ // pair to distinguish an existing relation from another concurrent state
transition.
+ PolicyTagRelPO winner =
+ SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.getByPolicyIdAndTagId(policyId, tagId));
+ if (winner != null) {
+ throw relationAlreadyExists(tagIdentifier, policyIdentifier);
+ }
+ throw relationConflict(tagIdentifier);
+ }
+
+ private static String auditInfo() throws IOException {
+ String principal = PrincipalUtils.getCurrentPrincipal().getName();
+ Instant now = Instant.now();
+ AuditInfo auditInfo =
AuditInfo.builder().withCreator(principal).withCreateTime(now).build();
+ return JsonUtils.anyFieldMapper().writeValueAsString(auditInfo);
+ }
+
+ private static OptimisticLockException relationConflict(NameIdentifier
tagIdentifier) {
+ return new OptimisticLockException(
+ "A policy-to-tag relation for tag %s was modified concurrently; retry
the operation",
+ tagIdentifier);
+ }
+
+ private static EntityAlreadyExistsException relationAlreadyExists(
+ NameIdentifier tagIdentifier, NameIdentifier policyIdentifier) {
+ return new EntityAlreadyExistsException(
+ "The policy-to-tag relation between tag %s and policy %s already
exists",
+ tagIdentifier, policyIdentifier);
+ }
+
+ private static void validatePolicyTarget(String metalake, RelationEdgeTarget
target) {
+ Preconditions.checkArgument(target != null, "Policy relation target cannot
be null");
+ Preconditions.checkArgument(
+ target.entityType() == Entity.EntityType.POLICY,
+ "Policy-to-tag relation target must be POLICY, but is %s",
+ target.entityType());
+ Preconditions.checkArgument(
+ target.nameIdentifier().namespace().length() > 0
+ && metalake.equals(target.nameIdentifier().namespace().level(0)),
+ "Policy and tag must belong to the same metalake");
+ }
+
+ private static void validatePolicyTargets(String metalake,
RelationEdgeTarget[] targets) {
+ for (RelationEdgeTarget target : targets) {
+ validatePolicyTarget(metalake, target);
+ }
+ }
+
+ private static void validateNoOverlappingTargets(
+ RelationEdgeTarget[] targetsToAdd, RelationEdgeTarget[] targetsToRemove)
{
+ Set<String> policyNamesToAdd =
+ Arrays.stream(targetsToAdd)
+ .map(target -> target.nameIdentifier().name())
+ .collect(Collectors.toSet());
+ for (RelationEdgeTarget target : targetsToRemove) {
+ Preconditions.checkArgument(
+ !policyNamesToAdd.contains(target.nameIdentifier().name()),
+ "Policy-to-tag relation target %s cannot be both added and removed",
+ target.nameIdentifier());
+ }
+ }
+
+ private static TagPO lockTag(NameIdentifier tagIdentifier) {
+ String metalake = tagIdentifier.namespace().level(0);
+ return SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper -> {
+ TagPO observed = mapper.selectTagMetaByMetalakeAndName(metalake,
tagIdentifier.name());
+ if (observed == null) {
+ throw noSuchEntity(Entity.EntityType.TAG, tagIdentifier.name());
+ }
+ TagPO locked = mapper.selectTagByTagIdForUpdate(observed.getTagId());
+ if (locked == null
+ || !Objects.equals(locked.getTagName(), tagIdentifier.name())
+ || !Objects.equals(locked.getMetalakeId(),
observed.getMetalakeId())) {
+ throw noSuchEntity(Entity.EntityType.TAG, tagIdentifier.name());
+ }
+ return locked;
+ });
+ }
+
+ private static Map<String, Long> resolvePolicyIds(
+ String metalake, RelationEdgeTarget[] targetsToAdd, RelationEdgeTarget[]
targetsToRemove) {
+ Set<String> policyNames = new LinkedHashSet<>();
+ Arrays.stream(targetsToAdd)
+ .map(target -> target.nameIdentifier().name())
+ .forEach(policyNames::add);
+ Arrays.stream(targetsToRemove)
+ .map(target -> target.nameIdentifier().name())
+ .forEach(policyNames::add);
+ if (policyNames.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ List<PolicyPO> policies =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
+ mapper.listPolicyPOsByMetalakeAndPolicyNames(
+ metalake, new ArrayList<>(policyNames)));
+ Map<String, Long> policyIds =
+ policies.stream().collect(Collectors.toMap(PolicyPO::getPolicyName,
PolicyPO::getPolicyId));
+
+ for (String policyName : policyNames) {
+ if (!policyIds.containsKey(policyName)) {
+ throw noSuchEntity(Entity.EntityType.POLICY, policyName);
+ }
+ }
+ return policyIds;
+ }
+
+ private static NoSuchEntityException noSuchEntity(Entity.EntityType type,
String name) {
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
type.name().toLowerCase(), name);
+ }
+
+ private static void validateSameMetalake(List<NameIdentifier> identifiers) {
+ Preconditions.checkArgument(
+ identifiers.stream()
+ .allMatch(identifier -> identifier != null &&
identifier.namespace().length() > 0),
+ "All policy-to-tag relation anchors must have a metalake namespace");
+ String metalake = identifiers.get(0).namespace().level(0);
+ Preconditions.checkArgument(
+ identifiers.stream()
+ .allMatch(identifier ->
metalake.equals(identifier.namespace().level(0))),
+ "All policy-to-tag relation anchors must belong to the same metalake");
+ }
+
+ private static RelationEdgeTarget[] nullToEmpty(RelationEdgeTarget[]
targets) {
+ return targets == null ? new RelationEdgeTarget[0] :
Arrays.copyOf(targets, targets.length);
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
index 92b6362379..6726bc8de7 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
@@ -49,6 +49,7 @@ import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.meta.GenericEntity;
import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.metrics.Monitored;
+import org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper;
import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
import org.apache.gravitino.storage.relational.po.TagMetadataObjectRelPO;
@@ -504,6 +505,7 @@ public class TagMetaService {
public int deleteTagMetasByLegacyTimeline(long legacyTimeline, int limit) {
int[] tagDeletedCount = new int[] {0};
int[] tagMetadataObjectRelDeletedCount = new int[] {0};
+ int[] policyTagRelDeletedCount = new int[] {0};
SessionUtils.doMultipleWithCommit(
() ->
@@ -515,9 +517,14 @@ public class TagMetaService {
tagMetadataObjectRelDeletedCount[0] =
SessionUtils.getWithoutCommit(
TagMetadataObjectRelMapper.class,
- mapper ->
mapper.deleteTagEntityRelsByLegacyTimeline(legacyTimeline, limit)));
+ mapper ->
mapper.deleteTagEntityRelsByLegacyTimeline(legacyTimeline, limit)),
+ () ->
+ policyTagRelDeletedCount[0] =
+ SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class,
+ mapper -> mapper.deleteByLegacyTimeline(legacyTimeline,
limit)));
- return tagDeletedCount[0] + tagMetadataObjectRelDeletedCount[0];
+ return tagDeletedCount[0] + tagMetadataObjectRelDeletedCount[0] +
policyTagRelDeletedCount[0];
}
private static List<TagEntity> tagPOsToTagEntities(List<TagPO> tagPOs,
Namespace namespace) {
diff --git
a/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
b/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
index d3aa2fbbcc..ba06be927c 100644
--- a/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
+++ b/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
@@ -91,20 +91,51 @@ public class TestSQLScripts extends TestJDBCBackend {
for (List<File> scripts : versionScrips.values()) {
dropAllTables();
for (File scriptFile : scripts) {
- List<String> ddls = extractStatements(scriptFile.toPath());
-
- try (SqlSession sqlSession =
-
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true))
{
- try (Connection connection = sqlSession.getConnection()) {
- try (Statement statement = connection.createStatement()) {
- for (String ddl : ddls) {
- Assertions.assertDoesNotThrow(
- () -> statement.execute(ddl),
- "Failed to execute DDL in file " + scriptFile.getName() +
"ddl: " + ddl);
- }
- }
- }
- }
+ executeScript(scriptFile);
+ }
+ }
+ }
+
+ @TestTemplate
+ public void testUpgradeSQLScripts() throws SQLException, IOException {
+ String gravitinoHome = System.getenv("GRAVITINO_HOME");
+ Assertions.assertNotNull(gravitinoHome, "GRAVITINO_HOME environment
variable is not set");
+ Path scriptDir = Path.of(gravitinoHome, "scripts",
backendType.toLowerCase());
+ File[] scriptFiles = scriptDir.toFile().listFiles();
+ Assertions.assertNotNull(scriptFiles, "No script files found in " +
scriptDir);
+ Arrays.sort(scriptFiles, Comparator.comparing(File::getName));
+
+ Pattern upgradePattern =
+ Pattern.compile("upgrade-([\\d.]+)-to-([\\d.]+)-" +
backendType.toLowerCase() + "\\.sql");
+ for (File upgradeScript : scriptFiles) {
+ Matcher upgradeMatcher = upgradePattern.matcher(upgradeScript.getName());
+ if (!upgradeMatcher.matches()) {
+ continue;
+ }
+
+ String fromVersion = upgradeMatcher.group(1);
+ File sourceSchema =
+ scriptDir
+ .resolve("schema-" + fromVersion + "-" +
backendType.toLowerCase() + ".sql")
+ .toFile();
+ Assertions.assertTrue(
+ sourceSchema.isFile(), "No source schema found for " +
upgradeScript.getName());
+ dropAllTables();
+ executeScript(sourceSchema);
+ executeScript(upgradeScript);
+ }
+ }
+
+ private void executeScript(File scriptFile) throws IOException, SQLException
{
+ List<String> ddls = extractStatements(scriptFile.toPath());
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement()) {
+ for (String ddl : ddls) {
+ Assertions.assertDoesNotThrow(
+ () -> statement.execute(ddl),
+ "Failed to execute DDL in file " + scriptFile.getName() + " ddl: "
+ ddl);
}
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyTagRelService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyTagRelService.java
new file mode 100644
index 0000000000..3d2911f68e
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyTagRelService.java
@@ -0,0 +1,610 @@
+/*
+ * 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.storage.relational.service;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.RelationEdgeTarget;
+import org.apache.gravitino.RelationUpdate;
+import org.apache.gravitino.RelationalEntity;
+import org.apache.gravitino.SupportsRelationOperations;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.meta.TagEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper;
+import org.apache.gravitino.storage.relational.po.PolicyTagRelPO;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.apache.ibatis.session.SqlSession;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+
+/** Tests policy-to-tag relation persistence and lifecycle behavior. */
+public class TestPolicyTagRelService extends TestJDBCBackend {
+
+ private static final String METALAKE = "policy_tag_relation_metalake";
+ private static final String FINANCE_SELECTOR =
"{\"type\":\"TAG_VALUE\",\"value\":\"finance\"}";
+ private static final String RISK_SELECTOR =
"{\"type\":\"TAG_VALUE\",\"value\":\"risk\"}";
+
+ @TestTemplate
+ public void testSelectorCreateConflictBidirectionalReadAndIdempotentDelete()
throws IOException {
+ createAndInsertMakeLake(METALAKE);
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("domain")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE))
+ .withProperties(Collections.emptyMap())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE),
+ "retention",
+ AUDIT_INFO);
+ backend.insert(tag, false);
+ backend.insert(policy, false);
+
+ RelationEdgeTarget financeTarget =
+ RelationEdgeTarget.of(policy.nameIdentifier(),
Entity.EntityType.POLICY, FINANCE_SELECTOR);
+ RelationUpdate financeUpdate =
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {financeTarget},
+ new RelationEdgeTarget[0]);
+ backend.updateEntityRelations(financeUpdate);
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class, () ->
backend.updateEntityRelations(financeUpdate));
+
+ RelationEdgeTarget riskTarget =
+ RelationEdgeTarget.of(policy.nameIdentifier(),
Entity.EntityType.POLICY, RISK_SELECTOR);
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class,
+ () ->
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {riskTarget},
+ new RelationEdgeTarget[0])));
+
+ List<RelationalEntity<?>> byTag =
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(tag.nameIdentifier()),
+ Entity.EntityType.TAG);
+ Assertions.assertEquals(1, byTag.size());
+ Assertions.assertEquals(SupportsRelationOperations.Type.POLICY_TAG_REL,
byTag.get(0).type());
+ Assertions.assertEquals(tag.nameIdentifier(), byTag.get(0).source());
+ Assertions.assertEquals(Entity.EntityType.TAG, byTag.get(0).sourceType());
+ Assertions.assertEquals(policy, byTag.get(0).targetEntity());
+ Assertions.assertEquals(FINANCE_SELECTOR,
byTag.get(0).relationValue().orElse(null));
+ Assertions.assertEquals(
+ policy,
+ backend.getEntityByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ policy.nameIdentifier()));
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {riskTarget},
+ new RelationEdgeTarget[] {financeTarget})));
+ List<RelationalEntity<?>> byPolicy =
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(policy.nameIdentifier()),
+ Entity.EntityType.POLICY);
+ Assertions.assertEquals(1, byPolicy.size());
+ Assertions.assertEquals(policy.nameIdentifier(), byPolicy.get(0).source());
+ Assertions.assertEquals(Entity.EntityType.POLICY,
byPolicy.get(0).sourceType());
+ Assertions.assertEquals(tag, byPolicy.get(0).targetEntity());
+ Assertions.assertEquals(FINANCE_SELECTOR,
byPolicy.get(0).relationValue().orElse(null));
+
+ RelationUpdate removeUpdate =
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[0],
+ new RelationEdgeTarget[] {financeTarget});
+ backend.updateEntityRelations(removeUpdate);
+ backend.updateEntityRelations(removeUpdate);
+ Assertions.assertTrue(
+ backend
+ .batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(NameIdentifierUtil.ofTag(METALAKE,
tag.name())),
+ Entity.EntityType.TAG)
+ .isEmpty());
+ }
+
+ @TestTemplate
+ public void testBatchListRelationsByMultipleAnchors() throws IOException {
+ createAndInsertMakeLake(METALAKE);
+ TagEntity firstTag = createAssociation(METALAKE, "domain_a",
"retention_a");
+ TagEntity secondTag = createAssociation(METALAKE, "domain_b",
"retention_b");
+ String otherMetalake = METALAKE + "_other";
+ createAndInsertMakeLake(otherMetalake);
+ createAssociation(otherMetalake, "domain_a", "retention_a");
+
+ List<RelationalEntity<?>> byTags =
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Arrays.asList(
+ firstTag.nameIdentifier(),
+ secondTag.nameIdentifier(),
+ NameIdentifierUtil.ofTag(METALAKE, "missing_tag")),
+ Entity.EntityType.TAG);
+ Assertions.assertEquals(2, byTags.size());
+ Assertions.assertEquals(
+ Set.of("domain_a", "domain_b"),
+ byTags.stream().map(relation ->
relation.source().name()).collect(Collectors.toSet()));
+
+ List<RelationalEntity<?>> byPolicies =
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Arrays.asList(
+ NameIdentifierUtil.ofPolicy(METALAKE, "retention_a"),
+ NameIdentifierUtil.ofPolicy(METALAKE, "retention_b"),
+ NameIdentifierUtil.ofPolicy(METALAKE, "missing_policy")),
+ Entity.EntityType.POLICY);
+ Assertions.assertEquals(2, byPolicies.size());
+ Assertions.assertEquals(
+ Set.of("retention_a", "retention_b"),
+ byPolicies.stream().map(relation ->
relation.source().name()).collect(Collectors.toSet()));
+ }
+
+ @TestTemplate
+ public void testUpdateRelationsRollsBackOnFailure() throws IOException {
+ createAndInsertMakeLake(METALAKE);
+ TagEntity tag = createAssociation(METALAKE, "domain", "retention");
+ RelationEdgeTarget existingTarget =
+ RelationEdgeTarget.of(
+ NameIdentifierUtil.ofPolicy(METALAKE, "retention"),
Entity.EntityType.POLICY, null);
+ RelationEdgeTarget missingTarget =
+ RelationEdgeTarget.of(
+ NameIdentifierUtil.ofPolicy(METALAKE, "missing_policy"),
+ Entity.EntityType.POLICY,
+ null);
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {missingTarget},
+ new RelationEdgeTarget[] {existingTarget})));
+
+ List<RelationalEntity<?>> relations =
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(tag.nameIdentifier()),
+ Entity.EntityType.TAG);
+ Assertions.assertEquals(1, relations.size());
+ Assertions.assertEquals("retention",
relations.get(0).targetEntity().name());
+ }
+
+ @TestTemplate
+ public void testDuplicateAddRollsBackAllRelationWrites() throws IOException {
+ createAndInsertMakeLake(METALAKE);
+ TagEntity tag = createAssociation(METALAKE, "domain", "retention");
+ PolicyEntity newPolicy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE),
+ "new_policy",
+ AUDIT_INFO);
+ backend.insert(newPolicy, false);
+
+ RelationEdgeTarget newTarget =
+ RelationEdgeTarget.of(newPolicy.nameIdentifier(),
Entity.EntityType.POLICY, null);
+ RelationEdgeTarget existingTarget =
+ RelationEdgeTarget.of(
+ NameIdentifierUtil.ofPolicy(METALAKE, "retention"),
Entity.EntityType.POLICY, null);
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class,
+ () ->
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {newTarget, existingTarget},
+ new RelationEdgeTarget[0])));
+
+ List<RelationalEntity<?>> relations =
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(tag.nameIdentifier()),
+ Entity.EntityType.TAG);
+ Assertions.assertEquals(1, relations.size());
+ Assertions.assertEquals("retention",
relations.get(0).targetEntity().name());
+ Assertions.assertNull(
+ SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class,
+ mapper -> mapper.getByPolicyIdAndTagId(newPolicy.id(), tag.id())));
+ }
+
+ @TestTemplate
+ public void testEntityDeletesCascadePolicyTagRelations() throws IOException {
+ createAndInsertMakeLake(METALAKE);
+ TagEntity policyDeletedTag =
+ createAssociation(METALAKE, "policy_deleted_tag", "deleted_policy");
+ backend.delete(
+ NameIdentifierUtil.ofPolicy(METALAKE, "deleted_policy"),
Entity.EntityType.POLICY, false);
+ Assertions.assertTrue(
+ backend
+ .batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(policyDeletedTag.nameIdentifier()),
+ Entity.EntityType.TAG)
+ .isEmpty());
+
+ TagEntity deletedTag = createAssociation(METALAKE, "deleted_tag",
"surviving_policy");
+ backend.delete(deletedTag.nameIdentifier(), Entity.EntityType.TAG, false);
+ Assertions.assertTrue(
+ backend
+ .batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(
+ NameIdentifierUtil.ofPolicy(METALAKE, "surviving_policy")),
+ Entity.EntityType.POLICY)
+ .isEmpty());
+ }
+
+ @TestTemplate
+ public void testMetalakeCascadeDoesNotDeleteRelationsFromOtherMetalakes()
throws IOException {
+ BaseMetalake deletedMetalake = createAndInsertMakeLake(METALAKE);
+ createAssociation(METALAKE, "deleted_domain", "deleted_retention");
+ String survivingMetalake = METALAKE + "_surviving";
+ createAndInsertMakeLake(survivingMetalake);
+ TagEntity survivingTag =
+ createAssociation(survivingMetalake, "surviving_domain",
"surviving_retention");
+
+ backend.delete(deletedMetalake.nameIdentifier(),
Entity.EntityType.METALAKE, true);
+
+ List<RelationalEntity<?>> survivingRelations =
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(survivingTag.nameIdentifier()),
+ Entity.EntityType.TAG);
+ Assertions.assertEquals(1, survivingRelations.size());
+ Assertions.assertEquals("surviving_retention",
survivingRelations.get(0).targetEntity().name());
+ }
+
+ @TestTemplate
+ public void testConcurrentInsertsHaveOneWinner() throws Exception {
+ createAndInsertMakeLake(METALAKE);
+ RelationEndpoints endpoints = createEndpoints(METALAKE, "domain",
"retention");
+ PolicyTagRelPO relation = newRelation(endpoints, null);
+ CyclicBarrier bothObservedAbsent = new CyclicBarrier(2);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ List<Integer> affectedRows;
+ try {
+ Future<Integer> first = executor.submit(concurrentInsertTask(relation,
bothObservedAbsent));
+ Future<Integer> second = executor.submit(concurrentInsertTask(relation,
bothObservedAbsent));
+ affectedRows =
+ Arrays.asList(first.get(20, TimeUnit.SECONDS), second.get(20,
TimeUnit.SECONDS));
+ } finally {
+ executor.shutdownNow();
+ }
+ Collections.sort(affectedRows);
+
+ Assertions.assertEquals(Arrays.asList(0, 1), affectedRows);
+ PolicyTagRelPO persisted = getRelation(endpoints);
+ Assertions.assertNotNull(persisted);
+ Assertions.assertNotNull(persisted.getId());
+ Assertions.assertEquals(1L, persisted.getCurrentVersion());
+ Assertions.assertEquals(
+ 1L,
+ queryForLong(
+ "SELECT COUNT(*) FROM policy_tag_relation_meta WHERE policy_id = "
+ + endpoints.policy.id()
+ + " AND tag_id = "
+ + endpoints.tag.id()
+ + " AND deleted_at = 0"));
+ }
+
+ @TestTemplate
+ public void testStaleRelationDeleteAffectsNoRows() throws Exception {
+ createAndInsertMakeLake(METALAKE);
+ RelationEndpoints endpoints = createEndpoints(METALAKE, "domain",
"retention");
+ backend.updateEntityRelations(relationUpdate(endpoints, null, true));
+ PolicyTagRelPO observed = getRelation(endpoints);
+ PolicyTagRelPO replacement = copyWithVersion(observed, 2L);
+
+ Assertions.assertEquals(
+ Integer.valueOf(0),
+ SessionUtils.doWithCommitAndFetchResult(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByIdAndVersion(replacement)));
+
+ PolicyTagRelPO current = getRelation(endpoints);
+ Assertions.assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.doWithCommitAndFetchResult(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByIdAndVersion(current)));
+ Assertions.assertEquals(
+ 2L,
+ queryForLong(
+ "SELECT current_version FROM policy_tag_relation_meta WHERE id = "
+ observed.getId()));
+ Assertions.assertEquals(
+ 2L,
+ queryForLong(
+ "SELECT last_version FROM policy_tag_relation_meta WHERE id = " +
observed.getId()));
+ }
+
+ @TestTemplate
+ public void testStaleRelationCannotDeleteRecreatedRow() throws Exception {
+ createAndInsertMakeLake(METALAKE);
+ RelationEndpoints endpoints = createEndpoints(METALAKE, "domain",
"retention");
+ backend.updateEntityRelations(relationUpdate(endpoints, null, true));
+ PolicyTagRelPO stale = getRelation(endpoints);
+
+ backend.updateEntityRelations(relationUpdate(endpoints, null, false));
+ backend.updateEntityRelations(relationUpdate(endpoints, null, true));
+ PolicyTagRelPO recreated = getRelation(endpoints);
+ Assertions.assertNotEquals(stale.getId(), recreated.getId());
+
+ Assertions.assertEquals(
+ Integer.valueOf(0),
+ SessionUtils.doWithCommitAndFetchResult(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByIdAndVersion(stale)));
+
+ PolicyTagRelPO current = getRelation(endpoints);
+ Assertions.assertNotNull(current);
+ Assertions.assertEquals(recreated.getId(), current.getId());
+ }
+
+ @TestTemplate
+ public void testDeleteReAddDeleteRetainsHistory() throws Exception {
+ createAndInsertMakeLake(METALAKE);
+ RelationEndpoints endpoints = createEndpoints(METALAKE, "domain",
"retention");
+ backend.updateEntityRelations(relationUpdate(endpoints, null, true));
+ backend.updateEntityRelations(relationUpdate(endpoints, null, false));
+ backend.updateEntityRelations(relationUpdate(endpoints, null, true));
+ backend.updateEntityRelations(relationUpdate(endpoints, null, false));
+
+ String pairPredicate =
+ "policy_id = "
+ + endpoints.policy.id()
+ + " AND tag_id = "
+ + endpoints.tag.id()
+ + " AND deleted_at > 0";
+ Assertions.assertEquals(
+ 2L, queryForLong("SELECT COUNT(*) FROM policy_tag_relation_meta WHERE
" + pairPredicate));
+ }
+
+ @TestTemplate
+ public void testLegacyCleanupHonorsCutoffAndLimit() throws Exception {
+ createAndInsertMakeLake(METALAKE);
+ RelationEndpoints first = createEndpoints(METALAKE, "domain_a",
"retention_a");
+ RelationEndpoints second = createEndpoints(METALAKE, "domain_b",
"retention_b");
+ backend.updateEntityRelations(relationUpdate(first, null, true));
+ backend.updateEntityRelations(relationUpdate(second, null, true));
+ backend.updateEntityRelations(relationUpdate(first, null, false));
+ backend.updateEntityRelations(relationUpdate(second, null, false));
+ executeUpdate("UPDATE policy_tag_relation_meta SET deleted_at = 100 WHERE
deleted_at > 0");
+
+ Assertions.assertEquals(
+ 0, TagMetaService.getInstance().deleteTagMetasByLegacyTimeline(100L,
10));
+ Assertions.assertEquals(
+ 1, TagMetaService.getInstance().deleteTagMetasByLegacyTimeline(101L,
1));
+ Assertions.assertEquals(
+ 1L, queryForLong("SELECT COUNT(*) FROM policy_tag_relation_meta WHERE
deleted_at > 0"));
+ Assertions.assertEquals(
+ 1, TagMetaService.getInstance().deleteTagMetasByLegacyTimeline(101L,
10));
+ }
+
+ @TestTemplate
+ public void testRelationIdentifiersAndTypesAreValidated() throws Exception {
+ createAndInsertMakeLake(METALAKE);
+ RelationEndpoints endpoints = createEndpoints(METALAKE, "domain",
"retention");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ backend.batchListEntitiesByRelation(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ Collections.singletonList(NameIdentifier.of("domain")),
+ Entity.EntityType.TAG));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ endpoints.tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {
+ RelationEdgeTarget.of(
+ NameIdentifierUtil.ofPolicy(METALAKE + "_other",
"retention"),
+ Entity.EntityType.POLICY,
+ null)
+ },
+ new RelationEdgeTarget[0])));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ endpoints.tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {
+ RelationEdgeTarget.of(
+ endpoints.policy.nameIdentifier(),
Entity.EntityType.TAG, null)
+ },
+ new RelationEdgeTarget[0])));
+ }
+
+ private TagEntity createAssociation(String metalake, String tagName, String
policyName)
+ throws IOException {
+ RelationEndpoints endpoints = createEndpoints(metalake, tagName,
policyName);
+ backend.updateEntityRelations(relationUpdate(endpoints, null, true));
+ return endpoints.tag;
+ }
+
+ private RelationEndpoints createEndpoints(String metalake, String tagName,
String policyName)
+ throws IOException {
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(tagName)
+ .withNamespace(NamespaceUtil.ofTag(metalake))
+ .withProperties(Collections.emptyMap())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(metalake),
+ policyName,
+ AUDIT_INFO);
+ backend.insert(tag, false);
+ backend.insert(policy, false);
+ return new RelationEndpoints(tag, policy);
+ }
+
+ private RelationUpdate relationUpdate(RelationEndpoints endpoints, String
selector, boolean add) {
+ RelationEdgeTarget target =
+ RelationEdgeTarget.of(
+ endpoints.policy.nameIdentifier(), Entity.EntityType.POLICY,
selector);
+ return RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ endpoints.tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ add ? new RelationEdgeTarget[] {target} : new RelationEdgeTarget[0],
+ add ? new RelationEdgeTarget[0] : new RelationEdgeTarget[] {target});
+ }
+
+ private PolicyTagRelPO getRelation(RelationEndpoints endpoints) {
+ return SessionUtils.getWithoutCommit(
+ PolicyTagRelMapper.class,
+ mapper -> mapper.getByPolicyIdAndTagId(endpoints.policy.id(),
endpoints.tag.id()));
+ }
+
+ private PolicyTagRelPO copyWithVersion(PolicyTagRelPO relation, long
version) {
+ return PolicyTagRelPO.builder()
+ .withId(relation.getId())
+ .withPolicyId(relation.getPolicyId())
+ .withTagId(relation.getTagId())
+ .withSelector(relation.getSelector())
+ .withAuditInfo(relation.getAuditInfo())
+ .withCurrentVersion(version)
+ .withLastVersion(version)
+ .withDeletedAt(0L)
+ .build();
+ }
+
+ private PolicyTagRelPO newRelation(RelationEndpoints endpoints, String
selector) {
+ return PolicyTagRelPO.builder()
+ .withPolicyId(endpoints.policy.id())
+ .withTagId(endpoints.tag.id())
+ .withSelector(selector)
+ .withAuditInfo("{}")
+ .withCurrentVersion(1L)
+ .withLastVersion(1L)
+ .withDeletedAt(0L)
+ .build();
+ }
+
+ private Callable<Integer> concurrentInsertTask(
+ PolicyTagRelPO relation, CyclicBarrier bothObservedAbsent) {
+ return () -> {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(false))
{
+ PolicyTagRelMapper mapper =
sqlSession.getMapper(PolicyTagRelMapper.class);
+ PolicyTagRelPO observed =
+ mapper.getByPolicyIdAndTagId(relation.getPolicyId(),
relation.getTagId());
+ bothObservedAbsent.await(10, TimeUnit.SECONDS);
+ Assertions.assertNull(observed);
+ int inserted = mapper.insertIfAbsent(relation);
+ sqlSession.commit();
+ return inserted;
+ }
+ };
+ }
+
+ private long queryForLong(String sql) throws SQLException {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ Assertions.assertTrue(resultSet.next());
+ return resultSet.getLong(1);
+ }
+ }
+
+ private int executeUpdate(String sql) throws SQLException {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement()) {
+ return statement.executeUpdate(sql);
+ }
+ }
+
+ private static class RelationEndpoints {
+ private final TagEntity tag;
+ private final PolicyEntity policy;
+
+ private RelationEndpoints(TagEntity tag, PolicyEntity policy) {
+ this.tag = tag;
+ this.policy = policy;
+ }
+ }
+}
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index ff8aaa3d50..37e9732884 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -431,6 +431,20 @@ CREATE TABLE IF NOT EXISTS `policy_relation_meta` (
KEY `idx_prmid` (`metadata_object_id`)
) ENGINE=InnoDB;
+CREATE TABLE IF NOT EXISTS `policy_tag_relation_meta` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id',
+ `tag_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'tag id',
+ `selector` CLOB DEFAULT NULL COMMENT 'policy tag selector JSON, NULL
matches tag presence',
+ `audit_info` CLOB NOT NULL COMMENT 'policy tag relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'policy tag
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `policy_tag_relation_meta_uk_pid_tid_del` (`policy_id`,
`tag_id`, `deleted_at`),
+ KEY `policy_tag_relation_meta_idx_tag_id` (`tag_id`)
+) ENGINE=InnoDB;
+
CREATE TABLE IF NOT EXISTS `statistic_meta` (
`id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
`statistic_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'statistic id',
diff --git a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
index 677341c49c..19d7f054dc 100644
--- a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
+++ b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
@@ -41,6 +41,20 @@ CREATE INDEX IF NOT EXISTS `idx_tid_value` ON
`tag_relation_meta` (`tag_id`, `ta
ALTER TABLE `job_run_meta` ADD COLUMN `job_started_at` BIGINT(20) UNSIGNED NOT
NULL DEFAULT 0 COMMENT 'job started at' AFTER `job_run_status`;
+CREATE TABLE IF NOT EXISTS `policy_tag_relation_meta` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id',
+ `tag_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'tag id',
+ `selector` CLOB DEFAULT NULL COMMENT 'policy tag selector JSON, NULL
matches tag presence',
+ `audit_info` CLOB NOT NULL COMMENT 'policy tag relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'policy tag
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `policy_tag_relation_meta_uk_pid_tid_del` (`policy_id`,
`tag_id`, `deleted_at`),
+ KEY `policy_tag_relation_meta_idx_tag_id` (`tag_id`)
+) ENGINE=InnoDB;
+
CREATE TABLE IF NOT EXISTS `semantic_model_meta` (
`semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'semantic model
id',
`semantic_model_name` VARCHAR(128) NOT NULL COMMENT 'semantic model name',
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index 7f0c35a060..d6bdfb8f5e 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -422,6 +422,20 @@ CREATE TABLE IF NOT EXISTS `policy_relation_meta` (
KEY `policy_relation_meta_idx_mid` (`metadata_object_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'policy
metadata object relation';
+CREATE TABLE IF NOT EXISTS `policy_tag_relation_meta` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id',
+ `tag_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'tag id',
+ `selector` MEDIUMTEXT DEFAULT NULL COMMENT 'policy tag selector JSON, NULL
matches tag presence',
+ `audit_info` MEDIUMTEXT NOT NULL COMMENT 'policy tag relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'policy tag
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `policy_tag_relation_meta_uk_pid_tid_del` (`policy_id`,
`tag_id`, `deleted_at`),
+ KEY `policy_tag_relation_meta_idx_tag_id` (`tag_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'policy
tag relation';
+
CREATE TABLE IF NOT EXISTS `statistic_meta` (
`id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
`statistic_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'statistic id',
diff --git a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
index 654f6bb45a..d59ef44bcc 100644
--- a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
+++ b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
@@ -100,6 +100,20 @@ ALTER TABLE `table_version_info`
ALTER TABLE `job_run_meta`
ADD COLUMN `job_started_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT
'job started at' AFTER `job_run_status`;
+CREATE TABLE IF NOT EXISTS `policy_tag_relation_meta` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id',
+ `tag_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'tag id',
+ `selector` MEDIUMTEXT DEFAULT NULL COMMENT 'policy tag selector JSON, NULL
matches tag presence',
+ `audit_info` MEDIUMTEXT NOT NULL COMMENT 'policy tag relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'policy tag
relation last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'policy tag
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `policy_tag_relation_meta_uk_pid_tid_del` (`policy_id`,
`tag_id`, `deleted_at`),
+ KEY `policy_tag_relation_meta_idx_tag_id` (`tag_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'policy
tag relation';
+
CREATE TABLE IF NOT EXISTS `semantic_model_meta` (
`semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'semantic model
id',
`semantic_model_name` VARCHAR(128) NOT NULL COMMENT 'semantic model name',
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index 28bbdf77d3..c0b0c0f6e8 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -750,6 +750,30 @@ COMMENT ON COLUMN policy_relation_meta.current_version IS
'policy relation curre
COMMENT ON COLUMN policy_relation_meta.last_version IS 'policy relation last
version';
COMMENT ON COLUMN policy_relation_meta.deleted_at IS 'policy relation deleted
at';
+CREATE TABLE IF NOT EXISTS policy_tag_relation_meta (
+ id BIGSERIAL NOT NULL,
+ policy_id BIGINT NOT NULL,
+ tag_id BIGINT NOT NULL,
+ selector TEXT DEFAULT NULL,
+ audit_info TEXT NOT NULL,
+ current_version INT NOT NULL DEFAULT 1,
+ last_version INT NOT NULL DEFAULT 1,
+ deleted_at BIGINT NOT NULL DEFAULT 0,
+ PRIMARY KEY (id),
+ UNIQUE (policy_id, tag_id, deleted_at)
+);
+
+CREATE INDEX IF NOT EXISTS policy_tag_relation_meta_idx_tag_id ON
policy_tag_relation_meta (tag_id);
+COMMENT ON TABLE policy_tag_relation_meta IS 'policy tag relation';
+COMMENT ON COLUMN policy_tag_relation_meta.id IS 'auto increment id';
+COMMENT ON COLUMN policy_tag_relation_meta.policy_id IS 'policy id';
+COMMENT ON COLUMN policy_tag_relation_meta.tag_id IS 'tag id';
+COMMENT ON COLUMN policy_tag_relation_meta.selector IS 'policy tag selector
JSON, NULL matches tag presence';
+COMMENT ON COLUMN policy_tag_relation_meta.audit_info IS 'policy tag relation
audit info';
+COMMENT ON COLUMN policy_tag_relation_meta.current_version IS 'policy tag
relation current version';
+COMMENT ON COLUMN policy_tag_relation_meta.last_version IS 'policy tag
relation last version';
+COMMENT ON COLUMN policy_tag_relation_meta.deleted_at IS 'policy tag relation
deleted at';
+
CREATE TABLE IF NOT EXISTS statistic_meta (
id BIGINT NOT NULL GENERATED BY DEFAULT AS IDENTITY,
statistic_id BIGINT NOT NULL,
diff --git a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
index 59b4c9e7ae..ea6edd173b 100644
--- a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
+++ b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
@@ -49,6 +49,30 @@ CREATE INDEX IF NOT EXISTS
tag_relation_meta_idx_tag_id_value ON tag_relation_me
ALTER TABLE job_run_meta ADD COLUMN IF NOT EXISTS job_started_at BIGINT NOT
NULL DEFAULT 0;
COMMENT ON COLUMN job_run_meta.job_started_at IS 'job run started at';
+CREATE TABLE IF NOT EXISTS policy_tag_relation_meta (
+ id BIGSERIAL NOT NULL,
+ policy_id BIGINT NOT NULL,
+ tag_id BIGINT NOT NULL,
+ selector TEXT DEFAULT NULL,
+ audit_info TEXT NOT NULL,
+ current_version INT NOT NULL DEFAULT 1,
+ last_version INT NOT NULL DEFAULT 1,
+ deleted_at BIGINT NOT NULL DEFAULT 0,
+ PRIMARY KEY (id),
+ UNIQUE (policy_id, tag_id, deleted_at)
+);
+
+CREATE INDEX IF NOT EXISTS policy_tag_relation_meta_idx_tag_id ON
policy_tag_relation_meta (tag_id);
+COMMENT ON TABLE policy_tag_relation_meta IS 'policy tag relation';
+COMMENT ON COLUMN policy_tag_relation_meta.id IS 'auto increment id';
+COMMENT ON COLUMN policy_tag_relation_meta.policy_id IS 'policy id';
+COMMENT ON COLUMN policy_tag_relation_meta.tag_id IS 'tag id';
+COMMENT ON COLUMN policy_tag_relation_meta.selector IS 'policy tag selector
JSON, NULL matches tag presence';
+COMMENT ON COLUMN policy_tag_relation_meta.audit_info IS 'policy tag relation
audit info';
+COMMENT ON COLUMN policy_tag_relation_meta.current_version IS 'policy tag
relation current version';
+COMMENT ON COLUMN policy_tag_relation_meta.last_version IS 'policy tag
relation last version';
+COMMENT ON COLUMN policy_tag_relation_meta.deleted_at IS 'policy tag relation
deleted at';
+
CREATE TABLE IF NOT EXISTS semantic_model_meta (
semantic_model_id BIGINT NOT NULL,
semantic_model_name VARCHAR(128) NOT NULL,
