arunkumarucet commented on code in PR #19865:
URL: https://github.com/apache/pinot/pull/19865#discussion_r4290886061
##########
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:
Done in `3f7bbbfa17`. `QueryScopedCache` now declares
`getOrComputeSharedValue`, so `QueryContext` implements it directly and the
forwarding method is gone. Worth noting your instinct was right about it being
the established name: the other eight call sites in pinot-core already used it,
so nothing else had to change.
##########
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:
Done in `3f7bbbfa17`. You're right that the old wording overstated it. The
interface now describes atomic computation per `(type, key)`, states that a
mapper which throws or returns `null` leaves nothing cached so a later call
runs it again, and points at `ConcurrentHashMap#computeIfAbsent` as the
reference semantics. The reader's comment now says the string is parsed once
per *effective parsing configuration* rather than once per segment, and notes
that one request can span segments configured differently.
##########
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:
Done in `3f7bbbfa17`. The private one-argument method is now
`parseQueryWithoutOptions`, so the chain reads `buildQuery` -> `parseQuery` ->
`parseQueryWithoutOptions` and the caching step is distinguishable from the
parsing steps by name. I also folded a stray sentence back into the opening
paragraph of that comment while I was there.
##########
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:
Done in `3f7bbbfa17`. Good catch — opening a reader costs far more than the
parse under test, so the old gate could let an early worker populate the cache
while the others were still in their constructor. Each worker now counts down a
readiness latch after opening its reader, and the main thread waits for all of
them with a bounded timeout before releasing the start gate.
--
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]
