jerryshao commented on code in PR #12602: URL: https://github.com/apache/gravitino/pull/12602#discussion_r3891625969
########## core/src/main/java/org/apache/gravitino/storage/relational/service/SemanticModelMetaService.java: ########## @@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; +import static org.apache.gravitino.storage.relational.po.SemanticModelPO.fromSemanticModelPO; +import static org.apache.gravitino.storage.relational.po.SemanticModelPO.initializeSemanticModelPO; + +import com.google.common.base.Preconditions; +import java.io.IOException; +import java.util.Locale; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.meta.SemanticModelEntity; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper; +import org.apache.gravitino.storage.relational.po.SemanticModelPO; +import org.apache.gravitino.storage.relational.po.SemanticModelVersionInfoPO; +import org.apache.gravitino.storage.relational.utils.ExceptionUtils; +import org.apache.gravitino.storage.relational.utils.SessionUtils; +import org.apache.gravitino.utils.NameIdentifierUtil; + +/** Provides relational create and load operations for Semantic Model metadata. */ +public class SemanticModelMetaService { + + private static final SemanticModelMetaService INSTANCE = new SemanticModelMetaService(); + + private final BasePOStorageOps<SemanticModelPO, SemanticModelMetaMapper> ops; + + /** Returns the singleton Semantic Model metadata service. */ + public static SemanticModelMetaService getInstance() { + return INSTANCE; + } + + private SemanticModelMetaService() { + this.ops = new HierarchicalConversionPOStorageOps<>(new SemanticModelPOStorageOps()); + } + + /** + * Resolves a Semantic Model stable ID by schema ID and name. + * + * @param schemaId The parent schema ID. + * @param semanticModelName The Semantic Model name. + * @return The stable Semantic Model ID. + * @throws NoSuchEntityException If the Semantic Model does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getSemanticModelIdBySchemaIdAndName") + public Long getSemanticModelIdBySchemaIdAndName(long schemaId, String semanticModelName) { + Long semanticModelId = + SessionUtils.getWithoutCommit( + SemanticModelMetaMapper.class, + mapper -> mapper.selectSemanticModelIdBySchemaIdAndName(schemaId, semanticModelName)); + if (semanticModelId == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.SEMANTIC_MODEL.name().toLowerCase(Locale.ROOT), + semanticModelName); + } + return semanticModelId; + } + + /** + * Loads the current Semantic Model by identifier. + * + * @param identifier The Semantic Model identifier. + * @return The current Semantic Model entity. + * @throws NoSuchEntityException If the Semantic Model does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getSemanticModelByIdentifier") + public SemanticModelEntity getSemanticModelByIdentifier(NameIdentifier identifier) { + SemanticModelPO semanticModelPO = getSemanticModelPOByIdentifier(identifier); + return fromSemanticModelPO(semanticModelPO, identifier.namespace()); + } + + /** + * Inserts a Semantic Model identity and its version-one snapshot atomically. + * + * @param semanticModelEntity The Semantic Model entity. + * @param overwrite Whether to overwrite rows for the same stable ID or natural key. + * @throws IOException If relational persistence fails. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertSemanticModel") + public void insertSemanticModel(SemanticModelEntity semanticModelEntity, boolean overwrite) + throws IOException { + NameIdentifierUtil.checkSemanticModel(semanticModelEntity.nameIdentifier()); + try { + SemanticModelPO po = + initializeSemanticModelPO(semanticModelEntity, SemanticModelPO.builder()); + AtomicReference<SemanticModelPO> persistedPO = new AtomicReference<>(); + AtomicReference<SemanticModelVersionInfoPO> persistedVersionInfoPO = + new AtomicReference<>(po.getSemanticModelVersionInfoPO()); + SessionUtils.doMultipleWithCommit( + () -> + SchemaMetaService.getInstance() + .lockSchemaForEntityWrite( + semanticModelEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId()), + () -> + SessionUtils.doWithoutCommit( + SemanticModelMetaMapper.class, + mapper -> { + if (overwrite) { Review Comment: **correctness**: `insertSemanticModel`'s overwrite path is a check-then-act (SELECT-by-name, then `SELECT...FOR UPDATE`, then INSERT) rather than an atomic upsert, so two concurrent overwrites that each start from their own unresolved id can leave an orphaned version-info row. Two concurrent `insertSemanticModel(entity, overwrite=true)` calls target the same `(schema_id, semantic_model_name)` but each carries its own freshly-generated id. Both `selectSemanticModelIdBySchemaIdAndName` calls (lines 130-132) run before either commits and return `null`, so both treat this as a fresh row and skip `identityForOverwrite`/`versionInfoForOverwrite`. Thread A's INSERT commits first; Thread B's INSERT (with a different `semantic_model_id`) collides with Thread A's row on the `uk_sid_smn_del` unique key. On MySQL, `ON DUPLICATE KEY UPDATE` silently resolves this by updating Thread A's row (keeping Thread A's id) — but Thread B's version-info insert already used Thread B's own (now-unused) `semantic_model_id`, producing a permanently orphaned `semantic_model_version_info` row with no matching identity row. On PostgreSQL the same race instead throws an unhandled unique-constraint violation, since `ON CONFLICT (semantic_model_id)` only covers the PK. `TableMetaServi ce.insertTable` avoids this by inserting first and reading the DB-resolved identity back afterward; this code does the opposite (pre-fetch/lock/recompute), reintroducing the hazard that pattern was designed to avoid. ########## core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java: ########## @@ -58,6 +58,7 @@ public abstract class BaseEntityCache implements EntityCache { Entity.EntityType.CATALOG, Entity.EntityType.SCHEMA, Entity.EntityType.TABLE, + Entity.EntityType.SEMANTIC_MODEL, Review Comment: **correctness**: `SEMANTIC_MODEL` was added to `CACHEABLE_TYPES` even though this class's own Javadoc says entities with a load-bearing version pointer (like `MODEL` and `FUNCTION`, which are deliberately excluded) must not be cached — and Semantic Model has exactly that structure (`current_version` pointing into a separate version-snapshot table). With `Coherence.LOCAL_PER_NODE` (the default, per this class's own doc), node A caches a `SemanticModelEntity` at version 1. A concurrent overwrite on node B bumps `current_version` to 2 with a new definition. Node A keeps serving the stale version-1 definition from cache until invalidation propagates — exactly the failure mode the class's Javadoc says `MODEL`/`MODEL_VERSION`/`FUNCTION` were excluded to avoid, since Semantic Model's storage layout (`SemanticModelPO` + `SemanticModelVersionInfoPO` joined via `current_version`) is structurally identical to Function's. ########## core/src/main/java/org/apache/gravitino/storage/relational/po/SemanticModelPO.java: ########## @@ -0,0 +1,225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.po; + +import static org.apache.gravitino.storage.relational.utils.POConverters.DEFAULT_DELETED_AT; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; +import java.util.Collections; +import java.util.Map; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.json.JsonUtils; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.NamespacedEntityId; +import org.apache.gravitino.meta.SemanticModelEntity; +import org.apache.gravitino.semantic.SemanticModelDefinition; +import org.apache.gravitino.storage.relational.service.EntityIdService; + +/** The persistent object for Semantic Model identity metadata and its current version snapshot. */ +@Getter +@EqualsAndHashCode(exclude = "semanticModelVersionInfoPO") +@ToString +public class SemanticModelPO { + + /** The initial version allocated to a newly created Semantic Model. */ + public static final Integer INITIAL_VERSION = 1; + + private Long semanticModelId; + private String semanticModelName; + private Long metalakeId; + private Long catalogId; + private Long schemaId; + private String auditInfo; + private Integer currentVersion; + private Integer lastVersion; + private Long deletedAt; + private SemanticModelVersionInfoPO semanticModelVersionInfoPO; + + /** Creates an empty persistent object for MyBatis. */ + public SemanticModelPO() {} + + /** A Lombok builder for {@link SemanticModelPO}. */ + public static class SemanticModelPOBuilder { + // Lombok generates the builder methods. + } + + @lombok.Builder(setterPrefix = "with") + private SemanticModelPO( + Long semanticModelId, + String semanticModelName, + Long metalakeId, + Long catalogId, + Long schemaId, + String auditInfo, + Integer currentVersion, + Integer lastVersion, + Long deletedAt, + SemanticModelVersionInfoPO semanticModelVersionInfoPO) { + Preconditions.checkArgument(semanticModelId != null, "Semantic Model id is required"); + Preconditions.checkArgument(semanticModelName != null, "Semantic Model name 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(auditInfo != null, "Audit info is required"); + Preconditions.checkArgument(currentVersion != null, "Current version is required"); + Preconditions.checkArgument(lastVersion != null, "Last version is required"); + Preconditions.checkArgument(deletedAt != null, "Deleted at is required"); + + this.semanticModelId = semanticModelId; + this.semanticModelName = semanticModelName; + this.metalakeId = metalakeId; + this.catalogId = catalogId; + this.schemaId = schemaId; + this.auditInfo = auditInfo; + this.currentVersion = currentVersion; + this.lastVersion = lastVersion; + this.deletedAt = deletedAt; + this.semanticModelVersionInfoPO = semanticModelVersionInfoPO; + } + + /** + * Converts a persistent object and its current version snapshot to a Semantic Model entity. + * + * @param semanticModelPO The persistent object to convert. + * @param namespace The Semantic Model namespace. + * @return The converted Semantic Model entity. + */ + public static SemanticModelEntity fromSemanticModelPO( + SemanticModelPO semanticModelPO, Namespace namespace) { + try { + SemanticModelVersionInfoPO versionPO = semanticModelPO.getSemanticModelVersionInfoPO(); + SemanticModelDefinition definition = + SemanticModelDefinitionSerDe.deserialize(versionPO.semanticModelDefinition()); + Map<String, String> properties = + versionPO.properties() == null + ? Collections.emptyMap() + : JsonUtils.anyFieldMapper() + .readValue( + versionPO.properties(), + JsonUtils.anyFieldMapper() + .getTypeFactory() + .constructMapType(Map.class, String.class, String.class)); + + return SemanticModelEntity.builder() + .withId(semanticModelPO.getSemanticModelId()) + .withName(versionPO.semanticModelName()) + .withNamespace(namespace) + .withComment(versionPO.semanticModelComment()) + .withDefinition(definition) + .withProperties(properties) + .withAuditInfo( + JsonUtils.anyFieldMapper().readValue(semanticModelPO.getAuditInfo(), AuditInfo.class)) + .build(); + } catch (JsonProcessingException e) { + throw new RuntimeException("Failed to deserialize Semantic Model JSON", e); + } + } + + /** + * Initializes a new Semantic Model identity and version-one snapshot. + * + * @param semanticModelEntity The Semantic Model entity. + * @param builder The identity persistent-object builder. + * @return The initialized persistent object. + */ + public static SemanticModelPO initializeSemanticModelPO( + SemanticModelEntity semanticModelEntity, SemanticModelPOBuilder builder) { + return buildSemanticModelPO(semanticModelEntity, builder, INITIAL_VERSION); + } + + /** + * Creates a complete version snapshot for a Semantic Model entity. + * + * @param semanticModelEntity The Semantic Model entity. + * @param namespacedEntityId The resolved schema and ancestor IDs. + * @param version The version to allocate. + * @return The version snapshot persistent object. + */ + public static SemanticModelVersionInfoPO initializeSemanticModelVersionInfoPO( + SemanticModelEntity semanticModelEntity, + NamespacedEntityId namespacedEntityId, + Integer version) { + try { + String definitionJson = + SemanticModelDefinitionSerDe.serialize(semanticModelEntity.definition()); Review Comment: **correctness**: `buildSemanticModelPO` calls `SemanticModelDefinitionSerDe.serialize(semanticModelEntity.definition())` with no null check; if `definition()` is null, Jackson serializes it to the literal string `"null"`, which passes the `StringUtils.isNotBlank` precondition meant to reject a missing definition. `SemanticModelEntity.validate()` (line 192) only delegates to `Entity.super.validate()` and never checks `definition` for null, and `Builder.withDefinition` (line 298) accepts null unchecked. If an entity is built without a definition, `serialize(null)` returns the 4-character string `"null"` (valid, non-blank text), which satisfies `SemanticModelVersionInfoPO`'s constructor check `Preconditions.checkArgument(StringUtils.isNotBlank(semanticModelDefinition), ...)`. The row persists with a literal `"null"` definition string instead of failing fast, and a later read deserializes it back into a broken/null definition, silently propagating a malformed Semantic Model instead of rejecting it at write time. ########## core/src/main/java/org/apache/gravitino/storage/relational/service/SemanticModelMetaService.java: ########## @@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; +import static org.apache.gravitino.storage.relational.po.SemanticModelPO.fromSemanticModelPO; +import static org.apache.gravitino.storage.relational.po.SemanticModelPO.initializeSemanticModelPO; + +import com.google.common.base.Preconditions; +import java.io.IOException; +import java.util.Locale; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.meta.SemanticModelEntity; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper; +import org.apache.gravitino.storage.relational.po.SemanticModelPO; +import org.apache.gravitino.storage.relational.po.SemanticModelVersionInfoPO; +import org.apache.gravitino.storage.relational.utils.ExceptionUtils; +import org.apache.gravitino.storage.relational.utils.SessionUtils; +import org.apache.gravitino.utils.NameIdentifierUtil; + +/** Provides relational create and load operations for Semantic Model metadata. */ +public class SemanticModelMetaService { + + private static final SemanticModelMetaService INSTANCE = new SemanticModelMetaService(); + + private final BasePOStorageOps<SemanticModelPO, SemanticModelMetaMapper> ops; + + /** Returns the singleton Semantic Model metadata service. */ + public static SemanticModelMetaService getInstance() { + return INSTANCE; + } + + private SemanticModelMetaService() { + this.ops = new HierarchicalConversionPOStorageOps<>(new SemanticModelPOStorageOps()); + } + + /** + * Resolves a Semantic Model stable ID by schema ID and name. + * + * @param schemaId The parent schema ID. + * @param semanticModelName The Semantic Model name. + * @return The stable Semantic Model ID. + * @throws NoSuchEntityException If the Semantic Model does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getSemanticModelIdBySchemaIdAndName") + public Long getSemanticModelIdBySchemaIdAndName(long schemaId, String semanticModelName) { + Long semanticModelId = + SessionUtils.getWithoutCommit( + SemanticModelMetaMapper.class, + mapper -> mapper.selectSemanticModelIdBySchemaIdAndName(schemaId, semanticModelName)); + if (semanticModelId == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.SEMANTIC_MODEL.name().toLowerCase(Locale.ROOT), + semanticModelName); + } + return semanticModelId; + } + + /** + * Loads the current Semantic Model by identifier. + * + * @param identifier The Semantic Model identifier. + * @return The current Semantic Model entity. + * @throws NoSuchEntityException If the Semantic Model does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getSemanticModelByIdentifier") + public SemanticModelEntity getSemanticModelByIdentifier(NameIdentifier identifier) { + SemanticModelPO semanticModelPO = getSemanticModelPOByIdentifier(identifier); + return fromSemanticModelPO(semanticModelPO, identifier.namespace()); + } + + /** + * Inserts a Semantic Model identity and its version-one snapshot atomically. + * + * @param semanticModelEntity The Semantic Model entity. + * @param overwrite Whether to overwrite rows for the same stable ID or natural key. + * @throws IOException If relational persistence fails. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertSemanticModel") + public void insertSemanticModel(SemanticModelEntity semanticModelEntity, boolean overwrite) + throws IOException { + NameIdentifierUtil.checkSemanticModel(semanticModelEntity.nameIdentifier()); + try { + SemanticModelPO po = + initializeSemanticModelPO(semanticModelEntity, SemanticModelPO.builder()); + AtomicReference<SemanticModelPO> persistedPO = new AtomicReference<>(); + AtomicReference<SemanticModelVersionInfoPO> persistedVersionInfoPO = + new AtomicReference<>(po.getSemanticModelVersionInfoPO()); + SessionUtils.doMultipleWithCommit( + () -> + SchemaMetaService.getInstance() + .lockSchemaForEntityWrite( + semanticModelEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId()), + () -> + SessionUtils.doWithoutCommit( + SemanticModelMetaMapper.class, + mapper -> { + if (overwrite) { + Long persistedId = Review Comment: **efficiency**: The overwrite path costs 3 sequential DB round trips (SELECT id-by-name, `SELECT...FOR UPDATE`, then INSERT) where the established pattern used by `TableMetaService.insertTable` and `FilesetMetaService` (insert first via the atomic `ON DUPLICATE KEY UPDATE`, then a single read-back SELECT) costs 2, and reimplements version-resolution logic in Java that duplicates what the SQL's `ON DUPLICATE KEY UPDATE` already computes atomically. Every `overwrite=true` call to `insertSemanticModel` does one extra query versus the equivalent Table/Fileset path for no additional correctness benefit, since the upsert statement already atomically locks and updates the row via the natural key. This also duplicates the "+1" version-bump logic in two places (Java's `versionInfoForOverwrite` and the SQL's `current_version + 1`), which must be kept in lockstep — the exact kind of drift that produced the finding above on `identityForOverwrite`. ########## core/src/main/java/org/apache/gravitino/storage/relational/po/SemanticModelPO.java: ########## @@ -0,0 +1,225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.po; + +import static org.apache.gravitino.storage.relational.utils.POConverters.DEFAULT_DELETED_AT; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; +import java.util.Collections; +import java.util.Map; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.json.JsonUtils; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.NamespacedEntityId; +import org.apache.gravitino.meta.SemanticModelEntity; +import org.apache.gravitino.semantic.SemanticModelDefinition; +import org.apache.gravitino.storage.relational.service.EntityIdService; + +/** The persistent object for Semantic Model identity metadata and its current version snapshot. */ +@Getter +@EqualsAndHashCode(exclude = "semanticModelVersionInfoPO") +@ToString +public class SemanticModelPO { + + /** The initial version allocated to a newly created Semantic Model. */ + public static final Integer INITIAL_VERSION = 1; + + private Long semanticModelId; + private String semanticModelName; + private Long metalakeId; + private Long catalogId; + private Long schemaId; + private String auditInfo; + private Integer currentVersion; + private Integer lastVersion; + private Long deletedAt; + private SemanticModelVersionInfoPO semanticModelVersionInfoPO; + + /** Creates an empty persistent object for MyBatis. */ + public SemanticModelPO() {} + + /** A Lombok builder for {@link SemanticModelPO}. */ + public static class SemanticModelPOBuilder { + // Lombok generates the builder methods. + } + + @lombok.Builder(setterPrefix = "with") + private SemanticModelPO( + Long semanticModelId, + String semanticModelName, + Long metalakeId, + Long catalogId, + Long schemaId, + String auditInfo, + Integer currentVersion, + Integer lastVersion, + Long deletedAt, + SemanticModelVersionInfoPO semanticModelVersionInfoPO) { + Preconditions.checkArgument(semanticModelId != null, "Semantic Model id is required"); + Preconditions.checkArgument(semanticModelName != null, "Semantic Model name 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(auditInfo != null, "Audit info is required"); + Preconditions.checkArgument(currentVersion != null, "Current version is required"); + Preconditions.checkArgument(lastVersion != null, "Last version is required"); + Preconditions.checkArgument(deletedAt != null, "Deleted at is required"); + + this.semanticModelId = semanticModelId; + this.semanticModelName = semanticModelName; + this.metalakeId = metalakeId; + this.catalogId = catalogId; + this.schemaId = schemaId; + this.auditInfo = auditInfo; + this.currentVersion = currentVersion; + this.lastVersion = lastVersion; + this.deletedAt = deletedAt; + this.semanticModelVersionInfoPO = semanticModelVersionInfoPO; + } + + /** + * Converts a persistent object and its current version snapshot to a Semantic Model entity. + * + * @param semanticModelPO The persistent object to convert. + * @param namespace The Semantic Model namespace. + * @return The converted Semantic Model entity. + */ + public static SemanticModelEntity fromSemanticModelPO( + SemanticModelPO semanticModelPO, Namespace namespace) { + try { + SemanticModelVersionInfoPO versionPO = semanticModelPO.getSemanticModelVersionInfoPO(); + SemanticModelDefinition definition = + SemanticModelDefinitionSerDe.deserialize(versionPO.semanticModelDefinition()); + Map<String, String> properties = + versionPO.properties() == null + ? Collections.emptyMap() + : JsonUtils.anyFieldMapper() Review Comment: **simplification**: Deserializing the properties column manually constructs a Jackson `MapType` via `getTypeFactory().constructMapType(Map.class, String.class, String.class)`, instead of the simpler `JsonUtils.anyFieldMapper().readValue(xxxPO.getProperties(), Map.class)` idiom used for the same properties column by every other PO in `POConverters.java` (Metalake, Catalog, Schema, Table, Fileset, Topic). Not a bug, but a heavier, non-conforming reimplementation of a one-line pattern used everywhere else in the module — a reader has to notice this is the one PO that builds its own `MapType` instead of following the established `readValue(..., Map.class)` shortcut, adding unnecessary cognitive overhead for no behavioral difference. ########## core/src/main/java/org/apache/gravitino/storage/relational/po/SemanticModelPO.java: ########## @@ -0,0 +1,225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.po; + +import static org.apache.gravitino.storage.relational.utils.POConverters.DEFAULT_DELETED_AT; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; +import java.util.Collections; +import java.util.Map; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.json.JsonUtils; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.NamespacedEntityId; +import org.apache.gravitino.meta.SemanticModelEntity; +import org.apache.gravitino.semantic.SemanticModelDefinition; +import org.apache.gravitino.storage.relational.service.EntityIdService; + +/** The persistent object for Semantic Model identity metadata and its current version snapshot. */ +@Getter +@EqualsAndHashCode(exclude = "semanticModelVersionInfoPO") +@ToString +public class SemanticModelPO { + + /** The initial version allocated to a newly created Semantic Model. */ + public static final Integer INITIAL_VERSION = 1; + + private Long semanticModelId; + private String semanticModelName; + private Long metalakeId; + private Long catalogId; + private Long schemaId; + private String auditInfo; + private Integer currentVersion; + private Integer lastVersion; + private Long deletedAt; + private SemanticModelVersionInfoPO semanticModelVersionInfoPO; + + /** Creates an empty persistent object for MyBatis. */ + public SemanticModelPO() {} + + /** A Lombok builder for {@link SemanticModelPO}. */ + public static class SemanticModelPOBuilder { + // Lombok generates the builder methods. + } + + @lombok.Builder(setterPrefix = "with") + private SemanticModelPO( + Long semanticModelId, + String semanticModelName, + Long metalakeId, + Long catalogId, + Long schemaId, + String auditInfo, + Integer currentVersion, + Integer lastVersion, + Long deletedAt, + SemanticModelVersionInfoPO semanticModelVersionInfoPO) { + Preconditions.checkArgument(semanticModelId != null, "Semantic Model id is required"); + Preconditions.checkArgument(semanticModelName != null, "Semantic Model name 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(auditInfo != null, "Audit info is required"); + Preconditions.checkArgument(currentVersion != null, "Current version is required"); + Preconditions.checkArgument(lastVersion != null, "Last version is required"); + Preconditions.checkArgument(deletedAt != null, "Deleted at is required"); + + this.semanticModelId = semanticModelId; + this.semanticModelName = semanticModelName; + this.metalakeId = metalakeId; + this.catalogId = catalogId; + this.schemaId = schemaId; + this.auditInfo = auditInfo; + this.currentVersion = currentVersion; + this.lastVersion = lastVersion; + this.deletedAt = deletedAt; + this.semanticModelVersionInfoPO = semanticModelVersionInfoPO; + } + + /** + * Converts a persistent object and its current version snapshot to a Semantic Model entity. + * + * @param semanticModelPO The persistent object to convert. + * @param namespace The Semantic Model namespace. + * @return The converted Semantic Model entity. + */ + public static SemanticModelEntity fromSemanticModelPO( Review Comment: **reuse**: PO-to-entity and entity-to-PO conversion logic for Semantic Model is implemented as static methods directly on `SemanticModelPO`, instead of being added to the shared `org.apache.gravitino.storage.relational.utils.POConverters` class where every other entity's equivalent conversion logic lives (`fromModelPO`, `fromPolicyPO`, `initializePolicyPOWithVersion`, etc.). This PR forks a second, per-entity location for conversion logic that the rest of the module centralizes in `POConverters.java`. A future maintainer looking for "how does entity X convert to/from its PO" will find every other type in `POConverters` except Semantic Model, making the codebase harder to navigate and increasing the chance that a future shared fix to conversion logic (e.g. a properties-map deserialization bugfix) misses this one entity because it isn't in the usual place. ########## core/src/main/java/org/apache/gravitino/storage/relational/service/SemanticModelMetaService.java: ########## @@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; +import static org.apache.gravitino.storage.relational.po.SemanticModelPO.fromSemanticModelPO; +import static org.apache.gravitino.storage.relational.po.SemanticModelPO.initializeSemanticModelPO; + +import com.google.common.base.Preconditions; +import java.io.IOException; +import java.util.Locale; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.meta.SemanticModelEntity; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.mapper.SemanticModelMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SemanticModelVersionInfoMapper; +import org.apache.gravitino.storage.relational.po.SemanticModelPO; +import org.apache.gravitino.storage.relational.po.SemanticModelVersionInfoPO; +import org.apache.gravitino.storage.relational.utils.ExceptionUtils; +import org.apache.gravitino.storage.relational.utils.SessionUtils; +import org.apache.gravitino.utils.NameIdentifierUtil; + +/** Provides relational create and load operations for Semantic Model metadata. */ +public class SemanticModelMetaService { + + private static final SemanticModelMetaService INSTANCE = new SemanticModelMetaService(); + + private final BasePOStorageOps<SemanticModelPO, SemanticModelMetaMapper> ops; + + /** Returns the singleton Semantic Model metadata service. */ + public static SemanticModelMetaService getInstance() { + return INSTANCE; + } + + private SemanticModelMetaService() { + this.ops = new HierarchicalConversionPOStorageOps<>(new SemanticModelPOStorageOps()); + } + + /** + * Resolves a Semantic Model stable ID by schema ID and name. + * + * @param schemaId The parent schema ID. + * @param semanticModelName The Semantic Model name. + * @return The stable Semantic Model ID. + * @throws NoSuchEntityException If the Semantic Model does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getSemanticModelIdBySchemaIdAndName") + public Long getSemanticModelIdBySchemaIdAndName(long schemaId, String semanticModelName) { + Long semanticModelId = + SessionUtils.getWithoutCommit( + SemanticModelMetaMapper.class, + mapper -> mapper.selectSemanticModelIdBySchemaIdAndName(schemaId, semanticModelName)); + if (semanticModelId == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.SEMANTIC_MODEL.name().toLowerCase(Locale.ROOT), + semanticModelName); + } + return semanticModelId; + } + + /** + * Loads the current Semantic Model by identifier. + * + * @param identifier The Semantic Model identifier. + * @return The current Semantic Model entity. + * @throws NoSuchEntityException If the Semantic Model does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getSemanticModelByIdentifier") + public SemanticModelEntity getSemanticModelByIdentifier(NameIdentifier identifier) { + SemanticModelPO semanticModelPO = getSemanticModelPOByIdentifier(identifier); + return fromSemanticModelPO(semanticModelPO, identifier.namespace()); + } + + /** + * Inserts a Semantic Model identity and its version-one snapshot atomically. + * + * @param semanticModelEntity The Semantic Model entity. + * @param overwrite Whether to overwrite rows for the same stable ID or natural key. + * @throws IOException If relational persistence fails. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertSemanticModel") + public void insertSemanticModel(SemanticModelEntity semanticModelEntity, boolean overwrite) + throws IOException { + NameIdentifierUtil.checkSemanticModel(semanticModelEntity.nameIdentifier()); + try { + SemanticModelPO po = + initializeSemanticModelPO(semanticModelEntity, SemanticModelPO.builder()); + AtomicReference<SemanticModelPO> persistedPO = new AtomicReference<>(); + AtomicReference<SemanticModelVersionInfoPO> persistedVersionInfoPO = + new AtomicReference<>(po.getSemanticModelVersionInfoPO()); + SessionUtils.doMultipleWithCommit( + () -> + SchemaMetaService.getInstance() + .lockSchemaForEntityWrite( + semanticModelEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId()), + () -> + SessionUtils.doWithoutCommit( + SemanticModelMetaMapper.class, + mapper -> { + if (overwrite) { + Long persistedId = + mapper.selectSemanticModelIdBySchemaIdAndName( + po.getSchemaId(), po.getSemanticModelName()); + SemanticModelPO existingPO = + mapper.selectSemanticModelMetaByIdForUpdate( + persistedId == null ? po.getSemanticModelId() : persistedId); + persistedPO.set(existingPO); + if (existingPO != null) { + persistedVersionInfoPO.set( + versionInfoForOverwrite( + po.getSemanticModelVersionInfoPO(), existingPO)); + } + } + ops.insertPO( + mapper, + persistedPO.get() == null + ? po + : identityForOverwrite(po, persistedPO.get()), + overwrite); + }), + () -> + SessionUtils.doWithoutCommit( + SemanticModelVersionInfoMapper.class, + mapper -> { + if (overwrite) { + mapper.insertSemanticModelVersionInfoOnDuplicateKeyUpdate( + persistedVersionInfoPO.get()); + } else { + mapper.insertSemanticModelVersionInfo(po.getSemanticModelVersionInfoPO()); + } + })); + } catch (RuntimeException re) { + ExceptionUtils.checkSQLException( + re, Entity.EntityType.SEMANTIC_MODEL, semanticModelEntity.nameIdentifier().toString()); + throw re; + } + } + + /** Returns the persistent-object operations used by this service. */ + public BasePOStorageOps<SemanticModelPO, SemanticModelMetaMapper> ops() { + return ops; + } + + private static SemanticModelPO identityForOverwrite( + SemanticModelPO source, SemanticModelPO persistedPO) { + return SemanticModelPO.builder() + .withSemanticModelId(persistedPO.getSemanticModelId()) + .withSemanticModelName(source.getSemanticModelName()) + .withMetalakeId(source.getMetalakeId()) + .withCatalogId(source.getCatalogId()) + .withSchemaId(source.getSchemaId()) + .withAuditInfo(source.getAuditInfo()) + .withCurrentVersion(source.getCurrentVersion()) Review Comment: **correctness (currently masked)**: `identityForOverwrite` builds the overwritten identity row's `currentVersion`/`lastVersion` from `source` (the freshly-built `po`, always version 1) instead of from `persistedPO` (the actual locked existing row) — inconsistent with the sibling `versionInfoForOverwrite`, which correctly uses `persistedPO.getCurrentVersion() + 1`. Currently masked: `insertSemanticModelMetaOnDuplicateKeyUpdate`'s SQL (`SemanticModelMetaBaseSQLProvider.java:178-179`) hardcodes `current_version = current_version + 1` / `last_version = current_version + 1` in the `ON DUPLICATE KEY UPDATE` clause, ignoring the bound Java values entirely, so the wrong value never reaches the DB today. But if a future contributor "simplifies" that SQL to trust the bound `#{semanticModelMeta.currentVersion}` parameter instead (a natural-looking cleanup since the parameter appears to already carry the right value), every overwrite would silently reset `semantic_model_meta.current_version`/`last_version` to 1 while `semantic_model_version_info` keeps accumulating rows at version 2, 3, 4... Reads join on `smm.current_version = smvi.version`, so they'd always resolve to the version-1 snapshot, silently hiding every subsequent overwrite. No existing unit test asserts `identityForOverwrite`'s version fields directly. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
