This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new aa663b24ee [#14626] feat(core): Persist Semantic Model dataset source
associations (#14627)
aa663b24ee is described below
commit aa663b24ee23ec23bec58f3dc0d3ef25fb7708de
Author: mchades <[email protected]>
AuthorDate: Thu Nov 5 23:13:37 2026 +0800
[#14626] feat(core): Persist Semantic Model dataset source associations
(#14627)
### What changes were proposed in this pull request?
Add relational storage for Semantic Model dataset-to-source associations
on H2, MySQL, and PostgreSQL. Maintain explicitly supplied associations
in the model transaction, soft-delete them on replacement or
model/parent deletion, and purge retired rows through bounded garbage
collection.
### Why are the changes needed?
Implement the storage foundation for source lookup approved in #14378.
Fixes #14626.
### Does this PR introduce _any_ user-facing change?
No public API changes. Model API write-path integration and lookup will
follow in separate PRs.
### How was this patch tested?
Unit and H2/MySQL/PostgreSQL tests passed, including schema upgrades and
transaction rollback. Spotless and Javadoc passed.
---
.../apache/gravitino/meta/SemanticModelEntity.java | 24 +
.../gravitino/semantic/SemanticModelSource.java | 51 ++
.../semantic/SemanticModelSourceUpdate.java | 100 ++++
.../mapper/SemanticModelSourceMapper.java | 77 +++
.../SemanticModelSourceSQLProviderFactory.java | 73 +++
.../provider/DefaultMapperPackageProvider.java | 2 +
.../base/SemanticModelSourceBaseSQLProvider.java | 83 +++
.../SemanticModelSourcePostgreSQLProvider.java | 42 ++
.../relational/service/CatalogMetaService.java | 7 +-
.../relational/service/MetalakeMetaService.java | 7 +-
.../relational/service/SchemaMetaService.java | 7 +-
.../service/SemanticModelMetaService.java | 44 +-
.../semantic/TestSemanticModelSourceUpdate.java | 82 +++
.../apache/gravitino/storage/TestSQLScripts.java | 19 +-
.../TestSemanticModelSourceAssociations.java | 589 +++++++++++++++++++++
scripts/h2/schema-2.0.0-h2.sql | 14 +
scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql | 14 +
scripts/mysql/schema-2.0.0-mysql.sql | 15 +
scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql | 15 +
scripts/postgresql/schema-2.0.0-postgresql.sql | 21 +
.../upgrade-1.3.0-to-2.0.0-postgresql.sql | 21 +
21 files changed, 1293 insertions(+), 14 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/meta/SemanticModelEntity.java
b/core/src/main/java/org/apache/gravitino/meta/SemanticModelEntity.java
index 1f4684d671..b4b89fa623 100644
--- a/core/src/main/java/org/apache/gravitino/meta/SemanticModelEntity.java
+++ b/core/src/main/java/org/apache/gravitino/meta/SemanticModelEntity.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.meta;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.Collections;
@@ -32,6 +33,7 @@ import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.semantic.SemanticModel;
import org.apache.gravitino.semantic.SemanticModelDefinition;
+import org.apache.gravitino.semantic.SemanticModelSourceUpdate;
/** A metadata-store entity representing a schema-scoped Semantic Model. */
@ToString
@@ -77,10 +79,21 @@ public class SemanticModelEntity implements Entity,
Auditable, HasIdentifier, Se
private Map<String, String> properties = Collections.emptyMap();
private AuditInfo auditInfo;
+ @JsonIgnore @ToString.Exclude
+ private transient SemanticModelSourceUpdate sourceUpdate =
SemanticModelSourceUpdate.preserve();
+
@Nullable private String comment;
private SemanticModelEntity() {}
+ /**
+ * @return The internal association update instruction for this write.
+ */
+ @JsonIgnore
+ public SemanticModelSourceUpdate sourceUpdate() {
+ return sourceUpdate == null ? SemanticModelSourceUpdate.preserve() :
sourceUpdate;
+ }
+
/**
* Returns the fields and values of this Semantic Model entity.
*
@@ -300,6 +313,17 @@ public class SemanticModelEntity implements Entity,
Auditable, HasIdentifier, Se
return this;
}
+ /**
+ * Attaches the source associations resolved for this write; they are not
snapshot fields.
+ *
+ * @param update The explicit replacement or preserve instruction.
+ * @return This builder.
+ */
+ public Builder withSourceUpdate(SemanticModelSourceUpdate update) {
+ semanticModel.sourceUpdate = Objects.requireNonNull(update,
"sourceUpdate");
+ return this;
+ }
+
/**
* Sets the Gravitino-specific properties.
*
diff --git
a/core/src/main/java/org/apache/gravitino/semantic/SemanticModelSource.java
b/core/src/main/java/org/apache/gravitino/semantic/SemanticModelSource.java
new file mode 100644
index 0000000000..694927ab3c
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/semantic/SemanticModelSource.java
@@ -0,0 +1,51 @@
+/*
+ * 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.semantic;
+
+import com.google.common.base.Preconditions;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import org.apache.gravitino.Entity;
+
+/** A dataset's source resolved during caller-authorized definition
validation. */
+@Getter
+@EqualsAndHashCode
+public final class SemanticModelSource {
+ private final String datasetName;
+ private final Entity.EntityType sourceType;
+ private final long sourceId;
+
+ /**
+ * Creates a resolved source association.
+ *
+ * @param datasetName The exact dataset name in the definition.
+ * @param sourceType TABLE or VIEW.
+ * @param sourceId The registered Gravitino entity ID.
+ */
+ public SemanticModelSource(String datasetName, Entity.EntityType sourceType,
long sourceId) {
+ Preconditions.checkArgument(
+ datasetName != null && !datasetName.isEmpty(), "Dataset name is
required");
+ Preconditions.checkArgument(
+ sourceType == Entity.EntityType.TABLE || sourceType ==
Entity.EntityType.VIEW,
+ "Source must be a table or view");
+ this.datasetName = datasetName;
+ this.sourceType = sourceType;
+ this.sourceId = sourceId;
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/semantic/SemanticModelSourceUpdate.java
b/core/src/main/java/org/apache/gravitino/semantic/SemanticModelSourceUpdate.java
new file mode 100644
index 0000000000..12d4928acb
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/semantic/SemanticModelSourceUpdate.java
@@ -0,0 +1,100 @@
+/*
+ * 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.semantic;
+
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import lombok.EqualsAndHashCode;
+
+/** Internal write intent for source associations; never part of a public
model definition. */
+@EqualsAndHashCode
+public final class SemanticModelSourceUpdate {
+ private static final SemanticModelSourceUpdate PRESERVE =
+ new SemanticModelSourceUpdate(false, List.of());
+ private final boolean replace;
+ private final List<SemanticModelSource> sources;
+
+ private SemanticModelSourceUpdate(boolean replace, List<SemanticModelSource>
sources) {
+ this.replace = replace;
+ this.sources = List.copyOf(sources);
+ }
+
+ /**
+ * @return An instruction to preserve existing associations during
metadata-only updates.
+ */
+ public static SemanticModelSourceUpdate preserve() {
+ return PRESERVE;
+ }
+
+ /**
+ * Creates an instruction to replace the entire association set.
+ *
+ * @param sources All resolved datasets of the new definition.
+ * @return The immutable replacement instruction.
+ * @throws IllegalArgumentException If sources is null or contains duplicate
dataset names.
+ */
+ public static SemanticModelSourceUpdate replace(List<SemanticModelSource>
sources) {
+ Preconditions.checkArgument(sources != null, "sources must not be null");
+ Set<String> names = new HashSet<>();
+ sources.forEach(
+ source ->
+ Preconditions.checkArgument(
+ names.add(source.getDatasetName()),
+ "Duplicate dataset association: %s",
+ source.getDatasetName()));
+ return new SemanticModelSourceUpdate(true, sources);
+ }
+
+ /**
+ * @return Whether this write replaces the associations, even for an
identical definition.
+ */
+ public boolean replacesSources() {
+ return replace;
+ }
+
+ /**
+ * @return The complete immutable replacement set, or an empty list for
preserve.
+ */
+ public List<SemanticModelSource> sources() {
+ return sources;
+ }
+
+ /**
+ * Checks that a replacement describes exactly the datasets being committed.
+ *
+ * @param definition The definition written in the same transaction.
+ */
+ public void validate(SemanticModelDefinition definition) {
+ if (replace) {
+ Set<String> datasets =
+
Arrays.stream(definition.datasets()).map(Dataset::name).collect(Collectors.toSet());
+ Preconditions.checkArgument(
+ datasets.size() == definition.datasets().length, "Duplicate dataset
names in definition");
+ Set<String> associations =
+
sources.stream().map(SemanticModelSource::getDatasetName).collect(Collectors.toSet());
+ Preconditions.checkArgument(
+ datasets.equals(associations),
+ "Source associations must match the definition's dataset names");
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelSourceMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelSourceMapper.java
new file mode 100644
index 0000000000..a325649cf1
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelSourceMapper.java
@@ -0,0 +1,77 @@
+/*
+ * 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.semantic.SemanticModelSource;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.UpdateProvider;
+
+/** Persists dataset-to-source associations and their retirement state. */
+public interface SemanticModelSourceMapper {
+ /** The dataset source association table name. */
+ String TABLE_NAME = "semantic_model_source_rel";
+
+ /** Inserts live associations with database-generated row IDs. */
+ @Insert(
+ """
+ <script>INSERT INTO semantic_model_source_rel
+ (metalake_id, semantic_model_id, dataset_name, source_type, source_id,
deleted_at) VALUES
+ <foreach collection='sources' item='source' separator=','>
+ (#{metalakeId}, #{modelId}, #{source.datasetName}, #{source.sourceType},
#{source.sourceId}, 0)
+ </foreach></script>
+ """)
+ void insertSources(
+ @Param("metalakeId") long metalakeId,
+ @Param("modelId") long modelId,
+ @Param("sources") List<SemanticModelSource> sources);
+
+ /** Retires all live associations of a model. */
+ @UpdateProvider(
+ type = SemanticModelSourceSQLProviderFactory.class,
+ method = "softDeleteByModelId")
+ int softDeleteByModelId(@Param("modelId") Long modelId);
+
+ /** Retires live associations owned by models in the given schemas. */
+ @UpdateProvider(
+ type = SemanticModelSourceSQLProviderFactory.class,
+ method = "softDeleteBySchemaIds")
+ int softDeleteBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
+
+ /** Retires live associations owned by models in a catalog. */
+ @UpdateProvider(
+ type = SemanticModelSourceSQLProviderFactory.class,
+ method = "softDeleteByCatalogId")
+ int softDeleteByCatalogId(@Param("catalogId") Long catalogId);
+
+ /** Retires live associations owned by models in a metalake. */
+ @UpdateProvider(
+ type = SemanticModelSourceSQLProviderFactory.class,
+ method = "softDeleteByMetalakeId")
+ int softDeleteByMetalakeId(@Param("metalakeId") Long metalakeId);
+
+ /** Permanently deletes retired associations older than a timeline. */
+ @DeleteProvider(
+ type = SemanticModelSourceSQLProviderFactory.class,
+ method = "deleteByLegacyTimeline")
+ int deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelSourceSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelSourceSQLProviderFactory.java
new file mode 100644
index 0000000000..34d6edafeb
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelSourceSQLProviderFactory.java
@@ -0,0 +1,73 @@
+/*
+ * 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.SemanticModelSourceBaseSQLProvider;
+import
org.apache.gravitino.storage.relational.mapper.provider.postgresql.SemanticModelSourcePostgreSQLProvider;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/** Selects database-specific SQL for retiring and purging dataset source
associations. */
+public class SemanticModelSourceSQLProviderFactory {
+ private static final Map<JDBCBackendType,
SemanticModelSourceBaseSQLProvider> PROVIDERS =
+ ImmutableMap.of(
+ JDBCBackendType.MYSQL, new SemanticModelSourceBaseSQLProvider(),
+ JDBCBackendType.H2, new SemanticModelSourceBaseSQLProvider(),
+ JDBCBackendType.POSTGRESQL, new
SemanticModelSourcePostgreSQLProvider());
+
+ /** Returns the SQL provider for the configured relational backend. */
+ public static SemanticModelSourceBaseSQLProvider getProvider() {
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+ return PROVIDERS.get(JDBCBackendType.fromString(databaseId));
+ }
+
+ /** Retires all live associations of a model. */
+ public static String softDeleteByModelId(@Param("modelId") Long modelId) {
+ return getProvider().softDeleteByModelId(modelId);
+ }
+
+ /** Retires live associations owned by models in the given schemas. */
+ public static String softDeleteBySchemaIds(@Param("schemaIds") List<Long>
schemaIds) {
+ return getProvider().softDeleteBySchemaIds(schemaIds);
+ }
+
+ /** Retires live associations owned by models in a catalog. */
+ public static String softDeleteByCatalogId(@Param("catalogId") Long
catalogId) {
+ return getProvider().softDeleteByCatalogId(catalogId);
+ }
+
+ /** Retires live associations owned by models in a metalake. */
+ public static String softDeleteByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ return getProvider().softDeleteByMetalakeId(metalakeId);
+ }
+
+ /** Permanently deletes retired associations older than a timeline. */
+ public static String deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return getProvider().deleteByLegacyTimeline(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 32f289750c..ab433bb5cf 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
@@ -45,6 +45,7 @@ import
org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelMetricMapper;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelSourceMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -89,6 +90,7 @@ public class DefaultMapperPackageProvider implements
MapperPackageProvider {
SecurableObjectMapper.class,
SemanticModelMetaMapper.class,
SemanticModelMetricMapper.class,
+ SemanticModelSourceMapper.class,
SemanticModelVersionInfoMapper.class,
StatisticMetaMapper.class,
TableColumnMapper.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SemanticModelSourceBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SemanticModelSourceBaseSQLProvider.java
new file mode 100644
index 0000000000..99f9718924
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SemanticModelSourceBaseSQLProvider.java
@@ -0,0 +1,83 @@
+/*
+ * 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.SemanticModelSourceMapper.TABLE_NAME;
+
+import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
+import org.apache.ibatis.annotations.Param;
+
+/** Provides MySQL-compatible SQL for retiring dataset source associations. */
+public class SemanticModelSourceBaseSQLProvider {
+
+ /** Returns SQL for retiring a model's live associations. */
+ public String softDeleteByModelId(@Param("modelId") Long modelId) {
+ return softDeleteBy("semantic_model_id = #{modelId}");
+ }
+
+ /** Returns SQL for retiring associations owned by models in the given
schemas. */
+ public String softDeleteBySchemaIds(@Param("schemaIds") List<Long>
schemaIds) {
+ return "<script>"
+ + softDeleteBy(
+ "semantic_model_id IN (SELECT semantic_model_id FROM "
+ + SemanticModelMetaMapper.TABLE_NAME
+ + " WHERE schema_id IN ("
+ + "<foreach collection='schemaIds' item='schemaId'
separator=','>"
+ + "#{schemaId}</foreach>))")
+ + "</script>";
+ }
+
+ /** Returns SQL for retiring associations owned by models in a catalog. */
+ public String softDeleteByCatalogId(@Param("catalogId") Long catalogId) {
+ return softDeleteBy(
+ "semantic_model_id IN (SELECT semantic_model_id FROM "
+ + SemanticModelMetaMapper.TABLE_NAME
+ + " WHERE catalog_id = #{catalogId})");
+ }
+
+ /** Returns SQL for retiring associations owned by models in a metalake. */
+ public String softDeleteByMetalakeId(@Param("metalakeId") Long metalakeId) {
+ return softDeleteBy("metalake_id = #{metalakeId}");
+ }
+
+ /** Returns SQL for purging retired rows without depending on model or
parent metadata. */
+ public String deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return "DELETE FROM "
+ + TABLE_NAME
+ + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit}";
+ }
+
+ /** Returns the database expression for the current time in epoch
milliseconds. */
+ protected String currentTimestampMillis() {
+ return "(UNIX_TIMESTAMP() * 1000.0)"
+ + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000";
+ }
+
+ private String softDeleteBy(String condition) {
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET deleted_at = "
+ + currentTimestampMillis()
+ + " WHERE "
+ + condition
+ + " AND deleted_at = 0";
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SemanticModelSourcePostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SemanticModelSourcePostgreSQLProvider.java
new file mode 100644
index 0000000000..ceb500ba9c
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SemanticModelSourcePostgreSQLProvider.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper.provider.postgresql;
+
+import static
org.apache.gravitino.storage.relational.mapper.SemanticModelSourceMapper.TABLE_NAME;
+
+import
org.apache.gravitino.storage.relational.mapper.provider.base.SemanticModelSourceBaseSQLProvider;
+import org.apache.ibatis.annotations.Param;
+
+/** Provides PostgreSQL SQL for retiring and purging dataset source
associations. */
+public class SemanticModelSourcePostgreSQLProvider extends
SemanticModelSourceBaseSQLProvider {
+ @Override
+ public String deleteByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return "DELETE FROM "
+ + TABLE_NAME
+ + " WHERE id IN (SELECT id FROM "
+ + TABLE_NAME
+ + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit})";
+ }
+
+ @Override
+ protected String currentTimestampMillis() {
+ return "CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)";
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
index 14aed3a43c..42ad99f9e8 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
@@ -50,6 +50,7 @@ import
org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelMetricMapper;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelSourceMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -359,7 +360,11 @@ public class CatalogMetaService {
() ->
SessionUtils.doWithoutCommit(
SemanticModelMetricMapper.class,
- mapper ->
mapper.softDeleteSemanticModelMetricsByCatalogId(catalogId)));
+ mapper ->
mapper.softDeleteSemanticModelMetricsByCatalogId(catalogId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelSourceMapper.class,
+ mapper -> mapper.softDeleteByCatalogId(catalogId)));
} else {
SessionUtils.doMultipleWithCommit(
() -> {
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 f4280ceb8a..beeba0556f 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
@@ -56,6 +56,7 @@ import
org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelMetricMapper;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelSourceMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -348,7 +349,11 @@ public class MetalakeMetaService {
() ->
SessionUtils.doWithoutCommit(
SemanticModelMetricMapper.class,
- mapper ->
mapper.softDeleteSemanticModelMetricsByMetalakeId(metalakeId)));
+ mapper ->
mapper.softDeleteSemanticModelMetricsByMetalakeId(metalakeId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelSourceMapper.class,
+ mapper -> mapper.softDeleteByMetalakeId(metalakeId)));
} else {
SessionUtils.doMultipleWithCommit(
() -> {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
index 6ddfe16616..ee924f51ea 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
@@ -60,6 +60,7 @@ import
org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelMetricMapper;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelSourceMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -410,7 +411,11 @@ public class SchemaMetaService {
() ->
SessionUtils.doWithoutCommit(
SemanticModelMetricMapper.class,
- mapper ->
mapper.softDeleteSemanticModelMetricsBySchemaIds(schemaIds.get())));
+ mapper ->
mapper.softDeleteSemanticModelMetricsBySchemaIds(schemaIds.get())),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelSourceMapper.class,
+ mapper -> mapper.softDeleteBySchemaIds(schemaIds.get())));
} else {
SessionUtils.doMultipleWithCommit(
() -> {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/SemanticModelMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/SemanticModelMetaService.java
index 582481f4c7..ffe5e6ccb2 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/SemanticModelMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/SemanticModelMetaService.java
@@ -48,10 +48,12 @@ import org.apache.gravitino.meta.SemanticModelEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.semantic.Metric;
import org.apache.gravitino.semantic.SemanticModelDefinition;
+import org.apache.gravitino.semantic.SemanticModelSource;
import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.relational.EntityChangeLogWriter;
import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelMetricMapper;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelSourceMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.po.SemanticModelMetricPO;
import org.apache.gravitino.storage.relational.po.SemanticModelPO;
@@ -69,6 +71,7 @@ public class SemanticModelMetaService {
private static final Logger LOG =
LoggerFactory.getLogger(SemanticModelMetaService.class);
private static final int METRIC_INSERT_BATCH_SIZE = 1000;
+ private static final int SOURCE_INSERT_BATCH_SIZE = 1000;
private static final SemanticModelMetaService INSTANCE = new
SemanticModelMetaService();
private final BasePOStorageOps<SemanticModelPO, SemanticModelMetaMapper> ops;
@@ -139,7 +142,8 @@ public class SemanticModelMetaService {
}
/**
- * Inserts or overwrites a Semantic Model, its snapshot, and its metric
identities atomically.
+ * Inserts or overwrites a Semantic Model, its snapshot, metric identities,
and supplied source
+ * associations atomically.
*
* @param semanticModelEntity The Semantic Model entity.
* @param overwrite Whether to overwrite rows for the same stable ID or
natural key.
@@ -186,6 +190,7 @@ public class SemanticModelMetaService {
persistedPO.set(replacementPO);
}),
() -> reconcileMetricIdentities(persistedPO.get(),
semanticModelEntity.definition()),
+ () -> replaceSources(persistedPO.get(), semanticModelEntity),
() ->
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
@@ -272,6 +277,7 @@ public class SemanticModelMetaService {
reconcileMetricIdentities(newSemanticModelPO,
newEntity.definition());
}
},
+ () -> replaceSources(newSemanticModelPO, newEntity),
() ->
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
@@ -295,7 +301,8 @@ public class SemanticModelMetaService {
}
/**
- * Soft-deletes a Semantic Model, its metric identities, and all of its
version snapshots.
+ * Soft-deletes a Semantic Model, its metric identities, source
associations, and version
+ * snapshots.
*
* @param identifier The Semantic Model identifier.
* @return {@code true} when an active identity was deleted.
@@ -312,8 +319,8 @@ public class SemanticModelMetaService {
}
/**
- * Permanently deletes retired metric identities and deleted models and
snapshots older than a
- * timeline.
+ * Permanently deletes retired source associations, metric identities, and
deleted models and
+ * snapshots older than a timeline.
*
* @param legacyTimeline The exclusive deletion timeline in epoch
milliseconds.
* @param limit The maximum rows to delete from each table.
@@ -323,6 +330,10 @@ public class SemanticModelMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteSemanticModelMetasByLegacyTimeline")
public int deleteSemanticModelMetasByLegacyTimeline(Long legacyTimeline, int
limit) {
+ int sourceDeletedCount =
+ SessionUtils.doWithCommitAndFetchResult(
+ SemanticModelSourceMapper.class,
+ mapper -> mapper.deleteByLegacyTimeline(legacyTimeline, limit));
int metricDeletedCount =
SessionUtils.doWithCommitAndFetchResult(
SemanticModelMetricMapper.class,
@@ -335,7 +346,7 @@ public class SemanticModelMetaService {
SessionUtils.doWithCommitAndFetchResult(
SemanticModelMetaMapper.class,
mapper ->
mapper.deleteSemanticModelMetasByLegacyTimeline(legacyTimeline, limit));
- return metricDeletedCount + versionDeletedCount + metaDeletedCount;
+ return sourceDeletedCount + metricDeletedCount + versionDeletedCount +
metaDeletedCount;
}
/**
@@ -416,11 +427,34 @@ public class SemanticModelMetaService {
SessionUtils.doWithoutCommit(
SemanticModelMetricMapper.class,
mapper ->
mapper.softDeleteSemanticModelMetricsBySemanticModelId(semanticModelId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelSourceMapper.class,
+ mapper -> mapper.softDeleteByModelId(semanticModelId)),
() ->
EntityChangeLogWriter.append(
semanticModelIdent, Entity.EntityType.SEMANTIC_MODEL,
OperateType.DROP));
}
+ private void replaceSources(SemanticModelPO po, SemanticModelEntity entity) {
+ if (!entity.sourceUpdate().replacesSources()) {
+ return;
+ }
+ entity.sourceUpdate().validate(entity.definition());
+ SessionUtils.doWithoutCommit(
+ SemanticModelSourceMapper.class,
+ mapper -> {
+ mapper.softDeleteByModelId(po.getSemanticModelId());
+ List<SemanticModelSource> sources = entity.sourceUpdate().sources();
+ for (int start = 0; start < sources.size(); start +=
SOURCE_INSERT_BATCH_SIZE) {
+ mapper.insertSources(
+ po.getMetalakeId(),
+ po.getSemanticModelId(),
+ sources.subList(start, Math.min(start +
SOURCE_INSERT_BATCH_SIZE, sources.size())));
+ }
+ });
+ }
+
private static void reconcileMetricIdentities(
SemanticModelPO model, SemanticModelDefinition definition) {
Set<String> names = metricNames(definition);
diff --git
a/core/src/test/java/org/apache/gravitino/semantic/TestSemanticModelSourceUpdate.java
b/core/src/test/java/org/apache/gravitino/semantic/TestSemanticModelSourceUpdate.java
new file mode 100644
index 0000000000..fde5a99bd8
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/semantic/TestSemanticModelSourceUpdate.java
@@ -0,0 +1,82 @@
+/*
+ * 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.semantic;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.junit.jupiter.api.Test;
+
+class TestSemanticModelSourceUpdate {
+ @Test
+ void testImmutableCompleteReplacement() {
+ SemanticModelSource source = new SemanticModelSource("Orders",
Entity.EntityType.TABLE, 123);
+ List<SemanticModelSource> mutable = new ArrayList<>(List.of(source));
+ SemanticModelSourceUpdate update =
SemanticModelSourceUpdate.replace(mutable);
+ mutable.clear();
+ assertEquals(List.of(source), update.sources());
+ assertTrue(update.replacesSources());
+ assertThrows(UnsupportedOperationException.class, () ->
update.sources().clear());
+ update.validate(definition("Orders"));
+ assertThrows(IllegalArgumentException.class, () ->
update.validate(definition("orders")));
+ Dataset dataset = definition("Orders").datasets()[0];
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ update.validate(
+ SemanticModelDefinition.builder()
+ .withDatasets(new Dataset[] {dataset, dataset})
+ .build()));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> SemanticModelSourceUpdate.replace(List.of(source, source)));
+ assertFalse(SemanticModelSourceUpdate.preserve().replacesSources());
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new SemanticModelSource("Orders", Entity.EntityType.CATALOG,
123));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new SemanticModelSource("", Entity.EntityType.TABLE, 123));
+ }
+
+ @Test
+ void testNullSourcesRejected() {
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
SemanticModelSourceUpdate.replace(null));
+ assertEquals("sources must not be null", exception.getMessage());
+ }
+
+ private SemanticModelDefinition definition(String name) {
+ return SemanticModelDefinition.builder()
+ .withDatasets(
+ new Dataset[] {
+ Dataset.builder()
+ .withName(name)
+ .withSource(NameIdentifier.of("catalog", "schema", "orders"))
+ .build()
+ })
+ .build();
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
b/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
index 6b04a69b0f..55f4fff8b5 100644
--- a/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
+++ b/core/src/test/java/org/apache/gravitino/storage/TestSQLScripts.java
@@ -104,7 +104,7 @@ public class TestSQLScripts extends TestJDBCBackend {
executeScript(scriptFile);
}
if ("2.0.0".equals(versionScripts.getKey())) {
- assertRoleRelationsIndexedByRoleId("schema " +
versionScripts.getKey());
+ assertCleanupIndexes("schema " + versionScripts.getKey());
}
}
}
@@ -137,7 +137,7 @@ public class TestSQLScripts extends TestJDBCBackend {
executeScript(sourceSchema);
executeScript(upgradeScript);
if ("2.0.0".equals(upgradeMatcher.group(2))) {
- assertRoleRelationsIndexedByRoleId(upgradeScript.getName());
+ assertCleanupIndexes(upgradeScript.getName());
}
}
}
@@ -605,12 +605,19 @@ public class TestSQLScripts extends TestJDBCBackend {
}
}
- private void assertRoleRelationsIndexedByRoleId(String source) throws
SQLException {
+ private void assertCleanupIndexes(String source) throws SQLException {
+ Map<String, String> leadingColumns =
+ Map.of(
+ "user_role_rel", "role_id",
+ "group_role_rel", "role_id",
+ "semantic_model_source_rel", "deleted_at");
try (SqlSession sqlSession =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
Connection connection = sqlSession.getConnection()) {
DatabaseMetaData metaData = connection.getMetaData();
- for (String table : List.of("user_role_rel", "group_role_rel")) {
+ for (Map.Entry<String, String> entry : leadingColumns.entrySet()) {
+ String table = entry.getKey();
+ String column = entry.getValue();
String tableName =
metaData.storesUpperCaseIdentifiers() ?
table.toUpperCase(Locale.ROOT) : table;
boolean indexed = false;
@@ -618,14 +625,14 @@ public class TestSQLScripts extends TestJDBCBackend {
metaData.getIndexInfo(connection.getCatalog(), null, tableName,
false, false)) {
while (indexes.next()) {
if (indexes.getShort("ORDINAL_POSITION") == 1
- &&
"role_id".equalsIgnoreCase(indexes.getString("COLUMN_NAME"))) {
+ && column.equalsIgnoreCase(indexes.getString("COLUMN_NAME"))) {
indexed = true;
break;
}
}
}
Assertions.assertTrue(
- indexed, "No index leading with role_id on " + table + " after " +
source);
+ indexed, "No index leading with " + column + " on " + table + "
after " + source);
}
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelSourceAssociations.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelSourceAssociations.java
new file mode 100644
index 0000000000..07634e026a
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelSourceAssociations.java
@@ -0,0 +1,589 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.SemanticModelEntity;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.semantic.Dataset;
+import org.apache.gravitino.semantic.SemanticModelDefinition;
+import org.apache.gravitino.semantic.SemanticModelSource;
+import org.apache.gravitino.semantic.SemanticModelSourceUpdate;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.ibatis.session.SqlSession;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+
+/** Verifies association writes, retirement, rollback and GC on every
relational backend. */
+public class TestSemanticModelSourceAssociations extends TestJDBCBackend {
+ private Namespace sourceNamespace;
+ private Namespace modelNamespace;
+ private TableEntity source;
+ private long metalakeId;
+
+ @BeforeEach
+ void prepareSources() throws Exception {
+ metalakeId = createAndInsertMakeLake("lake").id();
+ createAndInsertCatalog("lake", "sources");
+ createAndInsertCatalog("lake", "models");
+ createAndInsertSchema("lake", "sources", "hr");
+ createAndInsertSchema("lake", "models", "semantic");
+ sourceNamespace = Namespace.of("lake", "sources", "hr");
+ modelNamespace = Namespace.of("lake", "models", "semantic");
+ source = createAndInsertTableEntity(sourceNamespace, "employees");
+ }
+
+ @TestTemplate
+ void testSharedSourcesPreserveExactDatasetNames() throws Exception {
+ String[] names = {"employee", "Employee", "manager.with.dot",
"x".repeat(1200), "员工'"};
+ SemanticModelEntity model = model("staff", source.id(), names);
+ backend.insert(model, false);
+ List<Association> live = live(model.id());
+ assertEquals(
+ Set.of(names),
live.stream().map(Association::datasetName).collect(Collectors.toSet()));
+ assertEquals(names.length,
live.stream().map(Association::id).distinct().count());
+ assertTrue(live.stream().allMatch(row -> row.metalakeId() == metalakeId));
+ assertTrue(live.stream().allMatch(row -> row.sourceId() == source.id()));
+ assertTrue(live.stream().allMatch(row ->
row.sourceType().equals("TABLE")));
+ SemanticModelEntity loaded =
service().getSemanticModelByIdentifier(model.nameIdentifier());
+ assertEquals(model, loaded);
+ assertFalse(loaded.sourceUpdate().replacesSources());
+ }
+
+ @TestTemplate
+ void testTableAndViewAssociationsKeepTheirTypes() throws Exception {
+ SemanticModelEntity model = model("staff", source.id(), "table", "view");
+ List<SemanticModelSource> sources =
+ List.of(
+ new SemanticModelSource("table", Entity.EntityType.TABLE,
source.id()),
+ new SemanticModelSource("view", Entity.EntityType.VIEW,
source.id()));
+ backend.insert(
+ copy(model, model.name(), model.definition(),
SemanticModelSourceUpdate.replace(sources)),
+ false);
+ assertEquals(Set.copyOf(sources), sourceSet(model.id()));
+ }
+
+ @TestTemplate
+ void testReplaceEntireSetAndRetirePreviousRows() throws Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee",
"manager");
+ backend.insert(original, false);
+ Set<Long> initialIds = liveIds(original.id());
+ SemanticModelEntity changed = model("staff", source.id(), "manager",
"renamed", "added");
+ backend.update(
+ original.nameIdentifier(),
+ original.type(),
+ old -> copy(original, "staff", changed.definition(),
changed.sourceUpdate()));
+ assertEquals(Set.of("manager", "renamed", "added"),
datasetNames(original.id()));
+
assertTrue(liveIds(original.id()).stream().noneMatch(initialIds::contains));
+ List<Association> retired =
+ rows(original.id()).stream().filter(row -> row.deletedAt() >
0).toList();
+ assertEquals(2, retired.size());
+
+ SemanticModelEntity reordered = model("staff", source.id(), "added",
"renamed", "manager");
+ backend.update(
+ original.nameIdentifier(),
+ original.type(),
+ old -> copy(original, "staff", reordered.definition(),
reordered.sourceUpdate()));
+ assertEquals(Set.of("manager", "renamed", "added"),
datasetNames(original.id()));
+ assertTrue(rows(original.id()).containsAll(retired));
+ assertEquals(8, rows(original.id()).size());
+ assertEquals(
+ 3,
service().getSemanticModelPOByIdentifier(original.nameIdentifier()).getCurrentVersion());
+ }
+
+ @TestTemplate
+ void testMetadataUpdatesPreserveRowsAndExplicitReplacementRebinds() throws
Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ List<Association> initial = rows(original.id());
+ assertTrue(backend.delete(source.nameIdentifier(), source.type(), false));
+ TableEntity replacement = createAndInsertTableEntity(sourceNamespace,
source.name());
+ assertNotEquals(source.id(), replacement.id());
+ SemanticModelEntity renamed =
+ copy(original, "renamed", original.definition(),
SemanticModelSourceUpdate.preserve());
+ backend.update(original.nameIdentifier(), original.type(), old -> renamed);
+ assertEquals(initial, rows(original.id()));
+ assertEquals(renamed,
service().getSemanticModelByIdentifier(renamed.nameIdentifier()));
+
+ SemanticModelSourceUpdate rebound =
+ SemanticModelSourceUpdate.replace(
+ List.of(new SemanticModelSource("employee", source.type(),
replacement.id())));
+ backend.update(
+ renamed.nameIdentifier(),
+ renamed.type(),
+ old -> copy(renamed, renamed.name(), renamed.definition(), rebound));
+ assertEquals(Set.copyOf(rebound.sources()), sourceSet(original.id()));
+ assertEquals(2, rows(original.id()).size());
+ assertTrue(rows(original.id()).get(0).deletedAt() > 0);
+ }
+
+ @TestTemplate
+ void testPreserveKeepsRowsWhenDatasetNamesChange() throws Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ List<Association> initial = rows(original.id());
+ assertEquals(1, initial.size());
+
+ SemanticModelDefinition changedDefinition = model("staff", source.id(),
"manager").definition();
+ SemanticModelEntity changed =
+ copy(original, original.name(), changedDefinition,
SemanticModelSourceUpdate.preserve());
+ backend.update(original.nameIdentifier(), original.type(), old -> changed);
+
+ assertEquals(changed,
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ assertEquals(initial, rows(original.id()));
+ assertEquals(2, snapshotCount(original.id()));
+ }
+
+ @TestTemplate
+ void testSourceRenamePreservesAssociationsAndDefinition() throws Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ List<Association> initial = rows(original.id());
+ TableEntity renamed =
+ createTableEntity(source.id(), sourceNamespace, "employees_v2",
AUDIT_INFO);
+ backend.update(source.nameIdentifier(), source.type(), old -> renamed);
+ assertEquals(initial, rows(original.id()));
+ assertEquals(original,
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ }
+
+ @TestTemplate
+ void testOverwriteUsesPersistedModelIdentity() throws Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ SemanticModelEntity overwrite = model("staff", source.id(), "manager");
+ assertNotEquals(original.id(), overwrite.id());
+ backend.insert(overwrite, true);
+ assertEquals(
+ original.id(),
service().getSemanticModelByIdentifier(original.nameIdentifier()).id());
+ assertEquals(Set.of("manager"), datasetNames(original.id()));
+ assertEquals(2, rows(original.id()).size());
+ assertTrue(rows(overwrite.id()).isEmpty());
+ }
+
+ @TestTemplate
+ void testInvalidAssociationSetRollsBackCreateAndUpdate() throws Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ List<Association> initial = rows(original.id());
+ SemanticModelSourceUpdate invalid =
+ SemanticModelSourceUpdate.replace(
+ List.of(new SemanticModelSource("wrong", source.type(),
source.id())));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ backend.update(
+ original.nameIdentifier(),
+ original.type(),
+ old -> copy(original, "renamed", original.definition(),
invalid)));
+ assertEquals(original,
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ assertEquals(initial, rows(original.id()));
+ assertEquals(
+ 1,
service().getSemanticModelPOByIdentifier(original.nameIdentifier()).getCurrentVersion());
+ SemanticModelEntity invalidCreate =
+ copy(model("invalid", source.id(), "employee"), "invalid",
original.definition(), invalid);
+ assertThrows(IllegalArgumentException.class, () ->
backend.insert(invalidCreate, false));
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
service().getSemanticModelByIdentifier(invalidCreate.nameIdentifier()));
+ assertTrue(rows(invalidCreate.id()).isEmpty());
+ }
+
+ @TestTemplate
+ void testBatchedInsertAndFailureRollBackRetirementAndSnapshot() throws
Exception {
+ String[] names = IntStream.range(0, 1001).mapToObj(i -> "dataset_" +
i).toArray(String[]::new);
+ SemanticModelEntity original = model("staff", source.id(), names);
+ backend.insert(original, false);
+ List<Association> initial = rows(original.id());
+ assertEquals(1001, initial.size());
+ TableEntity blocked = createAndInsertTableEntity(sourceNamespace,
"blocked");
+ List<SemanticModelSource> sources = new
ArrayList<>(original.sourceUpdate().sources());
+ sources.set(1000, new SemanticModelSource(names[1000], source.type(),
blocked.id()));
+ addConstraint("reject_source", "source_id <> " + blocked.id());
+ try {
+ IOException failure =
+ assertThrows(
+ IOException.class,
+ () ->
+ backend.update(
+ original.nameIdentifier(),
+ original.type(),
+ old ->
+ copy(
+ original,
+ "renamed",
+ original.definition(),
+ SemanticModelSourceUpdate.replace(sources))));
+ assertTrue(
+
ExceptionUtils.getStackTrace(failure).toLowerCase(Locale.ROOT).contains("reject_source"));
+ assertEquals(initial, rows(original.id()));
+ assertEquals(original,
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ assertEquals(
+ 1,
+
service().getSemanticModelPOByIdentifier(original.nameIdentifier()).getCurrentVersion());
+ assertEquals(1, snapshotCount(original.id()));
+ assertThrows(
+ IOException.class,
+ () ->
+ backend.insert(
+ copy(
+ original,
+ original.name(),
+ original.definition(),
+ SemanticModelSourceUpdate.replace(sources)),
+ true));
+ assertEquals(initial, rows(original.id()));
+ assertEquals(original,
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ assertEquals(1, snapshotCount(original.id()));
+ SemanticModelEntity invalidCreate = model("invalid", blocked.id(),
"employee");
+ assertThrows(IOException.class, () -> backend.insert(invalidCreate,
false));
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
service().getSemanticModelByIdentifier(invalidCreate.nameIdentifier()));
+ assertTrue(rows(invalidCreate.id()).isEmpty());
+ assertEquals(0, snapshotCount(invalidCreate.id()));
+ } finally {
+ dropConstraint("reject_source");
+ }
+ }
+
+ @TestTemplate
+ void testModelAndParentDeletionRetireOnlyOwnedAssociations() throws
Exception {
+ for (Entity.EntityType type :
+ List.of(
+ Entity.EntityType.SEMANTIC_MODEL,
+ Entity.EntityType.SCHEMA,
+ Entity.EntityType.CATALOG,
+ Entity.EntityType.METALAKE)) {
+ resetDatabase();
+ prepareSources();
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ NameIdentifier target = deletionTarget(original, type);
+ assertTrue(backend.delete(target, type, true));
+ assertTrue(live(original.id()).isEmpty());
+ assertEquals(1, rows(original.id()).size());
+ assertTrue(rows(original.id()).get(0).deletedAt() > 0);
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ }
+ }
+
+ @TestTemplate
+ void testRetirementFailureRollsBackModelAndEveryParentCascade() throws
Exception {
+ for (Entity.EntityType type :
+ List.of(
+ Entity.EntityType.SEMANTIC_MODEL,
+ Entity.EntityType.SCHEMA,
+ Entity.EntityType.CATALOG,
+ Entity.EntityType.METALAKE)) {
+ resetDatabase();
+ prepareSources();
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ List<Association> initial = rows(original.id());
+ NameIdentifier target = deletionTarget(original, type);
+ addConstraint("reject_retirement", "deleted_at = 0");
+ try {
+ Exception failure = assertThrows(Exception.class, () ->
backend.delete(target, type, true));
+ assertTrue(
+ ExceptionUtils.getStackTrace(failure)
+ .toLowerCase(Locale.ROOT)
+ .contains("reject_retirement"));
+ assertEquals(original,
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ assertTrue(backend.exists(target, type));
+ assertTrue(backend.exists(source.nameIdentifier(), source.type()));
+ assertEquals(initial, rows(original.id()));
+ assertEquals(1, snapshotCount(original.id()));
+ } finally {
+ dropConstraint("reject_retirement");
+ }
+ }
+ }
+
+ @TestTemplate
+ void testSchemaCascadeIncludesDescendantsAndPreservesOtherModels() throws
Exception {
+ SemanticModelEntity parent = model("parent", source.id(), "employee");
+ backend.insert(parent, false);
+ createAndInsertSchema("lake", "models", "semantic:child");
+ Namespace originalNamespace = modelNamespace;
+ modelNamespace = Namespace.of("lake", "models", "semantic:child");
+ SemanticModelEntity child = model("child", source.id(), "manager");
+ backend.insert(child, false);
+ createAndInsertSchema("lake", "models", "other");
+ modelNamespace = Namespace.of("lake", "models", "other");
+ SemanticModelEntity other = model("other", source.id(), "employee");
+ backend.insert(other, false);
+ List<Association> unrelated = rows(other.id());
+ assertTrue(
+ backend.delete(
+ NameIdentifier.of(originalNamespace.levels()),
Entity.EntityType.SCHEMA, true));
+ assertTrue(live(parent.id()).isEmpty());
+ assertTrue(live(child.id()).isEmpty());
+ assertEquals(unrelated, rows(other.id()));
+ }
+
+ @TestTemplate
+ void testSourceDeletionAndParentDeletionPreserveLiveAssociations() throws
Exception {
+ for (Entity.EntityType type :
+ List.of(Entity.EntityType.TABLE, Entity.EntityType.SCHEMA,
Entity.EntityType.CATALOG)) {
+ resetDatabase();
+ prepareSources();
+ SemanticModelEntity original = model("staff", source.id(), "employee");
+ backend.insert(original, false);
+ List<Association> initial = rows(original.id());
+ NameIdentifier target =
+ switch (type) {
+ case TABLE -> source.nameIdentifier();
+ case SCHEMA -> NameIdentifier.of(sourceNamespace.levels());
+ default -> NameIdentifier.of("lake", "sources");
+ };
+ assertTrue(backend.delete(target, type, true));
+ backend.hardDeleteLegacyData(type, Long.MAX_VALUE);
+ service().deleteSemanticModelMetasByLegacyTimeline(Long.MAX_VALUE, 100);
+ assertEquals(initial, rows(original.id()));
+ assertEquals(original,
service().getSemanticModelByIdentifier(original.nameIdentifier()));
+ }
+ }
+
+ @TestTemplate
+ void testGcTimelineLimitAndSnapshotRetention() throws Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee",
"manager");
+ backend.insert(original, false);
+ SemanticModelEntity replacement = model("staff", source.id(), "employee");
+ backend.update(
+ original.nameIdentifier(),
+ original.type(),
+ old ->
+ copy(original, original.name(), replacement.definition(),
replacement.sourceUpdate()));
+ List<Association> current = live(original.id());
+ execute("UPDATE semantic_model_source_rel SET deleted_at = 10 WHERE
deleted_at > 0");
+ assertEquals(0, service().deleteSemanticModelMetasByLegacyTimeline(10L,
1));
+ assertEquals(1, service().deleteSemanticModelMetasByLegacyTimeline(11L,
1));
+ assertEquals(2, rows(original.id()).size());
+ assertEquals(1, service().deleteSemanticModelMetasByLegacyTimeline(11L,
1));
+ assertEquals(1, service().deleteSemanticModelVersionsByRetentionCount(1L,
100));
+ service().deleteSemanticModelMetasByLegacyTimeline(Long.MAX_VALUE, 100);
+ assertEquals(current, rows(original.id()));
+ assertEquals(1, snapshotCount(original.id()));
+ }
+
+ @TestTemplate
+ void testGcAfterModelAndParentsHaveBeenPhysicallyRemoved() throws Exception {
+ SemanticModelEntity original = model("staff", source.id(), "employee",
"manager");
+ backend.insert(original, false);
+ backend.delete(NameIdentifier.of("lake"), Entity.EntityType.METALAKE,
true);
+ SessionUtils.doWithCommitAndFetchResult(
+ SemanticModelMetaMapper.class,
+ mapper ->
mapper.deleteSemanticModelMetasByLegacyTimeline(Long.MAX_VALUE, 100));
+
SchemaMetaService.getInstance().deleteSchemaMetasByLegacyTimeline(Long.MAX_VALUE,
100);
+
CatalogMetaService.getInstance().deleteCatalogMetasByLegacyTimeline(Long.MAX_VALUE,
100);
+
MetalakeMetaService.getInstance().deleteMetalakeMetasByLegacyTimeline(Long.MAX_VALUE,
100);
+ assertEquals(2, rows(original.id()).size());
+ execute("UPDATE semantic_model_source_rel SET deleted_at = 10 WHERE
deleted_at > 0");
+ assertEquals(1, service().deleteSemanticModelMetasByLegacyTimeline(11L,
1));
+ assertEquals(1, rows(original.id()).size());
+ assertEquals(1, service().deleteSemanticModelMetasByLegacyTimeline(11L,
1));
+ assertTrue(rows(original.id()).isEmpty());
+ }
+
+ private static SemanticModelMetaService service() {
+ return SemanticModelMetaService.getInstance();
+ }
+
+ private NameIdentifier deletionTarget(SemanticModelEntity model,
Entity.EntityType type) {
+ return switch (type) {
+ case SEMANTIC_MODEL -> model.nameIdentifier();
+ case SCHEMA -> NameIdentifier.of(modelNamespace.levels());
+ case CATALOG -> NameIdentifier.of("lake", "models");
+ default -> NameIdentifier.of("lake");
+ };
+ }
+
+ private void addConstraint(String name, String condition) throws Exception {
+ execute(
+ "ALTER TABLE semantic_model_source_rel ADD CONSTRAINT "
+ + name
+ + " CHECK ("
+ + condition
+ + ")");
+ }
+
+ private void dropConstraint(String name) throws Exception {
+ execute(
+ "ALTER TABLE semantic_model_source_rel DROP "
+ + ("mysql".equals(backendType) ? "CHECK " : "CONSTRAINT ")
+ + name);
+ }
+
+ private void execute(String sql) throws Exception {
+ try (SqlSession session =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession();
+ Statement statement = session.getConnection().createStatement()) {
+ statement.execute(sql);
+ session.commit();
+ }
+ }
+
+ private List<Association> rows(long modelId) throws Exception {
+ try (SqlSession session =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession();
+ PreparedStatement statement =
+ session
+ .getConnection()
+ .prepareStatement(
+ "SELECT id, metalake_id, dataset_name, source_type,
source_id, deleted_at"
+ + " FROM semantic_model_source_rel WHERE
semantic_model_id = ? ORDER BY id")) {
+ statement.setLong(1, modelId);
+ try (ResultSet result = statement.executeQuery()) {
+ List<Association> rows = new ArrayList<>();
+ while (result.next()) {
+ rows.add(
+ new Association(
+ result.getLong(1),
+ result.getLong(2),
+ result.getString(3),
+ result.getString(4),
+ result.getLong(5),
+ result.getLong(6)));
+ }
+ return rows;
+ }
+ }
+ }
+
+ private List<Association> live(long modelId) throws Exception {
+ return rows(modelId).stream().filter(row -> row.deletedAt() == 0).toList();
+ }
+
+ private Set<Long> liveIds(long modelId) throws Exception {
+ return
live(modelId).stream().map(Association::id).collect(Collectors.toSet());
+ }
+
+ private Set<String> datasetNames(long modelId) throws Exception {
+ return
live(modelId).stream().map(Association::datasetName).collect(Collectors.toSet());
+ }
+
+ private Set<SemanticModelSource> sourceSet(long modelId) throws Exception {
+ return live(modelId).stream()
+ .map(
+ row ->
+ new SemanticModelSource(
+ row.datasetName(),
Entity.EntityType.valueOf(row.sourceType()), row.sourceId()))
+ .collect(Collectors.toSet());
+ }
+
+ private int snapshotCount(long modelId) throws Exception {
+ try (SqlSession session =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession();
+ PreparedStatement statement =
+ session
+ .getConnection()
+ .prepareStatement(
+ "SELECT COUNT(*) FROM semantic_model_version_info WHERE
semantic_model_id = ? AND deleted_at = 0")) {
+ statement.setLong(1, modelId);
+ try (ResultSet result = statement.executeQuery()) {
+ result.next();
+ return result.getInt(1);
+ }
+ }
+ }
+
+ private SemanticModelEntity model(String name, long sourceId, String...
datasets) {
+ SemanticModelDefinition definition =
+ SemanticModelDefinition.builder()
+ .withDatasets(
+ Arrays.stream(datasets)
+ .map(
+ dataset ->
+ Dataset.builder()
+ .withName(dataset)
+ .withSource(NameIdentifier.of("sources", "hr",
"employees"))
+ .build())
+ .toArray(Dataset[]::new))
+ .build();
+ return SemanticModelEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(modelNamespace)
+ .withDefinition(definition)
+ .withProperties(Map.of())
+ .withAuditInfo(AUDIT_INFO)
+ .withSourceUpdate(
+ SemanticModelSourceUpdate.replace(
+ Arrays.stream(datasets)
+ .map(
+ dataset ->
+ new SemanticModelSource(dataset,
Entity.EntityType.TABLE, sourceId))
+ .toList()))
+ .build();
+ }
+
+ private SemanticModelEntity copy(
+ SemanticModelEntity original,
+ String name,
+ SemanticModelDefinition definition,
+ SemanticModelSourceUpdate update) {
+ return SemanticModelEntity.builder()
+ .withId(original.id())
+ .withName(name)
+ .withNamespace(original.namespace())
+ .withDefinition(definition)
+ .withComment("Updated comment")
+ .withProperties(Map.of("updated", "true"))
+ .withAuditInfo(AUDIT_INFO)
+ .withSourceUpdate(update)
+ .build();
+ }
+
+ @SuppressWarnings("UnusedVariable") // Error Prone does not recognize record
component accessors.
+ private record Association(
+ long id,
+ long metalakeId,
+ String datasetName,
+ String sourceType,
+ long sourceId,
+ long deletedAt) {}
+}
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index a1fc82b59d..035007284f 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -781,3 +781,17 @@ CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
) COMMENT='table snapshot metrics metadata';
CREATE INDEX IF NOT EXISTS `idx_tsm_tid_pid_sid` ON
`table_snapshot_metrics`(`table_id`, `policy_id`, `snapshot_id`);
CREATE INDEX IF NOT EXISTS `idx_tsm_created_at` ON
`table_snapshot_metrics`(`created_at`);
+
+CREATE TABLE IF NOT EXISTS `semantic_model_source_rel` (
+ `id` BIGINT AUTO_INCREMENT COMMENT 'auto increment id',
+ `metalake_id` BIGINT NOT NULL COMMENT 'containing metalake id',
+ `semantic_model_id` BIGINT NOT NULL COMMENT 'containing semantic model id',
+ `dataset_name` CLOB NOT NULL COMMENT 'exact dataset name in the current
definition',
+ `source_type` VARCHAR(64) NOT NULL COMMENT 'TABLE or VIEW',
+ `source_id` BIGINT NOT NULL COMMENT 'resolved source object id',
+ `deleted_at` BIGINT NOT NULL DEFAULT 0 COMMENT 'association retired at',
+ PRIMARY KEY (`id`)
+) COMMENT='current semantic model dataset source associations';
+CREATE INDEX IF NOT EXISTS `idx_smsr_model` ON `semantic_model_source_rel`
(`semantic_model_id`, `deleted_at`);
+CREATE INDEX IF NOT EXISTS `idx_smsr_source` ON `semantic_model_source_rel`
(`metalake_id`, `source_type`, `source_id`, `deleted_at`);
+CREATE INDEX IF NOT EXISTS `idx_smsr_deleted_at` ON
`semantic_model_source_rel` (`deleted_at`);
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 1151ba1755..d246e6cea0 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
@@ -232,3 +232,17 @@ ALTER TABLE `policy_meta`
-- Role deletion soft-deletes relation rows by role_id.
CREATE INDEX IF NOT EXISTS `group_role_rel_idx_role_id` ON `group_role_rel`
(`role_id`);
+
+CREATE TABLE IF NOT EXISTS `semantic_model_source_rel` (
+ `id` BIGINT AUTO_INCREMENT COMMENT 'auto increment id',
+ `metalake_id` BIGINT NOT NULL COMMENT 'containing metalake id',
+ `semantic_model_id` BIGINT NOT NULL COMMENT 'containing semantic model id',
+ `dataset_name` CLOB NOT NULL COMMENT 'exact dataset name in the current
definition',
+ `source_type` VARCHAR(64) NOT NULL COMMENT 'TABLE or VIEW',
+ `source_id` BIGINT NOT NULL COMMENT 'resolved source object id',
+ `deleted_at` BIGINT NOT NULL DEFAULT 0 COMMENT 'association retired at',
+ PRIMARY KEY (`id`)
+) COMMENT='current semantic model dataset source associations';
+CREATE INDEX IF NOT EXISTS `idx_smsr_model` ON `semantic_model_source_rel`
(`semantic_model_id`, `deleted_at`);
+CREATE INDEX IF NOT EXISTS `idx_smsr_source` ON `semantic_model_source_rel`
(`metalake_id`, `source_type`, `source_id`, `deleted_at`);
+CREATE INDEX IF NOT EXISTS `idx_smsr_deleted_at` ON
`semantic_model_source_rel` (`deleted_at`);
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index 93b12fd127..56ec88cd70 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -768,3 +768,18 @@ CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
KEY `idx_tsm_tid_pid_sid` (`table_id`, `policy_id`, `snapshot_id`),
KEY `idx_tsm_created_at` (`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
snapshot metrics metadata';
+
+CREATE TABLE IF NOT EXISTS `semantic_model_source_rel` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing metalake
id',
+ `semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing
semantic model id',
+ `dataset_name` MEDIUMTEXT NOT NULL COMMENT 'exact dataset name in the
current definition',
+ `source_type` VARCHAR(64) NOT NULL COMMENT 'TABLE or VIEW',
+ `source_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'resolved source object
id',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'association
retired at',
+ PRIMARY KEY (`id`),
+ KEY `idx_smsr_model` (`semantic_model_id`, `deleted_at`),
+ KEY `idx_smsr_source` (`metalake_id`, `source_type`, `source_id`,
`deleted_at`),
+ KEY `idx_smsr_deleted_at` (`deleted_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
+ COMMENT 'current semantic model dataset source associations';
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 90cab77bea..bf43eae536 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
@@ -697,3 +697,18 @@ SET @ddl = IF(EXISTS(SELECT 1 FROM
information_schema.statistics
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
+
+CREATE TABLE IF NOT EXISTS `semantic_model_source_rel` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing metalake
id',
+ `semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing
semantic model id',
+ `dataset_name` MEDIUMTEXT NOT NULL COMMENT 'exact dataset name in the
current definition',
+ `source_type` VARCHAR(64) NOT NULL COMMENT 'TABLE or VIEW',
+ `source_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'resolved source object
id',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'association
retired at',
+ PRIMARY KEY (`id`),
+ KEY `idx_smsr_model` (`semantic_model_id`, `deleted_at`),
+ KEY `idx_smsr_source` (`metalake_id`, `source_type`, `source_id`,
`deleted_at`),
+ KEY `idx_smsr_deleted_at` (`deleted_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
+ COMMENT 'current semantic model dataset source associations';
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index f9e6bfe7dc..1612123497 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -1339,3 +1339,24 @@ COMMENT ON COLUMN table_snapshot_metrics.policy_id IS
'policy id from policy_met
COMMENT ON COLUMN table_snapshot_metrics.snapshot_id IS 'Iceberg snapshot id';
COMMENT ON COLUMN table_snapshot_metrics.metrics_value IS 'metrics value as
JSON';
COMMENT ON COLUMN table_snapshot_metrics.created_at IS 'creation timestamp in
milliseconds';
+
+CREATE TABLE IF NOT EXISTS semantic_model_source_rel (
+ id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+ metalake_id BIGINT NOT NULL,
+ semantic_model_id BIGINT NOT NULL,
+ dataset_name TEXT NOT NULL,
+ source_type VARCHAR(64) NOT NULL,
+ source_id BIGINT NOT NULL,
+ deleted_at BIGINT NOT NULL DEFAULT 0
+);
+CREATE INDEX IF NOT EXISTS idx_smsr_model ON semantic_model_source_rel
(semantic_model_id, deleted_at);
+CREATE INDEX IF NOT EXISTS idx_smsr_source ON semantic_model_source_rel
(metalake_id, source_type, source_id, deleted_at);
+CREATE INDEX IF NOT EXISTS idx_smsr_deleted_at ON semantic_model_source_rel
(deleted_at);
+COMMENT ON TABLE semantic_model_source_rel IS 'current semantic model dataset
source associations';
+COMMENT ON COLUMN semantic_model_source_rel.id IS 'auto increment id';
+COMMENT ON COLUMN semantic_model_source_rel.metalake_id IS 'containing
metalake id';
+COMMENT ON COLUMN semantic_model_source_rel.semantic_model_id IS 'containing
semantic model id';
+COMMENT ON COLUMN semantic_model_source_rel.dataset_name IS 'exact dataset
name in the current definition';
+COMMENT ON COLUMN semantic_model_source_rel.source_type IS 'TABLE or VIEW';
+COMMENT ON COLUMN semantic_model_source_rel.source_id IS 'resolved source
object id';
+COMMENT ON COLUMN semantic_model_source_rel.deleted_at IS 'association retired
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 28906b2de3..3424170694 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
@@ -344,3 +344,24 @@ COMMENT ON COLUMN policy_meta.occ_version IS 'policy
optimistic concurrency vers
-- Role deletion soft-deletes relation rows by role_id.
CREATE INDEX IF NOT EXISTS user_role_rel_idx_role_id ON user_role_rel
(role_id);
CREATE INDEX IF NOT EXISTS group_role_rel_idx_role_id ON group_role_rel
(role_id);
+
+CREATE TABLE IF NOT EXISTS semantic_model_source_rel (
+ id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+ metalake_id BIGINT NOT NULL,
+ semantic_model_id BIGINT NOT NULL,
+ dataset_name TEXT NOT NULL,
+ source_type VARCHAR(64) NOT NULL,
+ source_id BIGINT NOT NULL,
+ deleted_at BIGINT NOT NULL DEFAULT 0
+);
+CREATE INDEX IF NOT EXISTS idx_smsr_model ON semantic_model_source_rel
(semantic_model_id, deleted_at);
+CREATE INDEX IF NOT EXISTS idx_smsr_source ON semantic_model_source_rel
(metalake_id, source_type, source_id, deleted_at);
+CREATE INDEX IF NOT EXISTS idx_smsr_deleted_at ON semantic_model_source_rel
(deleted_at);
+COMMENT ON TABLE semantic_model_source_rel IS 'current semantic model dataset
source associations';
+COMMENT ON COLUMN semantic_model_source_rel.id IS 'auto increment id';
+COMMENT ON COLUMN semantic_model_source_rel.metalake_id IS 'containing
metalake id';
+COMMENT ON COLUMN semantic_model_source_rel.semantic_model_id IS 'containing
semantic model id';
+COMMENT ON COLUMN semantic_model_source_rel.dataset_name IS 'exact dataset
name in the current definition';
+COMMENT ON COLUMN semantic_model_source_rel.source_type IS 'TABLE or VIEW';
+COMMENT ON COLUMN semantic_model_source_rel.source_id IS 'resolved source
object id';
+COMMENT ON COLUMN semantic_model_source_rel.deleted_at IS 'association retired
at';
