Add ignore param to createDocuments for silent duplicate handling

premtsd-code · premtsd-code · commit 287420b01b49 · 2026-04-08T17:14:22.000+01:00
- Add `bool $ignore` param to `createDocuments` across all adapters
  (MariaDB: INSERT IGNORE, Postgres: ON CONFLICT DO NOTHING, SQLite:
  INSERT OR IGNORE, MongoDB: ordered:false outside transaction)
- Pre-fetch existing document IDs to skip relationship writes and
  filter known duplicates from adapter batches
- Deduplicate intra-batch documents by ID (first occurrence wins)
  to prevent ACL drift and overcounted results
- Handle tenant-per-document mode with composite keys for pre-fetch
- Chunk all find queries by maxQueryValues
- Optimize upsertDocuments: batch-fetch existing docs with find
  instead of per-row getDocument calls (~2x speedup)
- Add ignore param to Mirror::createDocuments
- Add e2e tests: mixed duplicates, all duplicates, intra-batch
  duplicates with onNext callback assertions
diff --git a/src/Database/Adapter.php b/src/Database/Adapter.php
@@ -729,12 +729,13 @@ abstract public function createDocument(Document $collection, Document $document
      *
      * @param Document $collection
      * @param array<Document> $documents
+     * @param bool $ignore If true, silently ignore duplicate documents instead of throwing
      *
      * @return array<Document>
      *
      * @throws DatabaseException
      */
-    abstract public function createDocuments(Document $collection, array $documents): array;
+    abstract public function createDocuments(Document $collection, array $documents, bool $ignore = false): array;
 
     /**
      * Update Document
diff --git a/src/Database/Adapter/Mongo.php b/src/Database/Adapter/Mongo.php
@@ -1460,11 +1460,18 @@ public function castingBefore(Document $collection, Document $document): Documen
      * @throws DuplicateException
      * @throws DatabaseException
      */
-    public function createDocuments(Document $collection, array $documents): array
+    public function createDocuments(Document $collection, array $documents, bool $ignore = false): array
     {
         $name = $this->getNamespace() . '_' . $this->filter($collection->getId());
 
-        $options = $this->getTransactionOptions();
+        if ($ignore) {
+            // Run outside transaction — MongoDB aborts transactions on any write error,
+            // so ordered:false + session would roll back even successfully inserted docs.
+            $options = ['ordered' => false];
+        } else {
+            $options = $this->getTransactionOptions();
+        }
+
         $records = [];
         $hasSequence = null;
         $documents = \array_map(fn ($doc) => clone $doc, $documents);
@@ -1490,7 +1497,16 @@ public function createDocuments(Document $collection, array $documents): array
         try {
             $documents = $this->client->insertMany($name, $records, $options);
         } catch (MongoException $e) {
-            throw $this->processException($e);
+            $processed = $this->processException($e);
+
+            if ($ignore && $processed instanceof DuplicateException) {
+                // Race condition: a doc was inserted between pre-filter and insertMany.
+                // With ordered:false outside transaction, non-duplicate inserts persist.
+                // Return empty — we cannot determine which docs succeeded without querying.
+                return [];
+            }
+
+            throw $processed;
         }
 
         foreach ($documents as $index => $document) {
diff --git a/src/Database/Adapter/Pool.php b/src/Database/Adapter/Pool.php
@@ -268,7 +268,7 @@ public function createDocument(Document $collection, Document $document): Docume
         return $this->delegate(__FUNCTION__, \func_get_args());
     }
 
-    public function createDocuments(Document $collection, array $documents): array
+    public function createDocuments(Document $collection, array $documents, bool $ignore = false): array
     {
         return $this->delegate(__FUNCTION__, \func_get_args());
     }
diff --git a/src/Database/Adapter/Postgres.php b/src/Database/Adapter/Postgres.php
@@ -1365,6 +1365,35 @@ public function updateDocument(Document $collection, string $id, Document $docum
         return $document;
     }
 
+    protected function getInsertKeyword(bool $ignore): string
+    {
+        return 'INSERT INTO';
+    }
+
+    protected function getInsertSuffix(bool $ignore, string $table): string
+    {
+        if (!$ignore) {
+            return '';
+        }
+
+        $conflictTarget = $this->sharedTables ? '("_uid", "_tenant")' : '("_uid")';
+
+        return "ON CONFLICT {$conflictTarget} DO NOTHING";
+    }
+
+    protected function getInsertPermissionsSuffix(bool $ignore): string
+    {
+        if (!$ignore) {
+            return '';
+        }
+
+        $conflictTarget = $this->sharedTables
+            ? '("_type", "_permission", "_document", "_tenant")'
+            : '("_type", "_permission", "_document")';
+
+        return "ON CONFLICT {$conflictTarget} DO NOTHING";
+    }
+
     /**
      * @param string $tableName
      * @param string $columns
diff --git a/src/Database/Adapter/SQL.php b/src/Database/Adapter/SQL.php
@@ -2471,7 +2471,7 @@ protected function execute(mixed $stmt): bool
      * @throws DuplicateException
      * @throws \Throwable
      */
-    public function createDocuments(Document $collection, array $documents): array
+    public function createDocuments(Document $collection, array $documents, bool $ignore = false): array
     {
         if (empty($documents)) {
             return $documents;
@@ -2573,8 +2573,9 @@ public function createDocuments(Document $collection, array $documents): array
             $batchKeys = \implode(', ', $batchKeys);
 
             $stmt = $this->getPDO()->prepare("
-                INSERT INTO {$this->getSQLTable($name)} {$columns}
+                {$this->getInsertKeyword($ignore)} {$this->getSQLTable($name)} {$columns}
                 VALUES {$batchKeys}
+                {$this->getInsertSuffix($ignore, $name)}
             ");
 
             foreach ($bindValues as $key => $value) {
@@ -2588,8 +2589,9 @@ public function createDocuments(Document $collection, array $documents): array
                 $permissions = \implode(', ', $permissions);
 
                 $sqlPermissions = "
-                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$tenantColumn})
-                    VALUES {$permissions};
+                    {$this->getInsertKeyword($ignore)} {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$tenantColumn})
+                    VALUES {$permissions}
+                    {$this->getInsertPermissionsSuffix($ignore)}
                 ";
 
                 $stmtPermissions = $this->getPDO()->prepare($sqlPermissions);
@@ -2608,6 +2610,33 @@ public function createDocuments(Document $collection, array $documents): array
         return $documents;
     }
 
+    /**
+     * Returns the INSERT keyword, optionally with IGNORE for duplicate handling.
+     * Override in adapter subclasses for DB-specific syntax.
+     */
+    protected function getInsertKeyword(bool $ignore): string
+    {
+        return $ignore ? 'INSERT IGNORE INTO' : 'INSERT INTO';
+    }
+
+    /**
+     * Returns a suffix appended after VALUES clause for duplicate handling.
+     * Override in adapter subclasses (e.g., Postgres uses ON CONFLICT DO NOTHING).
+     */
+    protected function getInsertSuffix(bool $ignore, string $table): string
+    {
+        return '';
+    }
+
+    /**
+     * Returns a suffix for the permissions INSERT statement when ignoring duplicates.
+     * Override in adapter subclasses for DB-specific syntax.
+     */
+    protected function getInsertPermissionsSuffix(bool $ignore): string
+    {
+        return '';
+    }
+
     /**
      * @param Document $collection
      * @param string $attribute
diff --git a/src/Database/Adapter/SQLite.php b/src/Database/Adapter/SQLite.php
@@ -34,6 +34,11 @@
  */
 class SQLite extends MariaDB
 {
+    protected function getInsertKeyword(bool $ignore): string
+    {
+        return $ignore ? 'INSERT OR IGNORE INTO' : 'INSERT INTO';
+    }
+
     /**
      * @inheritDoc
      */
diff --git a/src/Database/Database.php b/src/Database/Database.php
@@ -5621,6 +5621,7 @@ public function createDocument(string $collection, Document $document): Document
      * @param int $batchSize
      * @param (callable(Document): void)|null $onNext
      * @param (callable(Throwable): void)|null $onError
+     * @param bool $ignore If true, silently ignore duplicate documents instead of throwing
      * @return int
      * @throws AuthorizationException
      * @throws StructureException
@@ -5633,6 +5634,7 @@ public function createDocuments(
         int $batchSize = self::INSERT_BATCH_SIZE,
         ?callable $onNext = null,
         ?callable $onError = null,
+        bool $ignore = false,
     ): int {
         if (!$this->adapter->getSharedTables() && $this->adapter->getTenantPerDocument()) {
             throw new DatabaseException('Shared tables must be enabled if tenant per document is enabled.');
@@ -5653,6 +5655,71 @@ public function createDocuments(
         $time = DateTime::now();
         $modified = 0;
 
+        // Deduplicate intra-batch documents by ID when ignore mode is on.
+        // Keeps the first occurrence, mirrors upsertDocuments' seenIds check.
+        if ($ignore) {
+            $seenIds = [];
+            $deduplicated = [];
+            foreach ($documents as $document) {
+                $docId = $document->getId();
+                if ($docId !== '' && isset($seenIds[$docId])) {
+                    continue;
+                }
+                if ($docId !== '') {
+                    $seenIds[$docId] = true;
+                }
+                $deduplicated[] = $document;
+            }
+            $documents = $deduplicated;
+        }
+
+        // When ignore mode is on and relationships are being resolved,
+        // pre-fetch existing document IDs so we skip relationship writes for duplicates
+        $preExistingIds = [];
+        $tenantPerDocument = $this->adapter->getSharedTables() && $this->adapter->getTenantPerDocument();
+        if ($ignore) {
+            if ($tenantPerDocument) {
+                $idsByTenant = [];
+                foreach ($documents as $doc) {
+                    $idsByTenant[$doc->getTenant()][] = $doc->getId();
+                }
+                foreach ($idsByTenant as $tenant => $tenantIds) {
+                    $tenantIds = \array_values(\array_unique($tenantIds));
+                    foreach (\array_chunk($tenantIds, \max(1, $this->maxQueryValues)) as $idChunk) {
+                        $existing = $this->authorization->skip(fn () => $this->withTenant($tenant, fn () => $this->silent(fn () => $this->find(
+                            $collection->getId(),
+                            [
+                                Query::equal('$id', $idChunk),
+                                Query::select(['$id']),
+                                Query::limit(\count($idChunk)),
+                            ]
+                        ))));
+                        foreach ($existing as $doc) {
+                            $preExistingIds[$tenant . ':' . $doc->getId()] = true;
+                        }
+                    }
+                }
+            } else {
+                $inputIds = \array_values(\array_unique(\array_filter(
+                    \array_map(fn (Document $doc) => $doc->getId(), $documents)
+                )));
+
+                foreach (\array_chunk($inputIds, \max(1, $this->maxQueryValues)) as $idChunk) {
+                    $existing = $this->authorization->skip(fn () => $this->silent(fn () => $this->find(
+                        $collection->getId(),
+                        [
+                            Query::equal('$id', $idChunk),
+                            Query::select(['$id']),
+                            Query::limit(\count($idChunk)),
+                        ]
+                    )));
+                    foreach ($existing as $doc) {
+                        $preExistingIds[$doc->getId()] = true;
+                    }
+                }
+            }
+        }
+
         foreach ($documents as $document) {
             $createdAt = $document->getCreatedAt();
             $updatedAt = $document->getUpdatedAt();
@@ -5693,15 +5760,33 @@ public function createDocuments(
             }
 
             if ($this->resolveRelationships) {
-                $document = $this->silent(fn () => $this->createDocumentRelationships($collection, $document));
+                $preExistKey = $tenantPerDocument
+                    ? $document->getTenant() . ':' . $document->getId()
+                    : $document->getId();
+
+                if (!isset($preExistingIds[$preExistKey])) {
+                    $document = $this->silent(fn () => $this->createDocumentRelationships($collection, $document));
+                }
             }
 
             $document = $this->adapter->castingBefore($collection, $document);
         }
 
         foreach (\array_chunk($documents, $batchSize) as $chunk) {
-            $batch = $this->withTransaction(function () use ($collection, $chunk) {
-                return $this->adapter->createDocuments($collection, $chunk);
+            if ($ignore && !empty($preExistingIds)) {
+                $chunk = \array_values(\array_filter($chunk, function (Document $doc) use ($preExistingIds, $tenantPerDocument) {
+                    $key = $tenantPerDocument
+                        ? $doc->getTenant() . ':' . $doc->getId()
+                        : $doc->getId();
+                    return !isset($preExistingIds[$key]);
+                }));
+                if (empty($chunk)) {
+                    continue;
+                }
+            }
+
+            $batch = $this->withTransaction(function () use ($collection, $chunk, $ignore) {
+                return $this->adapter->createDocuments($collection, $chunk, $ignore);
             });
 
             $batch = $this->adapter->getSequences($collection->getId(), $batch);
@@ -7116,18 +7201,53 @@ public function upsertDocumentsWithIncrease(
         $created = 0;
         $updated = 0;
         $seenIds = [];
-        foreach ($documents as $key => $document) {
-            if ($this->getSharedTables() && $this->getTenantPerDocument()) {
-                $old = $this->authorization->skip(fn () => $this->withTenant($document->getTenant(), fn () => $this->silent(fn () => $this->getDocument(
-                    $collection->getId(),
-                    $document->getId(),
-                ))));
+
+        // Batch-fetch existing documents in one query instead of N individual getDocument() calls
+        $ids = \array_filter(\array_map(fn ($doc) => $doc->getId(), $documents));
+        $existingDocs = [];
+        $upsertTenantPerDocument = $this->getSharedTables() && $this->getTenantPerDocument();
+
+        if (!empty($ids)) {
+            $uniqueIds = \array_values(\array_unique($ids));
+
+            if ($upsertTenantPerDocument) {
+                // Group IDs by tenant and fetch each group separately
+                // Use composite key tenant:id to avoid cross-tenant collisions
+                $idsByTenant = [];
+                foreach ($documents as $doc) {
+                    $tenant = $doc->getTenant();
+                    $idsByTenant[$tenant][] = $doc->getId();
+                }
+                foreach ($idsByTenant as $tenant => $tenantIds) {
+                    $tenantIds = \array_values(\array_unique($tenantIds));
+                    foreach (\array_chunk($tenantIds, \max(1, $this->maxQueryValues)) as $idChunk) {
+                        $fetched = $this->authorization->skip(fn () => $this->withTenant($tenant, fn () => $this->silent(fn () => $this->find(
+                            $collection->getId(),
+                            [Query::equal('$id', $idChunk), Query::limit(\count($idChunk))],
+                        ))));
+                        foreach ($fetched as $doc) {
+                            $existingDocs[$tenant . ':' . $doc->getId()] = $doc;
+                        }
+                    }
+                }
             } else {
-                $old = $this->authorization->skip(fn () => $this->silent(fn () => $this->getDocument(
-                    $collection->getId(),
-                    $document->getId(),
-                )));
+                foreach (\array_chunk($uniqueIds, \max(1, $this->maxQueryValues)) as $idChunk) {
+                    $fetched = $this->authorization->skip(fn () => $this->silent(fn () => $this->find(
+                        $collection->getId(),
+                        [Query::equal('$id', $idChunk), Query::limit(\count($idChunk))],
+                    )));
+                    foreach ($fetched as $doc) {
+                        $existingDocs[$doc->getId()] = $doc;
+                    }
+                }
             }
+        }
+
+        foreach ($documents as $key => $document) {
+            $lookupKey = $upsertTenantPerDocument
+                ? $document->getTenant() . ':' . $document->getId()
+                : $document->getId();
+            $old = $existingDocs[$lookupKey] ?? new Document();
 
             // Extract operators early to avoid comparison issues
             $documentArray = $document->getArrayCopy();
diff --git a/src/Database/Mirror.php b/src/Database/Mirror.php
@@ -600,13 +600,15 @@ public function createDocuments(
         int $batchSize = self::INSERT_BATCH_SIZE,
         ?callable $onNext = null,
         ?callable $onError = null,
+        bool $ignore = false,
     ): int {
         $modified = $this->source->createDocuments(
             $collection,
             $documents,
             $batchSize,
             $onNext,
             $onError,
+            $ignore,
         );
 
         if (
@@ -645,6 +647,7 @@ public function createDocuments(
                     $collection,
                     $clones,
                     $batchSize,
+                    ignore: $ignore,
                 )
             );
 
diff --git a/tests/e2e/Adapter/Scopes/DocumentTests.php b/tests/e2e/Adapter/Scopes/DocumentTests.php

Original file line number	Diff line number	Diff line change
`@@ -729,12 +729,13 @@ abstract public function createDocument(Document $collection, Document $document`
`729`	`729`	`*`
`730`	`730`	`* @param Document $collection`
`731`	`731`	`* @param array<Document> $documents`
	`732`	`+ * @param bool $ignore If true, silently ignore duplicate documents instead of throwing`
`732`	`733`	`*`
`733`	`734`	`* @return array<Document>`
`734`	`735`	`*`
`735`	`736`	`* @throws DatabaseException`
`736`	`737`	`*/`
`737`		`- abstract public function createDocuments(Document $collection, array $documents): array;`
	`738`	`+ abstract public function createDocuments(Document $collection, array $documents, bool $ignore = false): array;`
`738`	`739`
`739`	`740`	`/**`
`740`	`741`	`* Update Document`
Original file line number	Diff line number	Diff line change
`@@ -268,7 +268,7 @@ public function createDocument(Document $collection, Document $document): Docume`
`268`	`268`	`return $this->delegate(__FUNCTION__, \func_get_args());`
`269`	`269`	`}`
`270`	`270`
`271`		`- public function createDocuments(Document $collection, array $documents): array`
	`271`	`+ public function createDocuments(Document $collection, array $documents, bool $ignore = false): array`
`272`	`272`	`{`
`273`	`273`	`return $this->delegate(__FUNCTION__, \func_get_args());`
`274`	`274`	`}`