Skip to content

Update overlay migration code to use DbMutationBatchStore #6268

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
May 12, 2022
Merged

Update overlay migration code to use DbMutationBatchStore #6268

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8d2536e
wip.
ehsannas May 11, 2022
2575fcc
remove console.log.
ehsannas May 11, 2022
a6a4a5d
remove .only for test.
ehsannas May 11, 2022
2cf4c72
lint fixes.
ehsannas May 11, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion packages/firestore/src/local/indexeddb_schema.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,8 @@ export const INDEXING_SCHEMA_VERSION = 15;
* 12. Add document overlays.
* 13. Rewrite the keys of the remote document cache to allow for efficient
* document lookup via `getAll()`.
* 14. Add indexing support.
* 14. Add overlays.
* 15. Add indexing support.
*/

export const SCHEMA_VERSION = INDEXING_ENABLED ? INDEXING_SCHEMA_VERSION : 14;
Expand Down
103 changes: 39 additions & 64 deletions packages/firestore/src/local/indexeddb_schema_converter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import { User } from '../auth/user';
import { ListenSequence } from '../core/listen_sequence';
import { SnapshotVersion } from '../core/snapshot_version';
import { documentKeySet } from '../model/collections';
import { DocumentKeySet, documentKeySet } from '../model/collections';
import { DocumentKey } from '../model/document_key';
import { ResourcePath } from '../model/path';
import { debugAssert, fail, hardAssert } from '../util/assert';
Expand Down Expand Up @@ -471,9 +471,6 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
db: IDBDatabase,
transaction: SimpleDbTransaction
): PersistencePromise<void> {
const queuesStore = transaction.store<DbMutationQueueKey, DbMutationQueue>(
DbMutationQueueStore
);
const mutationsStore = transaction.store<
DbMutationBatchKey,
DbMutationBatch
Expand All @@ -482,78 +479,56 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
const remoteDocumentCache = newIndexedDbRemoteDocumentCache(
this.serializer
);
const memoryPersistence = new MemoryPersistence(
MemoryEagerDelegate.factory,
this.serializer.remoteSerializer
);

const promises: Array<PersistencePromise<void>> = [];
let userIds = new Set<string>();
const userToDocumentSet = new Map<string, DocumentKeySet>();

return queuesStore
return mutationsStore
.loadAll()
.next(queues => {
for (const queue of queues) {
const userId = queue.userId;
if (userIds.has(userId)) {
// We have already processed this user.
continue;
}
userIds = userIds.add(userId);
.next(dbBatches => {
dbBatches.forEach(dbBatch => {
let documentSet =
userToDocumentSet.get(dbBatch.userId) ?? documentKeySet();
const batch = fromDbMutationBatch(this.serializer, dbBatch);
batch.keys().forEach(key => (documentSet = documentSet.add(key)));
userToDocumentSet.set(dbBatch.userId, documentSet);
});
})
.next(() => {
userToDocumentSet.forEach((allDocumentKeysForUser, userId) => {
const user = new User(userId);
const documentOverlayCache = IndexedDbDocumentOverlayCache.forUser(
this.serializer,
user
);
let allDocumentKeysForUser = documentKeySet();
const range = IDBKeyRange.bound(
[userId, BATCHID_UNKNOWN],
[userId, Number.POSITIVE_INFINITY]
// NOTE: The index manager and the reference delegate are
// irrelevant for the purpose of recalculating and saving
// overlays. We can therefore simply use the memory
// implementation.
const indexManager = memoryPersistence.getIndexManager(user);
const mutationQueue = IndexedDbMutationQueue.forUser(
user,
this.serializer,
indexManager,
memoryPersistence.referenceDelegate
);
const localDocumentsView = new LocalDocumentsView(
remoteDocumentCache,
mutationQueue,
documentOverlayCache,
indexManager
);
promises.push(
mutationsStore
.loadAll(DbMutationBatchUserMutationsIndex, range)
.next(dbBatches => {
dbBatches.forEach(dbBatch => {
hardAssert(
dbBatch.userId === userId,
`Cannot process batch ${dbBatch.batchId} from unexpected user`
);
const batch = fromDbMutationBatch(this.serializer, dbBatch);
batch
.keys()
.forEach(
key =>
(allDocumentKeysForUser =
allDocumentKeysForUser.add(key))
);
});
})
.next(() => {
// NOTE: The index manager and the reference delegate are
// irrelevant for the purpose of recalculating and saving
// overlays. We can therefore simply use the memory
// implementation.
const memoryPersistence = new MemoryPersistence(
MemoryEagerDelegate.factory,
this.serializer.remoteSerializer
);
const indexManager = memoryPersistence.getIndexManager(user);
const mutationQueue = IndexedDbMutationQueue.forUser(
user,
this.serializer,
indexManager,
memoryPersistence.referenceDelegate
);
const localDocumentsView = new LocalDocumentsView(
remoteDocumentCache,
mutationQueue,
documentOverlayCache,
indexManager
);
return localDocumentsView.recalculateAndSaveOverlaysForDocumentKeys(
new IndexedDbTransaction(transaction, ListenSequence.INVALID),
allDocumentKeysForUser
);
})
localDocumentsView.recalculateAndSaveOverlaysForDocumentKeys(
new IndexedDbTransaction(transaction, ListenSequence.INVALID),
allDocumentKeysForUser
)
);
}
});
})
.next(() => PersistencePromise.waitFor(promises));
}
Expand Down
26 changes: 1 addition & 25 deletions packages/firestore/test/unit/local/indexeddb_persistence.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1138,11 +1138,7 @@ describe('IndexedDbSchema: createOrUpgradeDb', () => {
DbDocumentMutationKey,
DbDocumentMutation
>(DbDocumentMutationStore);
const mutationQueuesStore = txn.store<
DbMutationQueueKey,
DbMutationQueue
>(DbMutationQueueStore);
// Manually populate the mutation queue and create all indicies.
// Manually populate the mutations.
return PersistencePromise.forEach(
testMutations,
(testMutation: DbMutationBatch) => {
Expand All @@ -1163,25 +1159,6 @@ describe('IndexedDbSchema: createOrUpgradeDb', () => {
);
});
}
).next(() =>
// Populate the mutation queues' metadata
PersistencePromise.waitFor([
mutationQueuesStore.put({
userId: 'user1',
lastAcknowledgedBatchId: -1,
lastStreamToken: ''
}),
mutationQueuesStore.put({
userId: 'user2',
lastAcknowledgedBatchId: -1,
lastStreamToken: ''
}),
mutationQueuesStore.put({
userId: 'user3',
lastAcknowledgedBatchId: -1,
lastStreamToken: ''
})
])
);
}
);
Expand All @@ -1202,7 +1179,6 @@ describe('IndexedDbSchema: createOrUpgradeDb', () => {
// We should have a total of 4 overlays:
// For user1: testWriteFoo
// For user2: testWriteBar, testWriteBaz, and testWritePending
// For user3: NO OVERLAYS!
let p = documentOverlayStore.count().next(count => {
expect(count).to.equal(4);
});
Expand Down