MuhammadTahaNaveed commented on code in PR #2376:
URL: https://github.com/apache/age/pull/2376#discussion_r3104903625
##########
age--1.7.0--y.y.y.sql:
##########
@@ -408,3 +408,13 @@ $function$;
COMMENT ON FUNCTION ag_catalog.age_pg_upgrade_status() IS
'Returns the current pg_upgrade readiness status of the AGE installation.';
+
+--
+-- VLE cache invalidation trigger function
+-- Installed on graph label tables to catch SQL-level mutations
+-- and increment the per-graph version counter for VLE cache invalidation.
+--
+CREATE FUNCTION ag_catalog.age_invalidate_graph_cache()
+ RETURNS trigger
+ LANGUAGE c
+AS 'MODULE_PATHNAME';
Review Comment:
upgrade script should also attach the trigger to pre-existing label tables
##########
regress/sql/age_global_graph.sql:
##########
@@ -146,6 +146,103 @@ RESET client_min_messages;
SELECT * FROM drop_graph('ag_graph_1', true);
SELECT * FROM drop_graph('ag_graph_2', true);
SELECT * FROM drop_graph('ag_graph_3', true);
+
+-----------------------------------------------------------------------------------------------------------------------------
+--
+-- VLE cache invalidation tests
+--
+-- These tests verify that the graph version counter properly invalidates
+-- the VLE hash table cache when the graph is mutated, and that thin
+-- entry lazy property fetch returns correct data.
+--
+
+-- Setup: create a graph with a chain a->b->c->d
+SELECT * FROM create_graph('vle_cache_test');
+
+SELECT * FROM cypher('vle_cache_test', $$
+ CREATE (a:Node {name: 'a'})-[:Edge]->(b:Node {name: 'b'})-[:Edge]->(c:Node
{name: 'c'})-[:Edge]->(d:Node {name: 'd'})
+$$) AS (v agtype);
+
+-- VLE query: find all paths from a's neighbors (should find b, b->c, b->c->d)
+SELECT * FROM cypher('vle_cache_test', $$
+ MATCH (a:Node {name: 'a'})-[:Edge*1..3]->(n:Node)
+ RETURN n.name
+ ORDER BY n.name
+$$) AS (name agtype);
+
+-- Now add a new node e connected to d. This should invalidate the cache.
+SELECT * FROM cypher('vle_cache_test', $$
+ MATCH (d:Node {name: 'd'})
+ CREATE (d)-[:Edge]->(:Node {name: 'e'})
+$$) AS (v agtype);
+
+-- VLE query again: should now also find e via a->b->c->d->e (4 hops won't
reach,
+-- but d->e is 1 hop from d, and a->b->c->d->e would be 4 hops from a).
+-- Increase range to *1..4 to include e
+SELECT * FROM cypher('vle_cache_test', $$
+ MATCH (a:Node {name: 'a'})-[:Edge*1..4]->(n:Node)
+ RETURN n.name
+ ORDER BY n.name
+$$) AS (name agtype);
+
+-- Test cache invalidation on DELETE: remove node c and its edges
+SELECT * FROM cypher('vle_cache_test', $$
+ MATCH (c:Node {name: 'c'})
+ DETACH DELETE c
+$$) AS (v agtype);
+
+-- VLE query: should only find b now (c is gone, so b->c path is broken)
+SELECT * FROM cypher('vle_cache_test', $$
+ MATCH (a:Node {name: 'a'})-[:Edge*1..4]->(n:Node)
+ RETURN n.name
+ ORDER BY n.name
+$$) AS (name agtype);
+
+-- Test cache invalidation on SET: change b's name property
+SELECT * FROM cypher('vle_cache_test', $$
+ MATCH (b:Node {name: 'b'})
+ SET b.name = 'b_modified'
+ RETURN b.name
+$$) AS (name agtype);
+
+-- VLE query: verify the updated property is returned via lazy fetch
+SELECT * FROM cypher('vle_cache_test', $$
+ MATCH (a:Node {name: 'a'})-[:Edge*1..4]->(n:Node)
+ RETURN n.name
+ ORDER BY n.name
+$$) AS (name agtype);
+
+-- Test VLE with edge properties (exercises thin entry edge property fetch)
+SELECT * FROM drop_graph('vle_cache_test', true);
+SELECT * FROM create_graph('vle_cache_test2');
+
+SELECT * FROM cypher('vle_cache_test2', $$
+ CREATE (a:N {name: 'a'})-[:E {weight: 1}]->(b:N {name: 'b'})-[:E {weight:
2}]->(c:N {name: 'c'})
+$$) AS (v agtype);
+
+-- VLE path output to verify edge properties are fetched correctly via
+-- thin entry lazy fetch. Returning the full path forces build_path()
+-- to call get_edge_entry_properties() for each edge in the result.
+-- The output must contain the correct weight values (1 and 2).
+SELECT * FROM cypher('vle_cache_test2', $$
+ MATCH p=(a:N {name: 'a'})-[:E *1..2]->(n:N)
+ RETURN p
+ ORDER BY n.name
+$$) AS (p agtype);
+
+-- VLE edge properties via UNWIND + relationships() to individually verify
+-- each edge's properties are correctly fetched from the heap via TID.
+SELECT * FROM cypher('vle_cache_test2', $$
+ MATCH p=(a:N {name: 'a'})-[:E *1..2]->(n:N)
+ WITH p, n
+ UNWIND relationships(p) AS e
+ RETURN n.name, e.weight
+ ORDER BY n.name, e.weight
+$$) AS (name agtype, weight agtype);
+
+-- Cleanup
+SELECT * FROM drop_graph('vle_cache_test2', true);
+
-----------------------------------------------------------------------------------------------------------------------------
Review Comment:
Recommended to add a few direct SQL INSERT/UPDATE/TRUNCATE tests to test
trigger and cache invalidation.
##########
src/backend/utils/adt/age_global_graph.c:
##########
@@ -1531,3 +1640,305 @@ Datum age_graph_stats(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(agtype_value_to_agtype(result.res));
}
+
+/*
+ * ============================================================================
+ * Graph Version Counter Implementation
+ *
+ * Provides per-graph monotonic version counters in shared memory for
+ * cross-backend VLE cache invalidation. Three modes are supported:
+ *
+ * DSM (PG 17+): Uses GetNamedDSMSegment — works without shared_preload_libs
+ * SHMEM (PG <17): Uses shmem_request/startup hooks — needs shared_preload_libs
+ * SNAPSHOT: Falls back to original snapshot-based invalidation
+ * ============================================================================
+ */
+
+#if PG_VERSION_NUM >= 170000
+/*
+ * DSM path: GetNamedDSMSegment init callback.
+ * Called once when the DSM segment is first created.
+ */
+static void age_dsm_init_callback(void *ptr)
+{
+ GraphVersionState *state = (GraphVersionState *) ptr;
+
+ LWLockInitialize(&state->lock,
+ LWLockNewTrancheId());
+ LWLockRegisterTranche(state->lock.tranche, "age_graph_version");
+ state->num_entries = 0;
+ memset(state->entries, 0, sizeof(state->entries));
+}
+
+/*
+ * Get the shared GraphVersionState via DSM registry.
+ * The segment is created on first access and persists until server shutdown.
+ */
+static GraphVersionState *get_version_state_dsm(void)
+{
+ bool found;
+
+ return (GraphVersionState *)
+ GetNamedDSMSegment("age_graph_versions",
+ sizeof(GraphVersionState),
+ age_dsm_init_callback,
+ &found);
+}
+#endif /* PG_VERSION_NUM >= 170000 */
+
+/*
+ * SHMEM path: request and startup hooks for PG < 17.
+ * These are registered in _PG_init when shared_preload_libraries is used.
+ * On PG 17+, DSM is used instead and these functions are not called.
+ */
+#if PG_VERSION_NUM < 170000
+void age_graph_version_shmem_request(void)
+{
+ RequestAddinShmemSpace(MAXALIGN(sizeof(GraphVersionState)));
+}
+
+void age_graph_version_shmem_startup(void)
+{
+ bool found;
+
+ LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
+
+ shmem_version_state =
+ (GraphVersionState *) ShmemInitStruct("AGE Graph Version State",
+ sizeof(GraphVersionState),
+ &found);
+ if (!found)
+ {
+ LWLockInitialize(&shmem_version_state->lock,
+ LWLockNewTrancheId());
+ LWLockRegisterTranche(shmem_version_state->lock.tranche,
+ "age_graph_version");
+ shmem_version_state->num_entries = 0;
+ memset(shmem_version_state->entries, 0,
+ sizeof(shmem_version_state->entries));
+ }
+
+ LWLockRelease(AddinShmemInitLock);
+}
+#endif /* PG_VERSION_NUM < 170000 */
+
+/*
+ * Detect which version mode to use. Called once per backend on first access.
+ * Emits a DEBUG1 log message indicating the chosen mode.
+ */
+static void detect_version_mode(void)
+{
+#if PG_VERSION_NUM >= 170000
+ version_mode = VERSION_MODE_DSM;
+ elog(DEBUG1, "AGE: VLE cache using DSM version counter");
+#else
+ if (shmem_version_state != NULL)
+ {
+ version_mode = VERSION_MODE_SHMEM;
+ elog(DEBUG1, "AGE: VLE cache using SHMEM version counter");
+ }
+ else
+ {
+ version_mode = VERSION_MODE_SNAPSHOT;
+ elog(DEBUG1, "AGE: VLE cache using snapshot-based invalidation "
+ "(add AGE to shared_preload_libraries for better caching)");
+ }
+#endif
+}
+
+/*
+ * Get a pointer to the GraphVersionState, regardless of mode.
+ * Returns NULL only in SNAPSHOT mode (no shared memory available).
+ */
+static GraphVersionState *get_version_state(void)
+{
+ if (version_mode == VERSION_MODE_UNKNOWN)
+ {
+ detect_version_mode();
+ }
+
+#if PG_VERSION_NUM >= 170000
+ if (version_mode == VERSION_MODE_DSM)
+ {
+ return get_version_state_dsm();
+ }
+#endif
+
+ if (version_mode == VERSION_MODE_SHMEM)
+ {
+ return shmem_version_state;
+ }
+
+ return NULL;
+}
+
+/*
+ * Get the current version counter for a graph.
+ * Returns 0 if the graph has never been tracked or if shared memory
+ * is not available. Lock-free read via pg_atomic_read_u64.
+ */
+uint64 get_graph_version(Oid graph_oid)
+{
+ GraphVersionState *state = get_version_state();
+ int i;
+
+ if (state == NULL)
+ {
+ return 0;
+ }
+
+ /* lock-free scan of the array */
+ for (i = 0; i < state->num_entries; i++)
+ {
+ if (state->entries[i].graph_oid == graph_oid)
+ {
+ return pg_atomic_read_u64(&state->entries[i].version);
+ }
+ }
+
+ return 0;
+}
+
+/*
+ * Increment the version counter for a graph.
+ * Called after any graph mutation (Cypher or SQL trigger).
+ * Lock-free for existing entries; acquires LWLock only to allocate new slots.
+ */
+void increment_graph_version(Oid graph_oid)
+{
+ GraphVersionState *state = get_version_state();
+ int i;
+
+ if (state == NULL)
+ {
+ return;
+ }
+
+ /* try to find existing entry (lock-free) */
+ for (i = 0; i < state->num_entries; i++)
+ {
+ if (state->entries[i].graph_oid == graph_oid)
+ {
+ pg_atomic_fetch_add_u64(&state->entries[i].version, 1);
+ return;
+ }
+ }
+
+ /* new graph — need lock to allocate slot */
+ LWLockAcquire(&state->lock, LW_EXCLUSIVE);
+
+ /* re-check after acquiring lock (another backend may have added it) */
+ for (i = 0; i < state->num_entries; i++)
+ {
+ if (state->entries[i].graph_oid == graph_oid)
+ {
+ LWLockRelease(&state->lock);
+ pg_atomic_fetch_add_u64(&state->entries[i].version, 1);
+ return;
+ }
+ }
+
+ /* add new entry */
+ if (state->num_entries < AGE_MAX_GRAPHS)
+ {
+ int idx = state->num_entries;
+
+ state->entries[idx].graph_oid = graph_oid;
+ pg_atomic_init_u64(&state->entries[idx].version, 1);
+
+ /*
+ * Write barrier ensures the entry fields are fully visible to
+ * other backends before num_entries is incremented. This prevents
+ * readers on weak memory-ordering architectures (e.g., ARM) from
+ * seeing the incremented count before the entry is initialized.
+ */
+ pg_write_barrier();
+ state->num_entries++;
+ }
+ else
+ {
+ elog(WARNING, "AGE: graph version counter table full (%d graphs)",
+ AGE_MAX_GRAPHS);
+ }
+
+ LWLockRelease(&state->lock);
+}
+
+/*
+ * Helper function to look up the graph OID for a given label table OID.
+ * Scans ag_label catalog for a matching relation column.
+ * Returns InvalidOid if the table is not a graph label table.
+ */
+Oid get_graph_oid_for_table(Oid table_oid)
+{
+ ScanKeyData scan_key;
+ Relation ag_label_rel;
+ TableScanDesc scan;
+ HeapTuple tuple;
+ Oid graph_oid = InvalidOid;
+
+ ag_label_rel = table_open(ag_label_relation_id(), AccessShareLock);
Review Comment:
label cache can be used here. (search_label_relation_cache)
--
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]
