Copilot commented on code in PR #12374:
URL: https://github.com/apache/gravitino/pull/12374#discussion_r3719029460
##########
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:
The backend now unconditionally starts/commits/rolls back a transaction for
cacheable updates. This breaks caller-managed transaction semantics unless
`SessionUtils` explicitly supports nested transactions with correct ownership
(e.g., savepoints or reference counting). In this PR, tests also begin a
transaction and then call `backend.update()`, which implies nesting is
expected. Recommendation (mandatory): make transaction ownership explicit by
only beginning/committing/rolling back when there is no existing transaction
(e.g., via a `SessionUtils.isInTransaction()`/`inTransaction()` check), and
otherwise participate in the caller’s transaction without committing it.
##########
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:
Reconstructing `NameIdentifier` by splitting on '.' is only correct if name
segments are guaranteed not to contain dots. If dots are allowed in any
segment, this will invalidate the wrong key(s) and can leave stale cache
entries. Recommendation (mandatory): either enforce 'no dots in names' at
validation time (and reference that invariant here), or change the change-log
encoding/parsing to a lossless format (e.g., store structured name levels or an
escaped/encoded representation).
##########
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:
The Javadoc claims 'operational entities' are read straight from the store,
but `BaseEntityCache` now explicitly includes `EntityType.JOB` in the cacheable
allowlist. Recommendation (mandatory): update this Javadoc to match actual
behavior (e.g., remove 'operational entities' or spell out which operational
types are excluded while noting that jobs are cacheable).
##########
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:
Needing `!(cache instanceof NoOpsCache)` alongside `coherence() ==
LOCAL_PER_NODE` suggests the `Coherence` model can’t fully express 'no
caching'. Recommendation (optional): have `NoOpsCache` override `coherence()`
to a value that naturally skips listener registration (e.g., introduce a `NONE`
coherence or define `NoOpsCache` as `SHARED`/non-propagating), so the
registration condition doesn’t require type checks.
##########
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:
Asserting `entityChanges.size() == 1` makes the tests sensitive to any
additional change-log rows written after `lastConsumedId` (e.g., if tests ever
run concurrently against a shared DB, or if background activity writes to the
log). Recommendation (optional but strongly suggested): assert that the
expected record exists (filter/match on metalake/type/fullName/operateType)
rather than requiring the list to contain exactly one element.
--
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]
