Skip to content

Commit eb99cf1

Browse files
premtsd-codepremtsd-code
committed
skipDuplicates: simplify by moving pre-filter to orchestrator only
The SQL adapter previously carried a pre-filter SELECT plus a _createdAt reconciliation pass to compensate for INSERT IGNORE not telling us which rows were actually inserted vs skipped. The Mongo adapter pulled the same trick via Client::upsertWithCounts() (mongo PR #37). Both layers were solving the same problem twice. Move the pre-filter back to Database::createDocuments where it ran originally, and let each adapter's INSERT IGNORE / ON CONFLICT DO NOTHING / upsert+\$setOnInsert serve only as a race-safety net. The adapter just returns its input as the inserted set. - Database.php: restore \$preExistingIds pre-fetch + chunk filter - SQL.php: drop pre-filter, drop _createdAt reconciliation, drop buildUidTenantLookup helper (-173 lines) - Mongo.php: revert to plain Client::upsert(), return input docs (withTransaction skipDuplicates bypass stays — still needed to avoid WriteConflict under snapshot isolation) - composer.json/lock: back to stable utopia-php/mongo 1.* Trade-off: in the race window between the orchestrator's pre-fetch SELECT and the adapter INSERT, \$modified over-reports by N and onNext fires for N skipped docs. Common case stays correct.
1 parent 3a483f2 commit eb99cf1

5 files changed

Lines changed: 45 additions & 235 deletions

File tree

‎composer.json‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,7 @@
44
"type": "library",
55
"keywords": ["php","framework", "upf", "utopia", "database"],
66
"license": "MIT",
7-
"minimum-stability": "dev",
8-
"prefer-stable": true,
7+
"minimum-stability": "stable",
98
"autoload": {
109
"psr-4": {"Utopia\\Database\\": "src/Database"}
1110
},
@@ -42,7 +41,7 @@
4241
"utopia-php/console": "0.1.*",
4342
"utopia-php/cache": "1.*",
4443
"utopia-php/pools": "1.*",
45-
"utopia-php/mongo": "dev-upsert-return-upserted-count"
44+
"utopia-php/mongo": "1.*"
4645
},
4746
"require-dev": {
4847
"fakerphp/faker": "1.23.*",
@@ -60,12 +59,6 @@
6059
"mongodb/mongodb": "Needed to support MongoDB Database Adapter"
6160

6261
},
63-
"repositories": [
64-
{
65-
"type": "vcs",
66-
"url": "https://github.com/utopia-php/mongo.git"
67-
}
68-
],
6962
"config": {
7063
"allow-plugins": {
7164
"php-http/discovery": false,

‎composer.lock‎

Lines changed: 12 additions & 32 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎src/Database/Adapter/Mongo.php‎

Lines changed: 3 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1511,8 +1511,7 @@ public function createDocuments(Document $collection, array $documents): array
15111511
}
15121512

15131513
$operations = [];
1514-
$opIndexToDocIndex = [];
1515-
foreach ($records as $i => $record) {
1514+
foreach ($records as $record) {
15161515
$filter = ['_uid' => $record['_uid'] ?? ''];
15171516
if ($this->sharedTables) {
15181517
$filter['_tenant'] = $record['_tenant'] ?? $this->getTenant();
@@ -1527,31 +1526,19 @@ public function createDocuments(Document $collection, array $documents): array
15271526
continue;
15281527
}
15291528

1530-
$opIndexToDocIndex[\count($operations)] = $i;
15311529
$operations[] = [
15321530
'filter' => $filter,
15331531
'update' => ['$setOnInsert' => $setOnInsert],
15341532
];
15351533
}
15361534

15371535
try {
1538-
$result = $this->client->upsertWithCounts($name, $operations, $options);
1536+
$this->client->upsert($name, $operations, $options);
15391537
} catch (MongoException $e) {
15401538
throw $this->processException($e);
15411539
}
15421540

1543-
// Map MongoDB's upserted entries back to the originating documents
1544-
// so the caller sees an exact "actually inserted" list — matched
1545-
// (pre-existing) docs are silently dropped.
1546-
$inserted = [];
1547-
foreach ($result['upserted'] as $entry) {
1548-
$docIndex = $opIndexToDocIndex[$entry['index']] ?? null;
1549-
if ($docIndex !== null) {
1550-
$inserted[] = $documents[$docIndex];
1551-
}
1552-
}
1553-
1554-
return $inserted;
1541+
return $documents;
15551542
}
15561543

15571544
try {

‎src/Database/Adapter/SQL.php‎

Lines changed: 0 additions & 173 deletions
Original file line numberDiff line numberDiff line change
@@ -2460,68 +2460,6 @@ protected function execute(mixed $stmt): bool
24602460
return $stmt->execute();
24612461
}
24622462

2463-
/**
2464-
* Build a `_uid IN (...) [AND _tenant ...]` WHERE clause (and matching binds)
2465-
* for a batch of documents. Used by the skipDuplicates pre-filter and the
2466-
* race-condition reconciliation path.
2467-
*
2468-
* @param array<Document> $documents
2469-
* @param string $bindPrefix Prefix for bind keys; must differ between concurrent uses.
2470-
* @return array{where: string, tenantSelect: string, binds: array<string, mixed>}
2471-
*/
2472-
private function buildUidTenantLookup(array $documents, string $bindPrefix): array
2473-
{
2474-
$binds = [];
2475-
$placeholders = [];
2476-
$uids = \array_values(\array_unique(\array_filter(
2477-
\array_map(fn (Document $doc) => $doc->getId(), $documents),
2478-
fn ($v) => $v !== null
2479-
)));
2480-
foreach ($uids as $i => $uid) {
2481-
$key = ':' . $bindPrefix . 'uid_' . $i;
2482-
$placeholders[] = $key;
2483-
$binds[$key] = $uid;
2484-
}
2485-
2486-
$tenantFilter = '';
2487-
$tenantSelect = '';
2488-
if ($this->sharedTables) {
2489-
if ($this->tenantPerDocument) {
2490-
$tenants = \array_values(\array_unique(\array_filter(
2491-
\array_map(fn (Document $doc) => $doc->getTenant(), $documents),
2492-
fn ($v) => $v !== null
2493-
)));
2494-
$tenantPlaceholders = [];
2495-
foreach ($tenants as $j => $tenant) {
2496-
$tKey = ':' . $bindPrefix . 'tenant_' . $j;
2497-
$tenantPlaceholders[] = $tKey;
2498-
$binds[$tKey] = $tenant;
2499-
}
2500-
$tenantFilter = ' AND _tenant IN (' . \implode(', ', $tenantPlaceholders) . ')';
2501-
$tenantSelect = ', _tenant';
2502-
} else {
2503-
$tenantKey = ':' . $bindPrefix . 'tenant';
2504-
$tenantFilter = ' AND _tenant = ' . $tenantKey;
2505-
$binds[$tenantKey] = $this->getTenant();
2506-
}
2507-
}
2508-
2509-
// No UIDs to match → return a clause that matches nothing without leaving dangling binds.
2510-
if (empty($placeholders)) {
2511-
return [
2512-
'where' => '1=0',
2513-
'tenantSelect' => $tenantSelect,
2514-
'binds' => [],
2515-
];
2516-
}
2517-
2518-
return [
2519-
'where' => '_uid IN (' . \implode(', ', $placeholders) . ')' . $tenantFilter,
2520-
'tenantSelect' => $tenantSelect,
2521-
'binds' => $binds,
2522-
];
2523-
}
2524-
25252463
/**
25262464
* Create Documents in batches
25272465
*
@@ -2539,51 +2477,6 @@ public function createDocuments(Document $collection, array $documents): array
25392477
return $documents;
25402478
}
25412479

2542-
// Pre-filter existing UIDs to prevent race-condition duplicates.
2543-
if ($this->skipDuplicates) {
2544-
$name = $this->filter($collection->getId());
2545-
$lookup = $this->buildUidTenantLookup($documents, '_dup_');
2546-
2547-
if (!empty($lookup['binds'])) {
2548-
try {
2549-
$sql = 'SELECT _uid' . $lookup['tenantSelect']
2550-
. ' FROM ' . $this->getSQLTable($name)
2551-
. ' WHERE ' . $lookup['where'];
2552-
2553-
$stmt = $this->getPDO()->prepare($sql);
2554-
foreach ($lookup['binds'] as $k => $v) {
2555-
$stmt->bindValue($k, $v, $this->getPDOType($v));
2556-
}
2557-
$stmt->execute();
2558-
$rows = $stmt->fetchAll();
2559-
$stmt->closeCursor();
2560-
2561-
if ($this->sharedTables && $this->tenantPerDocument) {
2562-
$existingKeys = [];
2563-
foreach ($rows as $row) {
2564-
$existingKeys[$row['_tenant'] . ':' . $row['_uid']] = true;
2565-
}
2566-
$documents = \array_values(\array_filter(
2567-
$documents,
2568-
fn (Document $doc) => !isset($existingKeys[$doc->getTenant() . ':' . $doc->getId()])
2569-
));
2570-
} else {
2571-
$existingUids = \array_flip(\array_column($rows, '_uid'));
2572-
$documents = \array_values(\array_filter(
2573-
$documents,
2574-
fn (Document $doc) => !isset($existingUids[$doc->getId()])
2575-
));
2576-
}
2577-
} catch (PDOException $e) {
2578-
throw $this->processException($e);
2579-
}
2580-
}
2581-
2582-
if (empty($documents)) {
2583-
return [];
2584-
}
2585-
}
2586-
25872480
$spatialAttributes = $this->getSpatialAttributes($collection);
25882481
$collection = $collection->getId();
25892482
try {
@@ -2692,72 +2585,6 @@ public function createDocuments(Document $collection, array $documents): array
26922585

26932586
$this->execute($stmt);
26942587

2695-
// Reconcile returned docs with actual inserts when a race condition skipped rows.
2696-
if ($this->skipDuplicates && $stmt->rowCount() < \count($documents)) {
2697-
$expectedTimestamps = [];
2698-
foreach ($documents as $doc) {
2699-
$eKey = ($this->sharedTables && $this->tenantPerDocument)
2700-
? $doc->getTenant() . ':' . $doc->getId()
2701-
: $doc->getId();
2702-
$expectedTimestamps[$eKey] = $doc->getCreatedAt();
2703-
}
2704-
2705-
$lookup = $this->buildUidTenantLookup($documents, '_vfy_');
2706-
$verifySql = 'SELECT _uid' . $lookup['tenantSelect'] . ', ' . $this->quote($this->filter('_createdAt'))
2707-
. ' FROM ' . $this->getSQLTable($name)
2708-
. ' WHERE ' . $lookup['where'];
2709-
2710-
$verifyStmt = $this->getPDO()->prepare($verifySql);
2711-
foreach ($lookup['binds'] as $k => $v) {
2712-
$verifyStmt->bindValue($k, $v, $this->getPDOType($v));
2713-
}
2714-
$verifyStmt->execute();
2715-
$rows = $verifyStmt->fetchAll();
2716-
$verifyStmt->closeCursor();
2717-
2718-
// Normalise timestamps — Postgres omits .000 for round seconds.
2719-
$normalizeTimestamp = fn (?string $ts): string => $ts !== null
2720-
? (new \DateTime($ts))->format('Y-m-d H:i:s.v')
2721-
: '';
2722-
2723-
$actualTimestamps = [];
2724-
foreach ($rows as $row) {
2725-
$key = ($this->sharedTables && $this->tenantPerDocument)
2726-
? $row['_tenant'] . ':' . $row['_uid']
2727-
: $row['_uid'];
2728-
$actualTimestamps[$key] = $normalizeTimestamp($row['_createdAt']);
2729-
}
2730-
2731-
$insertedDocs = [];
2732-
foreach ($documents as $doc) {
2733-
$key = ($this->sharedTables && $this->tenantPerDocument)
2734-
? $doc->getTenant() . ':' . $doc->getId()
2735-
: $doc->getId();
2736-
if (isset($actualTimestamps[$key]) && $actualTimestamps[$key] === $normalizeTimestamp($expectedTimestamps[$key] ?? null)) {
2737-
$insertedDocs[] = $doc;
2738-
}
2739-
}
2740-
$documents = $insertedDocs;
2741-
2742-
// Rebuild permissions for actually-inserted docs only
2743-
$permissions = [];
2744-
$bindValuesPermissions = [];
2745-
foreach ($documents as $index => $document) {
2746-
foreach (Database::PERMISSIONS as $type) {
2747-
foreach ($document->getPermissionsByType($type) as $permission) {
2748-
$tenantBind = $this->sharedTables ? ", :_tenant_{$index}" : '';
2749-
$permission = \str_replace('"', '', $permission);
2750-
$permission = "('{$type}', '{$permission}', :_uid_{$index} {$tenantBind})";
2751-
$permissions[] = $permission;
2752-
$bindValuesPermissions[":_uid_{$index}"] = $document->getId();
2753-
if ($this->sharedTables) {
2754-
$bindValuesPermissions[":_tenant_{$index}"] = $document->getTenant();
2755-
}
2756-
}
2757-
}
2758-
}
2759-
}
2760-
27612588
if (!empty($permissions)) {
27622589
$tenantColumn = $this->sharedTables ? ', _tenant' : '';
27632590
$permissions = \implode(', ', $permissions);

0 commit comments

Comments
 (0)