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 4b6216d284 [#14140] feat(core): Persist Semantic Model metric
identities (#14141)
4b6216d284 is described below
commit 4b6216d284ec34bf7d192d732efd8ef18055727f
Author: mchades <[email protected]>
AuthorDate: Tue Oct 27 20:43:05 2026 +0800
[#14140] feat(core): Persist Semantic Model metric identities (#14141)
### What changes were proposed in this pull request?
Add `semantic_model_metric_meta` and reconcile metric identities in the
existing model transaction. Exact-name matches retain IDs; removed names
are retired, and reintroduced names receive new IDs. Integrate
model/parent deletion and GC across H2, MySQL, and PostgreSQL.
Skip identity reconciliation when metric names are unchanged, while
preserving duplicate-name validation and complete definition snapshots.
Insert new identities in batches of up to 1,000 rows.
### Why are the changes needed?
Provide the identity foundation for schema-level metric discovery
described in #14038, under #14036.
Builds on #14139, now merged into `main`.
Fixes #14140.
### Does this PR introduce _any_ user-facing change?
No public API changes.
### How was this patch tested?
30 focused H2 tests passed: 16 metric-identity tests and 14 Semantic
Model storage tests. Coverage includes definition replacement, stable
IDs, whitespace-only names, duplicate rejection, 1,001-row insertion,
and rollback across batches.
Ran `spotlessApply` and `:core:coreH2Test` with
`TestSemanticModelMetricIdentities` and `TestSemanticModelJDBCBackend`
filters. Compilation and `git diff --check` passed.
Local MySQL/PostgreSQL tests were not rerun because the container IPs
remain unreachable; these backends are covered by the dedicated Fork CI
lanes.
---
.../mapper/SemanticModelMetricMapper.java | 85 +++
.../SemanticModelMetricSQLProviderFactory.java | 96 ++++
.../provider/DefaultMapperPackageProvider.java | 2 +
.../base/SemanticModelMetricBaseSQLProvider.java | 114 ++++
.../SemanticModelMetricPostgreSQLProvider.java | 43 ++
.../relational/po/SemanticModelMetricPO.java | 69 +++
.../relational/service/CatalogMetaService.java | 7 +-
.../relational/service/MetalakeMetaService.java | 7 +-
.../relational/service/SchemaMetaService.java | 7 +-
.../service/SemanticModelMetaService.java | 102 +++-
.../service/TestSemanticModelMetaService.java | 38 +-
.../service/TestSemanticModelMetricIdentities.java | 628 +++++++++++++++++++++
design-docs/gravitino-semantic-model-design.md | 29 +-
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 | 23 +
.../upgrade-1.3.0-to-2.0.0-postgresql.sql | 23 +
19 files changed, 1322 insertions(+), 11 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelMetricMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelMetricMapper.java
new file mode 100644
index 0000000000..3bf1319c92
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelMetricMapper.java
@@ -0,0 +1,85 @@
+/*
+ * 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.SemanticModelMetricPO;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+import org.apache.ibatis.annotations.UpdateProvider;
+
+/** A MyBatis mapper for identities of metrics contained in Semantic Models. */
+public interface SemanticModelMetricMapper {
+ /** The metric identity table name. */
+ String TABLE_NAME = "semantic_model_metric_meta";
+
+ /** Inserts new metric identities in one statement. */
+ @InsertProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "insertSemanticModelMetrics")
+ void insertSemanticModelMetrics(@Param("metrics")
List<SemanticModelMetricPO> metricPOs);
+
+ /** Locks and reads the live identities of a Semantic Model. */
+ @SelectProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "selectSemanticModelMetricsByModelIdForUpdate")
+ List<SemanticModelMetricPO> selectSemanticModelMetricsByModelIdForUpdate(
+ @Param("semanticModelId") Long semanticModelId);
+
+ /** Retires selected live identities belonging to a Semantic Model. */
+ @UpdateProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "softDeleteSemanticModelMetricsByIds")
+ Integer softDeleteSemanticModelMetricsByIds(
+ @Param("semanticModelId") Long semanticModelId, @Param("metricIds")
List<Long> metricIds);
+
+ /** Retires all live metric identities of a Semantic Model. */
+ @UpdateProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "softDeleteSemanticModelMetricsBySemanticModelId")
+ Integer softDeleteSemanticModelMetricsBySemanticModelId(
+ @Param("semanticModelId") Long semanticModelId);
+
+ /** Retires live metric identities under schemas. */
+ @UpdateProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "softDeleteSemanticModelMetricsBySchemaIds")
+ Integer softDeleteSemanticModelMetricsBySchemaIds(@Param("schemaIds")
List<Long> schemaIds);
+
+ /** Retires live metric identities under a catalog. */
+ @UpdateProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "softDeleteSemanticModelMetricsByCatalogId")
+ Integer softDeleteSemanticModelMetricsByCatalogId(@Param("catalogId") Long
catalogId);
+
+ /** Retires live metric identities under a metalake. */
+ @UpdateProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "softDeleteSemanticModelMetricsByMetalakeId")
+ Integer softDeleteSemanticModelMetricsByMetalakeId(@Param("metalakeId") Long
metalakeId);
+
+ /** Permanently deletes retired identities older than a timeline. */
+ @DeleteProvider(
+ type = SemanticModelMetricSQLProviderFactory.class,
+ method = "deleteSemanticModelMetricsByLegacyTimeline")
+ Integer deleteSemanticModelMetricsByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelMetricSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelMetricSQLProviderFactory.java
new file mode 100644
index 0000000000..4649b0bcfd
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SemanticModelMetricSQLProviderFactory.java
@@ -0,0 +1,96 @@
+/*
+ * 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.SemanticModelMetricBaseSQLProvider;
+import
org.apache.gravitino.storage.relational.mapper.provider.postgresql.SemanticModelMetricPostgreSQLProvider;
+import org.apache.gravitino.storage.relational.po.SemanticModelMetricPO;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/** Selects database-specific SQL providers for Semantic Model metric
identities. */
+public class SemanticModelMetricSQLProviderFactory {
+ private static final Map<JDBCBackendType,
SemanticModelMetricBaseSQLProvider> PROVIDERS =
+ ImmutableMap.of(
+ JDBCBackendType.MYSQL, new SemanticModelMetricBaseSQLProvider(),
+ JDBCBackendType.H2, new SemanticModelMetricBaseSQLProvider(),
+ JDBCBackendType.POSTGRESQL, new
SemanticModelMetricPostgreSQLProvider());
+
+ /** Returns the SQL provider for the configured relational backend. */
+ public static SemanticModelMetricBaseSQLProvider getProvider() {
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+ return PROVIDERS.get(JDBCBackendType.fromString(databaseId));
+ }
+
+ /** Returns SQL for inserting new metric identities in one statement. */
+ public static String insertSemanticModelMetrics(
+ @Param("metrics") List<SemanticModelMetricPO> metricPOs) {
+ return getProvider().insertSemanticModelMetrics(metricPOs);
+ }
+
+ /** Locks and reads the live identities of a Semantic Model. */
+ public static String selectSemanticModelMetricsByModelIdForUpdate(
+ @Param("semanticModelId") Long semanticModelId) {
+ return
getProvider().selectSemanticModelMetricsByModelIdForUpdate(semanticModelId);
+ }
+
+ /** Retires selected live identities belonging to a Semantic Model. */
+ public static String softDeleteSemanticModelMetricsByIds(
+ @Param("semanticModelId") Long semanticModelId, @Param("metricIds")
List<Long> metricIds) {
+ return getProvider().softDeleteSemanticModelMetricsByIds(semanticModelId,
metricIds);
+ }
+
+ /** Retires all live metric identities of a Semantic Model. */
+ public static String softDeleteSemanticModelMetricsBySemanticModelId(
+ @Param("semanticModelId") Long semanticModelId) {
+ return
getProvider().softDeleteSemanticModelMetricsBySemanticModelId(semanticModelId);
+ }
+
+ /** Retires live metric identities under schemas. */
+ public static String softDeleteSemanticModelMetricsBySchemaIds(
+ @Param("schemaIds") List<Long> schemaIds) {
+ return getProvider().softDeleteSemanticModelMetricsBySchemaIds(schemaIds);
+ }
+
+ /** Retires live metric identities under a catalog. */
+ public static String softDeleteSemanticModelMetricsByCatalogId(
+ @Param("catalogId") Long catalogId) {
+ return getProvider().softDeleteSemanticModelMetricsByCatalogId(catalogId);
+ }
+
+ /** Retires live metric identities under a metalake. */
+ public static String softDeleteSemanticModelMetricsByMetalakeId(
+ @Param("metalakeId") Long metalakeId) {
+ return
getProvider().softDeleteSemanticModelMetricsByMetalakeId(metalakeId);
+ }
+
+ /** Permanently deletes retired identities older than a timeline. */
+ public static String deleteSemanticModelMetricsByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return
getProvider().deleteSemanticModelMetricsByLegacyTimeline(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 ae8773fd75..32f289750c 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
@@ -44,6 +44,7 @@ import
org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
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.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -87,6 +88,7 @@ public class DefaultMapperPackageProvider implements
MapperPackageProvider {
SchemaMetaMapper.class,
SecurableObjectMapper.class,
SemanticModelMetaMapper.class,
+ SemanticModelMetricMapper.class,
SemanticModelVersionInfoMapper.class,
StatisticMetaMapper.class,
TableColumnMapper.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SemanticModelMetricBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SemanticModelMetricBaseSQLProvider.java
new file mode 100644
index 0000000000..f424a31fb9
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SemanticModelMetricBaseSQLProvider.java
@@ -0,0 +1,114 @@
+/*
+ * 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.SemanticModelMetricMapper.TABLE_NAME;
+
+import java.util.List;
+import org.apache.gravitino.storage.relational.po.SemanticModelMetricPO;
+import org.apache.ibatis.annotations.Param;
+
+/** Provides MySQL-compatible SQL for Semantic Model metric identities. */
+public class SemanticModelMetricBaseSQLProvider {
+
+ /** Returns SQL for inserting new identities in one statement. */
+ public String insertSemanticModelMetrics(
+ @Param("metrics") List<SemanticModelMetricPO> metricPOs) {
+ return "<script>INSERT INTO "
+ + TABLE_NAME
+ + " (metric_id, metalake_id, catalog_id, schema_id, semantic_model_id,
metric_name, deleted_at)"
+ + " VALUES "
+ + "<foreach collection='metrics' item='metric' separator=','>"
+ + "(#{metric.metricId}, #{metric.metalakeId}, #{metric.catalogId},"
+ + " #{metric.schemaId}, #{metric.semanticModelId},
#{metric.metricName}, #{metric.deletedAt})"
+ + "</foreach></script>";
+ }
+
+ /** Returns a locking read of live identities after the containing model's
concurrency check. */
+ public String selectSemanticModelMetricsByModelIdForUpdate(
+ @Param("semanticModelId") Long semanticModelId) {
+ return "SELECT metric_id as metricId, metalake_id as metalakeId,
catalog_id as catalogId,"
+ + " schema_id as schemaId, semantic_model_id as semanticModelId,
metric_name as metricName,"
+ + " deleted_at as deletedAt FROM "
+ + TABLE_NAME
+ + " WHERE semantic_model_id = #{semanticModelId} AND deleted_at = 0
FOR UPDATE";
+ }
+
+ /** Returns SQL for retiring selected live identities of a model. */
+ public String softDeleteSemanticModelMetricsByIds(
+ @Param("semanticModelId") Long semanticModelId, @Param("metricIds")
List<Long> metricIds) {
+ return "<script>"
+ + softDeleteBy(
+ "semantic_model_id = #{semanticModelId} AND metric_id IN ("
+ + "<foreach collection='metricIds' item='metricId'
separator=','>"
+ + "#{metricId}</foreach>)")
+ + "</script>";
+ }
+
+ /** Returns SQL for retiring all live identities of a model. */
+ public String softDeleteSemanticModelMetricsBySemanticModelId(
+ @Param("semanticModelId") Long semanticModelId) {
+ return softDeleteBy("semantic_model_id = #{semanticModelId}");
+ }
+
+ /** Returns SQL for retiring live identities under schemas. */
+ public String softDeleteSemanticModelMetricsBySchemaIds(
+ @Param("schemaIds") List<Long> schemaIds) {
+ return "<script>"
+ + softDeleteBy(
+ "schema_id IN ("
+ + "<foreach collection='schemaIds' item='schemaId'
separator=','>"
+ + "#{schemaId}</foreach>)")
+ + "</script>";
+ }
+
+ /** Returns SQL for retiring live identities under a catalog. */
+ public String softDeleteSemanticModelMetricsByCatalogId(@Param("catalogId")
Long catalogId) {
+ return softDeleteBy("catalog_id = #{catalogId}");
+ }
+
+ /** Returns SQL for retiring live identities under a metalake. */
+ public String
softDeleteSemanticModelMetricsByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ return softDeleteBy("metalake_id = #{metalakeId}");
+ }
+
+ /** Returns SQL for permanently deleting old retired identities. */
+ public String deleteSemanticModelMetricsByLegacyTimeline(
+ @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/SemanticModelMetricPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SemanticModelMetricPostgreSQLProvider.java
new file mode 100644
index 0000000000..6eb59bf351
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SemanticModelMetricPostgreSQLProvider.java
@@ -0,0 +1,43 @@
+/*
+ * 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.SemanticModelMetricMapper.TABLE_NAME;
+
+import
org.apache.gravitino.storage.relational.mapper.provider.base.SemanticModelMetricBaseSQLProvider;
+import org.apache.ibatis.annotations.Param;
+
+/** Provides PostgreSQL SQL for Semantic Model metric identities. */
+public class SemanticModelMetricPostgreSQLProvider extends
SemanticModelMetricBaseSQLProvider {
+
+ @Override
+ public String deleteSemanticModelMetricsByLegacyTimeline(
+ @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ return "DELETE FROM "
+ + TABLE_NAME
+ + " WHERE metric_id IN (SELECT metric_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/po/SemanticModelMetricPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/SemanticModelMetricPO.java
new file mode 100644
index 0000000000..453103bb8e
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/po/SemanticModelMetricPO.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.po;
+
+import com.google.common.base.Preconditions;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+import org.apache.commons.lang3.StringUtils;
+
+/** The persistent identity and lifecycle state of a metric contained in a
Semantic Model. */
+@EqualsAndHashCode
+@Getter
+@ToString
+@Accessors(fluent = true)
+public class SemanticModelMetricPO {
+ private Long metricId;
+ private Long metalakeId;
+ private Long catalogId;
+ private Long schemaId;
+ private Long semanticModelId;
+ private String metricName;
+ private Long deletedAt;
+
+ /** Creates an empty persistent object for MyBatis. */
+ public SemanticModelMetricPO() {}
+
+ @lombok.Builder(setterPrefix = "with")
+ private SemanticModelMetricPO(
+ Long metricId,
+ Long metalakeId,
+ Long catalogId,
+ Long schemaId,
+ Long semanticModelId,
+ String metricName,
+ Long deletedAt) {
+ Preconditions.checkArgument(metricId != null, "Metric id is required");
+ Preconditions.checkArgument(metalakeId != null, "Metalake id is required");
+ Preconditions.checkArgument(catalogId != null, "Catalog id is required");
+ Preconditions.checkArgument(schemaId != null, "Schema id is required");
+ Preconditions.checkArgument(semanticModelId != null, "Semantic Model id is
required");
+ Preconditions.checkArgument(StringUtils.isNotEmpty(metricName), "Metric
name cannot be empty");
+ Preconditions.checkArgument(deletedAt != null, "Deleted at is required");
+ this.metricId = metricId;
+ this.metalakeId = metalakeId;
+ this.catalogId = catalogId;
+ this.schemaId = schemaId;
+ this.semanticModelId = semanticModelId;
+ this.metricName = metricName;
+ this.deletedAt = deletedAt;
+ }
+}
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 8ef3740548..45b12caf2a 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
@@ -49,6 +49,7 @@ import
org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
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.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -362,7 +363,11 @@ public class CatalogMetaService {
() ->
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
- mapper ->
mapper.softDeleteSemanticModelVersionsByCatalogId(catalogId)));
+ mapper ->
mapper.softDeleteSemanticModelVersionsByCatalogId(catalogId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.softDeleteSemanticModelMetricsByCatalogId(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 c4eb17fe3f..02f94121ae 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
@@ -55,6 +55,7 @@ import
org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
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.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -346,7 +347,11 @@ public class MetalakeMetaService {
() ->
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
- mapper ->
mapper.softDeleteSemanticModelVersionsByMetalakeId(metalakeId)));
+ mapper ->
mapper.softDeleteSemanticModelVersionsByMetalakeId(metalakeId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.softDeleteSemanticModelMetricsByMetalakeId(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 951f3c799b..01ce984a66 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
@@ -57,6 +57,7 @@ import
org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
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.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
@@ -388,7 +389,11 @@ public class SchemaMetaService {
() ->
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
- mapper ->
mapper.softDeleteSemanticModelVersionsBySchemaIds(schemaIds.get())));
+ mapper ->
mapper.softDeleteSemanticModelVersionsBySchemaIds(schemaIds.get())),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.softDeleteSemanticModelMetricsBySchemaIds(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 ce8d32246e..582481f4c7 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
@@ -24,15 +24,21 @@ import static
org.apache.gravitino.storage.relational.po.SemanticModelPO.fromSem
import static
org.apache.gravitino.storage.relational.po.SemanticModelPO.initializeSemanticModelPO;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
+import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
@@ -40,9 +46,14 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.OptimisticLockException;
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.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.SemanticModelVersionInfoMapper;
+import org.apache.gravitino.storage.relational.po.SemanticModelMetricPO;
import org.apache.gravitino.storage.relational.po.SemanticModelPO;
import org.apache.gravitino.storage.relational.po.SemanticModelVersionInfoPO;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
@@ -57,6 +68,7 @@ import org.slf4j.LoggerFactory;
public class SemanticModelMetaService {
private static final Logger LOG =
LoggerFactory.getLogger(SemanticModelMetaService.class);
+ private static final int METRIC_INSERT_BATCH_SIZE = 1000;
private static final SemanticModelMetaService INSTANCE = new
SemanticModelMetaService();
private final BasePOStorageOps<SemanticModelPO, SemanticModelMetaMapper> ops;
@@ -127,7 +139,7 @@ public class SemanticModelMetaService {
}
/**
- * Inserts or overwrites a Semantic Model identity and its snapshot
atomically.
+ * Inserts or overwrites a Semantic Model, its snapshot, and its metric
identities atomically.
*
* @param semanticModelEntity The Semantic Model entity.
* @param overwrite Whether to overwrite rows for the same stable ID or
natural key.
@@ -173,6 +185,7 @@ public class SemanticModelMetaService {
}
persistedPO.set(replacementPO);
}),
+ () -> reconcileMetricIdentities(persistedPO.get(),
semanticModelEntity.definition()),
() ->
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
@@ -250,6 +263,15 @@ public class SemanticModelMetaService {
throw semanticModelWriteFailure(identifier, oldSemanticModelPO);
}
},
+ () -> {
+ if (!Objects.equals(
+
oldSemanticModelPO.getSemanticModelVersionInfoPO().metrics(),
+
newSemanticModelPO.getSemanticModelVersionInfoPO().metrics())
+ && !metricNames(oldSemanticModelEntity.definition())
+ .equals(metricNames(newEntity.definition()))) {
+ reconcileMetricIdentities(newSemanticModelPO,
newEntity.definition());
+ }
+ },
() ->
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
@@ -273,7 +295,7 @@ public class SemanticModelMetaService {
}
/**
- * Soft-deletes a Semantic Model identity and all of its version snapshots.
+ * Soft-deletes a Semantic Model, its metric identities, and all of its
version snapshots.
*
* @param identifier The Semantic Model identifier.
* @return {@code true} when an active identity was deleted.
@@ -290,7 +312,8 @@ public class SemanticModelMetaService {
}
/**
- * Permanently deletes soft-deleted Semantic Model identities and snapshots
older than a timeline.
+ * Permanently deletes retired 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.
@@ -300,6 +323,10 @@ public class SemanticModelMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteSemanticModelMetasByLegacyTimeline")
public int deleteSemanticModelMetasByLegacyTimeline(Long legacyTimeline, int
limit) {
+ int metricDeletedCount =
+ SessionUtils.doWithCommitAndFetchResult(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.deleteSemanticModelMetricsByLegacyTimeline(legacyTimeline, limit));
int versionDeletedCount =
SessionUtils.doWithCommitAndFetchResult(
SemanticModelVersionInfoMapper.class,
@@ -308,7 +335,7 @@ public class SemanticModelMetaService {
SessionUtils.doWithCommitAndFetchResult(
SemanticModelMetaMapper.class,
mapper ->
mapper.deleteSemanticModelMetasByLegacyTimeline(legacyTimeline, limit));
- return versionDeletedCount + metaDeletedCount;
+ return metricDeletedCount + versionDeletedCount + metaDeletedCount;
}
/**
@@ -385,11 +412,78 @@ public class SemanticModelMetaService {
SessionUtils.doWithoutCommit(
SemanticModelVersionInfoMapper.class,
mapper ->
mapper.softDeleteSemanticModelVersionsBySemanticModelId(semanticModelId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.softDeleteSemanticModelMetricsBySemanticModelId(semanticModelId)),
() ->
EntityChangeLogWriter.append(
semanticModelIdent, Entity.EntityType.SEMANTIC_MODEL,
OperateType.DROP));
}
+ private static void reconcileMetricIdentities(
+ SemanticModelPO model, SemanticModelDefinition definition) {
+ Set<String> names = metricNames(definition);
+
+ // The caller has already inserted or version-checked the model in this
transaction. A locking
+ // read also observes the latest identities on backends using
repeatable-read isolation.
+ SessionUtils.doWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper -> {
+ Map<String, SemanticModelMetricPO> live =
+ mapper
+
.selectSemanticModelMetricsByModelIdForUpdate(model.getSemanticModelId())
+ .stream()
+ .collect(
+ Collectors.toMap(SemanticModelMetricPO::metricName,
Function.identity()));
+ List<String> addedNames = new ArrayList<>();
+ for (String name : names) {
+ if (live.remove(name) == null) {
+ addedNames.add(name);
+ }
+ }
+ if (!addedNames.isEmpty()) {
+ IdGenerator generator = GravitinoEnv.getInstance().idGenerator();
+ Preconditions.checkState(
+ generator != null, "Metric identity ID generator is not
initialized");
+ List<SemanticModelMetricPO> addedMetrics = new
ArrayList<>(addedNames.size());
+ for (String name : addedNames) {
+ addedMetrics.add(
+ SemanticModelMetricPO.builder()
+ .withMetricId(generator.nextId())
+ .withMetalakeId(model.getMetalakeId())
+ .withCatalogId(model.getCatalogId())
+ .withSchemaId(model.getSchemaId())
+ .withSemanticModelId(model.getSemanticModelId())
+ .withMetricName(name)
+ .withDeletedAt(0L)
+ .build());
+ }
+ Lists.partition(addedMetrics, METRIC_INSERT_BATCH_SIZE)
+ .forEach(mapper::insertSemanticModelMetrics);
+ }
+ if (!live.isEmpty()) {
+ mapper.softDeleteSemanticModelMetricsByIds(
+ model.getSemanticModelId(),
+ live.values().stream()
+ .map(SemanticModelMetricPO::metricId)
+ .collect(Collectors.toList()));
+ }
+ });
+ }
+
+ private static Set<String> metricNames(SemanticModelDefinition definition) {
+ Set<String> names = new LinkedHashSet<>();
+ Metric[] metrics = definition.metrics();
+ if (metrics != null) {
+ for (Metric metric : metrics) {
+ Preconditions.checkArgument(
+ names.add(metric.name()), "Duplicate metric name: %s",
metric.name());
+ }
+ }
+ return names;
+ }
+
private SemanticModelPO updateSemanticModelPO(
SemanticModelPO oldSemanticModelPO, SemanticModelEntity newEntity) {
int previousVersion =
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelMetaService.java
index d89969d301..e2f4d94f01 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelMetaService.java
@@ -52,6 +52,9 @@ import org.apache.gravitino.meta.SemanticModelEntity;
import org.apache.gravitino.semantic.AIContext;
import org.apache.gravitino.semantic.CustomExtension;
import org.apache.gravitino.semantic.Dataset;
+import org.apache.gravitino.semantic.DialectExpression;
+import org.apache.gravitino.semantic.Expression;
+import org.apache.gravitino.semantic.Metric;
import org.apache.gravitino.semantic.SemanticModelDefinition;
import org.apache.gravitino.storage.RandomIdGenerator;
import
org.apache.gravitino.storage.relational.EntityChangeLogNameIdentifierCodec;
@@ -59,6 +62,7 @@ import
org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelMetricMapper;
import
org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper;
import org.apache.gravitino.storage.relational.po.SchemaPO;
import org.apache.gravitino.storage.relational.po.SemanticModelPO;
@@ -306,6 +310,11 @@ public class TestSemanticModelMetaService extends
TestJDBCBackend {
assertEquals(0, countSemanticModelIdentities(semanticModel.id()));
assertTrue(listSemanticModelVersions(semanticModel.id()).isEmpty());
+ assertTrue(
+ SessionUtils.getWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.selectSemanticModelMetricsByModelIdForUpdate(semanticModel.id()))
+ .isEmpty());
}
@TestTemplate
@@ -362,6 +371,11 @@ public class TestSemanticModelMetaService extends
TestJDBCBackend {
Map<Integer, VersionState> versions =
listSemanticModelVersions(original.id());
assertEquals(1, versions.size());
assertTrue(versions.get(1).deletedAt > 0L);
+ assertTrue(
+ SessionUtils.getWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.selectSemanticModelMetricsByModelIdForUpdate(original.id()))
+ .isEmpty());
}
@TestTemplate
@@ -496,7 +510,7 @@ public class TestSemanticModelMetaService extends
TestJDBCBackend {
int deleted =
SemanticModelMetaService.getInstance()
.deleteSemanticModelMetasByLegacyTimeline(Instant.now().toEpochMilli() + 1000,
100);
- assertEquals(4, deleted);
+ assertEquals(7, deleted);
assertEquals(0, listSemanticModelVersions(finalCurrent.id()).size());
assertEquals(0, countSemanticModelIdentities(finalCurrent.id()));
}
@@ -518,6 +532,22 @@ public class TestSemanticModelMetaService extends
TestJDBCBackend {
.withSource(NameIdentifier.of("sales", "mart",
datasetName))
.build()
})
+ .withMetrics(
+ new Metric[] {
+ Metric.builder()
+ .withName("metric_" + datasetName)
+ .withExpression(
+ Expression.builder()
+ .withDialects(
+ new DialectExpression[] {
+ DialectExpression.builder()
+ .withDialect("ansi")
+ .withExpression("COUNT(*)")
+ .build()
+ })
+ .build())
+ .build()
+ })
.withCustomExtensions(
new CustomExtension[] {
CustomExtension.builder()
@@ -576,6 +606,12 @@ public class TestSemanticModelMetaService extends
TestJDBCBackend {
SemanticModelVersionInfoMapper.class,
mapper ->
mapper.softDeleteSemanticModelVersionsBySchemaIds(
+ List.of(observedSchemaPO.getSchemaId()))),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
+ mapper.softDeleteSemanticModelMetricsBySchemaIds(
List.of(observedSchemaPO.getSchemaId()))));
return null;
} catch (Throwable throwable) {
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelMetricIdentities.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelMetricIdentities.java
new file mode 100644
index 0000000000..09376b147a
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSemanticModelMetricIdentities.java
@@ -0,0 +1,628 @@
+/*
+ * 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.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+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.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.SemanticModelEntity;
+import org.apache.gravitino.semantic.Dataset;
+import org.apache.gravitino.semantic.DialectExpression;
+import org.apache.gravitino.semantic.Expression;
+import org.apache.gravitino.semantic.Metric;
+import org.apache.gravitino.semantic.SemanticModelDefinition;
+import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelMetricMapper;
+import
org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper;
+import org.apache.gravitino.storage.relational.po.SemanticModelMetricPO;
+import org.apache.gravitino.storage.relational.po.SemanticModelPO;
+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.TestTemplate;
+import org.mockito.MockedStatic;
+
+/** Tests metric identity rules, atomic writes, and cleanup on every
relational backend. */
+public class TestSemanticModelMetricIdentities extends TestJDBCBackend {
+
+ @TestTemplate
+ void testExactNamesContentChangesAndReintroduction() throws Exception {
+ Namespace namespace = parents();
+ String longName = "metric_" + "x".repeat(300);
+ SemanticModelEntity original = model(namespace, "sales", "Revenue",
"revenue", "收入'", longName);
+ backend.insert(original, false);
+ Map<String, Long> initial = liveIds(original.id());
+ assertEquals(Set.of("Revenue", "revenue", "收入'", longName),
initial.keySet());
+ assertEquals(4, Set.copyOf(initial.values()).size());
+ SemanticModelPO stored =
service().getSemanticModelPOByIdentifier(original.nameIdentifier());
+ for (SemanticModelMetricPO metric : live(original.id())) {
+ assertEquals(stored.getMetalakeId(), metric.metalakeId());
+ assertEquals(stored.getCatalogId(), metric.catalogId());
+ assertEquals(stored.getSchemaId(), metric.schemaId());
+ assertEquals(original.id(), metric.semanticModelId());
+ assertEquals(0L, metric.deletedAt());
+ }
+
+ Metric[] reordered =
+ Arrays.stream(new String[] {longName, "收入'", "revenue", "Revenue"})
+ .map(
+ name ->
+ Metric.builder()
+ .withName(name)
+ .withExpression(expression("SUM(orders.amount) * 2"))
+ .withDescription("Updated description")
+ .build())
+ .toArray(Metric[]::new);
+ SemanticModelEntity renamed = copy(original, original.id(), "renamed",
reordered);
+ try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class,
CALLS_REAL_METHODS)) {
+ backend.update(original.nameIdentifier(), original.type(), old ->
renamed);
+ sessions.verify(
+ () ->
SessionUtils.doWithoutCommit(eq(SemanticModelMetricMapper.class), any()),
never());
+ }
+ assertEquals(renamed, backend.get(renamed.nameIdentifier(),
renamed.type()));
+ assertEquals(
+ 2,
service().getSemanticModelPOByIdentifier(renamed.nameIdentifier()).getCurrentVersion());
+ assertEquals(2, count("semantic_model_version_info", original.id()));
+ assertEquals(initial, liveIds(original.id()));
+ assertEquals(4, count("semantic_model_metric_meta", original.id()));
+
+ SemanticModelEntity reduced = copy(renamed, renamed.id(), renamed.name(),
metrics("Revenue"));
+ backend.update(renamed.nameIdentifier(), renamed.type(), old -> reduced);
+ assertEquals(Map.of("Revenue", initial.get("Revenue")),
liveIds(original.id()));
+ SemanticModelEntity reintroduced =
+ copy(reduced, reduced.id(), reduced.name(), metrics("Revenue",
"revenue"));
+ backend.update(reduced.nameIdentifier(), reduced.type(), old ->
reintroduced);
+ Map<String, Long> current = liveIds(original.id());
+ assertEquals(initial.get("Revenue"), current.get("Revenue"));
+ assertNotEquals(initial.get("revenue"), current.get("revenue"));
+ assertEquals(5, count("semantic_model_metric_meta", original.id()));
+ assertSnapshotNames(reintroduced);
+ }
+
+ @TestTemplate
+ void testUnchangedMetricsSkipIdentityReconciliation() throws Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", "revenue");
+ backend.insert(original, false);
+ Map<String, Long> initial = liveIds(original.id());
+ SemanticModelEntity replacement =
+ copy(original, original.id(), original.name(),
original.definition().metrics());
+
+ try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class,
CALLS_REAL_METHODS)) {
+ backend.update(original.nameIdentifier(), original.type(), old ->
replacement);
+ sessions.verify(
+ () ->
SessionUtils.doWithoutCommit(eq(SemanticModelMetricMapper.class), any()),
never());
+ }
+
+ assertEquals(replacement, backend.get(replacement.nameIdentifier(),
replacement.type()));
+ assertEquals(
+ 2,
+
service().getSemanticModelPOByIdentifier(replacement.nameIdentifier()).getCurrentVersion());
+ assertEquals(2, count("semantic_model_version_info", original.id()));
+ assertEquals(initial, liveIds(original.id()));
+ assertEquals(1, count("semantic_model_metric_meta", original.id()));
+ }
+
+ @TestTemplate
+ void testCreateWithMetricsAcrossInsertBatches() throws Exception {
+ String[] names = IntStream.range(0, 1001).mapToObj(i -> "metric_" +
i).toArray(String[]::new);
+ SemanticModelEntity original = model(parents(), "sales", names);
+ backend.insert(original, false);
+
+ Map<String, Long> identities = liveIds(original.id());
+ assertEquals(Set.copyOf(Arrays.asList(names)), identities.keySet());
+ assertEquals(names.length, Set.copyOf(identities.values()).size());
+ assertEquals(names.length, count("semantic_model_metric_meta",
original.id()));
+ assertEquals(original, backend.get(original.nameIdentifier(),
original.type()));
+ }
+
+ @TestTemplate
+ void testReconciliationWithoutNewNamesDoesNotRequireIdGenerator() throws
Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", "kept",
"removed");
+ backend.insert(original, false);
+ long keptId = liveIds(original.id()).get("kept");
+ SemanticModelEntity reduced = copy(original, original.id(),
original.name(), metrics("kept"));
+ SemanticModelEntity imported = model(namespace, original.name(), "kept");
+ SemanticModelEntity empty = copy(reduced, reduced.id(), reduced.name(),
null);
+
+ IdGenerator originalGenerator = GravitinoEnv.getInstance().idGenerator();
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator", null,
true);
+ try {
+ backend.update(original.nameIdentifier(), original.type(), old ->
reduced);
+ assertEquals(Map.of("kept", keptId), liveIds(original.id()));
+ backend.insert(imported, true);
+ assertEquals(Map.of("kept", keptId), liveIds(original.id()));
+ backend.update(original.nameIdentifier(), original.type(), old -> empty);
+ } finally {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator",
originalGenerator, true);
+ }
+
+ assertTrue(liveIds(original.id()).isEmpty());
+ assertEquals(2, count("semantic_model_metric_meta", original.id()));
+ assertEquals(empty, backend.get(original.nameIdentifier(),
original.type()));
+ assertEquals(
+ 4,
service().getSemanticModelPOByIdentifier(original.nameIdentifier()).getCurrentVersion());
+ }
+
+ @TestTemplate
+ void testWhitespaceOnlyNamesOnCreateReplaceAndOverwrite() throws Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", " ");
+ backend.insert(original, false);
+ Map<String, Long> initial = liveIds(original.id());
+ assertEquals(Set.of(" "), initial.keySet());
+ assertSnapshotNames(backend.get(original.nameIdentifier(),
original.type()));
+
+ SemanticModelEntity replacement =
+ copy(original, original.id(), original.name(), metrics(" ", "\t"));
+ backend.update(original.nameIdentifier(), original.type(), old ->
replacement);
+ Map<String, Long> replaced = liveIds(original.id());
+ assertEquals(Set.of(" ", "\t"), replaced.keySet());
+ assertEquals(initial.get(" "), replaced.get(" "));
+ assertSnapshotNames(backend.get(original.nameIdentifier(),
original.type()));
+
+ SemanticModelEntity imported = model(namespace, original.name(), " ",
"\t", "\n");
+ backend.insert(imported, true);
+ Map<String, Long> overwritten = liveIds(original.id());
+ assertEquals(Set.of(" ", "\t", "\n"), overwritten.keySet());
+ assertEquals(replaced.get(" "), overwritten.get(" "));
+ assertEquals(replaced.get("\t"), overwritten.get("\t"));
+ assertEquals(3, Set.copyOf(overwritten.values()).size());
+ assertEquals(0, count("semantic_model_metric_meta", imported.id()));
+ SemanticModelEntity loaded = backend.get(original.nameIdentifier(),
original.type());
+ assertEquals(original.id(), loaded.id());
+ assertSnapshotNames(loaded);
+ }
+
+ @TestTemplate
+ void testAbsentEmptyAndModelScopedNames() throws Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity absent = model(namespace, "absent", (String[]) null);
+ SemanticModelEntity empty = model(namespace, "empty");
+ backend.insert(absent, false);
+ backend.insert(empty, true);
+ assertEquals(0, count("semantic_model_metric_meta", absent.id()));
+ assertEquals(0, count("semantic_model_metric_meta", empty.id()));
+ SemanticModelEntity first = model(namespace, "first", "revenue");
+ SemanticModelEntity second = model(namespace, "second", "revenue");
+ backend.insert(first, false);
+ backend.insert(second, false);
+ assertNotEquals(liveIds(first.id()).get("revenue"),
liveIds(second.id()).get("revenue"));
+ SemanticModelEntity noMetrics = copy(first, first.id(), first.name(),
null);
+ backend.update(first.nameIdentifier(), first.type(), old -> noMetrics);
+ assertTrue(liveIds(first.id()).isEmpty());
+ assertEquals(Set.of("revenue"), liveIds(second.id()).keySet());
+ }
+
+ @TestTemplate
+ void testOverwriteUsesPersistedModelAndMetricIds() throws Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", "kept",
"removed");
+ backend.insert(original, true);
+ Map<String, Long> initial = liveIds(original.id());
+ SemanticModelEntity imported = model(namespace, original.name(), "kept",
"added");
+ backend.insert(imported, true);
+ assertEquals(Set.of("kept", "added"), liveIds(original.id()).keySet());
+ assertEquals(initial.get("kept"), liveIds(original.id()).get("kept"));
+ assertEquals(0, count("semantic_model_metric_meta", imported.id()));
+ assertEquals(
+ 2,
service().getSemanticModelPOByIdentifier(original.nameIdentifier()).getCurrentVersion());
+ SemanticModelEntity loaded = backend.get(original.nameIdentifier(),
original.type());
+ assertSnapshotNames(loaded);
+ }
+
+ @TestTemplate
+ void testDropAndRecreateAllocatesNewIdentities() throws Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", "revenue");
+ backend.insert(original, false);
+ long previousId = liveIds(original.id()).get("revenue");
+ assertTrue(backend.delete(original.nameIdentifier(), original.type(),
false));
+ assertTrue(live(original.id()).isEmpty());
+ SemanticModelEntity recreated = model(namespace, original.name(),
"revenue");
+ backend.insert(recreated, false);
+ assertNotEquals(previousId, liveIds(recreated.id()).get("revenue"));
+ assertEquals(1, count("semantic_model_metric_meta", original.id()));
+ }
+
+ @TestTemplate
+ void testDuplicateNamesRollBackCreateAndReplacement() throws Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity duplicate = model(namespace, "duplicate", "same",
"same");
+ assertThrows(IllegalArgumentException.class, () ->
backend.insert(duplicate, false));
+ assertEquals(0, count("semantic_model_meta", duplicate.id()));
+ assertEquals(0, count("semantic_model_metric_meta", duplicate.id()));
+ SemanticModelEntity original = model(namespace, "sales", "original");
+ backend.insert(original, false);
+ Map<String, Long> before = liveIds(original.id());
+ for (String duplicateName : new String[] {"same", "original"}) {
+ SemanticModelEntity invalid =
+ copy(original, original.id(), original.name(),
metrics(duplicateName, duplicateName));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> backend.update(original.nameIdentifier(), original.type(), old
-> invalid));
+ assertEquals(original, backend.get(original.nameIdentifier(),
original.type()));
+ assertEquals(before, liveIds(original.id()));
+ assertEquals(1, count("semantic_model_version_info", original.id()));
+ }
+ }
+
+ @TestTemplate
+ void testSnapshotFailureRollsBackCreateReplaceAndOverwrite() throws
Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", "kept",
"removed");
+ backend.insert(original, false);
+ Map<String, Long> before = liveIds(original.id());
+ SemanticModelEntity replacement =
+ copy(original, original.id(), original.name(), metrics("kept",
"added"));
+ SemanticModelEntity created = model(namespace, "new_model", "new_metric");
+ try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class,
CALLS_REAL_METHODS)) {
+ sessions
+ .when(() ->
SessionUtils.doWithoutCommit(eq(SemanticModelVersionInfoMapper.class), any()))
+ .thenThrow(new IllegalStateException("Injected snapshot failure"));
+ assertThrows(IllegalStateException.class, () -> backend.insert(created,
false));
+ assertThrows(
+ IllegalStateException.class,
+ () -> backend.update(original.nameIdentifier(), original.type(), old
-> replacement));
+ assertThrows(IllegalStateException.class, () ->
backend.insert(replacement, true));
+ }
+ assertEquals(original, backend.get(original.nameIdentifier(),
original.type()));
+ assertEquals(before, liveIds(original.id()));
+ assertEquals(2, count("semantic_model_metric_meta", original.id()));
+ assertEquals(1, count("semantic_model_version_info", original.id()));
+ assertEquals(0, count("semantic_model_meta", created.id()));
+ assertEquals(0, count("semantic_model_metric_meta", created.id()));
+ assertEquals(0, count("semantic_model_version_info", created.id()));
+ }
+
+ @TestTemplate
+ void testIdentityInsertFailureRollsBackAllBatches() throws Exception {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", "existing");
+ backend.insert(original, false);
+ Map<String, Long> before = liveIds(original.id());
+ IdGenerator originalGenerator = GravitinoEnv.getInstance().idGenerator();
+ for (int metricCount : new int[] {2, 1001}) {
+ String[] names =
+ IntStream.range(0, metricCount).mapToObj(i -> "metric_" +
i).toArray(String[]::new);
+ SemanticModelEntity replacement =
+ copy(original, original.id(), original.name(), metrics(names));
+ AtomicInteger allocated = new AtomicInteger();
+ IdGenerator duplicateGenerator =
+ () ->
+ allocated.incrementAndGet() < metricCount
+ ? originalGenerator.nextId()
+ : before.get("existing");
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator",
duplicateGenerator, true);
+ try {
+ assertThrows(
+ EntityAlreadyExistsException.class,
+ () -> backend.update(original.nameIdentifier(), original.type(),
old -> replacement));
+ } finally {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator",
originalGenerator, true);
+ }
+ assertEquals(metricCount, allocated.get());
+ assertEquals(before, liveIds(original.id()));
+ assertEquals(1, count("semantic_model_metric_meta", original.id()));
+ assertEquals(original, backend.get(original.nameIdentifier(),
original.type()));
+ assertEquals(1, count("semantic_model_version_info", original.id()));
+ }
+ }
+
+ @TestTemplate
+ void testStaleUpdateAndDropLeaveWinningIdentities() throws Exception {
+ SemanticModelEntity original = model(parents(), "sales", "original",
"kept");
+ backend.insert(original, false);
+ SemanticModelPO stale =
service().getSemanticModelPOByIdentifier(original.nameIdentifier());
+ long keptId = liveIds(original.id()).get("kept");
+ SemanticModelEntity winner =
+ copy(original, original.id(), original.name(), metrics("winner",
"kept"));
+ SemanticModelEntity loser = copy(original, original.id(), original.name(),
metrics("loser"));
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ backend.update(
+ original.nameIdentifier(),
+ original.type(),
+ old -> {
+ try {
+ backend.update(original.nameIdentifier(), original.type(),
ignored -> winner);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return loser;
+ }));
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
service().deleteSemanticModelWithVersion(original.nameIdentifier(), stale));
+ assertEquals(Set.of("winner", "kept"), liveIds(original.id()).keySet());
+ assertEquals(keptId, liveIds(original.id()).get("kept"));
+ assertEquals(3, count("semantic_model_metric_meta", original.id()));
+ assertEquals(winner, backend.get(original.nameIdentifier(),
original.type()));
+ }
+
+ @TestTemplate
+ void testConcurrentOverwritesKeepIdentityAndSnapshotConsistent() throws
Exception {
+ SemanticModelEntity original = model(parents(), "sales", "kept",
"original");
+ backend.insert(original, false);
+ long keptId = liveIds(original.id()).get("kept");
+ SemanticModelEntity first =
+ copy(original, original.id(), original.name(), metrics("kept",
"first"));
+ SemanticModelEntity second =
+ copy(original, original.id(), original.name(), metrics("kept",
"second"));
+ CountDownLatch ready = new CountDownLatch(2);
+ CountDownLatch start = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ List<Future<?>> writes =
+ List.of(first, second).stream()
+ .map(
+ model ->
+ executor.submit(
+ () -> {
+ ready.countDown();
+ assertTrue(start.await(30, TimeUnit.SECONDS));
+ backend.insert(model, true);
+ return null;
+ }))
+ .collect(Collectors.toList());
+ assertTrue(ready.await(30, TimeUnit.SECONDS));
+ start.countDown();
+ for (Future<?> write : writes) {
+ write.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ start.countDown();
+ executor.shutdownNow();
+ }
+ SemanticModelEntity current = backend.get(original.nameIdentifier(),
original.type());
+ assertTrue(current.equals(first) || current.equals(second));
+ assertSnapshotNames(current);
+ assertEquals(keptId, liveIds(original.id()).get("kept"));
+ assertEquals(4, count("semantic_model_metric_meta", original.id()));
+ }
+
+ @TestTemplate
+ void testParentCascadesRetireOnlyTheirMetricsAndAllowCleanup() throws
Exception {
+ SemanticModelEntity unrelated = model(parents(), "unrelated", "live");
+ backend.insert(unrelated, false);
+ Map<String, Long> liveBefore = liveIds(unrelated.id());
+ for (Entity.EntityType type :
+ List.of(Entity.EntityType.SCHEMA, Entity.EntityType.CATALOG,
Entity.EntityType.METALAKE)) {
+ Namespace namespace = parents();
+ SemanticModelEntity original = model(namespace, "sales", "revenue");
+ backend.insert(original, false);
+ NameIdentifier parent = parent(namespace, type);
+ assertTrue(backend.delete(parent, type, true));
+ assertTrue(live(original.id()).isEmpty());
+ assertEquals(1, count("semantic_model_metric_meta", original.id()));
+ // Purge the parent first: metric cleanup must not rely on joining a
surviving parent row.
+ backend.hardDeleteLegacyData(type, Long.MAX_VALUE);
+ backend.hardDeleteLegacyData(Entity.EntityType.SEMANTIC_MODEL,
Long.MAX_VALUE);
+ assertEquals(0, count("semantic_model_metric_meta", original.id()));
+ assertEquals(liveBefore, liveIds(unrelated.id()));
+ }
+ }
+
+ @TestTemplate
+ void testCascadeFailureRollsBackModelAndMetricDeletion() throws Exception {
+ SemanticModelEntity original = model(parents(), "sales", "revenue");
+ backend.insert(original, false);
+ Map<String, Long> before = liveIds(original.id());
+ NameIdentifier schema = parent(original.namespace(),
Entity.EntityType.SCHEMA);
+ assertThrows(
+ NonEmptyEntityException.class,
+ () -> backend.delete(schema, Entity.EntityType.SCHEMA, false));
+ try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class,
CALLS_REAL_METHODS)) {
+ sessions
+ .when(() ->
SessionUtils.doWithoutCommit(eq(SemanticModelMetricMapper.class), any()))
+ .thenThrow(new IllegalStateException("Injected identity retirement
failure"));
+ assertThrows(
+ IllegalStateException.class,
+ () -> backend.delete(schema, Entity.EntityType.SCHEMA, true));
+ assertThrows(
+ IllegalStateException.class,
+ () -> backend.delete(original.nameIdentifier(), original.type(),
false));
+ }
+ assertEquals(original, backend.get(original.nameIdentifier(),
original.type()));
+ assertEquals(before, liveIds(original.id()));
+ }
+
+ @TestTemplate
+ void testGarbageCollectionHonorsTimelineLimitAndLiveIdentities() throws
Exception {
+ SemanticModelEntity original = model(parents(), "sales", "kept",
"removed1", "removed2");
+ backend.insert(original, false);
+ long keptId = liveIds(original.id()).get("kept");
+ SemanticModelEntity updated = copy(original, original.id(),
original.name(), metrics("kept"));
+ backend.update(original.nameIdentifier(), original.type(), old -> updated);
+ try (SqlSession session =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ PreparedStatement statement =
+ session
+ .getConnection()
+ .prepareStatement(
+ "UPDATE semantic_model_metric_meta SET deleted_at = 10
WHERE deleted_at > 0")) {
+ assertEquals(2, statement.executeUpdate());
+ }
+ assertEquals(0, service().deleteSemanticModelMetasByLegacyTimeline(10L,
1));
+ assertEquals(1, service().deleteSemanticModelMetasByLegacyTimeline(11L,
1));
+ assertEquals(2, count("semantic_model_metric_meta", original.id()));
+ assertEquals(1, service().deleteSemanticModelMetasByLegacyTimeline(11L,
1));
+ assertEquals(1,
backend.deleteOldVersionData(Entity.EntityType.SEMANTIC_MODEL, 1));
+ backend.hardDeleteLegacyData(Entity.EntityType.SEMANTIC_MODEL,
Long.MAX_VALUE);
+ assertEquals(Map.of("kept", keptId), liveIds(original.id()));
+ assertEquals(1, count("semantic_model_metric_meta", original.id()));
+ assertEquals(1, count("semantic_model_version_info", original.id()));
+ }
+
+ private Namespace parents() throws IOException {
+ String suffix = Long.toUnsignedString(RandomIdGenerator.INSTANCE.nextId());
+ String metalake = "metric_metalake_" + suffix;
+ String catalog = "metric_catalog_" + suffix;
+ createAndInsertMakeLake(metalake);
+ createAndInsertCatalog(metalake, catalog);
+ createAndInsertSchema(metalake, catalog, "schema");
+ return Namespace.of(metalake, catalog, "schema");
+ }
+
+ private static NameIdentifier parent(Namespace namespace, Entity.EntityType
type) {
+ switch (type) {
+ case SCHEMA:
+ return NameIdentifier.of(namespace.levels());
+ case CATALOG:
+ return NameIdentifier.of(namespace.level(0), namespace.level(1));
+ case METALAKE:
+ return NameIdentifier.of(namespace.level(0));
+ default:
+ throw new IllegalArgumentException("Unsupported parent type: " + type);
+ }
+ }
+
+ private static SemanticModelEntity model(
+ Namespace namespace, String name, @Nullable String... names) {
+ return SemanticModelEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(namespace)
+ .withAuditInfo(AUDIT_INFO)
+ .withProperties(Map.of())
+ .withDefinition(
+ SemanticModelDefinition.builder()
+ .withDatasets(
+ new Dataset[] {
+ Dataset.builder()
+ .withName("orders")
+ .withSource(NameIdentifier.of("sales", "orders"))
+ .build()
+ })
+ .withMetrics(names == null ? null : metrics(names))
+ .build())
+ .build();
+ }
+
+ private static SemanticModelEntity copy(
+ SemanticModelEntity source, Long id, String name, @Nullable Metric[]
metrics) {
+ return SemanticModelEntity.builder()
+ .withId(id)
+ .withName(name)
+ .withNamespace(source.namespace())
+ .withComment("updated")
+ .withAuditInfo(AUDIT_INFO)
+ .withProperties(Map.of("revision", "updated"))
+ .withDefinition(
+ SemanticModelDefinition.builder()
+ .withDatasets(source.definition().datasets())
+ .withMetrics(metrics)
+ .build())
+ .build();
+ }
+
+ private static Expression expression(String sql) {
+ return Expression.builder()
+ .withDialects(
+ new DialectExpression[] {
+
DialectExpression.builder().withDialect("ansi").withExpression(sql).build()
+ })
+ .build();
+ }
+
+ private static Metric[] metrics(String... names) {
+ return Arrays.stream(names)
+ .map(
+ name ->
+ Metric.builder()
+ .withName(name)
+ .withExpression(expression("SUM(orders.amount)"))
+ .build())
+ .toArray(Metric[]::new);
+ }
+
+ private static SemanticModelMetaService service() {
+ return SemanticModelMetaService.getInstance();
+ }
+
+ private static List<SemanticModelMetricPO> live(Long modelId) {
+ return SessionUtils.getWithoutCommit(
+ SemanticModelMetricMapper.class,
+ mapper ->
mapper.selectSemanticModelMetricsByModelIdForUpdate(modelId));
+ }
+
+ private static Map<String, Long> liveIds(Long modelId) {
+ return live(modelId).stream()
+ .collect(
+ Collectors.toMap(SemanticModelMetricPO::metricName,
SemanticModelMetricPO::metricId));
+ }
+
+ private static void assertSnapshotNames(SemanticModelEntity model) {
+ Set<String> names =
+
Arrays.stream(model.definition().metrics()).map(Metric::name).collect(Collectors.toSet());
+ assertEquals(names, liveIds(model.id()).keySet());
+ }
+
+ private static int count(String table, long modelId) throws SQLException {
+ try (SqlSession session =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ PreparedStatement statement =
+ session
+ .getConnection()
+ .prepareStatement(
+ "SELECT COUNT(*) FROM " + table + " WHERE
semantic_model_id = ?")) {
+ statement.setLong(1, modelId);
+ try (ResultSet results = statement.executeQuery()) {
+ assertTrue(results.next());
+ return results.getInt(1);
+ }
+ }
+ }
+}
diff --git a/design-docs/gravitino-semantic-model-design.md
b/design-docs/gravitino-semantic-model-design.md
index 117c2a62d5..ac72ab07cc 100644
--- a/design-docs/gravitino-semantic-model-design.md
+++ b/design-docs/gravitino-semantic-model-design.md
@@ -600,14 +600,30 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
COMMENT 'semantic model version information';
+
+CREATE TABLE IF NOT EXISTS `semantic_model_metric_meta` (
+ `metric_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'stable internal metric
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'catalog id',
+ `schema_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'schema id',
+ `semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing
semantic model id',
+ `metric_name` MEDIUMTEXT NOT NULL COMMENT 'metric name',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'metric
retired at',
+ PRIMARY KEY (`metric_id`),
+ KEY `idx_smmm_model_live` (`semantic_model_id`, `deleted_at`),
+ KEY `idx_smmm_mid` (`metalake_id`),
+ KEY `idx_smmm_cid` (`catalog_id`),
+ KEY `idx_smmm_sid` (`schema_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model metric identities';
```
Each version row stores the complete definition in five JSON columns:
`datasets`, `relationships`,
`metrics`, `ai_context`, and `custom_extensions`. Typed converters preserve
the existing field
representations and collection order, including string-or-object AI context.
Absent optional fields
use SQL `NULL`, while explicit empty collections use JSON `[]`. Create and
alter still validate and
-persist all definition fields atomically in one snapshot. Child members are
not stored in separate
-tables because they do not have independent identity, lifecycle, or
authorization.
+persist all definition fields atomically in one snapshot. Metric definitions
remain in the snapshot;
+`semantic_model_metric_meta` stores only their internal IDs, exact names,
parent IDs, and retirement
+state. Other child members remain embedded. Editing and authorization remain
model-scoped.
These Semantic Model tables are introduced in 2.0.0. Before that release,
storage changes update the
2.0.0 schema and 1.3.0-to-2.0.0 upgrade scripts in place for all supported
databases. Released versions
@@ -615,7 +631,14 @@ have no Semantic Model data to migrate.
#### Version and Lifecycle Behavior
-- Create writes the identity row and version 1 snapshot in one transaction.
+- Create writes the model identity, version 1 snapshot, and metric identities
in one transaction.
+- Definition replacement and overwrite reconcile metric identities in the
model transaction after
+ the concurrency check. Exact-name matches retain their IDs, added names
receive new IDs, and
+ removed names are retired. Reintroducing a removed name allocates a new ID.
Model renames and
+ metric content or order changes preserve IDs; IDs are not exposed in public
APIs.
+- Model drops and schema/catalog/metalake cascades retire live metric
identities in the same
+ transaction as model deletion. Garbage collection purges retired identities
after the configured
+ retention period; snapshot retention never retires live metric identities.
- Alter follows Gravitino's existing version-based OCC mechanism. It
atomically writes a new
snapshot, increments `last_version`, and advances `current_version` with a
server-internal CAS. A
transaction-level CAS conflict rolls back and returns `409` with
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index 2c41c3e1e6..a3fda62695 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -612,6 +612,21 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB COMMENT 'semantic model version information';
+CREATE TABLE IF NOT EXISTS `semantic_model_metric_meta` (
+ `metric_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'stable internal metric
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'catalog id',
+ `schema_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'schema id',
+ `semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing
semantic model id',
+ `metric_name` CLOB NOT NULL COMMENT 'metric name',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'metric
retired at',
+ PRIMARY KEY (`metric_id`),
+ KEY `idx_smmm_model_live` (`semantic_model_id`, `deleted_at`),
+ KEY `idx_smmm_mid` (`metalake_id`),
+ KEY `idx_smmm_cid` (`catalog_id`),
+ KEY `idx_smmm_sid` (`schema_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model metric identities';
+
-- This schema extends version 1.1.0 with partition statistics storage support
-- The partition_statistic_meta table stores partition-level statistics for
tables
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 c979b98762..4a80bab24f 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
@@ -106,6 +106,21 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB COMMENT 'semantic model version information';
+CREATE TABLE IF NOT EXISTS `semantic_model_metric_meta` (
+ `metric_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'stable internal metric
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'catalog id',
+ `schema_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'schema id',
+ `semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing
semantic model id',
+ `metric_name` CLOB NOT NULL COMMENT 'metric name',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'metric
retired at',
+ PRIMARY KEY (`metric_id`),
+ KEY `idx_smmm_model_live` (`semantic_model_id`, `deleted_at`),
+ KEY `idx_smmm_mid` (`metalake_id`),
+ KEY `idx_smmm_cid` (`catalog_id`),
+ KEY `idx_smmm_sid` (`schema_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model metric identities';
+
CREATE TABLE IF NOT EXISTS `scheduled_expand_tasks` (
`task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
`task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index 6790855212..b8e79b9afa 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -603,6 +603,21 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model version information';
+CREATE TABLE IF NOT EXISTS `semantic_model_metric_meta` (
+ `metric_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'stable internal metric
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'catalog id',
+ `schema_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'schema id',
+ `semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing
semantic model id',
+ `metric_name` MEDIUMTEXT NOT NULL COMMENT 'metric name',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'metric
retired at',
+ PRIMARY KEY (`metric_id`),
+ KEY `idx_smmm_model_live` (`semantic_model_id`, `deleted_at`),
+ KEY `idx_smmm_mid` (`metalake_id`),
+ KEY `idx_smmm_cid` (`catalog_id`),
+ KEY `idx_smmm_sid` (`schema_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model metric identities';
+
-- This schema extends version 1.1.0 with partition statistics storage support
-- The partition_statistic_meta table stores partition-level statistics for
tables
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 8f1cc5e977..77ea3e487f 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
@@ -553,6 +553,21 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model version information';
+CREATE TABLE IF NOT EXISTS `semantic_model_metric_meta` (
+ `metric_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'stable internal metric
id',
+ `metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'catalog id',
+ `schema_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'schema id',
+ `semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'containing
semantic model id',
+ `metric_name` MEDIUMTEXT NOT NULL COMMENT 'metric name',
+ `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'metric
retired at',
+ PRIMARY KEY (`metric_id`),
+ KEY `idx_smmm_model_live` (`semantic_model_id`, `deleted_at`),
+ KEY `idx_smmm_mid` (`metalake_id`),
+ KEY `idx_smmm_cid` (`catalog_id`),
+ KEY `idx_smmm_sid` (`schema_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model metric identities';
+
CREATE TABLE IF NOT EXISTS `scheduled_expand_tasks` (
`task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
`task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index 7efb0a60ba..109b8bc456 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -1070,6 +1070,29 @@ COMMENT ON COLUMN semantic_model_version_info.properties
IS 'semantic model prop
COMMENT ON COLUMN semantic_model_version_info.audit_info IS 'semantic model
version audit info';
COMMENT ON COLUMN semantic_model_version_info.deleted_at IS 'version deleted
at';
+CREATE TABLE IF NOT EXISTS semantic_model_metric_meta (
+ metric_id BIGINT NOT NULL,
+ metalake_id BIGINT NOT NULL,
+ catalog_id BIGINT NOT NULL,
+ schema_id BIGINT NOT NULL,
+ semantic_model_id BIGINT NOT NULL,
+ metric_name TEXT NOT NULL,
+ deleted_at BIGINT NOT NULL DEFAULT 0,
+ PRIMARY KEY (metric_id)
+);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_model_live ON
semantic_model_metric_meta (semantic_model_id, deleted_at);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_metalake_id ON
semantic_model_metric_meta (metalake_id);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_catalog_id ON
semantic_model_metric_meta (catalog_id);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_schema_id ON
semantic_model_metric_meta (schema_id);
+COMMENT ON TABLE semantic_model_metric_meta IS 'semantic model metric
identities';
+COMMENT ON COLUMN semantic_model_metric_meta.metric_id IS 'stable internal
metric id';
+COMMENT ON COLUMN semantic_model_metric_meta.metalake_id IS 'metalake id';
+COMMENT ON COLUMN semantic_model_metric_meta.catalog_id IS 'catalog id';
+COMMENT ON COLUMN semantic_model_metric_meta.schema_id IS 'schema id';
+COMMENT ON COLUMN semantic_model_metric_meta.semantic_model_id IS 'containing
semantic model id';
+COMMENT ON COLUMN semantic_model_metric_meta.metric_name IS 'metric name';
+COMMENT ON COLUMN semantic_model_metric_meta.deleted_at IS 'metric retired at';
+
-- This schema extends version 1.1.0 with partition statistics storage support
-- The partition_statistic_meta table stores partition-level statistics for
tables
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 1bb1bd0556..b5a70c34ed 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
@@ -159,6 +159,29 @@ COMMENT ON COLUMN semantic_model_version_info.properties
IS 'semantic model prop
COMMENT ON COLUMN semantic_model_version_info.audit_info IS 'semantic model
version audit info';
COMMENT ON COLUMN semantic_model_version_info.deleted_at IS 'version deleted
at';
+CREATE TABLE IF NOT EXISTS semantic_model_metric_meta (
+ metric_id BIGINT NOT NULL,
+ metalake_id BIGINT NOT NULL,
+ catalog_id BIGINT NOT NULL,
+ schema_id BIGINT NOT NULL,
+ semantic_model_id BIGINT NOT NULL,
+ metric_name TEXT NOT NULL,
+ deleted_at BIGINT NOT NULL DEFAULT 0,
+ PRIMARY KEY (metric_id)
+);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_model_live ON
semantic_model_metric_meta (semantic_model_id, deleted_at);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_metalake_id ON
semantic_model_metric_meta (metalake_id);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_catalog_id ON
semantic_model_metric_meta (catalog_id);
+CREATE INDEX IF NOT EXISTS semantic_model_metric_meta_idx_schema_id ON
semantic_model_metric_meta (schema_id);
+COMMENT ON TABLE semantic_model_metric_meta IS 'semantic model metric
identities';
+COMMENT ON COLUMN semantic_model_metric_meta.metric_id IS 'stable internal
metric id';
+COMMENT ON COLUMN semantic_model_metric_meta.metalake_id IS 'metalake id';
+COMMENT ON COLUMN semantic_model_metric_meta.catalog_id IS 'catalog id';
+COMMENT ON COLUMN semantic_model_metric_meta.schema_id IS 'schema id';
+COMMENT ON COLUMN semantic_model_metric_meta.semantic_model_id IS 'containing
semantic model id';
+COMMENT ON COLUMN semantic_model_metric_meta.metric_name IS 'metric name';
+COMMENT ON COLUMN semantic_model_metric_meta.deleted_at IS 'metric retired at';
+
CREATE TABLE IF NOT EXISTS scheduled_expand_tasks (
task_name VARCHAR(100) NOT NULL,
task_instance VARCHAR(100) NOT NULL,
