jrgemignani commented on code in PR #2376:
URL: https://github.com/apache/age/pull/2376#discussion_r3112948873
##########
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:
added
--
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]
