This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 0b053a1f3c2 Fix stale realtime Lucene searchable document counts
(#19945)
0b053a1f3c2 is described below
commit 0b053a1f3c24551f232f41149f6797bcfc8b25c3
Author: Xiang Fu <[email protected]>
AuthorDate: Wed Oct 21 15:06:52 2026 -0700
Fix stale realtime Lucene searchable document counts (#19945)
* Fix stale realtime Lucene searchable document counts
* Use Lucene maxDoc for realtime document fences
* Isolate mock query instance data directories
---
.../testutils/MockInstanceDataManagerFactory.java | 4 +-
.../MockInstanceDataManagerFactoryTest.java | 76 ++++++++++++++++++++++
.../invertedindex/RealtimeLuceneTextIndex.java | 22 ++++++-
.../invertedindex/LuceneMutableTextIndexTest.java | 40 +++++++++++-
4 files changed, 139 insertions(+), 3 deletions(-)
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
index 990bb42cbdd..f4df35b81b6 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
@@ -70,6 +70,8 @@ public class MockInstanceDataManagerFactory {
private final Map<String, Schema> _registeredSchemaMap;
private final String _serverName;
+ // Parallel test classes can use the same server and table names, so each
factory must own its directories.
+ private final UUID _factoryId = UUID.randomUUID();
public MockInstanceDataManagerFactory(String serverName) {
// Parallel tests can use the same logical server name without sharing
their segment directories.
@@ -96,7 +98,7 @@ public class MockInstanceDataManagerFactory {
private void registerTableNameWithType(Schema schema, String
tableNameWithType) {
File tableDataDir = new File(FileUtils.getTempDirectory(),
- String.format("%s_%s_%s", DATA_DIR_PREFIX, _serverName,
tableNameWithType));
+ String.format("%s_%s_%s_%s", DATA_DIR_PREFIX, _serverName,
tableNameWithType, _factoryId));
FileUtils.deleteQuietly(tableDataDir);
_serverTableDataDirMap.put(tableNameWithType, tableDataDir);
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactoryTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactoryTest.java
new file mode 100644
index 00000000000..2c4cbffcdd0
--- /dev/null
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactoryTest.java
@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.testutils;
+
+import java.io.File;
+import java.util.List;
+import java.util.UUID;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// Verifies filesystem isolation between mock factories. Each test owns its
factories and segment files.
+public class MockInstanceDataManagerFactoryTest {
+ @Test
+ public void testFactoriesWithSameServerAndTableHaveIndependentDirectories()
+ throws Exception {
+ String serverName = "server_" + UUID.randomUUID();
+ String tableName = "test_OFFLINE";
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("test")
+ .addSingleValueDimension("id", DataType.INT)
+ .addDateTime("ts", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:HOURS")
+ .build();
+ GenericRow row = new GenericRow();
+ row.putValue("id", 1);
+ row.putValue("ts", 1704067200000L);
+ List<GenericRow> rows = List.of(row);
+ MockInstanceDataManagerFactory firstFactory = new
MockInstanceDataManagerFactory(serverName);
+ firstFactory.registerTable(schema, tableName);
+ ImmutableSegment firstSegment = firstFactory.addSegment(tableName, rows);
+ File firstSegmentDir = firstSegment.getSegmentMetadata().getIndexDir();
+ ImmutableSegment secondSegment = null;
+ try {
+ assertTrue(firstSegmentDir.isDirectory());
+ MockInstanceDataManagerFactory secondFactory = new
MockInstanceDataManagerFactory(serverName);
+ secondFactory.registerTable(schema, tableName);
+ assertTrue(firstSegmentDir.isDirectory(), "Registering another factory
must preserve existing segment files");
+
+ secondSegment = secondFactory.addSegment(tableName, rows);
+ File secondSegmentDir = secondSegment.getSegmentMetadata().getIndexDir();
+ assertTrue(secondSegmentDir.isDirectory());
+ assertNotEquals(firstSegmentDir.getParentFile(),
secondSegmentDir.getParentFile());
+ assertTrue(firstSegmentDir.isDirectory(), "Building another factory's
segment must preserve existing files");
+ } finally {
+ firstSegment.destroy();
+ FileUtils.deleteDirectory(firstSegmentDir.getParentFile());
+ if (secondSegment != null) {
+ File secondSegmentDir =
secondSegment.getSegmentMetadata().getIndexDir();
+ secondSegment.destroy();
+ FileUtils.deleteDirectory(secondSegmentDir.getParentFile());
+ }
+ }
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneTextIndex.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneTextIndex.java
index d2ea109f74b..9fe514de892 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneTextIndex.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneTextIndex.java
@@ -19,6 +19,7 @@
package org.apache.pinot.segment.local.realtime.impl.invertedindex;
import java.io.File;
+import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.Map;
import java.util.concurrent.Future;
@@ -339,8 +340,27 @@ public class RealtimeLuceneTextIndex implements
MutableTextIndex {
public SearcherManager getSearcherManager() {
return _searcherManager;
}
+
@Override
public int getSearchableDocCount() {
- return _refreshListener.getLastRefreshNumDocs();
+ // The refresh listener records a lower bound that can miss documents
added during a refresh.
+ SearcherManager searcherManager = _searcherManager;
+ IndexSearcher indexSearcher = null;
+ try {
+ indexSearcher = searcherManager.acquire();
+ // This is a doc-id fence, so deleted document slots must remain within
its bounds.
+ return indexSearcher.getIndexReader().maxDoc();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to get searchable document count for
realtime text index for segment "
+ + _segmentName + " for column " + _column, e);
+ } finally {
+ if (indexSearcher != null) {
+ try {
+ searcherManager.release(indexSearcher);
+ } catch (IOException e) {
+ LOGGER.error("Failed while releasing searcher manager for column
{}", _column, e);
+ }
+ }
+ }
}
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
index 8d71f6a21f0..247590e627e 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/LuceneMutableTextIndexTest.java
@@ -27,11 +27,13 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.io.FileUtils;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.core.KeywordTokenizer;
import org.apache.lucene.queryparser.classic.QueryParser;
+import org.apache.lucene.search.ReferenceManager.RefreshListener;
import org.apache.pinot.common.metrics.ServerMetrics;
import
org.apache.pinot.segment.local.segment.index.text.TextIndexConfigBuilder;
import org.apache.pinot.segment.spi.index.TextIndexConfig;
@@ -295,7 +297,7 @@ public class LuceneMutableTextIndexTest {
index.add(new String[]{"foo bar"});
index.add(new String[]{"baz qux"});
- // Block until the refresh attempt completes so the listener has
recorded the current doc count
+ // Block until the documents are visible to the current searcher.
index.getSearcherManager().maybeRefreshBlocking();
assertEquals(index.getSearchableDocCount(), 3);
@@ -304,6 +306,42 @@ public class LuceneMutableTextIndexTest {
}
}
+ @Test
+ public void testGetSearchableDocCountWhenDocumentIsAddedDuringRefresh()
+ throws IOException {
+ TextIndexConfig config = new
TextIndexConfigBuilder().withUseANDForMultiTermQueries(false).build();
+ RealtimeLuceneTextIndex index = new
RealtimeLuceneTextIndex(TEXT_COLUMN_NAME, INDEX_DIR,
+ "table__0__" + SEGMENT_NAME_SUFFIX_COUNTER.getAndIncrement() +
"__20240601T1818Z", config);
+ try {
+ index.add("hello world");
+ index.add("foo bar");
+ AtomicBoolean documentAdded = new AtomicBoolean();
+ index.getSearcherManager().addListener(new RefreshListener() {
+ @Override
+ public void beforeRefresh() {
+ // The production listener has captured two docs, but the refreshed
reader will include this third doc.
+ if (documentAdded.compareAndSet(false, true)) {
+ index.add("baz qux");
+ }
+ }
+
+ @Override
+ public void afterRefresh(boolean didRefresh) {
+ }
+ });
+ index.getSearcherManager().maybeRefreshBlocking();
+
+ assertEquals(index.getDocIds("baz"), ImmutableRoaringBitmap.bitmapOf(2));
+ assertEquals(index.getSearchableDocCount(), 3);
+
+ // A refresh with no further writes must retain the full searchable
document count.
+ index.getSearcherManager().maybeRefreshBlocking();
+ assertEquals(index.getSearchableDocCount(), 3);
+ } finally {
+ index.close();
+ }
+ }
+
@Test
public void testQueries() {
configureIndex(null, null, null, null);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
