From d277d6a5e71541aaf330e11a148c549ba93d14e9 Mon Sep 17 00:00:00 2001 From: Darek Date: Wed, 14 Oct 2026 06:45:15 +0200 Subject: [PATCH] fix(fts): skip rows COPY rejects with ignore_errors COPY indexes its rows in FTSIndex::finalize, after the whole file is read, but a row rejected for a duplicate primary key is deleted earlier, through NodeTable::delete_. FTSIndex::delete_ then looked for a document that was never indexed and threw "FTS index is inconsistent", failing the whole COPY. delete_ now skips a row past the index's indexed prefix when the docs table has no document for it, and finalize skips rows whose lookup fails instead of indexing the previous row's values under the deleted offset, which added a phantom document and skewed BM25 scores. Co-Authored-By: Claude Opus 5.5 --- fts/src/include/index/fts_index.h | 6 ++ fts/src/index/fts_index.cpp | 41 ++++++++--- fts/test/test_files/copy_ignore_errors.test | 82 +++++++++++++++++++++ 3 files changed, 119 insertions(+), 10 deletions(-) create mode 100644 fts/test/test_files/copy_ignore_errors.test diff --git a/fts/src/include/index/fts_index.h b/fts/src/include/index/fts_index.h index 0daf7cc2..4acfb74a 100644 --- a/fts/src/include/index/fts_index.h +++ b/fts/src/include/index/fts_index.h @@ -73,6 +73,9 @@ class FTSIndex final : public storage::Index { void finalize(main::ClientContext* context) override; std::pair getStats(const transaction::Transaction* transaction) const; + // Rows below this offset are in the index; COPY appends rows past it and indexes them in + // finalize(). + common::offset_t getNumIndexedNodes(const transaction::Transaction* transaction) const; void checkpoint(main::ClientContext*, storage::PageAllocator& pageAllocator, storage::ShadowFile& shadowFile) override; @@ -97,6 +100,9 @@ class FTSIndex final : public storage::Index { FTSInsertState& ftsInsertState, common::nodeID_t docID, common::table_id_t termsTableID) const; + bool containsDoc(transaction::Transaction* transaction, FTSDeleteState& deleteState, + common::nodeID_t nodeID) const; + common::nodeID_t deleteFromDocTable(transaction::Transaction* transaction, FTSDeleteState& deleteState, common::nodeID_t deletedNodeID, uint64_t& deletedDocLen) const; diff --git a/fts/src/index/fts_index.cpp b/fts/src/index/fts_index.cpp index 56f737c8..34a690c2 100644 --- a/fts/src/index/fts_index.cpp +++ b/fts/src/index/fts_index.cpp @@ -183,11 +183,18 @@ std::unique_ptr FTSIndex::initDeleteState(const Transaction* void FTSIndex::delete_(Transaction* transaction, const ValueVector& nodeIDVector, DeleteState& deleteState) { auto& ftsDeleteState = deleteState.cast(); + const auto numIndexedNodes = getNumIndexedNodes(transaction); uint64_t totalDeletedDocLen = 0; idx_t numDeletedDocs = 0; for (auto i = 0u; i < nodeIDVector.state->getSelSize(); i++) { auto pos = nodeIDVector.state->getSelVector()[i]; auto deletedNodeID = nodeIDVector.getValue(pos); + // A row COPY rejected (ignore_errors) is deleted before finalize() would index it. Above + // the indexed prefix the doc table decides, so an indexed row is still deleted. + if (deletedNodeID.offset >= numIndexedNodes && + !containsDoc(transaction, ftsDeleteState, deletedNodeID)) { + continue; + } ftsDeleteState.updateVectors.idVector.setValue(0, deletedNodeID); internalTableInfo.table->initScanState(transaction, *ftsDeleteState.indexTableState.scanState, deletedNodeID.tableID, deletedNodeID.offset); @@ -222,6 +229,16 @@ std::pair FTSIndex::getStats(const Transaction* transaction) cons return {static_cast(numDocs), numDocs == 0 ? 0 : totalDocLen / numDocs}; } +offset_t FTSIndex::getNumIndexedNodes(const Transaction* transaction) const { + const auto stats = storageInfo->cast().stats; + std::lock_guard lck{stats->mtx}; + auto numIndexedNodes = stats->numCheckpointedNodes; + if (const auto it = stats->deltas.find(transaction->getID()); it != stats->deltas.end()) { + numIndexedNodes = std::max(numIndexedNodes, it->second.numCheckpointedNodes); + } + return numIndexedNodes; +} + void FTSIndex::updateStats(Transaction* transaction, int64_t numDocs, double totalDocLen, offset_t numCheckpointedNodes) { if (numDocs == 0 && totalDocLen == 0 && numCheckpointedNodes == 0) { @@ -263,16 +280,8 @@ void FTSIndex::updateStats(Transaction* transaction, int64_t numDocs, double tot } void FTSIndex::finalize(main::ClientContext* context) { - const auto stats = storageInfo->cast().stats; auto transaction = transaction::Transaction::Get(*context); - common::offset_t numCheckpointedNodes; - { - std::lock_guard lck{stats->mtx}; - numCheckpointedNodes = stats->numCheckpointedNodes; - if (const auto it = stats->deltas.find(transaction->getID()); it != stats->deltas.end()) { - numCheckpointedNodes = std::max(numCheckpointedNodes, it->second.numCheckpointedNodes); - } - } + const auto numCheckpointedNodes = getNumIndexedNodes(transaction); const auto numTotalRows = internalTableInfo.table->getNumTotalRows(&DUMMY_CHECKPOINT_TRANSACTION); if (numTotalRows == numCheckpointedNodes) { @@ -288,7 +297,10 @@ void FTSIndex::finalize(main::ClientContext* context) { idVector.setValue(0, insertedNodeID); internalTableInfo.table->initScanState(transaction, *indexTableState.scanState, insertedNodeID.tableID, insertedNodeID.offset); - internalTableInfo.table->lookup(transaction, *indexTableState.scanState); + // Rows COPY rejected are already deleted; their vectors would still hold the previous row. + if (!internalTableInfo.table->lookup(transaction, *indexTableState.scanState)) { + continue; + } auto insertState = initInsertState(context, [](offset_t) { return true; }); insert(transaction, idVector, indexTableState.indexVectors, *insertState); } @@ -366,6 +378,15 @@ void FTSIndex::insertToAppearsInTable(Transaction* transaction, } } +bool FTSIndex::containsDoc(Transaction* transaction, FTSDeleteState& deleteState, + nodeID_t nodeID) const { + auto& pkVector = deleteState.updateVectors.int64PKVector; + pkVector.setValue(0, nodeID.offset); + offset_t docOffset = INVALID_OFFSET; + return internalTableInfo.docTable->lookupPK(transaction, &pkVector, 0 /* vectorPos */, + docOffset); +} + nodeID_t FTSIndex::deleteFromDocTable(Transaction* transaction, FTSDeleteState& deleteState, nodeID_t deletedNodeID, uint64_t& deletedDocLen) const { auto& pkVector = deleteState.updateVectors.int64PKVector; diff --git a/fts/test/test_files/copy_ignore_errors.test b/fts/test/test_files/copy_ignore_errors.test new file mode 100644 index 00000000..6614730c --- /dev/null +++ b/fts/test/test_files/copy_ignore_errors.test @@ -0,0 +1,82 @@ +-DATASET CSV empty-db + +-- + +# COPY with ignore_errors removes a rejected row before the FTS index has seen it. +-CASE CopySkipsDuplicateWithText +-SKIP_IN_MEM +-LOAD_DYNAMIC_EXTENSION fts +-STATEMENT CREATE NODE TABLE Doc(id INT64 PRIMARY KEY, body STRING); +---- ok +-STATEMENT CREATE (:Doc {id: 1, body: 'apple'}); +---- ok +-STATEMENT CALL CREATE_FTS_INDEX('Doc', 'idx', ['body']); +---- ok +-STATEMENT COPY (UNWIND [1, 2] AS i RETURN i, CASE i WHEN 1 THEN 'banana' ELSE 'cherry' END) TO '${LBUG_EXPORT_DB_DIRECTORY}_dup.csv' (header=false); +---- ok +-STATEMENT COPY Doc FROM '${LBUG_EXPORT_DB_DIRECTORY}_dup.csv' (header=false, ignore_errors=true); +---- ok +-STATEMENT MATCH (d:Doc) RETURN d.id, d.body ORDER BY d.id; +---- 2 +1|apple +2|cherry +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'cherry') RETURN node.id; +---- 1 +2 +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'banana') RETURN node.id; +---- 0 +-STATEMENT MATCH (d:Doc {id: 2}) DELETE d; +---- ok +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'cherry') RETURN node.id; +---- 0 + +# A rejected row without indexable text must not be indexed with the previous row's text. +-CASE CopySkipsDuplicateWithoutText +-SKIP_IN_MEM +-LOAD_DYNAMIC_EXTENSION fts +-STATEMENT CREATE NODE TABLE Doc(id INT64 PRIMARY KEY, body STRING); +---- ok +-STATEMENT CREATE (:Doc {id: 1, body: 'apple'}); +---- ok +-STATEMENT CALL CREATE_FTS_INDEX('Doc', 'idx', ['body']); +---- ok +-STATEMENT COPY (UNWIND [2, 1, 3] AS i RETURN i, CASE i WHEN 2 THEN 'banana cherry' WHEN 1 THEN '' ELSE 'date' END) TO '${LBUG_EXPORT_DB_DIRECTORY}_dup_empty.csv' (header=false); +---- ok +-STATEMENT COPY Doc FROM '${LBUG_EXPORT_DB_DIRECTORY}_dup_empty.csv' (header=false, ignore_errors=true); +---- ok +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'banana') RETURN node.id, score; +---- 1 +2|0.353634 +-STATEMENT CALL DROP_FTS_INDEX('Doc', 'idx'); +---- ok +-STATEMENT CALL CREATE_FTS_INDEX('Doc', 'idx', ['body']); +---- ok +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'banana') RETURN node.id, score; +---- 1 +2|0.353634 + +# The rejected row is the last one in the file; rows written after the COPY stay deletable. +-CASE CopyRejectsLastRow +-SKIP_IN_MEM +-LOAD_DYNAMIC_EXTENSION fts +-STATEMENT CREATE NODE TABLE Doc(id INT64 PRIMARY KEY, body STRING); +---- ok +-STATEMENT CREATE (:Doc {id: 1, body: 'apple'}); +---- ok +-STATEMENT CALL CREATE_FTS_INDEX('Doc', 'idx', ['body']); +---- ok +-STATEMENT COPY (UNWIND [2, 1] AS i RETURN i, CASE i WHEN 2 THEN 'cherry' ELSE 'banana' END) TO '${LBUG_EXPORT_DB_DIRECTORY}_dup_last.csv' (header=false); +---- ok +-STATEMENT COPY Doc FROM '${LBUG_EXPORT_DB_DIRECTORY}_dup_last.csv' (header=false, ignore_errors=true); +---- ok +-STATEMENT CREATE (:Doc {id: 3, body: 'date'}); +---- ok +-STATEMENT MATCH (d:Doc) WHERE d.id >= 2 DELETE d; +---- ok +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'cherry') RETURN node.id; +---- 0 +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'date') RETURN node.id; +---- 0 +-STATEMENT CALL QUERY_FTS_INDEX('Doc', 'idx', 'apple') RETURN node.id; +---- 1 +1