Jackie-Jiang commented on code in PR #19865:
URL: https://github.com/apache/pinot/pull/19865#discussion_r4288281436
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/text/LuceneTextIndexReader.java:
##########
@@ -297,8 +355,38 @@ public int getNumMatchingDocs(String searchQuery,
@Nullable String optionsString
/// Both [#getDocIds(String, String)] and [#getNumMatchingDocs] go through
this method, so a `count(*)` and
/// the filter it counts cannot interpret the same query differently. Keep
it that way: any new entry point
/// that parses a search string itself reintroduces that divergence.
+ /// Returns the parsed query, reusing the one already parsed for this
request when a cache is given.
+ ///
+ /// Every segment of a request parses the same string against the same
configuration, so without a
+ /// cache the work is repeated once per segment. The cache computes a key at
most once, so a cold
+ /// query parses once for the whole request rather than racing across
segment threads.
+ ///
+ /// The returned query may be shared: Lucene queries are immutable and
[Query#rewrite] does not
+ /// mutate them, so a multi-term query is still expanded per segment against
that segment's own
+ /// term dictionary.
+ @VisibleForTesting
+ Query buildQuery(String searchQuery, @Nullable String optionsString,
@Nullable QueryScopedCache queryScopedCache)
+ throws Exception {
+ if (queryScopedCache == null) {
+ return parseQuery(searchQuery, optionsString);
+ }
+ ParsedQueryKey key = new ParsedQueryKey(_parseConfig, searchQuery,
optionsString);
+ try {
+ return queryScopedCache.getOrCompute(Query.class, key, k -> {
+ try {
+ return parseQuery(k.searchQuery(), k.optionsString());
+ } catch (Exception e) {
+ throw new ParseFailure(e);
+ }
+ });
+ } catch (ParseFailure e) {
+ throw (Exception) e.getCause();
+ }
+ }
+
+ /// Parses unconditionally, bypassing any cache.
@VisibleForTesting
- Query buildQuery(String searchQuery, @Nullable String optionsString)
+ Query parseQuery(String searchQuery, @Nullable String optionsString)
Review Comment:
Nonblocking cleanup: clarify the parser helper names
The call sequence is now `buildQuery(searchQuery, options, cache) ->
parseQuery(searchQuery, options) -> buildQuery(searchQuery)`. Consider renaming
the private one-argument `buildQuery` method to `parseQueryWithoutOptions` so
the names make it clear which method handles caching and which performs parsing.
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/readers/text/LuceneTextIndexQueryCacheTest.java:
##########
@@ -0,0 +1,219 @@
+/**
+ * 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.segment.local.segment.index.readers.text;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+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.AtomicInteger;
+import java.util.function.Function;
+import org.apache.commons.io.FileUtils;
+import org.apache.lucene.search.Query;
+import
org.apache.pinot.segment.local.segment.creator.impl.text.LuceneTextIndexCreator;
+import
org.apache.pinot.segment.local.segment.index.text.TextIndexConfigBuilder;
+import org.apache.pinot.segment.spi.index.TextIndexConfig;
+import org.apache.pinot.segment.spi.index.reader.QueryScopedCache;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+
+
+/// Covers the query-scoped reuse of parsed queries in [LuceneTextIndexReader].
+public class LuceneTextIndexQueryCacheTest {
+ private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(),
+ LuceneTextIndexQueryCacheTest.class.getSimpleName() + "-" +
UUID.randomUUID());
+ private static final File INDEX_DIR = new File(TEMP_DIR, "index");
+ private static final String COLUMN = "querycachebody";
+
+ private static final String[] DOCS = {
+ "failed to place order for user alice",
+ "failed to charge card for user bob",
+ "connection refused by payment service",
+ "order confirmation email sent to carol"
+ };
+
+ /// Stands in for `QueryContext`, which lives in pinot-core and is not
visible here. It uses the same
+ /// `computeIfAbsent` contract, and counts how many times a key is actually
computed so a test can
+ /// assert that a parse happened once rather than once per caller.
+ private static final class CountingCache implements QueryScopedCache {
+ private final ConcurrentHashMap<Object, Object> _values = new
ConcurrentHashMap<>();
+ private final AtomicInteger _computeCount = new AtomicInteger();
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <K, V> V getOrCompute(Class<V> type, K key, Function<K, V> mapper) {
+ return (V) _values.computeIfAbsent(key, k -> {
+ _computeCount.incrementAndGet();
+ return mapper.apply((K) k);
+ });
+ }
+
+ int computeCount() {
+ return _computeCount.get();
+ }
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws IOException {
+ FileUtils.forceMkdir(INDEX_DIR);
+ TextIndexConfig config = new TextIndexConfigBuilder().build();
+ try (LuceneTextIndexCreator creator = new LuceneTextIndexCreator(COLUMN,
INDEX_DIR, true, false, null, null,
+ config)) {
+ for (String doc : DOCS) {
+ creator.add(doc);
+ }
+ creator.seal();
+ }
+ }
+
+ @AfterClass
+ public void tearDown() {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ }
+
+ private static TextIndexConfig configWithProperties(Map<String, String>
properties) {
+ return (TextIndexConfig) new
TextIndexConfigBuilder().withProperties(properties).build();
+ }
+
+ private static TextIndexConfig configWithStopWords(List<String>
stopWordsInclude) {
+ TextIndexConfig.AbstractBuilder builder = new TextIndexConfigBuilder();
+ builder.withStopWordsInclude(stopWordsInclude);
+ return (TextIndexConfig) builder.build();
+ }
+
+ private LuceneTextIndexReader reader(TextIndexConfig config) {
+ return new LuceneTextIndexReader(COLUMN, INDEX_DIR, DOCS.length, config);
+ }
+
+ @Test
+ public void testParseIsSharedWithinOneQuery()
+ throws Exception {
+ CountingCache cache = new CountingCache();
+ try (LuceneTextIndexReader first = reader(configWithProperties(Map.of()));
+ LuceneTextIndexReader second = reader(configWithProperties(Map.of())))
{
+ Query a = first.buildQuery("order", null, cache);
+ Query b = second.buildQuery("order", null, cache);
+ assertSame(b, a, "equivalent readers in one query should share the
parsed query");
+ assertEquals(cache.computeCount(), 1, "the query should have been parsed
exactly once");
+ assertNotSame(first.parseQuery("order", null), a, "parseQuery must
bypass the cache");
+ }
+ }
+
+ @Test
+ public void testNoCacheMeansNoSharing()
+ throws Exception {
+ try (LuceneTextIndexReader reader =
reader(configWithProperties(Map.of()))) {
+ Query a = reader.buildQuery("order", null, null);
+ Query b = reader.buildQuery("order", null, null);
+ assertNotSame(b, a);
+ assertEquals(b, a);
+ assertEquals(reader.getNumMatchingDocs("order", null, null), 2);
+ }
+ }
+
+ /// Regression test for a flattened cache key. `["failed", "order"]` and
`["failed, order"]` are
+ /// different stop-word sets -- the first makes `failed` a stop word, the
second does not -- yet both
+ /// render as `[failed, order]`. A key built from rendered text would let
the second configuration
+ /// reuse the first's empty query and silently lose matches.
+ @Test
+ public void testStopWordListsThatRenderIdenticallyDoNotCollide()
+ throws Exception {
+ CountingCache cache = new CountingCache();
+ try (LuceneTextIndexReader twoStopWords =
reader(configWithStopWords(List.of("failed", "order")));
+ LuceneTextIndexReader oneStopWord =
reader(configWithStopWords(List.of("failed, order")))) {
+ Query withFailedAsStopWord = twoStopWords.buildQuery("failed", null,
cache);
+ Query withFailedAsTerm = oneStopWord.buildQuery("failed", null, cache);
+ assertNotEquals(withFailedAsTerm, withFailedAsStopWord,
+ "configurations whose stop-word lists only render the same must not
share a cache entry");
+ assertEquals(cache.computeCount(), 2, "each configuration must be parsed
on its own");
+ assertEquals(withFailedAsStopWord, twoStopWords.parseQuery("failed",
null));
+ assertEquals(withFailedAsTerm, oneStopWord.parseQuery("failed", null));
+ }
+ }
+
+ @Test
+ public void testMultiTermConfigDoesNotCollide()
+ throws Exception {
+ CountingCache cache = new CountingCache();
+ try (LuceneTextIndexReader orReader = reader(
+
configWithProperties(Map.of(FieldConfig.TEXT_INDEX_USE_AND_FOR_MULTI_TERM_QUERIES,
"false")));
+ LuceneTextIndexReader andReader = reader(
+
configWithProperties(Map.of(FieldConfig.TEXT_INDEX_USE_AND_FOR_MULTI_TERM_QUERIES,
"true")))) {
+ assertNotEquals(andReader.buildQuery("failed order", null, cache),
+ orReader.buildQuery("failed order", null, cache));
+ assertEquals(orReader.getNumMatchingDocs("failed order", null, cache),
3);
+ assertEquals(andReader.getNumMatchingDocs("failed order", null, cache),
1);
+ }
+ }
+
+ @Test
+ public void testOptionsStringIsPartOfTheKey()
+ throws Exception {
+ CountingCache cache = new CountingCache();
+ try (LuceneTextIndexReader reader =
reader(configWithProperties(Map.of()))) {
+ assertNotEquals(reader.buildQuery("failed order",
"parser=MATCH,operator=AND", cache),
+ reader.buildQuery("failed order", null, cache));
+ assertEquals(cache.computeCount(), 2);
+ }
+ }
+
+ /// A cold query hit by every segment thread at once must still be parsed
once, not once per thread.
+ @Test
+ public void testConcurrentMissesParseOnce()
+ throws Exception {
+ int threads = 16;
+ CountingCache cache = new CountingCache();
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ try {
+ CountDownLatch start = new CountDownLatch(1);
+ Future<Query>[] futures = new Future[threads];
+ for (int i = 0; i < threads; i++) {
+ futures[i] = executor.submit(() -> {
+ try (LuceneTextIndexReader reader =
reader(configWithProperties(Map.of()))) {
+ start.await();
+ return reader.buildQuery("connection", null, cache);
+ }
+ });
+ }
+ start.countDown();
Review Comment:
Nonblocking cleanup: wait for worker readiness before opening the start gate
The start latch is released immediately after task submission, while workers
open their readers before reaching `start.await()`. A worker can therefore
parse and populate the cache before other workers have finished initialization,
so the test need not exercise concurrent misses. Add a readiness latch that
each worker decrements after opening its reader, wait for all workers with a
bounded timeout, then release the start latch. This would more reliably
exercise concurrent access without relying on reader initialization timing.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/reader/QueryScopedCache.java:
##########
@@ -0,0 +1,41 @@
+/**
+ * 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.segment.spi.index.reader;
+
+import java.util.function.Function;
+
+
+/// A cache whose lifetime is a single query, letting the segments of one
request share a value that
+/// is expensive to derive and identical for all of them.
+///
+/// Index readers live longer than a query and are shared by every query, so
they cannot hold
+/// per-query state themselves; the caller owns the cache and passes it in.
+///
+/// This interface exists so that segment-level code can use a query-scoped
cache without depending
+/// on the query engine. Implementations must compute a given key at most once
and be safe for
Review Comment:
Nonblocking cleanup: state the precise computation guarantee
Please describe atomic computation per `(type, key)` and clarify that failed
or null computations are not retained, matching
`ConcurrentHashMap.computeIfAbsent`. A mapper can run again after throwing or
returning null, so the unconditional "compute a given key at most once" wording
overstates the guarantee. The reader's corresponding comment should also say
once per effective parsing configuration, since one request can encounter
segments with different configurations.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java:
##########
@@ -620,6 +621,11 @@ public int
getSortAggregateSequentialCombineNumSegmentsThreshold() {
/// @param <K> the key type
/// @param <V> the value type
/// @return the shared value
+ @Override
+ public <K, V> V getOrCompute(Class<V> type, K key, Function<K, V> mapper) {
Review Comment:
Nonblocking cleanup: reuse the existing cache method name
Could `QueryScopedCache` declare `getOrComputeSharedValue` instead of
`getOrCompute`? The existing `QueryContext` method could then implement the
interface directly with `@Override`, removing this forwarding method and
keeping its documentation attached to the established API. The small SPI
interface itself is useful for preserving module layering.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/reader/TextIndexReader.java:
##########
@@ -61,6 +61,23 @@ default int getNumMatchingDocs(String searchQuery, @Nullable
String optionsStrin
return getDocIds(searchQuery, optionsString).getCardinality();
}
+ /// Returns the matching document ids, reusing a query-scoped parse when the
implementation supports it.
+ ///
+ /// Parsing a search string depends only on the string and the column's
index configuration, so every
+ /// segment of one request would otherwise repeat identical work.
Implementations that parse should
+ /// derive their cache key from the search string, the options and the
configuration that affects
+ /// parsing. A `null` cache means "do not share", and must behave exactly
like the two-argument form.
+ default MutableRoaringBitmap getDocIds(String searchQuery, @Nullable String
optionsString,
Review Comment:
[P2] Avoid ambiguous overloads for nullable multi-column calls
`MultiColumnTextIndexReader` already exposes `getDocIds(String column,
String searchQuery, @Nullable String optionsString)` and the corresponding
count method. Adding `(String, String, QueryScopedCache)` overloads makes
previously valid calls such as `reader.getDocIds("body", "error", null)` and
`reader.getNumMatchingDocs("body", "error", null)` ambiguous. I compiled the
same small caller against the old and new interfaces: it succeeds before this
change and fails with two ambiguous-method errors afterward. Existing binaries
remain compatible; this affects callers when recompiled. Please put the cache
parameter first or give these methods distinct names so the documented
null-options calls continue to compile.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
