Refactor InMemoryTable entity storage to track changes - #1258
Merged
Conversation
Replace the entityWithIndices row (latest + status + entityIndices) with three flat structures on Entity.t: - latestEntityChangeById: the latest Change per entity (checkpointId 0n marks a value loaded from the db that never becomes history) - prevEntityChanges: demotion-ordered superseded changes - indicesByEntityId: per-entity index set writeBatch regroups prevEntityChanges by entity id to rebuild the inMemoryStoreEntityUpdate boundary, so PgStorage/ClickHouse consumers are unchanged. Removes the now-unused inMemoryStoreEntityStatus type.
Contributor
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughThis PR converts in-memory entity storage to a change-log model (latestEntityChangeById + prevEntityChanges), adds loadedFromDbCheckpointId to mark DB-loaded changes, and propagates a flat per-entity changes array through InMemoryStore.writeBatch, Postgres/ClickHouse persistence, sink wiring, and test serialization/processing. ChangesEntity Storage and Persistence Refactor
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Comment |
Replace the per-entity-id {latestChange, history} grouping in the
writeBatch boundary (Persistence.updatedEntity) with a flat
`changes: array<Change.t>` log. Consumers regroup as needed:
- PgStorage/ClickHouse via Internal.groupChangesByEntityId
- TestIndexer locally over its serialized change form
Introduce Internal.loadedFromDbCheckpointId for the 0n sentinel marking
db-loaded changes, and rename InMemoryTable.Entity.rowToEntity to
mapChangeToEntity now that it maps a Change.
…o-g3QYt # Conflicts: # packages/envio/src/bindings/ClickHouse.res
entity_history insert order isn't meaningful (checkpointId is the ordering key), so groupChangesByEntityId no longer preserves it and the queryHistory test helper sorts by (entityId, checkpointId) for stable assertions.
- ClickHouse setUpdatesOrThrow compiles a single S.array converter instead of mapping each change, converting the whole batch in one call. - Inline the per-id grouping into PgStorage's setEntities (its only remaining user) as a single-pass dict of change groups, iterating group arrays by index to avoid intermediate update records and history slices. - Drop the now-unused Internal.groupChangesByEntityId and inMemoryStoreEntityUpdate.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
scenarios/test_codegen/test/helpers/MockIndexer.res (1)
427-456: 💤 Low valuePotential precision loss when comparing large checkpoint IDs.
BigInt.toFloaton line 451 can lose precision for checkpoint IDs exceeding 2^53 (~9 quadrillion). While unlikely in practice, this could cause non-deterministic ordering for very large IDs.Consider using
BigInt.compareor string comparison for full precision:Alternative using string comparison
->Array.toSorted((a, b) => { switch String.compare(a->Change.getEntityId, b->Change.getEntityId) { | 0. => - Float.compare( - a->Change.getCheckpointId->BigInt.toFloat, - b->Change.getCheckpointId->BigInt.toFloat, - ) + // Compare as strings with zero-padding for consistent ordering + let aId = a->Change.getCheckpointId->BigInt.toString + let bId = b->Change.getCheckpointId->BigInt.toString + String.compare(aId->String.padStart(20, "0"), bId->String.padStart(20, "0")) | order => order } })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scenarios/test_codegen/test/helpers/MockIndexer.res` around lines 427 - 456, The comparator passed to Array.toSorted currently converts checkpoint IDs via BigInt.toFloat (used on Change.getCheckpointId) which can lose precision for very large IDs; replace the Float conversion with a full-precision comparison (e.g., use BigInt.compare on a->Change.getCheckpointId and b->Change.getCheckpointId) or compare the checkpointId strings to determine order so stable ordering is preserved; update the comparator in the Array.toSorted block to call BigInt.compare (or String.compare on BigInt.toString) instead of BigInt.toFloat.packages/envio/src/TestIndexer.res (1)
161-161: 💤 Low valueMissing type annotation on
Utils.magic.Per coding guidelines,
Utils.magicshould have explicit type annotations.Suggested fix
- entityChange.sets->Array.push(parsedEntity->Utils.magic)->ignore + entityChange.sets->Array.push(parsedEntity->(Utils.magic: Internal.entity => unknown))->ignoreAs per coding guidelines: "When using
Utils.magicfor type casting in ReScript, always add explicit type annotations:value->(Utils.magic: inputType => outputType)"🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/envio/src/TestIndexer.res` at line 161, The call to Utils.magic used when pushing into entityChange.sets lacks an explicit ReScript type annotation; change parsedEntity->Utils.magic to use an explicit cast like parsedEntity->(Utils.magic: inputType => outputType) where inputType is the actual type of parsedEntity and outputType is the element type expected by entityChange.sets (the collection element type used by Array.push). Update the expression referenced (entityChange.sets->Array.push(parsedEntity->Utils.magic)->ignore) to include this explicit annotation so the compiler and readers know the intended conversion.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/envio/src/TestIndexer.res`:
- Around line 200-209: The current loop over changesByEntityId turns each
entityChanges into history that excludes the last change and only calls
processChange for the last change when history is empty, which drops the latest
change for multi-change entities; update the logic in the block that iterates
over changesByEntityId/Dict.valuesToArray/Array.forEach so that after computing
lastIdx and history you always call processChange for the last element
(entityChanges->Array.getUnsafe(lastIdx)) in addition to processing history (or
change the slice to include the last element), ensuring processChange is invoked
for the final/latest change for every entity.
---
Nitpick comments:
In `@packages/envio/src/TestIndexer.res`:
- Line 161: The call to Utils.magic used when pushing into entityChange.sets
lacks an explicit ReScript type annotation; change parsedEntity->Utils.magic to
use an explicit cast like parsedEntity->(Utils.magic: inputType => outputType)
where inputType is the actual type of parsedEntity and outputType is the element
type expected by entityChange.sets (the collection element type used by
Array.push). Update the expression referenced
(entityChange.sets->Array.push(parsedEntity->Utils.magic)->ignore) to include
this explicit annotation so the compiler and readers know the intended
conversion.
In `@scenarios/test_codegen/test/helpers/MockIndexer.res`:
- Around line 427-456: The comparator passed to Array.toSorted currently
converts checkpoint IDs via BigInt.toFloat (used on Change.getCheckpointId)
which can lose precision for very large IDs; replace the Float conversion with a
full-precision comparison (e.g., use BigInt.compare on a->Change.getCheckpointId
and b->Change.getCheckpointId) or compare the checkpointId strings to determine
order so stable ordering is preserved; update the comparator in the
Array.toSorted block to call BigInt.compare (or String.compare on
BigInt.toString) instead of BigInt.toFloat.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 16f31282-5a65-4b93-9f9c-5314af10de95
📒 Files selected for processing (11)
packages/envio/src/InMemoryStore.respackages/envio/src/InMemoryTable.respackages/envio/src/Internal.respackages/envio/src/LoadLayer.respackages/envio/src/Persistence.respackages/envio/src/PgStorage.respackages/envio/src/Sink.respackages/envio/src/TestIndexer.respackages/envio/src/TestIndexerProxyStorage.respackages/envio/src/bindings/ClickHouse.resscenarios/test_codegen/test/helpers/MockIndexer.res
TestIndexer.handleWriteBatch dropped an entity's latest change whenever it had history in the batch (a pre-existing quirk). Each (id, checkpointId) appears at most once in the flat change log, so just process every change into its checkpoint bucket. PgStorage.setEntities now builds the latest-per-id map, the entity-table sets/deletes, the history-table batches, and the backfill set in a single pass over the change log plus one pass over the unique ids, instead of grouping into per-id arrays and re-walking them. History batches are built eagerly alongside entitiesToSet so they survive transaction retries.
This was referenced Jun 1, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Refactored the
InMemoryTable.Entitymodule to store entity changes directly instead of wrapping them in an intermediateentityWithIndicesrecord. This simplifies the data structure and makes change tracking more explicit.Key Changes
Replaced
entityWithIndicesrecord with a flatter structure:entities: dict<entityWithIndices>→latestEntityChangeById: dict<Change.t<Internal.entity>>prevEntityChanges: array<Change.t<Internal.entity>>to track historical changesindicesByEntityId: dict<entityIndices>to separate index trackingRemoved
inMemoryStoreEntityStatustype fromInternal.res:UpdatedandLoadedvariants0nmarks database-loaded values that never become historyUpdated entity change handling:
setfunction now directly storesChange.t<Internal.entity>instead of constructing status wrappersInMemoryStore.writeBatchby collectingprevEntityChanges0nis used as a sentinel to distinguish database-loaded entities from user changesSimplified index management:
getOrCreateEntityIndicesnow takesself: tandentityIdinstead of a row referenceupdateIndicesanddeleteEntityFromIndicesno longer require passing row objectsindicesByEntityIddictionary directlyUpdated
rowToEntityhelper to extract entity fromChange.tinstead of accessing alatestfieldNotable Implementation Details
prevEntityChangesarray during persistence, rather than stored inline with each entity0neliminates the need for an explicitLoadedstatus varianthttps://claude.ai/code/session_01VnnXgvyEfHMFBCURZyhaqw
Summary by CodeRabbit