From ce27b796816ccc376cdfca3efd9147bc7fb90bad Mon Sep 17 00:00:00 2001 From: hoshinojyunn Date: Mon, 13 Jul 2026 14:23:44 +0800 Subject: [PATCH] [fix](clucene) Support block WAND in composite readers ### What problem does this PR solve? Issue Number: close #62955 Related PR: None Problem Summary: Implement block postings APIs in MultiTermDocs so composite readers globalize physical block docids, expose their block statistics, and cross child-reader boundaries. A cross-reader skip must invalidate the caller block cache even when the new child has fewer postings than the skip interval and reports no local block skip. Preserve short physical tail blocks without merging them across readers. Add composite-reader tests for 511/512/tail block layouts and transitions into short readers. ### Release note None ### Check List (For Author) - Test: Unit Test (be/ut_build_ASAN/bin/cl_test testReadRange) - Behavior changed: Yes (composite readers report cross-reader block state changes) - Does this need documentation: No --- src/core/CLucene/index/MultiSegmentReader.cpp | 75 +++++ src/core/CLucene/index/SegmentTermDocs.cpp | 3 +- src/core/CLucene/index/_MultiSegmentReader.h | 6 + src/test/index/TestReadRange.cpp | 277 ++++++++++++++++++ 4 files changed, 360 insertions(+), 1 deletion(-) diff --git a/src/core/CLucene/index/MultiSegmentReader.cpp b/src/core/CLucene/index/MultiSegmentReader.cpp index c102808b4a9..f56eb90afd2 100644 --- a/src/core/CLucene/index/MultiSegmentReader.cpp +++ b/src/core/CLucene/index/MultiSegmentReader.cpp @@ -757,6 +757,33 @@ bool MultiTermDocs::readRange(DocRange* docRange) { } } +bool MultiTermDocs::readBlock(DocRange* docRange) { + while (true) { + while (current == NULL) { + if (pointer < subReaders->length) { + base = starts[pointer]; + current = termDocs(pointer++); + } else { + return false; + } + } + if (!current->readBlock(docRange)) { + current = nullptr; + } else { + if (docRange->doc_many && docRange->doc_many_size_ > 0) { + auto begin = docRange->doc_many->begin(); + auto end = docRange->doc_many->begin() + docRange->doc_many_size_; + std::transform(begin, end, begin, [this](int32_t val) { return val + base; }); + } + if (docRange->type_ == DocRangeType::kRange) { + docRange->doc_range.first += base; + docRange->doc_range.second += base; + } + return true; + } + } +} + bool MultiTermDocs::skipTo(const int32_t target) { // do { // if (!next()) @@ -775,6 +802,54 @@ bool MultiTermDocs::skipTo(const int32_t target) { } } +bool MultiTermDocs::skipToBlock(const int32_t target) { + bool switched_reader = false; + while (true) { + while (current == NULL) { + if (pointer < subReaders->length) { + base = starts[pointer]; + current = termDocs(pointer++); + } else { + return true; + } + } + + if (target >= starts[pointer]) { + current = nullptr; + switched_reader = true; + continue; + } + + // A child without a skip list returns false even though switching readers + // invalidates the caller's cached block metadata. + return switched_reader || current->skipToBlock(target - base); + } +} + +int32_t MultiTermDocs::getMaxBlockFreq() { + return current != NULL ? current->getMaxBlockFreq() : -1; +} + +int32_t MultiTermDocs::getMaxBlockNorm() { + return current != NULL ? current->getMaxBlockNorm() : -1; +} + +int32_t MultiTermDocs::getLastDocInBlock() { + if (current == NULL) { + return LUCENE_INT32_MAX_SHOULDBE; + } + + int32_t lastDoc = current->getLastDocInBlock(); + if (lastDoc == -1) { + return -1; + } + if (lastDoc == LUCENE_INT32_MAX_SHOULDBE) { + return pointer < subReaders->length ? starts[pointer] - 1 + : LUCENE_INT32_MAX_SHOULDBE; + } + return base + lastDoc; +} + void MultiTermDocs::close() { //Func - Closes all MultiTermDocs managed by this instance //Pre - true diff --git a/src/core/CLucene/index/SegmentTermDocs.cpp b/src/core/CLucene/index/SegmentTermDocs.cpp index 58aa3c38dad..ee53328ecdf 100644 --- a/src/core/CLucene/index/SegmentTermDocs.cpp +++ b/src/core/CLucene/index/SegmentTermDocs.cpp @@ -307,7 +307,8 @@ bool SegmentTermDocs::skipToBlock(const int32_t target) { freqStream->seek(skipListReader->getFreqPointer()); skipProx(skipListReader->getProxPointer(), skipListReader->getPayloadLength()); _doc = skipListReader->getDoc(); - count = newCount; + // -1 means the target is in the first skip block and no postings were skipped. + count = newCount < 0 ? 0 : newCount; return true; } return false; diff --git a/src/core/CLucene/index/_MultiSegmentReader.h b/src/core/CLucene/index/_MultiSegmentReader.h index 02f718a66ce..356a9b485c2 100644 --- a/src/core/CLucene/index/_MultiSegmentReader.h +++ b/src/core/CLucene/index/_MultiSegmentReader.h @@ -173,9 +173,15 @@ class MultiTermDocs:public virtual TermDocs { int32_t read(int32_t* docs, int32_t* freqs, int32_t length); int32_t read(int32_t* docs, int32_t* freqs, int32_t* norms , int32_t length); bool readRange(DocRange* docRange) override; + bool readBlock(DocRange* docRange) override; /* A Possible future optimization could skip entire segments */ bool skipTo(const int32_t target); + bool skipToBlock(const int32_t target) override; + + int32_t getMaxBlockFreq() override; + int32_t getMaxBlockNorm() override; + int32_t getLastDocInBlock() override; void close(); diff --git a/src/test/index/TestReadRange.cpp b/src/test/index/TestReadRange.cpp index ea2bf8edd7a..5f8647bab8c 100644 --- a/src/test/index/TestReadRange.cpp +++ b/src/test/index/TestReadRange.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include @@ -986,6 +987,279 @@ void TestBlockBasedInterfaces(CuTest* tc) { std::cout << "\nTestBlockBasedInterfaces success" << std::endl; } +//============================================================================= +// Test: block APIs on MultiTermDocs through MultiReader +//============================================================================= +void TestMultiReaderBlockApis(CuTest* tc) { + constexpr int32_t firstReaderDocumentCount = 1149; + constexpr int32_t secondReaderDocumentCount = 1134; + const std::string fieldName = "content"; + const std::string termText = "target"; + + std::vector firstDocuments; + std::vector secondDocuments; + firstDocuments.reserve(firstReaderDocumentCount); + secondDocuments.reserve(secondReaderDocumentCount); + for (int32_t i = 0; i < firstReaderDocumentCount; ++i) { + firstDocuments.emplace_back(termText + " " + termText); + } + for (int32_t i = 0; i < secondReaderDocumentCount; ++i) { + secondDocuments.emplace_back(termText + " " + termText + " " + termText); + } + + RAMDirectory firstDirectory; + RAMDirectory secondDirectory; + writeTestIndex(fieldName, &firstDirectory, IndexVersion::kV4, firstDocuments); + writeTestIndex(fieldName, &secondDirectory, IndexVersion::kV4, secondDocuments); + + ValueArray readers(2); + readers[0] = IndexReader::open(&firstDirectory); + readers[1] = IndexReader::open(&secondDirectory); + auto* reader = _CLNEW MultiReader(&readers, true); + std::exception_ptr eptr; + Term* term = nullptr; + + try { + std::wstring fieldNameW = StringUtil::string_to_wstring(fieldName); + std::wstring termTextW = StringUtil::string_to_wstring(termText); + term = _CLNEW Term(fieldNameW.c_str(), termTextW.c_str()); + + TermDocs* nextDocs = reader->termDocs(); + nextDocs->seek(term); + TermDocsResult expected = readWithNext(nextDocs); + nextDocs->close(); + _CLDELETE(nextDocs); + + TermDocs* blockDocs = reader->termDocs(); + blockDocs->seek(term); + TermDocsResult actual; + DocRange docRange; + bool sawSecondReader = false; + bool sawBlockMetadata = false; + bool sawSecondReaderBlockBoundary = false; + bool blockDocsAreOrdered = true; + bool blockMetadataIsValid = true; + int32_t previousDoc = -1; + std::vector blockSizes; + while (blockDocs->readBlock(&docRange)) { + blockDocsAreOrdered = blockDocsAreOrdered && docRange.doc_many_size_ > 0; + blockSizes.push_back(docRange.doc_many_size_); + + int32_t actualMaxFreq = 0; + for (uint32_t i = 0; i < docRange.doc_many_size_; ++i) { + int32_t doc = (*docRange.doc_many)[i]; + int32_t freq = (*docRange.freq_many)[i]; + blockDocsAreOrdered = blockDocsAreOrdered && doc > previousDoc; + previousDoc = doc; + actual.docs.push_back(doc); + actual.freqs.push_back(freq); + actualMaxFreq = std::max(actualMaxFreq, freq); + sawSecondReader = sawSecondReader || doc >= firstReaderDocumentCount; + } + + int32_t maxBlockFreq = blockDocs->getMaxBlockFreq(); + int32_t maxBlockNorm = blockDocs->getMaxBlockNorm(); + int32_t lastDocInBlock = blockDocs->getLastDocInBlock(); + if (maxBlockFreq >= 0 && maxBlockNorm >= 0) { + sawBlockMetadata = true; + blockMetadataIsValid = blockMetadataIsValid && maxBlockFreq >= actualMaxFreq; + } + if (lastDocInBlock >= 0) { + const int32_t lastDocInRange = + (*docRange.doc_many)[docRange.doc_many_size_ - 1]; + blockMetadataIsValid = + blockMetadataIsValid && + lastDocInBlock >= lastDocInRange; + sawSecondReaderBlockBoundary = sawSecondReaderBlockBoundary || + (*docRange.doc_many)[0] >= firstReaderDocumentCount; + } + } + blockDocs->close(); + _CLDELETE(blockDocs); + + assertTrueMsg(_T("all composite block docs must match next()"), + compareTermDocsResults(expected, actual)); + assertTrueMsg(_T("composite block docs must remain ordered"), blockDocsAreOrdered); + assertTrueMsg(_T("composite block read must reach the second reader"), sawSecondReader); + assertTrueMsg(_T("composite block metadata must be available"), sawBlockMetadata); + assertTrueMsg(_T("composite block metadata must cross the reader boundary"), + sawSecondReaderBlockBoundary); + assertTrueMsg(_T("block metadata must cover the returned physical block"), + blockMetadataIsValid); + + const std::vector expectedBlockSizes = {511, 512, 126, 511, 512, 111}; + assertTrueMsg(_T("composite reader must preserve short physical tail blocks"), + blockSizes == expectedBlockSizes); + + for (int32_t target : {1023, firstReaderDocumentCount - 1, firstReaderDocumentCount, + firstReaderDocumentCount + 37}) { + TermDocs* skippedDocs = reader->termDocs(); + skippedDocs->seek(term); + skippedDocs->skipToBlock(target); + + std::vector actualTail; + DocRange skippedRange; + while (skippedDocs->readBlock(&skippedRange)) { + for (uint32_t i = 0; i < skippedRange.doc_many_size_; ++i) { + int32_t doc = (*skippedRange.doc_many)[i]; + if (target >= firstReaderDocumentCount) { + assertTrue(doc >= firstReaderDocumentCount); + } + if (doc >= target) { + actualTail.push_back(doc); + } + } + } + skippedDocs->close(); + _CLDELETE(skippedDocs); + + std::vector expectedTail; + for (int32_t doc : expected.docs) { + if (doc >= target) { + expectedTail.push_back(doc); + } + } + assertTrue(actualTail == expectedTail); + } + } catch (...) { + eptr = std::current_exception(); + } + + FINALLY(eptr, { + _CLDECDELETE(term); + reader->close(); + _CLDELETE(reader); + }) + + std::cout << "\nTestMultiReaderBlockApis success" << std::endl; +} + +//============================================================================= +// Test: changing child readers invalidates block state even when the new child +// has fewer postings than a skip block. +//============================================================================= +void TestMultiReaderBlockSkipAcrossShortReaders(CuTest* tc) { + constexpr int32_t firstReaderDocumentCount = 128; + constexpr int32_t secondReaderDocumentCount = 11; + const std::string fieldName = "content"; + const std::string termText = "target"; + + std::vector firstDocuments(firstReaderDocumentCount, termText); + std::vector secondDocuments(secondReaderDocumentCount, termText); + RAMDirectory firstDirectory; + RAMDirectory secondDirectory; + writeTestIndex(fieldName, &firstDirectory, IndexVersion::kV4, firstDocuments); + writeTestIndex(fieldName, &secondDirectory, IndexVersion::kV4, secondDocuments); + + ValueArray readers(2); + readers[0] = IndexReader::open(&firstDirectory); + readers[1] = IndexReader::open(&secondDirectory); + auto* reader = _CLNEW MultiReader(&readers, true); + std::exception_ptr eptr; + Term* term = nullptr; + + try { + std::wstring fieldNameW = StringUtil::string_to_wstring(fieldName); + std::wstring termTextW = StringUtil::string_to_wstring(termText); + term = _CLNEW Term(fieldNameW.c_str(), termTextW.c_str()); + + TermDocs* termDocs = reader->termDocs(); + termDocs->seek(term); + + DocRange firstRange; + assertTrue(termDocs->readBlock(&firstRange)); + assertEquals(firstReaderDocumentCount, firstRange.doc_many_size_); + assertTrue(!termDocs->skipToBlock(firstReaderDocumentCount - 1)); + + // The second child has no skip list, so only MultiTermDocs can report + // the state transition that invalidates a WAND block cache. + assertTrue(termDocs->skipToBlock(firstReaderDocumentCount)); + + DocRange secondRange; + assertTrue(termDocs->readBlock(&secondRange)); + assertEquals(secondReaderDocumentCount, secondRange.doc_many_size_); + for (uint32_t i = 0; i < secondRange.doc_many_size_; ++i) { + assertEquals(firstReaderDocumentCount + static_cast(i), + (*secondRange.doc_many)[i]); + } + assertTrue(!termDocs->readBlock(&secondRange)); + + termDocs->close(); + _CLDELETE(termDocs); + } catch (...) { + eptr = std::current_exception(); + } + + FINALLY(eptr, { + _CLDECDELETE(term); + reader->close(); + _CLDELETE(reader); + }) + + std::cout << "\nTestMultiReaderBlockSkipAcrossShortReaders success" << std::endl; +} + +//============================================================================= +// Test: after skipping a block in a large child, entering a short child still +// reports the reader transition to a block-WAND caller. +//============================================================================= +void TestMultiReaderBlockSkipIntoShortReader(CuTest* tc) { + constexpr int32_t firstReaderDocumentCount = 1149; + constexpr int32_t secondReaderDocumentCount = 11; + const std::string fieldName = "content"; + const std::string termText = "target"; + + std::vector firstDocuments(firstReaderDocumentCount, termText); + std::vector secondDocuments(secondReaderDocumentCount, termText); + RAMDirectory firstDirectory; + RAMDirectory secondDirectory; + writeTestIndex(fieldName, &firstDirectory, IndexVersion::kV4, firstDocuments); + writeTestIndex(fieldName, &secondDirectory, IndexVersion::kV4, secondDocuments); + + ValueArray readers(2); + readers[0] = IndexReader::open(&firstDirectory); + readers[1] = IndexReader::open(&secondDirectory); + auto* reader = _CLNEW MultiReader(&readers, true); + std::exception_ptr eptr; + Term* term = nullptr; + + try { + std::wstring fieldNameW = StringUtil::string_to_wstring(fieldName); + std::wstring termTextW = StringUtil::string_to_wstring(termText); + term = _CLNEW Term(fieldNameW.c_str(), termTextW.c_str()); + + TermDocs* termDocs = reader->termDocs(); + termDocs->seek(term); + + // This skip enters the first reader's final short physical block. + assertTrue(termDocs->skipToBlock(1023)); + // This one changes readers; the new reader has df < skipInterval. + assertTrue(termDocs->skipToBlock(firstReaderDocumentCount)); + + DocRange shortReaderRange; + assertTrue(termDocs->readBlock(&shortReaderRange)); + assertEquals(secondReaderDocumentCount, shortReaderRange.doc_many_size_); + for (uint32_t i = 0; i < shortReaderRange.doc_many_size_; ++i) { + assertEquals(firstReaderDocumentCount + static_cast(i), + (*shortReaderRange.doc_many)[i]); + } + assertTrue(!termDocs->readBlock(&shortReaderRange)); + + termDocs->close(); + _CLDELETE(termDocs); + } catch (...) { + eptr = std::current_exception(); + } + + FINALLY(eptr, { + _CLDECDELETE(term); + reader->close(); + _CLDELETE(reader); + }) + + std::cout << "\nTestMultiReaderBlockSkipIntoShortReader success" << std::endl; +} + //============================================================================= // Suite registration //============================================================================= @@ -1000,6 +1274,9 @@ CuSuite* testReadRange() { SUITE_ADD_TEST(suite, TestReadRangeVersions); SUITE_ADD_TEST(suite, TestReadRangeEdgeCases); SUITE_ADD_TEST(suite, TestBlockBasedInterfaces); + SUITE_ADD_TEST(suite, TestMultiReaderBlockApis); + SUITE_ADD_TEST(suite, TestMultiReaderBlockSkipAcrossShortReaders); + SUITE_ADD_TEST(suite, TestMultiReaderBlockSkipIntoShortReader); return suite; }