You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@geaflow.apache.org by "E2ern1ty (via GitHub)" <gi...@apache.org> on 2026/08/26 11:50:16 UTC
[PR] [ISSUE-844] fix(ai): invalidate an embedding vector when its text changes [geaflow]
E2ern1ty opened a new pull request, #858:
URL: https://github.com/apache/geaflow/pull/858
### What changes were proposed in this pull request?
Fixes #844.
`EmbeddingIndexStore` keyed each persisted vector by the entity key, a prefix plus the id and the label, with nothing derived from the value that was embedded. `initStore` read the presence of that key as "this entity is indexed", so an entity that kept its id and changed its value silently kept the vector of the value it no longer had. Retrieval for the old value still reached the entity while its current value was unreachable, and because the subgraph handed on is verbalized from the current graph, a query about the old value came back with a context showing the new one.
**A record now says which text produced it.** `EmbeddingResult` gains `contentHash`, SHA-256 over the entity's chunk list with the count and a separator folded in. On rebuild a record is accepted only when that fingerprint matches what the entity would be embedded from now; a mismatch counts as not indexed, so the entity is embedded again.
**Existing index files.** A record without a fingerprint cannot be shown to match the current value, and trusting it would keep exactly the staleness the check is for, so such records are dropped and their entities embedded once. `upgradeIndexFile` is offered for the case where a file is known to predate no change in the graph: it stamps the current fingerprint on, verifying only that the record count per entity matches the chunk count now, and its javadoc states plainly what it cannot check. This is the open question from the issue; happy to go the other way if the project would rather not spend a round of embedding on upgrade.
**Completeness.** A fingerprint covers all of an entity's chunks at once, so a record carries no ordinal and a set short of one chunk still matches every record it has. The count of accepted records is checked against the chunk count, otherwise an entity that lost a record to a half written flush would be held with part of its vectors for good.
**Compaction, and what is deliberately not compacted.** The file is rewritten when records are dropped, since it was append only and a superseded record otherwise stayed beside its replacement, growing the file with every change of value. Only records whose entity was found in the graph are rewritten from memory; a record belonging to no scanned entity is carried over untouched. From `initStore` a deleted entity and a graph that scanned empty or short look the same, and reading the second as the first would let one bad scan take the whole index with it, so pruning those is left to `upgradeIndexFile`, where the caller states that the graph it passes is the whole of it. The rewrite goes to a temporary file named by the file system, is flushed to the device and moved onto the target, and is skipped when nothing was dropped, so an unchanged run leaves the file byte for byte.
### One thing to look at: a test fixture is removed
`GraphMemoryTest` no longer loads an embedding store from `src/test/resources/index/LDBCEmbeddingIndexStore`, and that 4.6MB file is deleted. Two findings led there, both checkable before the deletion:
- It could not be carried over. Its records line up with the text the current code would embed for **14 of its 163 entities**; the other 149 disagree on chunk count, in a pattern consistent with the file having been written when the values were joined before splitting rather than split per value. So there was no honest fingerprint to stamp on it, and no way to embed it again offline.
- It could not have affected a single assertion. Every query in that test embeds to `double[0]` through `MockChatRobot`, and `EmbeddingVector.match` returns `0.0` on a length mismatch, below the `0.5` threshold. Everything asserted there comes from the keyword path, and those 22 assertions pass unchanged. Runtime for that test drops from 27.6s to 0.6s, the 27s having been model retries against a null URL.
If you would rather keep the file, I can drop only the wiring, though it would then be unreferenced. The `**/resources/index/**` rat exclusion in the root pom is now unused; I left it alone as it is a generic pattern, and can remove it on request.
### Not addressed here
The fingerprint covers the text only, so changing the embedding model or its dimension leaves every record accepted, and recall would quietly go to zero on a dimension change. That is the same class of staleness, but which properties of a model identify it is a separate decision; happy to open a follow-up issue.
### How was this PR tested?
- [x] Tests have Added for the changes
- [ ] Production environment verified
`EmbeddingIndexInvalidationTest` drives the store against a local `/v1/embeddings` endpoint answering with one hot vectors, one dimension per distinct text, so a stored vector says which text produced it: cosine 1 against its own text and 0 against any other. No key and no network needed. Nine cases:
| case | what it pins |
| --- | --- |
| changed value on a vertex | embedded again, the stale record leaves the file, the new one is then accepted |
| changed value on an edge | the same for edges |
| recall through `GraphMemoryServer` | the old value reaches nothing, the current value reaches the vertex |
| entity deleted | its record is kept, and pruned by `upgradeIndexFile` |
| graph that scans nothing | the file is left byte for byte |
| entity holding only some of its chunks | embedded again, no partial hold |
| record without a fingerprint | not taken on trust, replaced, not duplicated |
| upgrade of an unversioned file | spares the round of embedding, and a later change is still caught |
| upgrade of a file it cannot line up | nothing stamped, file untouched, entity embedded again |
Each case fails against the behaviour it pins, verified by disabling each check in turn. The unchanged-graph case asserts the file is byte identical, which is what would catch a fingerprint computed differently on the write and the verify path.
`mvn -pl geaflow-ai clean test -Pjdk8`: 15 tests, 0 failures, checkstyle and RAT clean.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-844] fix(ai): invalidate an embedding vector when its text changes [geaflow]
Posted by "E2ern1ty (via GitHub)" <gi...@apache.org>.
E2ern1ty commented on code in PR #858:
URL: https://github.com/apache/geaflow/pull/858#discussion_r3870191190
##########
geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java:
##########
@@ -96,24 +103,45 @@ public void initStore(GraphAccessor graphAccessor, VerbalizationFunction func,
if (line.isEmpty()) {
continue;
}
+ EmbeddingService.EmbeddingResult embedding;
try {
- EmbeddingService.EmbeddingResult embedding =
- new Gson().fromJson(line, EmbeddingService.EmbeddingResult.class);
- String key = embedding.input;
- GraphEntity entity = key2EntityMap.get(key);
- if (entity != null) {
- this.indexStoreMap.computeIfAbsent(entity, k -> new ArrayList<>()).add(embedding);
- }
- count++;
+ embedding = new Gson().fromJson(line, EmbeddingService.EmbeddingResult.class);
} catch (Throwable e) {
+ // Only the parse is guarded. What follows verbalises the entity, and a
+ // verbaliser that throws must not be reported as a malformed file.
LOGGER.info("Cannot parse embedding item: " + line);
+ continue;
}
+ String key = embedding == null ? null : embedding.input;
+ GraphEntity entity = key == null ? null : key2EntityMap.get(key);
+ if (entity != null) {
+ if (embedding.contentHash == null) {
+ unversionedRecords++;
+ } else if (embedding.contentHash.equals(
+ currentFingerprint(key, entity, key2Fingerprint))) {
+ this.indexStoreMap.computeIfAbsent(entity, k -> new ArrayList<>()).add(embedding);
Review Comment:
Done in 923347fc, keyed by `ModelUtils.getGraphEntityKey` as you suggest, though I want to correct the two reasons given, because they point at the wrong risk and I would rather the record be accurate.
**Lookups across instances already worked.** `GraphVertex.equals`/`hashCode` delegate to `Vertex`, which is `Objects.equals(id) && Objects.equals(label)`, and `GraphEdge` to `Edge`, which is `srcId`, `dstId` and `label`. So an instance handed out by a different `GraphAccessor` for the same entity is equal and hashes alike, and `getEntityIndex` found the records. I could not construct a case on master where it did not.
**What a key retained was not the graph.** `GraphVertex` holds a `Vertex`, and `Vertex` holds `id`, `label` and `values` — no reference to `MemoryGraph`, `GraphAccessor` or anything above it. So the retention was of the `Vertex` objects and their value strings, which is real but bounded by the entity count and small beside the vectors in the same map, at 1024 doubles each. Not a leak in the sense of holding a graph alive.
**The reason I think the change is right is narrower, and closer to what this PR is about.** With the entity as the key, what counts as the same indexed entity is decided by `GraphVertex.equals`, which answers on the id and the label. This PR exists because an entity's identity did not cover its value. Extending `Vertex.equals` to cover the values is a reasonable thing for someone to want next, and the moment they did, `getEntityIndex` would stop finding records and the scan in `initStore` would stop skipping what is already indexed — silently, both of them. Meanwhile the file has always been keyed by the entity key. Two notions of identity for one thing, one of them defined in a class this PR invites people to change. Keying the map by the same string as the file collapses them into one and takes the index out of reach of that change.
The call sites moved with it: the scan loop and the in-batch dedup set now hold keys, `indexBatch` writes under the key it already computes for `res.input`, and `getEntityIndex` looks up `getGraphEntityKey(entity)`. That last one builds a small string per lookup where before it hashed two fields, which on the query path is one allocation per vertex per query, against a 1024-dimension cosine for each of them.
There is a test for the scenario you describe: it indexes through one graph and reads through an entity object from another, asserting first that the two objects really are different. It holds on master as well — it is there to keep it holding.
##########
geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java:
##########
@@ -96,24 +103,45 @@ public void initStore(GraphAccessor graphAccessor, VerbalizationFunction func,
if (line.isEmpty()) {
continue;
}
+ EmbeddingService.EmbeddingResult embedding;
try {
- EmbeddingService.EmbeddingResult embedding =
- new Gson().fromJson(line, EmbeddingService.EmbeddingResult.class);
- String key = embedding.input;
- GraphEntity entity = key2EntityMap.get(key);
- if (entity != null) {
- this.indexStoreMap.computeIfAbsent(entity, k -> new ArrayList<>()).add(embedding);
Review Comment:
Addressed in 923347fc, but not with a `ConcurrentHashMap`, because I do not think that fixes the case you describe.
Take it as stated: initialization on the startup thread, a query on a worker thread. `initStore` assigned the map to the field up front and then mutated it for the whole of its run — the file read, then batch after batch of embedding. A worker reading in that window sees whatever has been filled in so far. With a `ConcurrentHashMap` it still does: every individual `get` is safe, no exception is thrown, and the answer is a partial index. For a recall path that is worse than an error, because it comes back as a plausible result rather than a failure. The same goes for a second `initStore` on a live store, which under either map empties the index for readers before filling it again.
So the map is built aside and assigned to the field in one go, at the end of `initStore`. The field is `volatile`, and starts as `Collections.emptyMap()`. A reader now sees either the index that was there before or the finished one, and a reader that arrives before any build has happened gets an empty list rather than the `NullPointerException` it would have got on master. `getEntityIndex` reads the field once per call, so it works against one map throughout.
Two notes on the trade-off, since it is not free:
- During a rebuild both maps are held, so peak memory is roughly doubled for that window. I think consistency is worth it here, and there is no caller in tree that rebuilds a live store — nothing in `geaflow-ai/src/main` constructs this store at all today.
- This does not make `initStore` safe to call concurrently with itself. Two builds at once still race on the index file, which is appended to. That was true before and I have not tried to fix it here; if you would like it enforced, the smallest thing would be to reject a concurrent call outright rather than to make it work.
There is a test: it holds the embeddings endpoint open with a latch, runs a second `initStore` on the same store on another thread, and asserts that a reader in that window gets the vector from before the build. It fails if the field is assigned before the build instead of after — with the assignment moved back, the reader gets zero vectors, which is the partial index a `ConcurrentHashMap` would also have shown.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-844] fix(ai): invalidate an embedding vector when its text changes [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #858:
URL: https://github.com/apache/geaflow/pull/858#discussion_r3869354834
##########
geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java:
##########
@@ -96,24 +103,45 @@ public void initStore(GraphAccessor graphAccessor, VerbalizationFunction func,
if (line.isEmpty()) {
continue;
}
+ EmbeddingService.EmbeddingResult embedding;
try {
- EmbeddingService.EmbeddingResult embedding =
- new Gson().fromJson(line, EmbeddingService.EmbeddingResult.class);
- String key = embedding.input;
- GraphEntity entity = key2EntityMap.get(key);
- if (entity != null) {
- this.indexStoreMap.computeIfAbsent(entity, k -> new ArrayList<>()).add(embedding);
- }
- count++;
+ embedding = new Gson().fromJson(line, EmbeddingService.EmbeddingResult.class);
} catch (Throwable e) {
+ // Only the parse is guarded. What follows verbalises the entity, and a
+ // verbaliser that throws must not be reported as a malformed file.
LOGGER.info("Cannot parse embedding item: " + line);
+ continue;
}
+ String key = embedding == null ? null : embedding.input;
+ GraphEntity entity = key == null ? null : key2EntityMap.get(key);
+ if (entity != null) {
+ if (embedding.contentHash == null) {
+ unversionedRecords++;
+ } else if (embedding.contentHash.equals(
+ currentFingerprint(key, entity, key2Fingerprint))) {
+ this.indexStoreMap.computeIfAbsent(entity, k -> new ArrayList<>()).add(embedding);
Review Comment:
I've observed a phenomenon: the code uses a `Map<GraphEntity, List<EmbeddingResult>>(indexStoreMap)` and populates it with entity instances in the `initStore`; however, subsequent locations can access the index using other `GraphAccessor`/`GraphEntity` instances (e.g., the `GraphEntity` instance generated when `getEntityIndex` is called externally might not be the same object reference). If the `GraphEntity` lacks appropriate `equals/hashCode` or object instance adjacency, it can lead to not finding loaded records (a logical error); furthermore, using a `GraphEntity` object as a long-term key maintains a strong reference to the graph, causing memory leaks.
Relevant code:
```
// Snippet from EmbeddingIndexStore.initStore (already exists in the PR)
GraphEntity entity = key2EntityMap.get(key);
if (entity != null) {
this.indexStoreMap.computeIfAbsent(entity, k -> new ArrayList<>()).add(embedding);
}
```
Suggested solution (prioritize stable string keys instead of objects): Change the indexStoreMap to a Map<String, List<EmbeddingResult>> with the entity key (ModelUtils.getGraphEntityKey(entity)) as the key, and replace the key with this string in all reads. This avoids memory issues caused by referencing entity objects and ensures consistency across Accessors/instances.
##########
geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java:
##########
@@ -96,24 +103,45 @@ public void initStore(GraphAccessor graphAccessor, VerbalizationFunction func,
if (line.isEmpty()) {
continue;
}
+ EmbeddingService.EmbeddingResult embedding;
try {
- EmbeddingService.EmbeddingResult embedding =
- new Gson().fromJson(line, EmbeddingService.EmbeddingResult.class);
- String key = embedding.input;
- GraphEntity entity = key2EntityMap.get(key);
- if (entity != null) {
- this.indexStoreMap.computeIfAbsent(entity, k -> new ArrayList<>()).add(embedding);
Review Comment:
If the indexStoreMap is read and written by different threads during runtime (e.g., index initialization on the startup thread, querying on the worker thread), concurrent read/write safety must be guaranteed. Replace the map with a ConcurrentHashMap.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org