This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-domain-dev
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-domain-dev by this push:
new bb3ecc18b9 [#13883] feat(core): Add sub-domain relation storage
(#14350)
bb3ecc18b9 is described below
commit bb3ecc18b9783ed7af48324370fbb9a85de68c11
Author: Jerry Shao <[email protected]>
AuthorDate: Mon Nov 2 16:17:17 2026 +0800
[#13883] feat(core): Add sub-domain relation storage (#14350)
### What changes were proposed in this pull request?
Store the parent of each domain in a new `domain_rel` table and resolve
it on domain reads.
- Add the `domain_rel` table (one live parent per domain, indexed by
parent) to the H2, MySQL and PostgreSQL 2.0.0 schema and upgrade
scripts.
- Add `DomainRelPO`, `DomainRelMapper` and its SQL providers.
- Add `DomainRelService` to insert, re-point and remove a parent edge,
to list sub-domains, and to resolve parents singly or in batch.
Hierarchy changes lock the metalake row first, then the domain rows in
ID order, so they keep the lock order of a metalake drop. The shape of
the hierarchy (one level deep, C1–C6) is enforced by the domain manager
(#13884).
- Fill `DomainEntity.parent` on domain get, list and batch get, and
reject a domain update that changes the parent.
- Remove a domain's own edge when it is deleted, and refuse the delete
while it has live sub-domains.
- Soft-delete edges on metalake delete and hard-delete them in GC.
### Why are the changes needed?
Domains form a hierarchy (epic #13864). The domain manager (#13884)
needs storage for the parent edge with consistent locking before it can
implement attach, move and detach.
Fixes #13883
### Does this PR introduce _any_ user-facing change?
No. It adds the `domain_rel` table to the 2.0.0 schema and upgrade
scripts; no API change.
### How was this patch tested?
- New `TestDomainRelPO` and `TestDomainRelService`, covering edge insert
/ re-point / remove, delete refusal while sub-domains exist, metalake
delete (cascade and non-cascade), GC, and concurrency (parent delete and
attach serialize).
- Updated `TestDomainPO` and `TestDomainMetaService`.
- Regression coverage for cached pre-lock parent reads, consistent
get/list/batch/sub-domain snapshots, response hydration failures, caught
nested delete refusals, and repeated detach/reattach/delete within one
transaction.
- Passed `compileJava compileTestJava spotlessCheck`, `:core:javadoc`,
the complete `:core:coreH2Test` suite (647 tests), and domain
`:core:coreUnitTest` tests (14 tests). Native MySQL and PostgreSQL
verification is left to CI.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../relational/mapper/DomainMetaMapper.java | 46 +-
.../mapper/DomainMetaSQLProviderFactory.java | 19 +-
.../storage/relational/mapper/DomainRelMapper.java | 118 +++
.../mapper/DomainRelSQLProviderFactory.java | 102 +++
.../provider/DefaultMapperPackageProvider.java | 2 +
.../provider/base/DomainMetaBaseSQLProvider.java | 84 +-
.../provider/base/DomainRelBaseSQLProvider.java | 186 +++++
.../postgresql/DomainMetaPostgreSQLProvider.java | 5 +
...vider.java => DomainRelPostgreSQLProvider.java} | 34 +-
.../gravitino/storage/relational/po/DomainPO.java | 36 +-
.../storage/relational/po/DomainRelPO.java | 154 ++++
.../relational/service/DomainMetaService.java | 88 +-
.../relational/service/DomainRelService.java | 350 ++++++++
.../relational/service/MetalakeMetaService.java | 9 +
.../storage/relational/po/TestDomainPO.java | 27 +-
.../storage/relational/po/TestDomainRelPO.java | 112 +++
.../relational/service/TestDomainRelService.java | 921 +++++++++++++++++++++
scripts/h2/schema-2.0.0-h2.sql | 15 +
scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql | 15 +
scripts/mysql/schema-2.0.0-mysql.sql | 15 +
scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql | 15 +
scripts/postgresql/schema-2.0.0-postgresql.sql | 24 +
.../upgrade-1.3.0-to-2.0.0-postgresql.sql | 24 +
23 files changed, 2342 insertions(+), 59 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaMapper.java
index 59cf4ad530..daf3774f7f 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaMapper.java
@@ -22,12 +22,19 @@ import java.util.List;
import org.apache.gravitino.storage.relational.po.DomainPO;
import org.apache.ibatis.annotations.DeleteProvider;
import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Options;
+import org.apache.ibatis.annotations.Options.FlushCachePolicy;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.SelectProvider;
import org.apache.ibatis.annotations.UpdateProvider;
-/** MyBatis mapper of the {@code domain_meta} table. */
+/**
+ * MyBatis mapper of the {@code domain_meta} table.
+ *
+ * <p>Reads flush the session cache so that a lookup performed after taking a
domain lock observes
+ * committed state even when an enclosing transaction previously read the same
row or edge.
+ */
public interface DomainMetaMapper {
/** The name of the domain table. */
@@ -39,6 +46,7 @@ public interface DomainMetaMapper {
* @param domainId The domain ID.
* @return The number of deleted rows with this ID.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@Select(
"SELECT COUNT(*) FROM "
+ DOMAIN_TABLE_NAME
@@ -51,6 +59,7 @@ public interface DomainMetaMapper {
* @param metalakeName The metalake name.
* @return The active domains.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(type = DomainMetaSQLProviderFactory.class, method =
"listDomainPOsByMetalake")
List<DomainPO> listDomainPOsByMetalake(@Param("metalakeName") String
metalakeName);
@@ -61,6 +70,7 @@ public interface DomainMetaMapper {
* @param domainName The domain name.
* @return The domain, or null if it is not active.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(
type = DomainMetaSQLProviderFactory.class,
method = "selectDomainMetaByMetalakeAndName")
@@ -75,6 +85,7 @@ public interface DomainMetaMapper {
* @param domainName The domain name.
* @return The domain, or null if it is not active.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(
type = DomainMetaSQLProviderFactory.class,
method = "selectDomainMetaByMetalakeIdAndName")
@@ -89,6 +100,7 @@ public interface DomainMetaMapper {
* @param domainName The domain name.
* @return The locked domain, or null if the natural key is not active.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(
type = DomainMetaSQLProviderFactory.class,
method = "selectDomainMetaByMetalakeIdAndNameForUpdate")
@@ -101,6 +113,7 @@ public interface DomainMetaMapper {
* @param domainId The domain ID.
* @return The domain, or null if it is not active.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(type = DomainMetaSQLProviderFactory.class, method =
"selectDomainMetaById")
DomainPO selectDomainMetaById(@Param("domainId") Long domainId);
@@ -111,11 +124,25 @@ public interface DomainMetaMapper {
* @param domainId The domain ID.
* @return The locked domain, or null if it is not active.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(
type = DomainMetaSQLProviderFactory.class,
method = "selectDomainMetaByIdForUpdate")
DomainPO selectDomainMetaByIdForUpdate(@Param("domainId") Long domainId);
+ /**
+ * Selects an active domain by ID and locks it for shared access. The
backing tag name is not
+ * read, so that no tag row is locked.
+ *
+ * @param domainId The domain ID.
+ * @return The locked domain, or null if it is not active.
+ */
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
+ @SelectProvider(
+ type = DomainMetaSQLProviderFactory.class,
+ method = "selectDomainMetaByIdForShare")
+ DomainPO selectDomainMetaByIdForShare(@Param("domainId") Long domainId);
+
/**
* Lists the active domains with the given IDs, for name lookups. The
backing tag name is not
* read.
@@ -123,6 +150,7 @@ public interface DomainMetaMapper {
* @param domainIds The domain IDs.
* @return The active domains.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(type = DomainMetaSQLProviderFactory.class, method =
"listDomainPOsByDomainIds")
List<DomainPO> listDomainPOsByDomainIds(@Param("domainIds") List<Long>
domainIds);
@@ -133,12 +161,28 @@ public interface DomainMetaMapper {
* @param domainNames The domain names.
* @return The active domains.
*/
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
@SelectProvider(
type = DomainMetaSQLProviderFactory.class,
method = "listDomainPOsByMetalakeAndDomainNames")
List<DomainPO> listDomainPOsByMetalakeAndDomainNames(
@Param("metalakeName") String metalakeName, @Param("domainNames")
List<String> domainNames);
+ /**
+ * Lists the active sub-domains of a domain.
+ *
+ * @param metalakeName The metalake name.
+ * @param parentName The parent domain name.
+ * @return The active sub-domains, or one row with a null domain ID for a
childless parent. An
+ * absent parent produces no rows.
+ */
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
+ @SelectProvider(
+ type = DomainMetaSQLProviderFactory.class,
+ method = "listDomainPOsByParentIdentifier")
+ List<DomainPO> listDomainPOsByParentIdentifier(
+ @Param("metalakeName") String metalakeName, @Param("parentName") String
parentName);
+
/**
* Inserts a domain.
*
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaSQLProviderFactory.java
index e81342d898..f9c9eb49cd 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainMetaSQLProviderFactory.java
@@ -56,7 +56,13 @@ public class DomainMetaSQLProviderFactory {
static class DomainMetaMySQLProvider extends DomainMetaBaseSQLProvider {}
- static class DomainMetaH2Provider extends DomainMetaBaseSQLProvider {}
+ static class DomainMetaH2Provider extends DomainMetaBaseSQLProvider {
+ @Override
+ public String selectDomainMetaByIdForShare(Long domainId) {
+ // H2 has no shared row-lock syntax, so H2 backends fall back to an
exclusive lock.
+ return selectDomainMetaByIdForUpdate(domainId);
+ }
+ }
/** Delegates {@link DomainMetaBaseSQLProvider#listDomainPOsByMetalake}. */
public static String listDomainPOsByMetalake(@Param("metalakeName") String
metalakeName) {
@@ -91,6 +97,11 @@ public class DomainMetaSQLProviderFactory {
return getProvider().selectDomainMetaByIdForUpdate(domainId);
}
+ /** Delegates {@link
DomainMetaBaseSQLProvider#selectDomainMetaByIdForShare}. */
+ public static String selectDomainMetaByIdForShare(@Param("domainId") Long
domainId) {
+ return getProvider().selectDomainMetaByIdForShare(domainId);
+ }
+
/** Delegates {@link DomainMetaBaseSQLProvider#listDomainPOsByDomainIds}. */
public static String listDomainPOsByDomainIds(@Param("domainIds") List<Long>
domainIds) {
return getProvider().listDomainPOsByDomainIds(domainIds);
@@ -102,6 +113,12 @@ public class DomainMetaSQLProviderFactory {
return getProvider().listDomainPOsByMetalakeAndDomainNames(metalakeName,
domainNames);
}
+ /** Delegates {@link
DomainMetaBaseSQLProvider#listDomainPOsByParentIdentifier}. */
+ public static String listDomainPOsByParentIdentifier(
+ @Param("metalakeName") String metalakeName, @Param("parentName") String
parentName) {
+ return getProvider().listDomainPOsByParentIdentifier(metalakeName,
parentName);
+ }
+
/** Delegates {@link DomainMetaBaseSQLProvider#insertDomainMeta}. */
public static String insertDomainMeta(@Param("domainMeta") DomainPO
domainPO) {
return getProvider().insertDomainMeta(domainPO);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainRelMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainRelMapper.java
new file mode 100644
index 0000000000..3d1f71899f
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainRelMapper.java
@@ -0,0 +1,118 @@
+/*
+ * 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.DomainRelPO;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Options;
+import org.apache.ibatis.annotations.Options.FlushCachePolicy;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+import org.apache.ibatis.annotations.UpdateProvider;
+
+/** MyBatis mapper of the {@code domain_rel} table, which holds the parent
edge of sub-domains. */
+public interface DomainRelMapper {
+
+ /** The name of the domain relation table. */
+ String DOMAIN_REL_TABLE_NAME = "domain_rel";
+
+ /**
+ * Selects the active parent edge of a domain. The parent name is not read.
+ *
+ * @param childId The ID of the domain.
+ * @return The edge, or null if the domain is top-level.
+ */
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
+ @SelectProvider(type = DomainRelSQLProviderFactory.class, method =
"selectDomainRelByChildId")
+ DomainRelPO selectDomainRelByChildId(@Param("childId") Long childId);
+
+ /**
+ * Lists the active parent edges of the given domains, with the name of each
parent.
+ *
+ * @param childIds The IDs of the domains.
+ * @return The edges of the domains that have a parent.
+ */
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
+ @SelectProvider(type = DomainRelSQLProviderFactory.class, method =
"listDomainRelsByChildIds")
+ List<DomainRelPO> listDomainRelsByChildIds(@Param("childIds") List<Long>
childIds);
+
+ /**
+ * Counts the active sub-domains of a domain.
+ *
+ * @param parentId The ID of the parent domain.
+ * @return The number of active edges to the domain.
+ */
+ @Options(useCache = false, flushCache = FlushCachePolicy.TRUE)
+ @SelectProvider(type = DomainRelSQLProviderFactory.class, method =
"countDomainRelsByParentId")
+ int countDomainRelsByParentId(@Param("parentId") Long parentId);
+
+ /**
+ * Inserts a parent edge.
+ *
+ * @param domainRelPO The edge to insert.
+ */
+ @InsertProvider(type = DomainRelSQLProviderFactory.class, method =
"insertDomainRel")
+ void insertDomainRel(@Param("domainRel") DomainRelPO domainRelPO);
+
+ /**
+ * Updates an active parent edge when its version still matches the observed
one.
+ *
+ * @param newDomainRelPO The new edge.
+ * @param oldDomainRelPO The edge observed by the caller.
+ * @return The number of affected rows.
+ */
+ @UpdateProvider(type = DomainRelSQLProviderFactory.class, method =
"updateDomainRel")
+ Integer updateDomainRel(
+ @Param("newDomainRel") DomainRelPO newDomainRelPO,
+ @Param("oldDomainRel") DomainRelPO oldDomainRelPO);
+
+ /**
+ * Soft-deletes the active parent edge of a domain.
+ *
+ * @param childId The ID of the domain.
+ * @return The number of affected rows.
+ */
+ @UpdateProvider(type = DomainRelSQLProviderFactory.class, method =
"softDeleteDomainRelByChildId")
+ Integer softDeleteDomainRelByChildId(@Param("childId") Long childId);
+
+ /**
+ * Soft-deletes all active parent edges of a metalake.
+ *
+ * @param metalakeId The metalake ID.
+ */
+ @UpdateProvider(
+ type = DomainRelSQLProviderFactory.class,
+ method = "softDeleteDomainRelsByMetalakeId")
+ void softDeleteDomainRelsByMetalakeId(@Param("metalakeId") Long metalakeId);
+
+ /**
+ * Hard-deletes parent edges soft-deleted before the given time.
+ *
+ * @param legacyTimeline The time before which soft-deleted rows are removed.
+ * @param limit The maximum number of rows to remove.
+ * @return The number of removed rows.
+ */
+ @DeleteProvider(
+ type = DomainRelSQLProviderFactory.class,
+ method = "deleteDomainRelsByLegacyTimeline")
+ Integer deleteDomainRelsByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainRelSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainRelSQLProviderFactory.java
new file mode 100644
index 0000000000..f519873ca7
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/DomainRelSQLProviderFactory.java
@@ -0,0 +1,102 @@
+/*
+ * 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.DomainRelBaseSQLProvider;
+import
org.apache.gravitino.storage.relational.mapper.provider.postgresql.DomainRelPostgreSQLProvider;
+import org.apache.gravitino.storage.relational.po.DomainRelPO;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/** Selects the SQL of the {@code domain_rel} table for the configured
backend. */
+public class DomainRelSQLProviderFactory {
+
+ private static final Map<JDBCBackendType, DomainRelBaseSQLProvider>
DOMAIN_REL_SQL_PROVIDER_MAP =
+ ImmutableMap.of(
+ JDBCBackendType.MYSQL, new DomainRelMySQLProvider(),
+ JDBCBackendType.H2, new DomainRelH2Provider(),
+ JDBCBackendType.POSTGRESQL, new DomainRelPostgreSQLProvider());
+
+ /**
+ * Returns the SQL provider of the configured backend.
+ *
+ * @return The SQL provider.
+ */
+ public static DomainRelBaseSQLProvider getProvider() {
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+
+ JDBCBackendType jdbcBackendType = JDBCBackendType.fromString(databaseId);
+ return DOMAIN_REL_SQL_PROVIDER_MAP.get(jdbcBackendType);
+ }
+
+ static class DomainRelMySQLProvider extends DomainRelBaseSQLProvider {}
+
+ static class DomainRelH2Provider extends DomainRelBaseSQLProvider {}
+
+ /** Delegates {@link DomainRelBaseSQLProvider#selectDomainRelByChildId}. */
+ public static String selectDomainRelByChildId(@Param("childId") Long
childId) {
+ return getProvider().selectDomainRelByChildId(childId);
+ }
+
+ /** Delegates {@link DomainRelBaseSQLProvider#listDomainRelsByChildIds}. */
+ public static String listDomainRelsByChildIds(@Param("childIds") List<Long>
childIds) {
+ return getProvider().listDomainRelsByChildIds(childIds);
+ }
+
+ /** Delegates {@link DomainRelBaseSQLProvider#countDomainRelsByParentId}. */
+ public static String countDomainRelsByParentId(@Param("parentId") Long
parentId) {
+ return getProvider().countDomainRelsByParentId(parentId);
+ }
+
+ /** Delegates {@link DomainRelBaseSQLProvider#insertDomainRel}. */
+ public static String insertDomainRel(@Param("domainRel") DomainRelPO
domainRelPO) {
+ return getProvider().insertDomainRel(domainRelPO);
+ }
+
+ /** Delegates {@link DomainRelBaseSQLProvider#updateDomainRel}. */
+ public static String updateDomainRel(
+ @Param("newDomainRel") DomainRelPO newDomainRelPO,
+ @Param("oldDomainRel") DomainRelPO oldDomainRelPO) {
+ return getProvider().updateDomainRel(newDomainRelPO, oldDomainRelPO);
+ }
+
+ /** Delegates {@link DomainRelBaseSQLProvider#softDeleteDomainRelByChildId}.
*/
+ public static String softDeleteDomainRelByChildId(@Param("childId") Long
childId) {
+ return getProvider().softDeleteDomainRelByChildId(childId);
+ }
+
+ /** Delegates {@link
DomainRelBaseSQLProvider#softDeleteDomainRelsByMetalakeId}. */
+ public static String softDeleteDomainRelsByMetalakeId(@Param("metalakeId")
Long metalakeId) {
+ return getProvider().softDeleteDomainRelsByMetalakeId(metalakeId);
+ }
+
+ /** Delegates {@link
DomainRelBaseSQLProvider#deleteDomainRelsByLegacyTimeline}. */
+ public static String deleteDomainRelsByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return getProvider().deleteDomainRelsByLegacyTimeline(legacyTimeline,
limit);
+ }
+}
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 dc2c908f9d..c949f4a4ac 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
@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableList;
import java.util.List;
import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.DomainRelMapper;
import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
@@ -66,6 +67,7 @@ public class DefaultMapperPackageProvider implements
MapperPackageProvider {
return ImmutableList.of(
CatalogMetaMapper.class,
DomainMetaMapper.class,
+ DomainRelMapper.class,
EntityChangeLogMapper.class,
FilesetMetaMapper.class,
FilesetVersionMapper.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainMetaBaseSQLProvider.java
index 0cf22b34ff..9dbf1358dc 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainMetaBaseSQLProvider.java
@@ -21,6 +21,7 @@ package
org.apache.gravitino.storage.relational.mapper.provider.base;
import static
org.apache.gravitino.storage.relational.mapper.DomainMetaMapper.DOMAIN_TABLE_NAME;
import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.DomainRelMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
@@ -46,8 +47,8 @@ public class DomainMetaBaseSQLProvider {
// Reads take the backing tag name from the live tag, so a deleted tag reads
as no backing tag.
// Locking reads do not join: they must lock the domain row only, and
PostgreSQL rejects FOR
// UPDATE on the nullable side of an outer join.
- private static final String DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME =
- DOMAIN_COLUMNS + ", bt.tag_name as backingTagName";
+ private static final String DOMAIN_COLUMNS_WITH_RESOLVED_NAMES =
+ DOMAIN_COLUMNS + ", bt.tag_name as backingTagName, pd.domain_name as
parentName";
private static final String BACKING_TAG_JOIN =
" LEFT JOIN "
@@ -55,6 +56,14 @@ public class DomainMetaBaseSQLProvider {
+ " bt ON dm.backing_tag_id = bt.tag_id AND dm.metalake_id =
bt.metalake_id"
+ " AND bt.deleted_at = 0";
+ private static final String PARENT_JOIN =
+ " LEFT JOIN "
+ + DomainRelMapper.DOMAIN_REL_TABLE_NAME
+ + " pr ON dm.domain_id = pr.child_id AND pr.deleted_at = 0"
+ + " LEFT JOIN "
+ + DOMAIN_TABLE_NAME
+ + " pd ON pr.parent_id = pd.domain_id AND pd.deleted_at = 0";
+
/**
* Returns SQL that lists the active domains of a metalake.
*
@@ -63,13 +72,14 @@ public class DomainMetaBaseSQLProvider {
*/
public String listDomainPOsByMetalake(@Param("metalakeName") String
metalakeName) {
return "SELECT "
- + DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME
+ + DOMAIN_COLUMNS_WITH_RESOLVED_NAMES
+ " FROM "
+ DOMAIN_TABLE_NAME
+ " dm JOIN "
+ MetalakeMetaMapper.TABLE_NAME
+ " mm ON dm.metalake_id = mm.metalake_id"
+ BACKING_TAG_JOIN
+ + PARENT_JOIN
+ " WHERE mm.metalake_name = #{metalakeName} AND dm.deleted_at = 0 AND
mm.deleted_at = 0";
}
@@ -83,13 +93,14 @@ public class DomainMetaBaseSQLProvider {
public String selectDomainMetaByMetalakeAndName(
@Param("metalakeName") String metalakeName, @Param("domainName") String
domainName) {
return "SELECT "
- + DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME
+ + DOMAIN_COLUMNS_WITH_RESOLVED_NAMES
+ " FROM "
+ DOMAIN_TABLE_NAME
+ " dm JOIN "
+ MetalakeMetaMapper.TABLE_NAME
+ " mm ON dm.metalake_id = mm.metalake_id"
+ BACKING_TAG_JOIN
+ + PARENT_JOIN
+ " WHERE mm.metalake_name = #{metalakeName} AND dm.domain_name =
#{domainName}"
+ " AND dm.deleted_at = 0 AND mm.deleted_at = 0";
}
@@ -131,11 +142,12 @@ public class DomainMetaBaseSQLProvider {
*/
public String selectDomainMetaById(@Param("domainId") Long domainId) {
return "SELECT "
- + DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME
+ + DOMAIN_COLUMNS_WITH_RESOLVED_NAMES
+ " FROM "
+ DOMAIN_TABLE_NAME
+ " dm"
+ BACKING_TAG_JOIN
+ + PARENT_JOIN
+ " WHERE dm.domain_id = #{domainId} AND dm.deleted_at = 0";
}
@@ -146,11 +158,17 @@ public class DomainMetaBaseSQLProvider {
* @return The SQL.
*/
public String selectDomainMetaByIdForUpdate(@Param("domainId") Long
domainId) {
- return "SELECT "
- + DOMAIN_COLUMNS
- + " FROM "
- + DOMAIN_TABLE_NAME
- + " dm WHERE dm.domain_id = #{domainId} AND dm.deleted_at = 0 FOR
UPDATE";
+ return selectLockableDomainMetaById(domainId) + " FOR UPDATE";
+ }
+
+ /**
+ * Returns SQL that selects an active domain by ID and locks it for shared
access.
+ *
+ * @param domainId The domain ID.
+ * @return The SQL.
+ */
+ public String selectDomainMetaByIdForShare(@Param("domainId") Long domainId)
{
+ return selectLockableDomainMetaById(domainId) + " LOCK IN SHARE MODE";
}
/**
@@ -185,13 +203,14 @@ public class DomainMetaBaseSQLProvider {
@Param("metalakeName") String metalakeName, @Param("domainNames")
List<String> domainNames) {
return "<script>"
+ "SELECT "
- + DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME
+ + DOMAIN_COLUMNS_WITH_RESOLVED_NAMES
+ " FROM "
+ DOMAIN_TABLE_NAME
+ " dm JOIN "
+ MetalakeMetaMapper.TABLE_NAME
+ " mm ON dm.metalake_id = mm.metalake_id"
+ BACKING_TAG_JOIN
+ + PARENT_JOIN
+ " WHERE mm.metalake_name = #{metalakeName}"
+ " AND dm.domain_name IN ("
+ "<foreach collection='domainNames' item='domainName' separator=','>"
@@ -202,6 +221,34 @@ public class DomainMetaBaseSQLProvider {
+ "</script>";
}
+ /**
+ * Returns SQL that resolves the live parent and lists its sub-domains in
one statement.
+ *
+ * @param metalakeName The metalake name.
+ * @param parentName The parent domain name.
+ * @return The SQL. An existing childless parent produces a row with a null
domain ID.
+ */
+ public String listDomainPOsByParentIdentifier(
+ @Param("metalakeName") String metalakeName, @Param("parentName") String
parentName) {
+ return "SELECT "
+ + DOMAIN_COLUMNS
+ + ", bt.tag_name as backingTagName, lp.domain_name as parentName"
+ + " FROM "
+ + DOMAIN_TABLE_NAME
+ + " lp JOIN "
+ + MetalakeMetaMapper.TABLE_NAME
+ + " mm ON lp.metalake_id = mm.metalake_id AND mm.deleted_at = 0"
+ + " LEFT JOIN "
+ + DomainRelMapper.DOMAIN_REL_TABLE_NAME
+ + " dr ON lp.domain_id = dr.parent_id AND dr.deleted_at = 0"
+ + " LEFT JOIN "
+ + DOMAIN_TABLE_NAME
+ + " dm ON dr.child_id = dm.domain_id AND dm.deleted_at = 0"
+ + BACKING_TAG_JOIN
+ + " WHERE mm.metalake_name = #{metalakeName} AND lp.domain_name =
#{parentName}"
+ + " AND lp.deleted_at = 0";
+ }
+
/**
* Returns SQL that inserts a domain.
*
@@ -299,4 +346,19 @@ public class DomainMetaBaseSQLProvider {
+ DOMAIN_TABLE_NAME
+ " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit}";
}
+
+ /**
+ * Returns SQL that selects an active domain by ID without joining the
backing tag, so that a
+ * locking clause appended to it locks the domain row only.
+ *
+ * @param domainId The domain ID.
+ * @return The SQL.
+ */
+ protected String selectLockableDomainMetaById(Long domainId) {
+ return "SELECT "
+ + DOMAIN_COLUMNS
+ + " FROM "
+ + DOMAIN_TABLE_NAME
+ + " dm WHERE dm.domain_id = #{domainId} AND dm.deleted_at = 0";
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainRelBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainRelBaseSQLProvider.java
new file mode 100644
index 0000000000..2d852ec9b3
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainRelBaseSQLProvider.java
@@ -0,0 +1,186 @@
+/*
+ * 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.DomainRelMapper.DOMAIN_REL_TABLE_NAME;
+
+import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
+import org.apache.gravitino.storage.relational.po.DomainRelPO;
+import org.apache.ibatis.annotations.Param;
+
+/** The SQL of the {@code domain_rel} table shared by all backends. */
+public class DomainRelBaseSQLProvider {
+
+ private static final String DOMAIN_REL_COLUMNS =
+ "dr.metalake_id as metalakeId, dr.parent_id as parentId, dr.child_id as
childId,"
+ + " dr.audit_info as auditInfo,"
+ + " dr.current_version as currentVersion,"
+ + " dr.last_version as lastVersion,"
+ + " dr.deleted_at as deletedAt";
+
+ /**
+ * Returns SQL that selects the active parent edge of a domain.
+ *
+ * @param childId The ID of the domain.
+ * @return The SQL.
+ */
+ public String selectDomainRelByChildId(@Param("childId") Long childId) {
+ return "SELECT "
+ + DOMAIN_REL_COLUMNS
+ + " FROM "
+ + DOMAIN_REL_TABLE_NAME
+ + " dr WHERE dr.child_id = #{childId} AND dr.deleted_at = 0";
+ }
+
+ /**
+ * Returns SQL that lists the active parent edges of the given domains, with
the name of each
+ * parent.
+ *
+ * @param childIds The IDs of the domains.
+ * @return The SQL.
+ */
+ public String listDomainRelsByChildIds(@Param("childIds") List<Long>
childIds) {
+ return "<script>"
+ + "SELECT "
+ + DOMAIN_REL_COLUMNS
+ + ", pd.domain_name as parentName"
+ + " FROM "
+ + DOMAIN_REL_TABLE_NAME
+ + " dr JOIN "
+ + DomainMetaMapper.DOMAIN_TABLE_NAME
+ + " pd ON dr.parent_id = pd.domain_id AND pd.deleted_at = 0"
+ + " WHERE dr.deleted_at = 0"
+ + " AND dr.child_id IN ("
+ + "<foreach collection='childIds' item='childId' separator=','>"
+ + "#{childId}"
+ + "</foreach>"
+ + ")"
+ + "</script>";
+ }
+
+ /**
+ * Returns SQL that counts the active sub-domains of a domain.
+ *
+ * @param parentId The ID of the parent domain.
+ * @return The SQL.
+ */
+ public String countDomainRelsByParentId(@Param("parentId") Long parentId) {
+ return "SELECT COUNT(*) FROM "
+ + DOMAIN_REL_TABLE_NAME
+ + " WHERE parent_id = #{parentId} AND deleted_at = 0";
+ }
+
+ /**
+ * Returns SQL that inserts a parent edge.
+ *
+ * @param domainRelPO The edge to insert.
+ * @return The SQL.
+ */
+ public String insertDomainRel(@Param("domainRel") DomainRelPO domainRelPO) {
+ return "INSERT INTO "
+ + DOMAIN_REL_TABLE_NAME
+ + " (metalake_id, parent_id, child_id, audit_info, current_version,
last_version,"
+ + " deleted_at)"
+ + " VALUES ("
+ + " #{domainRel.metalakeId},"
+ + " #{domainRel.parentId},"
+ + " #{domainRel.childId},"
+ + " #{domainRel.auditInfo},"
+ + " #{domainRel.currentVersion},"
+ + " #{domainRel.lastVersion},"
+ + " #{domainRel.deletedAt}"
+ + " )";
+ }
+
+ /**
+ * Returns SQL that updates an active parent edge when its version still
matches the observed one.
+ *
+ * @param newDomainRelPO The new edge.
+ * @param oldDomainRelPO The edge observed by the caller.
+ * @return The SQL.
+ */
+ public String updateDomainRel(
+ @Param("newDomainRel") DomainRelPO newDomainRelPO,
+ @Param("oldDomainRel") DomainRelPO oldDomainRelPO) {
+ return "UPDATE "
+ + DOMAIN_REL_TABLE_NAME
+ + " SET parent_id = #{newDomainRel.parentId},"
+ + " audit_info = #{newDomainRel.auditInfo},"
+ + " current_version = #{newDomainRel.currentVersion},"
+ + " last_version = #{newDomainRel.lastVersion},"
+ + " deleted_at = #{newDomainRel.deletedAt}"
+ + " WHERE child_id = #{oldDomainRel.childId}"
+ + " AND parent_id = #{oldDomainRel.parentId}"
+ + " AND current_version = #{oldDomainRel.currentVersion}"
+ + " AND deleted_at = 0";
+ }
+
+ /**
+ * Returns SQL that soft-deletes the active parent edge of a domain.
+ *
+ * @param childId The ID of the domain.
+ * @return The SQL.
+ */
+ // Advance beyond existing tombstones as well as database time: PostgreSQL
timestamps stay
+ // constant within a transaction, and millisecond clocks can collide on
every backend. The
+ // aggregate derived table also prevents MySQL from merging a read of the
UPDATE target.
+ public String softDeleteDomainRelByChildId(@Param("childId") Long childId) {
+ return "UPDATE "
+ + DOMAIN_REL_TABLE_NAME
+ + " SET deleted_at = "
+ + "GREATEST("
+ + DatabaseTimeSQL.MYSQL
+ + ", (SELECT COALESCE(last_deleted_at, 0) + 1 FROM"
+ + " (SELECT MAX(deleted_at) AS last_deleted_at FROM domain_rel WHERE
child_id = #{childId}) legacy))"
+ + " WHERE child_id = #{childId} AND deleted_at = 0";
+ }
+
+ /**
+ * Returns SQL that soft-deletes all active parent edges of a metalake.
+ *
+ * @param metalakeId The metalake ID.
+ * @return The SQL.
+ */
+ public String softDeleteDomainRelsByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ return "UPDATE "
+ + DOMAIN_REL_TABLE_NAME
+ + " SET deleted_at = "
+ + "GREATEST("
+ + DatabaseTimeSQL.MYSQL
+ + ", (SELECT COALESCE(last_deleted_at, 0) + 1 FROM"
+ + " (SELECT MAX(deleted_at) AS last_deleted_at FROM domain_rel WHERE
metalake_id = #{metalakeId}) legacy))"
+ + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ }
+
+ /**
+ * Returns SQL that hard-deletes parent edges soft-deleted before the given
time.
+ *
+ * @param legacyTimeline The time before which soft-deleted rows are removed.
+ * @param limit The maximum number of rows to remove.
+ * @return The SQL.
+ */
+ public String deleteDomainRelsByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return "DELETE FROM "
+ + DOMAIN_REL_TABLE_NAME
+ + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit}";
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainMetaPostgreSQLProvider.java
index 6aa2a0b6da..ba60386cee 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainMetaPostgreSQLProvider.java
@@ -27,6 +27,11 @@ import org.apache.ibatis.annotations.Param;
/** The SQL of the {@code domain_meta} table for PostgreSQL. */
public class DomainMetaPostgreSQLProvider extends DomainMetaBaseSQLProvider {
+ @Override
+ public String selectDomainMetaByIdForShare(@Param("domainId") Long domainId)
{
+ return selectLockableDomainMetaById(domainId) + " FOR SHARE";
+ }
+
@Override
public String softDeleteDomainMetaByIdAndVersion(
@Param("domainId") Long domainId, @Param("currentVersion") Long
currentVersion) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainRelPostgreSQLProvider.java
similarity index 60%
copy from
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainMetaPostgreSQLProvider.java
copy to
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainRelPostgreSQLProvider.java
index 6aa2a0b6da..184b79a6c1 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/DomainRelPostgreSQLProvider.java
@@ -18,42 +18,46 @@
*/
package org.apache.gravitino.storage.relational.mapper.provider.postgresql;
-import static
org.apache.gravitino.storage.relational.mapper.DomainMetaMapper.DOMAIN_TABLE_NAME;
+import static
org.apache.gravitino.storage.relational.mapper.DomainRelMapper.DOMAIN_REL_TABLE_NAME;
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
-import
org.apache.gravitino.storage.relational.mapper.provider.base.DomainMetaBaseSQLProvider;
+import
org.apache.gravitino.storage.relational.mapper.provider.base.DomainRelBaseSQLProvider;
import org.apache.ibatis.annotations.Param;
-/** The SQL of the {@code domain_meta} table for PostgreSQL. */
-public class DomainMetaPostgreSQLProvider extends DomainMetaBaseSQLProvider {
+/** The SQL of the {@code domain_rel} table for PostgreSQL. */
+public class DomainRelPostgreSQLProvider extends DomainRelBaseSQLProvider {
@Override
- public String softDeleteDomainMetaByIdAndVersion(
- @Param("domainId") Long domainId, @Param("currentVersion") Long
currentVersion) {
+ public String softDeleteDomainRelByChildId(@Param("childId") Long childId) {
return "UPDATE "
- + DOMAIN_TABLE_NAME
+ + DOMAIN_REL_TABLE_NAME
+ " SET deleted_at = "
+ + "GREATEST("
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE domain_id = #{domainId} AND current_version =
#{currentVersion}"
- + " AND deleted_at = 0";
+ + ", (SELECT COALESCE(last_deleted_at, 0) + 1 FROM"
+ + " (SELECT MAX(deleted_at) AS last_deleted_at FROM domain_rel WHERE
child_id = #{childId}) legacy))"
+ + " WHERE child_id = #{childId} AND deleted_at = 0";
}
@Override
- public String softDeleteDomainMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ public String softDeleteDomainRelsByMetalakeId(@Param("metalakeId") Long
metalakeId) {
return "UPDATE "
- + DOMAIN_TABLE_NAME
+ + DOMAIN_REL_TABLE_NAME
+ " SET deleted_at = "
+ + "GREATEST("
+ DatabaseTimeSQL.POSTGRESQL
+ + ", (SELECT COALESCE(last_deleted_at, 0) + 1 FROM"
+ + " (SELECT MAX(deleted_at) AS last_deleted_at FROM domain_rel WHERE
metalake_id = #{metalakeId}) legacy))"
+ " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
}
@Override
- public String deleteDomainMetasByLegacyTimeline(
+ public String deleteDomainRelsByLegacyTimeline(
@Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
return "DELETE FROM "
- + DOMAIN_TABLE_NAME
- + " WHERE domain_id IN (SELECT domain_id FROM "
- + DOMAIN_TABLE_NAME
+ + DOMAIN_REL_TABLE_NAME
+ + " WHERE id IN (SELECT id FROM "
+ + DOMAIN_REL_TABLE_NAME
+ " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit})";
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/DomainPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/DomainPO.java
index 6afb443ba6..fcf9e703ee 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/po/DomainPO.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/po/DomainPO.java
@@ -26,6 +26,7 @@ import com.google.common.base.Preconditions;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.ToString;
@@ -59,7 +60,8 @@ public class DomainPO {
private String comment;
private String status;
private Long backingTagId;
- private String backingTagName;
+ @EqualsAndHashCode.Exclude private String backingTagName;
+ @EqualsAndHashCode.Exclude private String parentName;
private String backingTagValue;
private String properties;
private String auditInfo;
@@ -135,15 +137,29 @@ public class DomainPO {
}
/**
- * Converts a domain persistent object to a domain entity. The parent is not
stored in the domain
- * row, so it is not set. The backing tag name is set only when the read
joined the live tag.
+ * Converts a domain persistent object to a domain entity. Parent and
backing tag names are
+ * populated by reads that join their live rows.
*
* @param domainPO The domain persistent object.
* @param namespace The namespace of the domain.
* @return The domain entity.
*/
- @SuppressWarnings("unchecked")
public static DomainEntity fromDomainPO(DomainPO domainPO, Namespace
namespace) {
+ return fromDomainPO(domainPO, namespace, domainPO.parentName());
+ }
+
+ /**
+ * Converts a domain persistent object to a domain entity with the given
parent. The parent is not
+ * stored in the domain row; it is read from the sub-domain relation.
+ *
+ * @param domainPO The domain persistent object.
+ * @param namespace The namespace of the domain.
+ * @param parentName The name of the parent domain, or null if the domain is
top-level.
+ * @return The domain entity.
+ */
+ @SuppressWarnings("unchecked")
+ public static DomainEntity fromDomainPO(
+ DomainPO domainPO, Namespace namespace, @Nullable String parentName) {
try {
return DomainEntity.builder()
.withId(domainPO.domainId())
@@ -151,6 +167,7 @@ public class DomainPO {
.withNamespace(namespace)
.withComment(domainPO.comment())
.withStatus(Domain.Status.valueOf(domainPO.status()))
+ .withParent(parentName)
.withBackingTagId(domainPO.backingTagId())
.withBackingTagName(domainPO.backingTagName())
.withBackingTagValue(domainPO.backingTagValue())
@@ -164,14 +181,19 @@ public class DomainPO {
}
/**
- * Converts domain persistent objects to domain entities.
+ * Converts domain persistent objects to domain entities with their parents.
*
* @param domainPOs The domain persistent objects.
* @param namespace The namespace of the domains.
+ * @param parentNames The name of the parent of each sub-domain, keyed by
domain ID. Domains that
+ * are not in the map are top-level.
* @return The domain entities.
*/
- public static List<DomainEntity> fromDomainPOs(List<DomainPO> domainPOs,
Namespace namespace) {
- return domainPOs.stream().map(po -> fromDomainPO(po,
namespace)).collect(Collectors.toList());
+ public static List<DomainEntity> fromDomainPOs(
+ List<DomainPO> domainPOs, Namespace namespace, Map<Long, String>
parentNames) {
+ return domainPOs.stream()
+ .map(po -> fromDomainPO(po, namespace, parentNames.get(po.domainId())))
+ .collect(Collectors.toList());
}
// Returns the built PO rather than the lombok builder: javadoc runs before
lombok, so a
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/DomainRelPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/DomainRelPO.java
new file mode 100644
index 0000000000..51b3429132
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/po/DomainRelPO.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 static
org.apache.gravitino.storage.relational.utils.POConverters.DEFAULT_DELETED_AT;
+import static
org.apache.gravitino.storage.relational.utils.POConverters.INIT_VERSION;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.json.JsonUtils;
+import org.apache.gravitino.meta.AuditInfo;
+
+/**
+ * The persistent object of a row in {@code domain_rel}, the edge from a
sub-domain to its parent
+ * domain. The parent domain name is not a column of the row: reads that join
the live parent fill
+ * it.
+ */
+@Getter
+@Accessors(fluent = true)
+@EqualsAndHashCode
+@ToString
+public class DomainRelPO {
+
+ private Long metalakeId;
+ private Long parentId;
+ @EqualsAndHashCode.Exclude private String parentName;
+ private Long childId;
+ private String auditInfo;
+ private Long currentVersion;
+ private Long lastVersion;
+ private Long deletedAt;
+
+ /** Creates an empty domain relation persistent object, used by MyBatis to
map result rows. */
+ public DomainRelPO() {}
+
+ @lombok.Builder(setterPrefix = "with")
+ private DomainRelPO(
+ Long metalakeId,
+ Long parentId,
+ Long childId,
+ String auditInfo,
+ Long currentVersion,
+ Long lastVersion,
+ Long deletedAt) {
+ Preconditions.checkArgument(metalakeId != null, "metalakeId cannot be
null");
+ Preconditions.checkArgument(parentId != null, "parentId cannot be null");
+ Preconditions.checkArgument(childId != null, "childId cannot be null");
+ Preconditions.checkArgument(
+ !parentId.equals(childId), "A domain cannot be its own parent: %s",
childId);
+ Preconditions.checkArgument(StringUtils.isNotBlank(auditInfo), "auditInfo
cannot be blank");
+ Preconditions.checkArgument(currentVersion != null, "currentVersion cannot
be null");
+ Preconditions.checkArgument(lastVersion != null, "lastVersion cannot be
null");
+ Preconditions.checkArgument(deletedAt != null, "deletedAt cannot be null");
+
+ this.metalakeId = metalakeId;
+ this.parentId = parentId;
+ this.childId = childId;
+ this.auditInfo = auditInfo;
+ this.currentVersion = currentVersion;
+ this.lastVersion = lastVersion;
+ this.deletedAt = deletedAt;
+ }
+
+ /**
+ * Creates the first version of the persistent object of a new parent edge.
+ *
+ * @param metalakeId The ID of the metalake of both domains.
+ * @param parentId The ID of the parent domain.
+ * @param childId The ID of the sub-domain.
+ * @param auditInfo The audit info of the edge.
+ * @return The domain relation persistent object.
+ */
+ public static DomainRelPO initializeDomainRelPO(
+ Long metalakeId, Long parentId, Long childId, AuditInfo auditInfo) {
+ return builder()
+ .withMetalakeId(metalakeId)
+ .withParentId(parentId)
+ .withChildId(childId)
+ .withAuditInfo(serializeAuditInfo(auditInfo))
+ .withCurrentVersion(INIT_VERSION)
+ .withLastVersion(INIT_VERSION)
+ .withDeletedAt(DEFAULT_DELETED_AT)
+ .build();
+ }
+
+ /**
+ * Creates the next version of the persistent object of an edge that is
re-pointed to another
+ * parent. The edge keeps its creator and creation time and takes the last
modifier and the last
+ * modified time from {@code modifiedAuditInfo}.
+ *
+ * @param oldDomainRelPO The edge observed by the caller.
+ * @param newParentId The ID of the new parent domain.
+ * @param modifiedAuditInfo The audit info of the change.
+ * @return The domain relation persistent object with the next version.
+ */
+ public static DomainRelPO updateDomainRelPO(
+ DomainRelPO oldDomainRelPO, Long newParentId, AuditInfo
modifiedAuditInfo) {
+ AuditInfo oldAuditInfo = deserializeAuditInfo(oldDomainRelPO.auditInfo());
+ AuditInfo newAuditInfo =
+ AuditInfo.builder()
+ .withCreator(oldAuditInfo.creator())
+ .withCreateTime(oldAuditInfo.createTime())
+ .withLastModifier(modifiedAuditInfo.lastModifier())
+ .withLastModifiedTime(modifiedAuditInfo.lastModifiedTime())
+ .build();
+ Long nextVersion = oldDomainRelPO.currentVersion() + 1;
+ return builder()
+ .withMetalakeId(oldDomainRelPO.metalakeId())
+ .withParentId(newParentId)
+ .withChildId(oldDomainRelPO.childId())
+ .withAuditInfo(serializeAuditInfo(newAuditInfo))
+ .withCurrentVersion(nextVersion)
+ .withLastVersion(nextVersion)
+ .withDeletedAt(DEFAULT_DELETED_AT)
+ .build();
+ }
+
+ private static String serializeAuditInfo(AuditInfo auditInfo) {
+ try {
+ return JsonUtils.anyFieldMapper().writeValueAsString(auditInfo);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Failed to serialize domain relation audit
info", e);
+ }
+ }
+
+ private static AuditInfo deserializeAuditInfo(String auditInfo) {
+ try {
+ return JsonUtils.anyFieldMapper().readValue(auditInfo, AuditInfo.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Failed to deserialize domain relation audit
info", e);
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
index ebee5d7bfd..07dde79f3d 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
@@ -35,9 +35,11 @@ import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.meta.DomainEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.DomainRelMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
@@ -76,7 +78,7 @@ public class DomainMetaService {
List<DomainPO> domainPOs =
SessionUtils.getWithoutCommit(
DomainMetaMapper.class, mapper ->
mapper.listDomainPOsByMetalake(metalakeName));
- return DomainPO.fromDomainPOs(domainPOs, ns);
+ return fromDomainPOsWithParents(domainPOs, ns);
}
/**
@@ -92,7 +94,7 @@ public class DomainMetaService {
public DomainEntity getDomainByIdentifier(NameIdentifier ident) {
String metalakeName = ident.namespace().level(0);
DomainPO domainPO = getDomainPOByMetalakeAndName(metalakeName,
ident.name());
- return DomainPO.fromDomainPO(domainPO, ident.namespace());
+ return fromDomainPOWithParent(domainPO, ident.namespace());
}
/**
@@ -118,7 +120,7 @@ public class DomainMetaService {
SessionUtils.getWithoutCommit(
DomainMetaMapper.class,
mapper ->
mapper.listDomainPOsByMetalakeAndDomainNames(metalakeName, domainNames));
- return DomainPO.fromDomainPOs(domainPOs, firstIdent.namespace());
+ return fromDomainPOsWithParents(domainPOs, firstIdent.namespace());
}
/**
@@ -185,7 +187,8 @@ public class DomainMetaService {
* changing the new domain; a later call with the name acts on the new
domain.
*
* <p>The returned domain is read back in the same transaction, so its
backing tag name comes from
- * the tag it is now bound to rather than from the updater's entity.
+ * the tag it is now bound to rather than from the updater's entity. The
parent cannot be changed
+ * by an update, an updater that changes it is rejected; {@link
DomainRelService} changes it.
*
* @param identifier The domain identifier.
* @param updater The function that returns the updated domain.
@@ -204,7 +207,7 @@ public class DomainMetaService {
try {
DomainPO observedDomainPO = getDomainPOByMetalakeAndName(metalakeName,
identifier.name());
- DomainPO[] storedDomainPO = new DomainPO[1];
+ DomainEntity[] storedDomain = new DomainEntity[1];
SessionUtils.doWithCommit(
DomainMetaMapper.class,
mapper -> {
@@ -216,20 +219,26 @@ public class DomainMetaService {
// name of its backing tag.
DomainPO lockedDomainPO =
mapper.selectDomainMetaById(observedDomainPO.domainId());
DomainEntity oldDomainEntity =
- DomainPO.fromDomainPO(lockedDomainPO, identifier.namespace());
+ fromDomainPOWithParent(lockedDomainPO, identifier.namespace());
DomainEntity updatedDomainEntity = (DomainEntity)
updater.apply((E) oldDomainEntity);
Preconditions.checkArgument(
Objects.equals(oldDomainEntity.id(), updatedDomainEntity.id()),
"The updated domain entity id: %s must have the same id as the
old entity id %s",
updatedDomainEntity.id(),
oldDomainEntity.id());
+ Preconditions.checkArgument(
+ Objects.equals(oldDomainEntity.parent(),
updatedDomainEntity.parent()),
+ "The parent of domain %s cannot be changed by an update, use
DomainRelService",
+ identifier);
DomainPO newDomainPO = DomainPO.updateDomainPO(lockedDomainPO,
updatedDomainEntity);
updateDomainWithVersion(identifier, mapper, lockedDomainPO,
newDomainPO);
- storedDomainPO[0] =
mapper.selectDomainMetaById(lockedDomainPO.domainId());
+ storedDomain[0] =
+ DomainPO.fromDomainPO(
+ mapper.selectDomainMetaById(lockedDomainPO.domainId()),
identifier.namespace());
});
- return DomainPO.fromDomainPO(storedDomainPO[0], identifier.namespace());
+ return storedDomain[0];
} catch (RuntimeException e) {
ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
identifier.toString());
throw e;
@@ -237,11 +246,12 @@ public class DomainMetaService {
}
/**
- * Deletes a domain, together with its owner and the privileges granted on
it. The backing tag and
- * its assignments are kept.
+ * Deletes a domain, together with its parent edge, its owner and the
privileges granted on it.
+ * The backing tag and its assignments are kept. A domain with sub-domains
is not deleted.
*
* @param identifier The domain identifier.
* @return True if the domain is deleted, false if it does not exist.
+ * @throws NonEmptyEntityException If the domain has sub-domains.
*/
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
@@ -257,19 +267,32 @@ public class DomainMetaService {
}
/**
- * Hard-deletes the domains soft-deleted before the given time.
+ * Hard-deletes the domains and the parent edges soft-deleted before the
given time.
*
- * @param legacyTimeline The time before which soft-deleted domains are
removed.
- * @param limit The maximum number of domains to remove.
- * @return The number of removed domains.
+ * @param legacyTimeline The time before which soft-deleted rows are removed.
+ * @param limit The maximum number of rows to remove from each table.
+ * @return The number of removed rows.
*/
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteDomainMetasByLegacyTimeline")
public int deleteDomainMetasByLegacyTimeline(long legacyTimeline, int limit)
{
- return SessionUtils.doWithCommitAndFetchResult(
- DomainMetaMapper.class,
- mapper -> mapper.deleteDomainMetasByLegacyTimeline(legacyTimeline,
limit));
+ int[] domainDeletedCount = new int[] {0};
+ int[] domainRelDeletedCount = new int[] {0};
+
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ domainDeletedCount[0] =
+ SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class,
+ mapper ->
mapper.deleteDomainMetasByLegacyTimeline(legacyTimeline, limit)),
+ () ->
+ domainRelDeletedCount[0] =
+ SessionUtils.getWithoutCommit(
+ DomainRelMapper.class,
+ mapper ->
mapper.deleteDomainRelsByLegacyTimeline(legacyTimeline, limit)));
+
+ return domainDeletedCount[0] + domainRelDeletedCount[0];
}
/**
@@ -311,15 +334,23 @@ public class DomainMetaService {
// The metalake lock comes first. Dropping the metalake soft-deletes the
privileges before the
// domains, while this delete changes the domain row before its
privileges; without the lock
- // the two could deadlock on a domain with privileges. The domain row is
locked next, and a row
- // that is gone or renamed makes the delete report the domain as missing
before any dependent
- // row is touched. The version-checked delete keeps the version as a
backstop. The cleanups that
- // follow are blanket soft deletes for this domain ID, so their row counts
carry no information
- // and are not checked.
+ // the two could deadlock on a domain with privileges. The domain row is
locked next, which also
+ // keeps a sub-domain from being attached while the sub-domain check runs.
A row that is gone or
+ // renamed makes the delete report the domain as missing before any
dependent row is touched.
+ // The version-checked delete keeps the version as a backstop. The
cleanups that follow are
+ // blanket soft deletes for this domain ID, so their row counts carry no
information and are not
+ // checked.
try {
SessionUtils.doMultipleWithCommit(
() -> lockMetalakeForDomainWrite(identifier.namespace().level(0),
domainPO.metalakeId()),
- () -> deleteDomainWithVersion(identifier,
lockAndGetDomainPO(identifier, domainPO)),
+ () -> {
+ DomainPO lockedDomainPO = lockAndGetDomainPO(identifier, domainPO);
+
DomainRelService.getInstance().checkDomainHasNoSubDomains(identifier, domainId);
+ deleteDomainWithVersion(identifier, lockedDomainPO);
+ },
+ () ->
+ SessionUtils.doWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.softDeleteDomainRelByChildId(domainId)),
() ->
SessionUtils.doWithoutCommit(
OwnerMetaMapper.class,
@@ -482,4 +513,15 @@ public class DomainMetaService {
}
return domainPO;
}
+
+ private DomainEntity fromDomainPOWithParent(DomainPO domainPO, Namespace
namespace) {
+ return DomainPO.fromDomainPO(domainPO, namespace);
+ }
+
+ private List<DomainEntity> fromDomainPOsWithParents(
+ List<DomainPO> domainPOs, Namespace namespace) {
+ return domainPOs.stream()
+ .map(po -> DomainPO.fromDomainPO(po, namespace))
+ .collect(Collectors.toList());
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainRelService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainRelService.java
new file mode 100644
index 0000000000..a2718d553a
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainRelService.java
@@ -0,0 +1,350 @@
+/*
+ * 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 static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.DomainEntity;
+import org.apache.gravitino.metrics.Monitored;
+import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.DomainRelMapper;
+import org.apache.gravitino.storage.relational.po.DomainPO;
+import org.apache.gravitino.storage.relational.po.DomainRelPO;
+import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+
+/**
+ * The service that stores the parent edge of sub-domains in the {@code
domain_rel} table. A domain
+ * has at most one active edge, enforced by the unique key on {@code
(child_id, deleted_at)}.
+ *
+ * <p>For parent-name-only lookups, this service provides the shared resolver
through {@link
+ * #getParentDomainName(Long)} and {@link #batchGetParentDomainNames(List)}.
Future inherited
+ * authorization checks can reuse these methods. Domain entity reads instead
join the live parent
+ * into the same query as the domain metadata, preserving a consistent
statement snapshot without a
+ * separate resolver lookup.
+ *
+ * <p>Every change of the hierarchy first locks the metalake row, then locks
the {@code domain_meta}
+ * rows of the domains involved in domain ID order: the sub-domain exclusively
and the parent
+ * shared. Taking the metalake first keeps the order of a metalake drop, which
locks the metalake
+ * row exclusively before its domain rows, so the two cannot deadlock. All
writers of an edge lock
+ * the sub-domain, so the edge read under that lock is current. Each locked
row is re-read and must
+ * still have the observed name, otherwise the change fails as if it did not
exist.
+ *
+ * <p>This service does not enforce the shape of the hierarchy. The domain
manager keeps it one
+ * level deep, which also keeps it acyclic.
+ */
+public class DomainRelService {
+
+ private static final DomainRelService INSTANCE = new DomainRelService();
+
+ private DomainRelService() {}
+
+ /**
+ * Returns the singleton service instance.
+ *
+ * @return The domain relation service.
+ */
+ public static DomainRelService getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Returns the name of the current parent of a domain.
+ *
+ * @param domainId The domain ID.
+ * @return The name of the parent domain, or empty if the domain is
top-level.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "getParentDomainName")
+ public Optional<String> getParentDomainName(Long domainId) {
+ return Optional.ofNullable(
+
batchGetParentDomainNames(Collections.singletonList(domainId)).get(domainId));
+ }
+
+ /**
+ * Returns the names of the current parents of the given domains.
+ *
+ * @param domainIds The domain IDs.
+ * @return The name of the parent of each domain that has one, keyed by
domain ID. Top-level
+ * domains are not in the map.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "batchGetParentDomainNames")
+ public Map<Long, String> batchGetParentDomainNames(List<Long> domainIds) {
+ if (domainIds.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ List<DomainRelPO> domainRelPOs =
+ SessionUtils.getWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.listDomainRelsByChildIds(domainIds));
+ return domainRelPOs.stream()
+ .collect(Collectors.toMap(DomainRelPO::childId,
DomainRelPO::parentName));
+ }
+
+ /**
+ * Lists the sub-domains of a domain.
+ *
+ * @param parentIdent The identifier of the parent domain.
+ * @return The sub-domains, each with the parent set.
+ * @throws NoSuchEntityException If the parent domain does not exist.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "listSubDomains")
+ public List<DomainEntity> listSubDomains(NameIdentifier parentIdent) {
+ List<DomainPO> subDomainPOs =
+ SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class,
+ mapper ->
+ mapper.listDomainPOsByParentIdentifier(
+ parentIdent.namespace().level(0), parentIdent.name()));
+ if (subDomainPOs.isEmpty()) {
+ throw new NoSuchEntityException("The domain %s does not exist",
parentIdent);
+ }
+ return subDomainPOs.stream()
+ .filter(po -> po.domainId() != null)
+ .map(po -> DomainPO.fromDomainPO(po, parentIdent.namespace()))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Inserts the parent edge of a top-level domain, which makes it a
sub-domain of {@code
+ * parentIdent}.
+ *
+ * @param childIdent The identifier of the domain.
+ * @param parentIdent The identifier of the parent domain, in the same
metalake.
+ * @param auditInfo The audit info of the edge.
+ * @throws IOException If the edge cannot be stored.
+ * @throws NoSuchEntityException If either domain does not exist.
+ * @throws EntityAlreadyExistsException If the domain already has a parent.
+ * @throws IllegalArgumentException If the parent is the domain itself or is
in another metalake.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "insertDomainRel")
+ public void insertDomainRel(
+ NameIdentifier childIdent, NameIdentifier parentIdent, AuditInfo
auditInfo)
+ throws IOException {
+ checkSameMetalakeAndDistinct(childIdent, parentIdent);
+ try {
+ DomainPO childPO = getDomainPO(childIdent);
+ DomainPO parentPO = getDomainPO(parentIdent);
+ SessionUtils.doMultipleWithCommit(
+ () -> lockMetalakeAndDomains(childIdent, childPO, parentPO),
+ () -> {
+ if (selectDomainRel(childPO.domainId()) != null) {
+ throw new EntityAlreadyExistsException(
+ "The domain %s already has a parent", childIdent);
+ }
+ DomainRelPO domainRelPO =
+ DomainRelPO.initializeDomainRelPO(
+ childPO.metalakeId(), parentPO.domainId(),
childPO.domainId(), auditInfo);
+ SessionUtils.doWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.insertDomainRel(domainRelPO));
+ });
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
childIdent.toString());
+ throw e;
+ }
+ }
+
+ /**
+ * Re-points the parent edge of a sub-domain to another parent. Re-pointing
it to its current
+ * parent changes nothing.
+ *
+ * @param childIdent The identifier of the sub-domain.
+ * @param newParentIdent The identifier of the new parent domain, in the
same metalake.
+ * @param auditInfo The audit info of the change; the edge keeps its creator.
+ * @throws IOException If the edge cannot be stored.
+ * @throws NoSuchEntityException If either domain does not exist, or the
domain has no parent.
+ * @throws IllegalArgumentException If the new parent is the domain itself
or is in another
+ * metalake.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "updateDomainRel")
+ public void updateDomainRel(
+ NameIdentifier childIdent, NameIdentifier newParentIdent, AuditInfo
auditInfo)
+ throws IOException {
+ checkSameMetalakeAndDistinct(childIdent, newParentIdent);
+ try {
+ DomainPO childPO = getDomainPO(childIdent);
+ DomainPO newParentPO = getDomainPO(newParentIdent);
+ SessionUtils.doMultipleWithCommit(
+ () -> lockMetalakeAndDomains(childIdent, childPO, newParentPO),
+ () -> {
+ DomainRelPO oldDomainRelPO = selectDomainRel(childPO.domainId());
+ if (oldDomainRelPO == null) {
+ throw new NoSuchEntityException("The domain %s has no parent",
childIdent);
+ }
+ if (Objects.equals(oldDomainRelPO.parentId(),
newParentPO.domainId())) {
+ return;
+ }
+ DomainRelPO newDomainRelPO =
+ DomainRelPO.updateDomainRelPO(oldDomainRelPO,
newParentPO.domainId(), auditInfo);
+ OccWriteSupport.updateWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ DomainRelMapper.class,
+ mapper -> mapper.updateDomainRel(newDomainRelPO,
oldDomainRelPO)),
+ () ->
ExceptionUtils.concurrentModification(Entity.EntityType.DOMAIN, childIdent));
+ });
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
childIdent.toString());
+ throw e;
+ }
+ }
+
+ /**
+ * Removes the parent edge of a domain, which makes it top-level.
+ *
+ * @param childIdent The identifier of the domain.
+ * @return True if the edge is removed, false if the domain is already
top-level.
+ * @throws IOException If the edge cannot be removed.
+ * @throws NoSuchEntityException If the domain does not exist.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "deleteDomainRel")
+ public boolean deleteDomainRel(NameIdentifier childIdent) throws IOException
{
+ int[] deleted = new int[1];
+ try {
+ DomainPO childPO = getDomainPO(childIdent);
+ SessionUtils.doMultipleWithCommit(
+ () -> lockMetalakeAndDomains(childIdent, childPO, null),
+ () -> deleted[0] = softDeleteDomainRel(childPO.domainId()));
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
childIdent.toString());
+ throw e;
+ }
+ return deleted[0] > 0;
+ }
+
+ /**
+ * Checks that a domain being deleted has no sub-domains, in the caller's
transaction. The caller
+ * must hold the exclusive lock on the domain row, so no sub-domain can be
attached to it
+ * meanwhile.
+ *
+ * <p>The sub-domain count is a plain read. It still sees every edge
attached before the lock was
+ * granted, because the store runs at read committed, where each statement
reads the data
+ * committed before it starts.
+ *
+ * @param identifier The identifier of the domain.
+ * @param domainId The domain ID.
+ * @throws NonEmptyEntityException If the domain has sub-domains.
+ */
+ void checkDomainHasNoSubDomains(NameIdentifier identifier, long domainId) {
+ int subDomainCount =
+ SessionUtils.getWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.countDomainRelsByParentId(domainId));
+ if (subDomainCount > 0) {
+ throw new NonEmptyEntityException(
+ "The domain %s has %d sub-domain(s), remove them first", identifier,
subDomainCount);
+ }
+ }
+
+ private void lockMetalakeAndDomains(
+ NameIdentifier childIdent, DomainPO childPO, @Nullable DomainPO
parentPO) {
+ MetalakeMetaService.getInstance()
+ .lockMetalakeForChildWrite(childIdent.namespace().level(0),
childPO.metalakeId());
+ if (parentPO == null) {
+ lockDomain(childPO, true);
+ } else if (childPO.domainId() < parentPO.domainId()) {
+ lockDomain(childPO, true);
+ lockDomain(parentPO, false);
+ } else {
+ lockDomain(parentPO, false);
+ lockDomain(childPO, true);
+ }
+ }
+
+ private void lockDomain(DomainPO observedDomainPO, boolean exclusive) {
+ OccWriteSupport.lockParentForChildWrite(
+ observedDomainPO.domainName(),
+ Entity.EntityType.DOMAIN,
+ () ->
+ SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class,
+ mapper ->
+ exclusive
+ ?
mapper.selectDomainMetaByIdForUpdate(observedDomainPO.domainId())
+ :
mapper.selectDomainMetaByIdForShare(observedDomainPO.domainId())),
+ null,
+ current ->
+ Objects.equals(current.domainName(), observedDomainPO.domainName())
+ && Objects.equals(current.metalakeId(),
observedDomainPO.metalakeId()));
+ }
+
+ @Nullable
+ private DomainRelPO selectDomainRel(Long childId) {
+ return SessionUtils.getWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.selectDomainRelByChildId(childId));
+ }
+
+ private int softDeleteDomainRel(Long childId) {
+ return SessionUtils.getWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.softDeleteDomainRelByChildId(childId));
+ }
+
+ private DomainPO getDomainPO(NameIdentifier identifier) {
+ DomainPO domainPO =
+ SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class,
+ mapper ->
+ mapper.selectDomainMetaByMetalakeAndName(
+ identifier.namespace().level(0), identifier.name()));
+ if (domainPO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.DOMAIN.name().toLowerCase(),
+ identifier.name());
+ }
+ return domainPO;
+ }
+
+ private static void checkSameMetalakeAndDistinct(
+ NameIdentifier childIdent, NameIdentifier parentIdent) {
+ Preconditions.checkArgument(
+ Objects.equals(childIdent.namespace(), parentIdent.namespace()),
+ "The domain %s and its parent %s must be in the same metalake",
+ childIdent,
+ parentIdent);
+ Preconditions.checkArgument(
+ !Objects.equals(childIdent.name(), parentIdent.name()),
+ "The domain %s cannot be its own parent",
+ childIdent);
+ }
+}
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 44346eedfd..000b564116 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
@@ -36,6 +36,7 @@ import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.DomainRelMapper;
import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper;
@@ -293,6 +294,10 @@ public class MetalakeMetaService {
SessionUtils.doWithoutCommit(
DomainMetaMapper.class,
mapper ->
mapper.softDeleteDomainMetasByMetalakeId(metalakeId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ DomainRelMapper.class,
+ mapper ->
mapper.softDeleteDomainRelsByMetalakeId(metalakeId)),
() ->
SessionUtils.doWithoutCommit(
TagMetadataObjectRelMapper.class,
@@ -403,6 +408,10 @@ public class MetalakeMetaService {
SessionUtils.doWithoutCommit(
DomainMetaMapper.class,
mapper ->
mapper.softDeleteDomainMetasByMetalakeId(metalakeId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ DomainRelMapper.class,
+ mapper ->
mapper.softDeleteDomainRelsByMetalakeId(metalakeId)),
() ->
SessionUtils.doWithoutCommit(
TagMetadataObjectRelMapper.class,
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestDomainPO.java
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestDomainPO.java
index b110212735..5e23801079 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestDomainPO.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestDomainPO.java
@@ -37,6 +37,20 @@ public class TestDomainPO {
private static final AuditInfo AUDIT_INFO =
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+ @Test
+ public void testDerivedNamesDoNotChangeEquality() throws Exception {
+ DomainPO plain = new DomainPO();
+ DomainPO joined = new DomainPO();
+ var parentName = DomainPO.class.getDeclaredField("parentName");
+ parentName.setAccessible(true);
+ parentName.set(joined, "derived name");
+ var backingTagName = DomainPO.class.getDeclaredField("backingTagName");
+ backingTagName.setAccessible(true);
+ backingTagName.set(joined, "derived name");
+ Assertions.assertEquals(plain, joined);
+ Assertions.assertEquals(plain.hashCode(), joined.hashCode());
+ }
+
@Test
public void testInitializeAndConvertDomainPO() {
DomainEntity domain = newDomain(10L, "Finance/Trading", "comment",
ImmutableMap.of("k", "v"));
@@ -86,6 +100,15 @@ public class TestDomainPO {
domain.toBuilder().withParent(null).withBackingTagName(null).build(),
converted);
}
+ @Test
+ public void testConvertDomainPOWithParent() {
+ DomainEntity domain = newDomain(10L, "Trading", null, null);
+ DomainEntity converted =
+ DomainPO.fromDomainPO(DomainPO.initializeDomainPO(domain, 1L),
NAMESPACE, "Finance");
+ Assertions.assertEquals("Finance", converted.parent().orElse(null));
+ Assertions.assertEquals(domain.toBuilder().withParent("Finance").build(),
converted);
+ }
+
@Test
public void testUpdateDomainPO() {
DomainPO oldDomainPO =
@@ -116,8 +139,10 @@ public class TestDomainPO {
Lists.newArrayList(
DomainPO.initializeDomainPO(finance, 1L),
DomainPO.initializeDomainPO(markets, 1L));
+ // Only sub-domains are in the parent map; the others are top-level.
Assertions.assertEquals(
- Lists.newArrayList(finance, markets),
DomainPO.fromDomainPOs(domainPOs, NAMESPACE));
+ Lists.newArrayList(finance,
markets.toBuilder().withParent("Finance").build()),
+ DomainPO.fromDomainPOs(domainPOs, NAMESPACE, ImmutableMap.of(11L,
"Finance")));
}
@Test
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestDomainRelPO.java
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestDomainRelPO.java
new file mode 100644
index 0000000000..f09ffe39b4
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestDomainRelPO.java
@@ -0,0 +1,112 @@
+/*
+ * 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.fasterxml.jackson.core.JsonProcessingException;
+import java.time.Instant;
+import org.apache.gravitino.json.JsonUtils;
+import org.apache.gravitino.meta.AuditInfo;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestDomainRelPO {
+
+ private static final Instant CREATE_TIME = Instant.ofEpochMilli(1000L);
+
+ private static final AuditInfo AUDIT_INFO =
+
AuditInfo.builder().withCreator("creator").withCreateTime(CREATE_TIME).build();
+
+ @Test
+ public void testDerivedNamesDoNotChangeEquality() throws Exception {
+ DomainRelPO plain = new DomainRelPO();
+ DomainRelPO joined = new DomainRelPO();
+ var parentName = DomainRelPO.class.getDeclaredField("parentName");
+ parentName.setAccessible(true);
+ parentName.set(joined, "derived name");
+ Assertions.assertEquals(plain, joined);
+ Assertions.assertEquals(plain.hashCode(), joined.hashCode());
+ }
+
+ @Test
+ public void testInitializeDomainRelPO() throws JsonProcessingException {
+ DomainRelPO domainRelPO = DomainRelPO.initializeDomainRelPO(1L, 10L, 11L,
AUDIT_INFO);
+
+ Assertions.assertEquals(1L, domainRelPO.metalakeId());
+ Assertions.assertEquals(10L, domainRelPO.parentId());
+ Assertions.assertEquals(11L, domainRelPO.childId());
+ Assertions.assertNull(domainRelPO.parentName());
+ Assertions.assertEquals(1L, domainRelPO.currentVersion());
+ Assertions.assertEquals(1L, domainRelPO.lastVersion());
+ Assertions.assertEquals(0L, domainRelPO.deletedAt());
+ Assertions.assertEquals(AUDIT_INFO, readAuditInfo(domainRelPO));
+ }
+
+ @Test
+ public void testUpdateDomainRelPOKeepsCreator() throws
JsonProcessingException {
+ DomainRelPO oldDomainRelPO = DomainRelPO.initializeDomainRelPO(1L, 10L,
11L, AUDIT_INFO);
+ Instant modifiedTime = Instant.ofEpochMilli(2000L);
+ AuditInfo modifiedAuditInfo =
+ AuditInfo.builder()
+ .withCreator("ignored")
+ .withCreateTime(Instant.ofEpochMilli(3000L))
+ .withLastModifier("modifier")
+ .withLastModifiedTime(modifiedTime)
+ .build();
+
+ DomainRelPO newDomainRelPO =
+ DomainRelPO.updateDomainRelPO(oldDomainRelPO, 12L, modifiedAuditInfo);
+ Assertions.assertEquals(1L, newDomainRelPO.metalakeId());
+ Assertions.assertEquals(12L, newDomainRelPO.parentId());
+ Assertions.assertEquals(11L, newDomainRelPO.childId());
+ Assertions.assertEquals(2L, newDomainRelPO.currentVersion());
+ Assertions.assertEquals(2L, newDomainRelPO.lastVersion());
+ Assertions.assertEquals(0L, newDomainRelPO.deletedAt());
+
+ AuditInfo storedAuditInfo = readAuditInfo(newDomainRelPO);
+ Assertions.assertEquals("creator", storedAuditInfo.creator());
+ Assertions.assertEquals(CREATE_TIME, storedAuditInfo.createTime());
+ Assertions.assertEquals("modifier", storedAuditInfo.lastModifier());
+ Assertions.assertEquals(modifiedTime, storedAuditInfo.lastModifiedTime());
+ }
+
+ @Test
+ public void testDomainRelPOValidation() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> DomainRelPO.initializeDomainRelPO(1L, 10L, 10L, AUDIT_INFO));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> DomainRelPO.initializeDomainRelPO(null, 10L, 11L, AUDIT_INFO));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> DomainRelPO.initializeDomainRelPO(1L, null, 11L, AUDIT_INFO));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> DomainRelPO.initializeDomainRelPO(1L, 10L, null, AUDIT_INFO));
+
+ DomainRelPO domainRelPO = DomainRelPO.initializeDomainRelPO(1L, 10L, 11L,
AUDIT_INFO);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> DomainRelPO.updateDomainRelPO(domainRelPO, 11L, AUDIT_INFO));
+ }
+
+ private AuditInfo readAuditInfo(DomainRelPO domainRelPO) throws
JsonProcessingException {
+ return JsonUtils.anyFieldMapper().readValue(domainRelPO.auditInfo(),
AuditInfo.class);
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainRelService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainRelService.java
new file mode 100644
index 0000000000..dd53110914
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainRelService.java
@@ -0,0 +1,921 @@
+/*
+ * 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.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.domain.Domain;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.json.JsonUtils;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.DomainEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.DomainRelMapper;
+import org.apache.gravitino.storage.relational.po.DomainRelPO;
+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.exceptions.PersistenceException;
+import org.apache.ibatis.executor.Executor;
+import org.apache.ibatis.mapping.MappedStatement;
+import org.apache.ibatis.plugin.Interceptor;
+import org.apache.ibatis.plugin.Intercepts;
+import org.apache.ibatis.plugin.Invocation;
+import org.apache.ibatis.plugin.Signature;
+import org.apache.ibatis.session.ResultHandler;
+import org.apache.ibatis.session.RowBounds;
+import org.apache.ibatis.session.SqlSession;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+
+public class TestDomainRelService extends TestJDBCBackend {
+
+ private static final String METALAKE_NAME =
"metalake_for_domain_rel_service_test";
+
+ private final DomainMetaService domainMetaService =
DomainMetaService.getInstance();
+
+ private final DomainRelService domainRelService =
DomainRelService.getInstance();
+
+ @TestTemplate
+ public void testInsertDomainRelAndResolveParent() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ DomainEntity tax = insertDomain(METALAKE_NAME, "Tax");
+ DomainEntity markets = insertDomain(METALAKE_NAME, "Markets");
+
+
Assertions.assertFalse(domainRelService.getParentDomainName(trading.id()).isPresent());
+
Assertions.assertTrue(domainRelService.listSubDomains(finance.nameIdentifier()).isEmpty());
+
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+ domainRelService.insertDomainRel(tax.nameIdentifier(),
finance.nameIdentifier(), AUDIT_INFO);
+
+ Assertions.assertEquals("Finance",
domainRelService.getParentDomainName(trading.id()).get());
+
Assertions.assertFalse(domainRelService.getParentDomainName(finance.id()).isPresent());
+ Assertions.assertEquals(
+ ImmutableMap.of(trading.id(), "Finance", tax.id(), "Finance"),
+ domainRelService.batchGetParentDomainNames(
+ Lists.newArrayList(finance.id(), trading.id(), tax.id(),
markets.id())));
+ Assertions.assertTrue(
+
domainRelService.batchGetParentDomainNames(Lists.newArrayList()).isEmpty());
+
+ DomainEntity tradingWithParent =
trading.toBuilder().withParent("Finance").build();
+ DomainEntity taxWithParent = tax.toBuilder().withParent("Finance").build();
+ List<DomainEntity> subDomains =
domainRelService.listSubDomains(finance.nameIdentifier());
+ Assertions.assertEquals(2, subDomains.size());
+ Assertions.assertTrue(subDomains.contains(tradingWithParent));
+ Assertions.assertTrue(subDomains.contains(taxWithParent));
+
Assertions.assertTrue(domainRelService.listSubDomains(trading.nameIdentifier()).isEmpty());
+
+ DomainRelPO domainRelPO = selectDomainRel(trading.id());
+ Assertions.assertEquals(finance.id(), domainRelPO.parentId());
+ Assertions.assertEquals(selectMetalakeId(trading),
domainRelPO.metalakeId());
+ Assertions.assertEquals(1L, domainRelPO.currentVersion());
+ Assertions.assertEquals(
+ AUDIT_INFO,
JsonUtils.anyFieldMapper().readValue(domainRelPO.auditInfo(), AuditInfo.class));
+ }
+
+ @TestTemplate
+ public void testDomainReadsFillParent() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+ DomainEntity tradingWithParent =
trading.toBuilder().withParent("Finance").build();
+
+ Assertions.assertEquals(
+ tradingWithParent,
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()));
+ Assertions.assertEquals(
+ tradingWithParent, backend.get(trading.nameIdentifier(),
Entity.EntityType.DOMAIN));
+ Assertions.assertEquals(
+ finance,
domainMetaService.getDomainByIdentifier(finance.nameIdentifier()));
+
+ List<DomainEntity> domains =
+
domainMetaService.listDomainsByNamespace(NamespaceUtil.ofDomain(METALAKE_NAME));
+ Assertions.assertEquals(2, domains.size());
+ Assertions.assertTrue(domains.contains(finance));
+ Assertions.assertTrue(domains.contains(tradingWithParent));
+ Assertions.assertEquals(
+ Lists.newArrayList(tradingWithParent),
+
domainMetaService.batchGetDomainByIdentifier(Lists.newArrayList(trading.nameIdentifier())));
+
+ // An update keeps the parent, which only the relation changes.
+ DomainEntity updated =
+ domainMetaService.updateDomain(
+ trading.nameIdentifier(),
+ entity -> ((DomainEntity)
entity).toBuilder().withComment("updated").build());
+ Assertions.assertEquals("Finance", updated.parent().orElse(null));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainMetaService.updateDomain(
+ trading.nameIdentifier(),
+ entity -> ((DomainEntity)
entity).toBuilder().withParent(null).build()));
+ Assertions.assertEquals(
+ "Finance",
+
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()).parent().orElse(null));
+
+ // The edge refers to the parent by ID, so a renamed parent is resolved by
its new name.
+ domainMetaService.updateDomain(
+ finance.nameIdentifier(),
+ entity -> ((DomainEntity)
entity).toBuilder().withName("Finance/Global").build());
+ Assertions.assertEquals(
+ "Finance/Global",
+
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()).parent().orElse(null));
+ }
+
+ @TestTemplate
+ public void testInsertDomainRelIsRejected() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ String anotherMetalakeName = METALAKE_NAME + "_another";
+ createAndInsertMakeLake(anotherMetalakeName);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ DomainEntity markets = insertDomain(METALAKE_NAME, "Markets");
+ DomainEntity elsewhere = insertDomain(anotherMetalakeName, "Elsewhere");
+ NameIdentifier missing = NameIdentifierUtil.ofDomain(METALAKE_NAME,
"missing");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.insertDomainRel(
+ finance.nameIdentifier(), finance.nameIdentifier(),
AUDIT_INFO));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.insertDomainRel(
+ elsewhere.nameIdentifier(), finance.nameIdentifier(),
AUDIT_INFO));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () -> domainRelService.insertDomainRel(missing,
finance.nameIdentifier(), AUDIT_INFO));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () -> domainRelService.insertDomainRel(trading.nameIdentifier(),
missing, AUDIT_INFO));
+ Assertions.assertThrows(
+ NoSuchEntityException.class, () ->
domainRelService.listSubDomains(missing));
+
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+ // A domain has at most one parent; moving it is a re-point.
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class,
+ () ->
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), markets.nameIdentifier(),
AUDIT_INFO));
+ Assertions.assertEquals(finance.id(),
selectDomainRel(trading.id()).parentId());
+
+ // The unique key keeps one active edge per domain even without the check.
+ DomainRelPO secondEdge =
+ DomainRelPO.initializeDomainRelPO(
+ selectMetalakeId(trading), markets.id(), trading.id(), AUDIT_INFO);
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ SessionUtils.doWithCommit(
+ DomainRelMapper.class, mapper ->
mapper.insertDomainRel(secondEdge)));
+ }
+
+ @TestTemplate
+ public void testUpdateDomainRel() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity markets = insertDomain(METALAKE_NAME, "Markets");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ domainRelService.updateDomainRel(
+ trading.nameIdentifier(), markets.nameIdentifier(),
AUDIT_INFO));
+
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+ Instant modifiedTime = Instant.now();
+ AuditInfo modifiedAuditInfo =
+
AuditInfo.builder().withLastModifier("mover").withLastModifiedTime(modifiedTime).build();
+ domainRelService.updateDomainRel(
+ trading.nameIdentifier(), markets.nameIdentifier(), modifiedAuditInfo);
+
+ Assertions.assertEquals("Markets",
domainRelService.getParentDomainName(trading.id()).get());
+
Assertions.assertTrue(domainRelService.listSubDomains(finance.nameIdentifier()).isEmpty());
+ Assertions.assertEquals(
+ Lists.newArrayList(trading.toBuilder().withParent("Markets").build()),
+ domainRelService.listSubDomains(markets.nameIdentifier()));
+
+ DomainRelPO domainRelPO = selectDomainRel(trading.id());
+ Assertions.assertEquals(2L, domainRelPO.currentVersion());
+ AuditInfo storedAuditInfo =
+ JsonUtils.anyFieldMapper().readValue(domainRelPO.auditInfo(),
AuditInfo.class);
+ Assertions.assertEquals(AUDIT_INFO.creator(), storedAuditInfo.creator());
+ Assertions.assertEquals("mover", storedAuditInfo.lastModifier());
+ Assertions.assertEquals(modifiedTime, storedAuditInfo.lastModifiedTime());
+ Assertions.assertEquals(1, countDomainRelRows(trading.id()));
+
+ // Re-pointing to the current parent changes nothing.
+ domainRelService.updateDomainRel(
+ trading.nameIdentifier(), markets.nameIdentifier(), modifiedAuditInfo);
+ Assertions.assertEquals(2L,
selectDomainRel(trading.id()).currentVersion());
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.updateDomainRel(
+ trading.nameIdentifier(), trading.nameIdentifier(),
AUDIT_INFO));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ domainRelService.updateDomainRel(
+ trading.nameIdentifier(),
+ NameIdentifierUtil.ofDomain(METALAKE_NAME, "missing"),
+ AUDIT_INFO));
+ }
+
+ @TestTemplate
+ public void testDeleteDomainRel() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+
+
Assertions.assertFalse(domainRelService.deleteDomainRel(trading.nameIdentifier()));
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+
Assertions.assertTrue(domainRelService.deleteDomainRel(trading.nameIdentifier()));
+
Assertions.assertFalse(domainRelService.deleteDomainRel(trading.nameIdentifier()));
+
Assertions.assertFalse(domainRelService.getParentDomainName(trading.id()).isPresent());
+
Assertions.assertTrue(domainRelService.listSubDomains(finance.nameIdentifier()).isEmpty());
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ domainRelService.deleteDomainRel(
+ NameIdentifierUtil.ofDomain(METALAKE_NAME, "missing")));
+
+ // A detached domain can be attached again.
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+ Assertions.assertEquals("Finance",
domainRelService.getParentDomainName(trading.id()).get());
+ }
+
+ @TestTemplate
+ public void testDeleteDomainWithSubDomains() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+
+ Assertions.assertThrows(
+ NonEmptyEntityException.class,
+ () -> domainMetaService.deleteDomain(finance.nameIdentifier()));
+ Assertions.assertThrows(
+ NonEmptyEntityException.class,
+ () -> backend.delete(finance.nameIdentifier(),
Entity.EntityType.DOMAIN, true));
+ Assertions.assertTrue(backend.exists(finance.nameIdentifier(),
Entity.EntityType.DOMAIN));
+ Assertions.assertEquals("Finance",
domainRelService.getParentDomainName(trading.id()).get());
+
+ // Deleting the sub-domain removes its own edge, after which the parent
can be deleted.
+
Assertions.assertTrue(domainMetaService.deleteDomain(trading.nameIdentifier()));
+ Assertions.assertNull(selectDomainRel(trading.id()));
+ Assertions.assertEquals(1, countDomainRelRows(trading.id()));
+
Assertions.assertTrue(domainMetaService.deleteDomain(finance.nameIdentifier()));
+ }
+
+ @TestTemplate
+ public void testDomainRelWithMetalakeDeleteAndGarbageCollection() throws
IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ String anotherMetalakeName = METALAKE_NAME + "_another";
+ createAndInsertMakeLake(anotherMetalakeName);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ DomainEntity anotherFinance = insertDomain(anotherMetalakeName, "Finance");
+ DomainEntity anotherTrading = insertDomain(anotherMetalakeName, "Trading");
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+ domainRelService.insertDomainRel(
+ anotherTrading.nameIdentifier(), anotherFinance.nameIdentifier(),
AUDIT_INFO);
+
+ Assertions.assertTrue(
+
MetalakeMetaService.getInstance().deleteMetalake(metalake.nameIdentifier(),
true));
+ Assertions.assertNull(selectDomainRel(trading.id()));
+ Assertions.assertEquals(1, countDomainRelRows(trading.id()));
+ Assertions.assertEquals(
+ "Finance",
domainRelService.getParentDomainName(anotherTrading.id()).get());
+
+ Assertions.assertTrue(
+ backend.hardDeleteLegacyData(Entity.EntityType.DOMAIN,
Instant.now().toEpochMilli() + 1000)
+ >= 3);
+ Assertions.assertEquals(0, countDomainRelRows(trading.id()));
+ Assertions.assertEquals(1, countDomainRelRows(anotherTrading.id()));
+ Assertions.assertEquals(
+ "Finance",
domainRelService.getParentDomainName(anotherTrading.id()).get());
+ }
+
+ @TestTemplate
+ public void testDomainRelWithNonCascadeMetalakeDelete() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ domainRelService.insertDomainRel(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
+
+ Assertions.assertTrue(
+
MetalakeMetaService.getInstance().deleteMetalake(metalake.nameIdentifier(),
false));
+ Assertions.assertNull(selectDomainRel(trading.id()));
+ Assertions.assertEquals(1, countDomainRelRows(trading.id()));
+ }
+
+ @TestTemplate
+ public void testDeleteParentWaitsForConcurrentAttach() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+
+ CountDownLatch attachWritten = new CountDownLatch(1);
+ CountDownLatch allowAttachCommit = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> attachResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> insertDomainRel(trading, finance),
+ () -> {
+ attachWritten.countDown();
+ await(allowAttachCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ Assertions.assertTrue(attachWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ domainMetaService.deleteDomain(finance.nameIdentifier());
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ // The attach holds the parent row, so the delete waits for it instead
of missing the edge.
+ Assertions.assertThrows(
+ TimeoutException.class, () -> deleteResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowAttachCommit.countDown();
+ Assertions.assertNull(attachResult.get(30, TimeUnit.SECONDS));
+ Throwable deleteFailure = deleteResult.get(30, TimeUnit.SECONDS);
+ Assertions.assertTrue(
+ deleteFailure instanceof NonEmptyEntityException,
String.valueOf(deleteFailure));
+ } finally {
+ allowAttachCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ Assertions.assertTrue(backend.exists(finance.nameIdentifier(),
Entity.EntityType.DOMAIN));
+ Assertions.assertEquals("Finance",
domainRelService.getParentDomainName(trading.id()).get());
+ }
+
+ @TestTemplate
+ public void testAttachWaitsForConcurrentParentDelete() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+
+ CountDownLatch deleteWritten = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ Assertions.assertTrue(
+
domainMetaService.deleteDomain(finance.nameIdentifier())),
+ () -> {
+ deleteWritten.countDown();
+ await(allowDeleteCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ Assertions.assertTrue(deleteWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> attachResult =
+ executor.submit(
+ () -> {
+ try {
+ insertDomainRel(trading, finance);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ // The delete holds the parent row, so the attach waits and then sees
the parent is gone.
+ Assertions.assertThrows(
+ TimeoutException.class, () -> attachResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowDeleteCommit.countDown();
+ Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS));
+ Throwable attachFailure = attachResult.get(30, TimeUnit.SECONDS);
+ Assertions.assertTrue(
+ attachFailure instanceof NoSuchEntityException,
String.valueOf(attachFailure));
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ Assertions.assertFalse(backend.exists(finance.nameIdentifier(),
Entity.EntityType.DOMAIN));
+ Assertions.assertNull(selectDomainRel(trading.id()));
+ Assertions.assertEquals(0, countDomainRelRows(trading.id()));
+ }
+
+ @TestTemplate
+ public void testCaughtNonEmptyDeleteDoesNotMutateParent() throws IOException
{
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity parent = insertDomain(METALAKE_NAME, "Parent");
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ insertDomainRel(child, parent);
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ Assertions.assertThrows(
+ NonEmptyEntityException.class,
+ () -> domainMetaService.deleteDomain(parent.nameIdentifier())),
+ () -> {
+ try {
+ domainMetaService.updateDomain(
+ child.nameIdentifier(),
+ entity -> ((DomainEntity)
entity).toBuilder().withComment("changed").build());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ Assertions.assertTrue(backend.exists(parent.nameIdentifier(),
Entity.EntityType.DOMAIN));
+ Assertions.assertEquals(
+ "Parent",
domainRelService.getParentDomainName(child.id()).orElse(null));
+ Assertions.assertEquals(
+ "changed",
domainMetaService.getDomainByIdentifier(child.nameIdentifier()).comment());
+ }
+
+ @TestTemplate
+ public void testRepeatedDetachWithinTransaction() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity parent = insertDomain(METALAKE_NAME, "Parent");
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ try {
+ for (int i = 0; i < 3; i++) {
+ insertDomainRel(child, parent);
+
Assertions.assertTrue(domainRelService.deleteDomainRel(child.nameIdentifier()));
+ }
+ insertDomainRel(child, parent);
+
Assertions.assertTrue(domainMetaService.deleteDomain(child.nameIdentifier()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ Assertions.assertEquals(4, countDomainRelRows(child.id()));
+ Assertions.assertNull(selectDomainRel(child.id()));
+ }
+
+ @TestTemplate
+ public void testLockedUpdateReadsPastCachedParent() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity first = insertDomain(METALAKE_NAME, "First");
+ DomainEntity second = insertDomain(METALAKE_NAME, "Second");
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ insertDomainRel(child, first);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ SessionUtils.beginTransaction();
+ try {
+ Assertions.assertEquals(
+ "First",
+
domainMetaService.getDomainByIdentifier(child.nameIdentifier()).parent().orElse(null));
+ Assertions.assertEquals(
+ "First",
domainRelService.getParentDomainName(child.id()).orElse(null));
+ executor
+ .submit(
+ () -> {
+ try {
+ domainRelService.updateDomainRel(
+ child.nameIdentifier(), second.nameIdentifier(),
AUDIT_INFO);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .get(30, TimeUnit.SECONDS);
+ Assertions.assertEquals(
+ "Second",
domainRelService.getParentDomainName(child.id()).orElse(null));
+ domainMetaService.updateDomain(
+ child.nameIdentifier(),
+ entity -> {
+ DomainEntity domain = (DomainEntity) entity;
+ Assertions.assertEquals("Second", domain.parent().orElse(null));
+ return entity;
+ });
+ SessionUtils.commitTransaction();
+ } finally {
+ if (SessionUtils.isInTransaction()) {
+ SessionUtils.rollbackTransaction();
+ }
+ executor.shutdownNow();
+ }
+ }
+
+ @TestTemplate
+ public void testResponseReadFailureRollsBackUpdate() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ installReadProbe();
+ AtomicBoolean written = new AtomicBoolean();
+ ReadProbe.AFTER_WRITE.set(
+ id -> {
+ if (id.endsWith(".updateDomainMeta")) {
+ written.set(true);
+ }
+ });
+ ReadProbe.BEFORE_READ.set(
+ id -> {
+ if (written.get()) {
+ throw new IllegalStateException("response read failed");
+ }
+ });
+ try {
+ Assertions.assertThrows(
+ PersistenceException.class,
+ () ->
+ domainMetaService.updateDomain(
+ child.nameIdentifier(),
+ entity ->
+ ((DomainEntity) entity).toBuilder().withComment("must
roll back").build()));
+ } finally {
+ ReadProbe.clear();
+ }
+ Assertions.assertTrue(written.get());
+ Assertions.assertNull(
+
domainMetaService.getDomainByIdentifier(child.nameIdentifier()).comment());
+ }
+
+ @TestTemplate
+ public void testReadPathsKeepMetadataAndParentTogether() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity first = insertDomain(METALAKE_NAME, "First");
+ DomainEntity second = insertDomain(METALAKE_NAME, "Second");
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ insertDomainRel(child, first);
+ installReadProbe();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ for (String path : new String[] {"get", "list", "batch", "subdomains"}) {
+ domainRelService.updateDomainRel(
+ child.nameIdentifier(), first.nameIdentifier(), AUDIT_INFO);
+ domainMetaService.updateDomain(first.nameIdentifier(), entity ->
entity);
+ domainMetaService.updateDomain(
+ child.nameIdentifier(),
+ entity -> ((DomainEntity)
entity).toBuilder().withComment("old").build());
+ AtomicBoolean triggered = new AtomicBoolean();
+ String method =
+ path.equals("get")
+ ? "selectDomainMetaByMetalakeAndName"
+ : path.equals("list")
+ ? "listDomainPOsByMetalake"
+ : path.equals("batch")
+ ? "listDomainPOsByMetalakeAndDomainNames"
+ : "listDomainPOsByParentIdentifier";
+ ReadProbe.AFTER_READ.set(
+ id -> {
+ if (id.endsWith("." + method) && triggered.compareAndSet(false,
true)) {
+ try {
+ executor
+ .submit(
+ () ->
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ try {
+ domainMetaService.updateDomain(
+ child.nameIdentifier(),
+ entity ->
+ ((DomainEntity) entity)
+
.toBuilder().withComment("new").build());
+ domainRelService.updateDomainRel(
+ child.nameIdentifier(),
+ second.nameIdentifier(),
+ AUDIT_INFO);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }))
+ .get(30, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ DomainEntity read;
+ try {
+ if (path.equals("get")) {
+ read =
domainMetaService.getDomainByIdentifier(child.nameIdentifier());
+ } else {
+ List<DomainEntity> domains =
+ path.equals("list")
+ ?
domainMetaService.listDomainsByNamespace(child.namespace())
+ : path.equals("batch")
+ ? domainMetaService.batchGetDomainByIdentifier(
+ Collections.singletonList(child.nameIdentifier()))
+ :
domainRelService.listSubDomains(first.nameIdentifier());
+ read = domains.stream().filter(d ->
d.id().equals(child.id())).findFirst().get();
+ }
+ } finally {
+ ReadProbe.clear();
+ }
+ Assertions.assertTrue(triggered.get(), path);
+ Assertions.assertEquals("old", read.comment(), path);
+ Assertions.assertEquals("First", read.parent().orElse(null), path);
+ }
+ } finally {
+ ReadProbe.clear();
+ executor.shutdownNow();
+ }
+ }
+
+ @TestTemplate
+ public void testSubDomainReadUsesCurrentParentName() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity parent = insertDomain(METALAKE_NAME, "Parent");
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ insertDomainRel(child, parent);
+ installReadProbe();
+ AtomicBoolean triggered = new AtomicBoolean();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ ReadProbe.AFTER_READ.set(
+ id -> {
+ if (id.endsWith(".listDomainPOsByParentIdentifier")
+ && triggered.compareAndSet(false, true)) {
+ try {
+ executor
+ .submit(
+ () ->
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ try {
+ domainMetaService.updateDomain(
+ parent.nameIdentifier(),
+ entity ->
+ ((DomainEntity) entity)
+
.toBuilder().withName("Renamed").build());
+ domainMetaService.updateDomain(
+ child.nameIdentifier(),
+ entity ->
+ ((DomainEntity) entity)
+
.toBuilder().withComment("new").build());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }))
+ .get(30, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ try {
+ DomainEntity read =
domainRelService.listSubDomains(parent.nameIdentifier()).get(0);
+ Assertions.assertTrue(triggered.get());
+ Assertions.assertNull(read.comment());
+ Assertions.assertEquals("Parent", read.parent().orElse(null));
+ } finally {
+ ReadProbe.clear();
+ executor.shutdownNow();
+ }
+ }
+
+ @TestTemplate
+ public void testRepeatedDetachThenMetalakeDelete() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity parent = insertDomain(METALAKE_NAME, "Parent");
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ insertDomainRel(child, parent);
+ try {
+
Assertions.assertTrue(domainRelService.deleteDomainRel(child.nameIdentifier()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ insertDomainRel(child, parent);
+ Assertions.assertTrue(
+
MetalakeMetaService.getInstance().deleteMetalake(metalake.nameIdentifier(),
false));
+ });
+ Assertions.assertEquals(2, countDomainRelRows(child.id()));
+ Assertions.assertNull(selectDomainRel(child.id()));
+ }
+
+ @TestTemplate
+ public void testSubDomainsResolveRecreatedParentInOneStatement() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity parent = insertDomain(METALAKE_NAME, "Parent");
+ DomainEntity child = insertDomain(METALAKE_NAME, "Child");
+ insertDomainRel(child, parent);
+ installReadProbe();
+ AtomicBoolean triggered = new AtomicBoolean();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ ReadProbe.BEFORE_READ.set(
+ id -> {
+ // Replace the parent after any preliminary lookup but before the
children query.
+ // A two-statement implementation would keep the original parent's
ID.
+ if (id.endsWith(".listDomainPOsByParentIdentifier")
+ && triggered.compareAndSet(false, true)) {
+ try {
+ executor
+ .submit(
+ () ->
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ Assertions.assertTrue(
+
domainMetaService.deleteDomain(child.nameIdentifier()));
+ Assertions.assertTrue(
+
domainMetaService.deleteDomain(parent.nameIdentifier()));
+ try {
+ DomainEntity replacement =
insertDomain(METALAKE_NAME, "Parent");
+ DomainEntity replacementChild =
+ insertDomain(METALAKE_NAME,
"ReplacementChild");
+ insertDomainRel(replacementChild,
replacement);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }))
+ .get(30, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ try {
+ List<DomainEntity> children =
domainRelService.listSubDomains(parent.nameIdentifier());
+ Assertions.assertTrue(triggered.get());
+ Assertions.assertEquals(1, children.size());
+ Assertions.assertEquals("ReplacementChild", children.get(0).name());
+ Assertions.assertEquals("Parent", children.get(0).parent().orElse(null));
+ } finally {
+ ReadProbe.clear();
+ executor.shutdownNow();
+ }
+ }
+
+ private void installReadProbe() {
+ var configuration =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().getConfiguration();
+ if (configuration.getInterceptors().stream().noneMatch(i -> i instanceof
ReadProbe)) {
+ configuration.addInterceptor(new ReadProbe());
+ }
+ }
+
+ /** Injects controlled interleavings and failures without changing
production mapper code. */
+ @Intercepts({
+ @Signature(
+ type = Executor.class,
+ method = "query",
+ args = {MappedStatement.class, Object.class, RowBounds.class,
ResultHandler.class}),
+ @Signature(
+ type = Executor.class,
+ method = "update",
+ args = {MappedStatement.class, Object.class})
+ })
+ public static class ReadProbe implements Interceptor {
+ private static final ThreadLocal<Consumer<String>> BEFORE_READ = new
ThreadLocal<>();
+ private static final ThreadLocal<Consumer<String>> AFTER_READ = new
ThreadLocal<>();
+ private static final ThreadLocal<Consumer<String>> AFTER_WRITE = new
ThreadLocal<>();
+
+ /** Creates an inactive probe. */
+ public ReadProbe() {}
+
+ /** {@inheritDoc} */
+ @Override
+ public Object intercept(Invocation invocation) throws Throwable {
+ String id = ((MappedStatement) invocation.getArgs()[0]).getId();
+ boolean read = invocation.getMethod().getName().equals("query");
+ if (read && BEFORE_READ.get() != null) {
+ BEFORE_READ.get().accept(id);
+ }
+ Object result = invocation.proceed();
+ Consumer<String> after = read ? AFTER_READ.get() : AFTER_WRITE.get();
+ if (after != null) {
+ after.accept(id);
+ }
+ return result;
+ }
+
+ private static void clear() {
+ BEFORE_READ.remove();
+ AFTER_READ.remove();
+ AFTER_WRITE.remove();
+ }
+ }
+
+ private DomainEntity insertDomain(String metalake, String name) throws
IOException {
+ Map<String, String> props = ImmutableMap.of("k1", "v1");
+ DomainEntity domain =
+ DomainEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(NamespaceUtil.ofDomain(metalake))
+ .withProperties(props)
+ .withStatus(Domain.Status.DRAFT)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ domainMetaService.insertDomain(domain, false);
+ return domain;
+ }
+
+ private void insertDomainRel(DomainEntity child, DomainEntity parent) {
+ try {
+ domainRelService.insertDomainRel(child.nameIdentifier(),
parent.nameIdentifier(), AUDIT_INFO);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ Assertions.assertTrue(latch.await(30, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private DomainRelPO selectDomainRel(long childId) {
+ return SessionUtils.getWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.selectDomainRelByChildId(childId));
+ }
+
+ private long selectMetalakeId(DomainEntity domain) {
+ return
MetalakeMetaService.getInstance().getMetalakeIdByName(domain.namespace().level(0));
+ }
+
+ private int countDomainRelRows(long childId) {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet rs =
+ statement.executeQuery(
+ String.format("SELECT count(*) FROM domain_rel WHERE child_id
= %d", childId))) {
+ if (rs.next()) {
+ return rs.getInt(1);
+ }
+ throw new RuntimeException("Doesn't contain data");
+ } catch (SQLException se) {
+ throw new RuntimeException("SQL execution failed", se);
+ }
+ }
+}
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index 7c943490fa..553862039f 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -783,3 +783,18 @@ CREATE TABLE IF NOT EXISTS `domain_meta` (
UNIQUE KEY `uk_mi_dn_del` (`metalake_id`, `domain_name`, `deleted_at`),
KEY `domain_meta_idx_backing_tag` (`metalake_id`, `backing_tag_id`,
`backing_tag_value`, `deleted_at`)
) ENGINE=InnoDB;
+
+CREATE TABLE IF NOT EXISTS `domain_rel` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `parent_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'parent domain id',
+ `child_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'sub-domain id',
+ `audit_info` CLOB NOT NULL COMMENT 'domain relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'domain
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_dr_ci_del` (`child_id`, `deleted_at`),
+ KEY `domain_rel_idx_parent` (`parent_id`, `deleted_at`),
+ KEY `domain_rel_idx_metalake` (`metalake_id`, `deleted_at`)
+) ENGINE=InnoDB;
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 ec4f85d1a5..72c2df350b 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
@@ -229,3 +229,18 @@ CREATE TABLE IF NOT EXISTS `domain_meta` (
UNIQUE KEY `uk_mi_dn_del` (`metalake_id`, `domain_name`, `deleted_at`),
KEY `domain_meta_idx_backing_tag` (`metalake_id`, `backing_tag_id`,
`backing_tag_value`, `deleted_at`)
) ENGINE=InnoDB;
+
+CREATE TABLE IF NOT EXISTS `domain_rel` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `parent_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'parent domain id',
+ `child_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'sub-domain id',
+ `audit_info` CLOB NOT NULL COMMENT 'domain relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'domain
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_dr_ci_del` (`child_id`, `deleted_at`),
+ KEY `domain_rel_idx_parent` (`parent_id`, `deleted_at`),
+ KEY `domain_rel_idx_metalake` (`metalake_id`, `deleted_at`)
+) ENGINE=InnoDB;
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index 9fc8275771..8bd3766d31 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -770,3 +770,18 @@ CREATE TABLE IF NOT EXISTS `domain_meta` (
UNIQUE KEY `uk_mi_dn_del` (`metalake_id`, `domain_name`, `deleted_at`),
KEY `domain_meta_idx_backing_tag` (`metalake_id`, `backing_tag_id`,
`backing_tag_value`, `deleted_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'domain
metadata';
+
+CREATE TABLE IF NOT EXISTS `domain_rel` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `parent_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'parent domain id',
+ `child_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'sub-domain id',
+ `audit_info` MEDIUMTEXT NOT NULL COMMENT 'domain relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'domain
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_dr_ci_del` (`child_id`, `deleted_at`),
+ KEY `domain_rel_idx_parent` (`parent_id`, `deleted_at`),
+ KEY `domain_rel_idx_metalake` (`metalake_id`, `deleted_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'domain
parent relation';
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 bdf4290e0e..8ff0f66857 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
@@ -688,3 +688,18 @@ CREATE TABLE IF NOT EXISTS `domain_meta` (
UNIQUE KEY `uk_mi_dn_del` (`metalake_id`, `domain_name`, `deleted_at`),
KEY `domain_meta_idx_backing_tag` (`metalake_id`, `backing_tag_id`,
`backing_tag_value`, `deleted_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'domain
metadata';
+
+CREATE TABLE IF NOT EXISTS `domain_rel` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `parent_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'parent domain id',
+ `child_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'sub-domain id',
+ `audit_info` MEDIUMTEXT NOT NULL COMMENT 'domain relation audit info',
+ `current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
current version',
+ `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'domain relation
last version',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'domain
relation deleted at',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_dr_ci_del` (`child_id`, `deleted_at`),
+ KEY `domain_rel_idx_parent` (`parent_id`, `deleted_at`),
+ KEY `domain_rel_idx_metalake` (`metalake_id`, `deleted_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'domain
parent relation';
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index 31797b3540..ed53d5e69e 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -1345,3 +1345,27 @@ COMMENT ON COLUMN domain_meta.audit_info IS 'domain
audit info';
COMMENT ON COLUMN domain_meta.current_version IS 'domain current version';
COMMENT ON COLUMN domain_meta.last_version IS 'domain last version';
COMMENT ON COLUMN domain_meta.deleted_at IS 'domain deleted at';
+
+CREATE TABLE IF NOT EXISTS domain_rel (
+ id BIGINT GENERATED BY DEFAULT AS IDENTITY,
+ metalake_id BIGINT NOT NULL,
+ parent_id BIGINT NOT NULL,
+ child_id BIGINT NOT 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),
+ CONSTRAINT uk_dr_ci_del UNIQUE (child_id, deleted_at)
+);
+CREATE INDEX IF NOT EXISTS domain_rel_idx_parent ON domain_rel (parent_id,
deleted_at);
+CREATE INDEX IF NOT EXISTS domain_rel_idx_metalake ON domain_rel (metalake_id,
deleted_at);
+COMMENT ON TABLE domain_rel IS 'domain parent relation';
+COMMENT ON COLUMN domain_rel.id IS 'auto increment id';
+COMMENT ON COLUMN domain_rel.metalake_id IS 'metalake id';
+COMMENT ON COLUMN domain_rel.parent_id IS 'parent domain id';
+COMMENT ON COLUMN domain_rel.child_id IS 'sub-domain id';
+COMMENT ON COLUMN domain_rel.audit_info IS 'domain relation audit info';
+COMMENT ON COLUMN domain_rel.current_version IS 'domain relation current
version';
+COMMENT ON COLUMN domain_rel.last_version IS 'domain relation last version';
+COMMENT ON COLUMN domain_rel.deleted_at IS 'domain relation deleted at';
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 9b9ba62b8d..ce63038efa 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
@@ -345,3 +345,27 @@ COMMENT ON COLUMN domain_meta.audit_info IS 'domain audit
info';
COMMENT ON COLUMN domain_meta.current_version IS 'domain current version';
COMMENT ON COLUMN domain_meta.last_version IS 'domain last version';
COMMENT ON COLUMN domain_meta.deleted_at IS 'domain deleted at';
+
+CREATE TABLE IF NOT EXISTS domain_rel (
+ id BIGINT GENERATED BY DEFAULT AS IDENTITY,
+ metalake_id BIGINT NOT NULL,
+ parent_id BIGINT NOT NULL,
+ child_id BIGINT NOT 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),
+ CONSTRAINT uk_dr_ci_del UNIQUE (child_id, deleted_at)
+);
+CREATE INDEX IF NOT EXISTS domain_rel_idx_parent ON domain_rel (parent_id,
deleted_at);
+CREATE INDEX IF NOT EXISTS domain_rel_idx_metalake ON domain_rel (metalake_id,
deleted_at);
+COMMENT ON TABLE domain_rel IS 'domain parent relation';
+COMMENT ON COLUMN domain_rel.id IS 'auto increment id';
+COMMENT ON COLUMN domain_rel.metalake_id IS 'metalake id';
+COMMENT ON COLUMN domain_rel.parent_id IS 'parent domain id';
+COMMENT ON COLUMN domain_rel.child_id IS 'sub-domain id';
+COMMENT ON COLUMN domain_rel.audit_info IS 'domain relation audit info';
+COMMENT ON COLUMN domain_rel.current_version IS 'domain relation current
version';
+COMMENT ON COLUMN domain_rel.last_version IS 'domain relation last version';
+COMMENT ON COLUMN domain_rel.deleted_at IS 'domain relation deleted at';
