yuqi1129 commented on code in PR #12374:
URL: https://github.com/apache/gravitino/pull/12374#discussion_r3719621220
##########
core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java:
##########
@@ -115,6 +121,14 @@ public void initialize(Config config) throws
RuntimeException {
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)),
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)),
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)));
+
+ // The coherence gate: a LOCAL_PER_NODE cache keeps its own copy per node,
so changes made on
+ // other nodes must be replayed here through the change log. A SHARED
cache (or a disabled
+ // NoOpsCache) has nothing per-node to invalidate, so no listener is
registered.
+ if (cache.coherence() == Coherence.LOCAL_PER_NODE && !(cache instanceof
NoOpsCache)) {
+ this.entityCacheChangeLogListener = new
EntityCacheChangeLogListener(cache);
+
this.entityChangeLogPoller.registerListener(entityCacheChangeLogListener);
+ }
Review Comment:
Fixed in 917df696fa. Added Coherence.NONE, made NoOpsCache return it, and
removed the instanceof special case from listener registration. Tests now
verify the cache coherence modes.
##########
core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java:
##########
@@ -64,20 +75,20 @@ private void assertEntityChange(
Entity.EntityType entityType,
String fullName,
OperateType operateType) {
- Assertions.assertTrue(
- listEntityChanges(lastConsumedId).stream()
- .anyMatch(
- record ->
- record.getMetalakeName().equals(metalakeName)
- && record.getEntityType().equals(entityType.name())
- && record.getFullName().equals(fullName)
- && record.getOperateType() == operateType),
- String.format("Missing %s %s changelog for %s", entityType,
operateType, fullName));
+ List<EntityChangeRecord> entityChanges = listEntityChanges(lastConsumedId);
+ Assertions.assertEquals(1, entityChanges.size());
+ EntityChangeRecord entityChange = entityChanges.get(0);
+ Assertions.assertEquals(metalakeName, entityChange.getMetalakeName());
+ Assertions.assertEquals(entityType.name(), entityChange.getEntityType());
+ Assertions.assertEquals(fullName, entityChange.getFullName());
+ Assertions.assertEquals(operateType, entityChange.getOperateType());
Review Comment:
Fixed in 917df696fa. The assertion now filters by metalake, entity type,
full name, and operation, then requires exactly one matching row while
tolerating unrelated rows.
##########
core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java:
##########
@@ -56,10 +54,9 @@
* schemas and tables under it).
*
* <p>Relation query results are NOT cached by this implementation; relation
and list operations
- * always fall back to the {@code EntityStore}. Entity types whose
materialized form embeds
- * relation-derived data ({@code USER}, {@code GROUP}, {@code ROLE}) are
excluded from caching
- * entirely by {@link BaseEntityCache#put}, because without relation tracking
their entries could
- * not be invalidated when the referenced entities change.
+ * always fall back to the {@code EntityStore}. Only the self-contained
metadata objects listed in
+ * {@link BaseEntityCache#isCacheable} are cached; every other type
(user/group/role, model/model
+ * version, function, and operational entities) is read straight from the
{@code EntityStore}.
Review Comment:
Fixed in 917df696fa. The Javadoc now describes the actual allowlist behavior
and explicitly lists the excluded types, without classifying all operational
entities as uncached.
##########
core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java:
##########
@@ -178,95 +183,45 @@ public boolean exists(NameIdentifier ident,
Entity.EntityType entityType) throws
@Override
public <E extends Entity & HasIdentifier> void insert(E e, boolean
overwritten)
throws EntityAlreadyExistsException, IOException {
- if (e instanceof BaseMetalake) {
- MetalakeMetaService.getInstance().insertMetalake((BaseMetalake) e,
overwritten);
- } else if (e instanceof CatalogEntity) {
- CatalogMetaService.getInstance().insertCatalog((CatalogEntity) e,
overwritten);
- } else if (e instanceof SchemaEntity) {
- SchemaMetaService.getInstance().insertSchema((SchemaEntity) e,
overwritten);
- } else if (e instanceof TableEntity) {
- TableMetaService.getInstance().insertTable((TableEntity) e, overwritten);
- } else if (e instanceof FilesetEntity) {
- FilesetMetaService.getInstance().insertFileset((FilesetEntity) e,
overwritten);
- } else if (e instanceof TopicEntity) {
- TopicMetaService.getInstance().insertTopic((TopicEntity) e, overwritten);
- } else if (e instanceof UserEntity) {
- UserMetaService.getInstance().insertUser((UserEntity) e, overwritten);
- } else if (e instanceof RoleEntity) {
- RoleMetaService.getInstance().insertRole((RoleEntity) e, overwritten);
- } else if (e instanceof GroupEntity) {
- GroupMetaService.getInstance().insertGroup((GroupEntity) e, overwritten);
- } else if (e instanceof TagEntity) {
- TagMetaService.getInstance().insertTag((TagEntity) e, overwritten);
- } else if (e instanceof ModelEntity) {
- ModelMetaService.getInstance().insertModel((ModelEntity) e, overwritten);
- } else if (e instanceof ModelVersionEntity) {
- if (overwritten) {
- LOG.warn(
- "'overwritten' is not supported for model version meta, ignoring
this flag and "
- + "inserting the new model version.");
+ if (!overwritten || !BaseEntityCache.isCacheable(e.type())) {
+ insertEntity(e, overwritten);
+ return;
+ }
+
+ SessionUtils.beginTransaction();
+ boolean committed = false;
+ try {
+ insertEntity(e, true);
+ insertEntityChange(e.nameIdentifier(), e.type(), OperateType.ALTER);
+ SessionUtils.commitTransaction();
+ committed = true;
+ } finally {
+ if (!committed) {
+ SessionUtils.rollbackTransaction();
}
-
ModelVersionMetaService.getInstance().insertModelVersion((ModelVersionEntity)
e);
- } else if (e instanceof FunctionEntity) {
- FunctionMetaService.getInstance().insertFunction((FunctionEntity) e,
overwritten);
- } else if (e instanceof PolicyEntity) {
- PolicyMetaService.getInstance().insertPolicy((PolicyEntity) e,
overwritten);
- } else if (e instanceof JobTemplateEntity) {
-
JobTemplateMetaService.getInstance().insertJobTemplate((JobTemplateEntity) e,
overwritten);
- } else if (e instanceof JobEntity) {
- JobMetaService.getInstance().insertJob((JobEntity) e, overwritten);
- } else if (e instanceof ViewEntity) {
- ViewMetaService.getInstance().insertView((ViewEntity) e, overwritten);
- } else if (e instanceof GenericEntity) {
- GenericEntity genericEntity = (GenericEntity) e;
- throw new UnsupportedEntityTypeException(
- "Unsupported entity type: %s for insert operation",
genericEntity.type());
- } else {
- throw new UnsupportedEntityTypeException(
- "Unsupported entity type: %s for insert operation", e.getClass());
}
}
@Override
public <E extends Entity & HasIdentifier> E update(
NameIdentifier ident, Entity.EntityType entityType, Function<E, E>
updater)
throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
- switch (entityType) {
- case METALAKE:
- return (E) MetalakeMetaService.getInstance().updateMetalake(ident,
updater);
- case CATALOG:
- return (E) CatalogMetaService.getInstance().updateCatalog(ident,
updater);
- case SCHEMA:
- return (E) SchemaMetaService.getInstance().updateSchema(ident,
updater);
- case TABLE:
- return (E) TableMetaService.getInstance().updateTable(ident, updater);
- case FILESET:
- return (E) FilesetMetaService.getInstance().updateFileset(ident,
updater);
- case TOPIC:
- return (E) TopicMetaService.getInstance().updateTopic(ident, updater);
- case USER:
- return (E) UserMetaService.getInstance().updateUser(ident, updater);
- case GROUP:
- return (E) GroupMetaService.getInstance().updateGroup(ident, updater);
- case ROLE:
- return (E) RoleMetaService.getInstance().updateRole(ident, updater);
- case TAG:
- return (E) TagMetaService.getInstance().updateTag(ident, updater);
- case MODEL:
- return (E) ModelMetaService.getInstance().updateModel(ident, updater);
- case MODEL_VERSION:
- return (E)
ModelVersionMetaService.getInstance().updateModelVersion(ident, updater);
- case FUNCTION:
- return (E) FunctionMetaService.getInstance().updateFunction(ident,
updater);
- case POLICY:
- return (E) PolicyMetaService.getInstance().updatePolicy(ident,
updater);
- case JOB_TEMPLATE:
- return (E)
JobTemplateMetaService.getInstance().updateJobTemplate(ident, updater);
- case VIEW:
- return (E) ViewMetaService.getInstance().updateView(ident, updater);
- default:
- throw new UnsupportedEntityTypeException(
- "Unsupported entity type: %s for update operation", entityType);
+ if (!BaseEntityCache.isCacheable(entityType)) {
+ return updateEntity(ident, entityType, updater);
+ }
+
+ SessionUtils.beginTransaction();
+ boolean committed = false;
+ try {
+ E updatedEntity = updateEntity(ident, entityType, updater);
+ insertEntityChange(ident, entityType, OperateType.ALTER);
+ SessionUtils.commitTransaction();
+ committed = true;
+ return updatedEntity;
+ } finally {
+ if (!committed) {
+ SessionUtils.rollbackTransaction();
+ }
}
Review Comment:
Fixed in 917df696fa. JDBCBackend now checks SessionUtils.isInTransaction()
and only begins, commits, or rolls back when it owns the transaction; otherwise
it participates in the caller-managed transaction. Added session-state coverage
and a service test that verifies the entity update and change-log row roll back
together.
##########
core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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;
+
+import com.google.common.base.Preconditions;
+import java.util.List;
+import java.util.Locale;
+import org.apache.gravitino.Entity.EntityType;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.cache.EntityCache;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Keeps a per-node {@link EntityCache} coherent across a multi-node cluster
by replaying {@code
+ * entity_change_log} rows written by other nodes.
+ *
+ * <p>Every ALTER/DROP row is replayed as a direct {@link
EntityCache#invalidate(NameIdentifier,
+ * EntityType)} for exactly the changed entity key. Because the cache indexes
its keys by identifier
+ * prefix, invalidating a container (for example a schema) cascades to its
cached children on the
+ * local node through that forward prefix scan; no reverse index is involved.
+ *
+ * <p>This listener is registered only for a {@link
+ * org.apache.gravitino.cache.Coherence#LOCAL_PER_NODE} cache: a shared cache
has a single
+ * cluster-wide copy and nothing per-node to invalidate. It is called
<em>synchronously</em> on the
+ * poller thread, so it performs only fast, in-memory, idempotent
invalidations.
+ */
+public class EntityCacheChangeLogListener implements EntityChangeLogListener {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(EntityCacheChangeLogListener.class);
+
+ private final EntityCache cache;
+
+ /**
+ * Creates a listener that invalidates the given entity store cache.
+ *
+ * @param cache the per-node entity store cache to keep coherent
+ */
+ public EntityCacheChangeLogListener(EntityCache cache) {
+ Preconditions.checkArgument(cache != null, "cache cannot be null");
+ this.cache = cache;
+ }
+
+ @Override
+ public void onEntityChange(List<EntityChangeRecord> changes) {
+ for (EntityChangeRecord change : changes) {
+ try {
+ EntityType type = entityType(change);
+ NameIdentifier ident = identifier(change);
+ if (type == null || ident == null) {
+ continue;
+ }
+
+ LOG.debug("Invalidating entity cache due to entity change log: {}
({})", ident, type);
+ cache.invalidate(ident, type);
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Failed to process entity change log record: fullName={},
entityType={}",
+ change.getFullName(),
+ change.getEntityType(),
+ e);
+ }
+ }
+ }
+
+ private EntityType entityType(EntityChangeRecord change) {
+ if (change.getEntityType() == null) {
+ LOG.warn("Invalid entity type in entity change log: null");
+ return null;
+ }
+ try {
+ return
EntityType.valueOf(change.getEntityType().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Unknown entity type in entity change log: {}",
change.getEntityType());
+ return null;
+ }
+ }
+
+ private NameIdentifier identifier(EntityChangeRecord change) {
+ String fullName = change.getFullName();
+ if (fullName == null || fullName.isEmpty()) {
+ LOG.warn("Invalid full name in entity change log: {}", fullName);
+ return null;
+ }
+ // The change log stores the entity's NameIdentifier#toString(), a
dot-joined full name. Split
+ // it back into levels the same way the catalog cache listener does; this
is exact as long as
+ // no name segment contains a dot.
+ return NameIdentifier.of(fullName.split("\\."));
Review Comment:
Fixed in 917df696fa with a lossless, versioned length-prefixed codec for
identifiers containing dots. Ordinary identifiers retain the legacy dot-joined
form for rolling compatibility. The entity cache, catalog, and JCasbin
consumers decode through the codec, with focused listener, codec, and H2
service tests.
--
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]
