RemanenetSpy commented on issue #11304:
URL: https://github.com/apache/seatunnel/issues/11304#issuecomment-5543544141
For enterprise agent memory, standard CDC `upsert` / `delete` sinks
introduce a major flaw: a CDC `DELETE` hard-deletes the entity from storage,
destroying causal reasoning history and corrupting active agent decisions that
relied on historical state.
The production pattern is a **Bi-Temporal CDC Memory Sink**:
```java
public class BiTemporalMemorySinkWriter implements SinkWriter<SeaTunnelRow,
Void, Void> {
@Override
public void write(SeaTunnelRow element) {
RowKind kind = element.getRowKind();
String entityId = element.getField(0).toString();
String factTupleJson = serializeSvoTriple(element);
Instant commitTime = Instant.ofEpochMilli(element.getTimestamp());
if (kind == RowKind.INSERT) {
// Ingest as active assertion
storeActiveFact(entityId, factTupleJson, commitTime /*
valid_from */, null /* valid_to */);
} else if (kind == RowKind.UPDATE_AFTER || kind == RowKind.DELETE) {
// Atomic in-process supersession: close validity window without
hard delete
closeFactValidity(entityId, commitTime /* valid_to */);
if (kind == RowKind.UPDATE_AFTER) {
storeActiveFact(entityId, factTupleJson, commitTime /*
valid_from */, null /* valid_to */);
}
}
}
}
```
1. **Zero Data Loss:** `DELETE` marks `valid_to = commit_time`, keeping
historical facts available for time-travel queries ("what was true last week?").
2. **Consistent Read Projections:** Agent pre-flight queries filter on
`WHERE valid_to IS NULL`, achieving sub-millisecond retrieval of current state
while preserving complete enterprise audit lineage.
--
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]
