diff --git a/changelog/unreleased/auxindexjoin-qparser.yml b/changelog/unreleased/auxindexjoin-qparser.yml new file mode 100644 index 000000000000..3548e368da76 --- /dev/null +++ b/changelog/unreleased/auxindexjoin-qparser.yml @@ -0,0 +1,10 @@ +title: > + Introducing {!auxIndexJoin} query for query-time join with auxiliary index. +type: added +authors: + - name: Mikhail Khludnev + nick: mkhl +links: + - name: SOLR-18307 + url: https://issues.apache.org/jira/browse/SOLR-18307 + diff --git a/solr/core/src/java/org/apache/solr/search/join/AuxIndexJoinQParserPlugin.java b/solr/core/src/java/org/apache/solr/search/join/AuxIndexJoinQParserPlugin.java new file mode 100644 index 000000000000..928e2f0dde5b --- /dev/null +++ b/solr/core/src/java/org/apache/solr/search/join/AuxIndexJoinQParserPlugin.java @@ -0,0 +1,276 @@ +/* + * 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.solr.search.join; + +import java.io.IOException; +import java.io.OutputStream; +import java.lang.invoke.MethodHandles; +import java.nio.file.Path; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.Query; +import org.apache.lucene.store.Directory; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.params.CommonParams; +import org.apache.solr.common.params.SolrParams; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.core.CloseHook; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.DirectoryFactory.DirContext; +import org.apache.solr.core.SolrCore; +import org.apache.solr.request.SolrQueryRequest; +import org.apache.solr.request.SolrQueryRequestBase; +import org.apache.solr.request.SolrRequestInfo; +import org.apache.solr.response.QueryResponseWriter; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.search.QParser; +import org.apache.solr.search.QParserPlugin; +import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.search.SyntaxError; +import org.apache.solr.search.join.auxindexjoin.AuxIndexJoinConfig; +import org.apache.solr.search.join.auxindexjoin.AuxIndexManager; +import org.apache.solr.util.RefCounted; +import org.apache.solr.util.plugin.SolrCoreAware; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Query parser exercising {@link AuxIndexManager} inside a {@link SolrCore}: it mimics {@link + * ScoreJoinQParserPlugin}'s local parameters, but resolves matches through the sidecar join index + * instead of {@link org.apache.lucene.search.join.JoinUtil}. Local parameters: + * + * + * + * Example: {@code q={!aijoin from=manu_id_s to=id fromIndex=products}foo}. + * + *

Unlike {@link ScoreJoinQParserPlugin.OtherCoreJoinQuery}, which only borrows the from-side + * searcher long enough to build a self-contained {@code Query} in {@code createWeight}, an {@link + * org.apache.solr.search.join.auxindexjoin.AuxIndexJoinQuery} keeps reading the from-side searcher + * on every {@code scorerSupplier} call (it may lazily build missing pair columns per to-segment), + * so a cross-core from-searcher is pinned open for the whole request via {@link + * SolrRequestInfo#addCloseHook}, the same mechanism {@link + * org.apache.solr.search.JoinQuery.JoinQueryWeight} uses for the regular {@code {!join}}. + * + *

One {@link AuxIndexManager} is opened per core in {@link #inform(SolrCore)}, backed by a + * directory under the core's dataDir (configurable via the {@code dir} init parameter, resolved + * relative to dataDir unless absolute), and closed when the core closes. The remaining init + * parameters mirror {@link AuxIndexJoinConfig}: {@code singleFieldPerSegment}, {@code + * blockingRefresh}, {@code useFromSideThreads}, and {@code sweepSamplingInterval} (seconds). This + * sidecar always belongs to the "to" side core -- the one this plugin is registered in. + * + *

Why this implements {@link QueryResponseWriter}: {@link + * org.apache.solr.core.SolrResourceLoader}'s {@code awareCompatibility} allowlist (see SOLR-8311) + * only lets specific plugin base types implement {@link SolrCoreAware}, and {@code QParserPlugin} + * isn't one of them, so a plain {@code implements SolrCoreAware} fails core load with "Invalid + * 'Aware' object". {@code QueryResponseWriter} is on the allowlist and happens to be the cheapest + * interface there to satisfy (two abstract methods, both unreachable stubs below -- this class is + * never registered as a {@code }). This is safe here specifically because + * {@code QParserPlugin} instances are loaded once per core load/reload via {@link + * org.apache.solr.core.PluginBag}, exactly like the already-whitelisted {@link + * org.apache.solr.handler.component.SearchComponent} -- never created ad-hoc per request ({@link + * QParser#getParser(String, SolrQueryRequest)} resolves the already registered instance via {@code + * req.getCore().getQueryPlugin(name)}). + */ +public class AuxIndexJoinQParserPlugin extends QParserPlugin + implements QueryResponseWriter, SolrCoreAware { + + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + /** + * Init parameter: directory holding the sidecar join index, resolved against the core's dataDir + * unless absolute. Defaults to {@value #DEFAULT_DIR}. + */ + public static final String DIR = "dir"; + + public static final String NAME = "auxIndexJoin"; + + public static final String DEFAULT_DIR = "aux-index-join"; + + /** Init parameter: whether each pair column is flushed into its own sidecar segment. */ + public static final String SINGLE_FIELD_PER_SEGMENT = "singleFieldPerSegment"; + + /** Init parameter: whether writing a batch of pair columns blocks until it is searchable. */ + public static final String BLOCKING_REFRESH = "blockingRefresh"; + + /** + * Init parameter: whether loading from-side leaves that feed a join build is parallelized across + * executor threads. Defaults to {@code true}. + */ + public static final String USE_FROM_SIDE_THREADS = "useFromSideThreads"; + + /** + * Init parameter: how often (in seconds) {@code AuxIndexManager.onCreateWeight} samples searcher + * state for the dead-pair reaper. Non-positive means sample on every call. Defaults to 60. + */ + public static final String SWEEP_SAMPLING_INTERVAL = "sweepSamplingInterval"; + + private String configuredDir = DEFAULT_DIR; + private final AuxIndexJoinConfig joinIndexConfig = new AuxIndexJoinConfig(); + + private volatile AuxIndexManager joinIndex; + + @Override + public void init(NamedList args) { + super.init(args); + if (args != null) { + SolrParams params = args.toSolrParams(); + configuredDir = params.get(DIR, DEFAULT_DIR); + joinIndexConfig.setSingleFieldPerSegment( + params.getBool(SINGLE_FIELD_PER_SEGMENT, joinIndexConfig.getSingleFieldPerSegment())); + joinIndexConfig.setBlockingRefresh( + params.getBool(BLOCKING_REFRESH, joinIndexConfig.getBlockingRefresh())); + joinIndexConfig.setUseFromSideThreads( + params.getBool(USE_FROM_SIDE_THREADS, joinIndexConfig.getUseFromSideThreads())); + joinIndexConfig.setSweepSamplingInterval( + params.getLong(SWEEP_SAMPLING_INTERVAL, 60), TimeUnit.SECONDS); + } + } + + @Override + public void inform(SolrCore core) { + Path path = Path.of(configuredDir); + if (!path.isAbsolute()) { + path = Path.of(core.getDataDir()).resolve(path); + } else { + core.getCoreContainer().assertPathAllowed(path); + } + Directory directory = null; + try { + directory = + core.getDirectoryFactory() + .get(path.toString(), DirContext.DEFAULT, core.getSolrConfig().indexConfig.lockType); + AuxIndexJoinConfig config = joinIndexConfig; + joinIndex = new AuxIndexManager(directory, config); + } catch (IOException | RuntimeException e) { + if (directory != null) { + try { + core.getDirectoryFactory().release(directory); + } catch (IOException releaseException) { + e.addSuppressed(releaseException); + } + } + throw new SolrException( + SolrException.ErrorCode.SERVER_ERROR, "Failed to open AuxIndexManager at " + path, e); + } + final Directory capturedDirectory = directory; + core.addCloseHook( + new CloseHook() { + @Override + public void preClose(SolrCore core) { + try { + joinIndex.close(); + } catch (IOException e) { + log.warn("Failed closing AuxIndexManager", e); + } finally { + try { + core.getDirectoryFactory().release(capturedDirectory); + } catch (IOException e) { + log.warn("Failed releasing AuxIndexManager directory {}", capturedDirectory, e); + } + } + } + }); + } + + // QueryResponseWriter stubs, unreachable: implemented only to satisfy SolrCoreAware's allowlist, + // see the class javadoc. This plugin is never registered as a . + + @Override + public void write( + OutputStream out, SolrQueryRequest request, SolrQueryResponse response, String contentType) { + throw new UnsupportedOperationException( + AuxIndexJoinQParserPlugin.class.getSimpleName() + + " is a QParserPlugin, not a QueryResponseWriter"); + } + + @Override + public String getContentType(SolrQueryRequest request, SolrQueryResponse response) { + throw new UnsupportedOperationException( + AuxIndexJoinQParserPlugin.class.getSimpleName() + + " is a QParserPlugin, not a QueryResponseWriter"); + } + + @Override + public QParser createParser( + String qstr, SolrParams localParams, SolrParams params, SolrQueryRequest req) { + return new QParser(qstr, localParams, params, req) { + @Override + public Query parse() throws SyntaxError { + if (joinIndex == null) { + throw new SolrException( + SolrException.ErrorCode.SERVER_ERROR, + "AuxIndexJoinQParserPlugin is not initialized; is it registered as a ?"); + } + final String fromField = getParam("from"); + final String toField = getParam("to"); + if (fromField == null || toField == null) { + throw new SyntaxError("auxIndexJoin query parser requires 'from' and 'to' local params"); + } + final String fromIndex = localParams.get("fromIndex"); + final String v = localParams.get(CommonParams.VALUE); + final String myCore = req.getCore().getCoreDescriptor().getName(); + + final Query fromQuery; + final IndexSearcher fromSearcher; + ExecutorService fromExecutor; + if (fromIndex != null && !fromIndex.equals(myCore)) { + CoreContainer container = req.getCoreContainer(); + String coreName = + ScoreJoinQParserPlugin.getCoreName( + fromIndex, container, req.getCore(), toField, fromField, localParams); + SolrCore fromCore = container.getCore(coreName); + if (fromCore == null) { + throw new SolrException( + SolrException.ErrorCode.BAD_REQUEST, "Cross-core join: no such core " + coreName); + } + SolrRequestInfo info = SolrRequestInfo.getRequestInfo(); + if (info == null) { + fromCore.close(); + throw new SolrException( + SolrException.ErrorCode.BAD_REQUEST, + "Cross-core auxIndexJoin must have SolrRequestInfo"); + } + // released once this request completes: the from-side searcher is read on every + // scorerSupplier() call, not just while building this query, so it must outlive parse() + info.addCloseHook(fromCore); + try (SolrQueryRequestBase otherReq = new SolrQueryRequestBase(fromCore, params)) { + fromQuery = QParser.getParser(v, otherReq).getQuery(); + } + RefCounted fromRef = fromCore.getSearcher(false, true, null); + info.addCloseHook(fromRef::decref); + fromSearcher = fromRef.get(); + fromExecutor = (ExecutorService) fromCore.getCoreContainer().getIndexSearcherExecutor(); + } else { + fromQuery = subQuery(v, null).getQuery(); + fromSearcher = req.getSearcher(); + fromExecutor = (ExecutorService) req.getCoreContainer().getIndexSearcherExecutor(); + } + + return joinIndex.newJoinQuery(fromField, fromQuery, fromSearcher, toField, fromExecutor); + } + }; + } +} diff --git a/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinConfig.java b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinConfig.java new file mode 100644 index 000000000000..779e5aac6d1a --- /dev/null +++ b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinConfig.java @@ -0,0 +1,107 @@ +/* + * 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.solr.search.join.auxindexjoin; + +import java.util.concurrent.TimeUnit; +import org.apache.lucene.store.Directory; + +/** + * Holds the configuration used to create an {@link AuxIndexManager}. Every setter returns {@link + * AuxIndexJoinConfig} to allow chaining settings conveniently, for example: + * + *

+ * AuxIndexJoinConfig config = new AuxIndexJoinConfig().setBlockingRefresh(false);
+ * AuxIndexManager joinIndex = new AuxIndexManager(joinDir, config);
+ * 
+ * + *

Once passed to {@link AuxIndexManager#AuxIndexManager(Directory, AuxIndexJoinConfig)}, changes + * to this object no longer affect the created {@link AuxIndexManager} instance. + */ +public final class AuxIndexJoinConfig { + + private boolean singleFieldPerSegment = false; + private boolean blockingRefresh = true; + private boolean useFromSideThreads = true; + private long sweepSamplingIntervalNanos = TimeUnit.MINUTES.toNanos(1); + + /** Sole constructor, using the default settings documented on each setter. */ + public AuxIndexJoinConfig() {} + + /** + * Whether each pair column is flushed into its own sidecar segment, rather than batching every + * pair column built in the same round into one segment. Default is {@code false}: many columns + * per segment, traded off against a longer sweep to reclaim any that become dead. + */ + public AuxIndexJoinConfig setSingleFieldPerSegment(boolean singleFieldPerSegment) { + this.singleFieldPerSegment = singleFieldPerSegment; + return this; + } + + /** Returns the current value set via {@link #setSingleFieldPerSegment}. */ + public boolean getSingleFieldPerSegment() { + return singleFieldPerSegment; + } + + /** + * Whether writing a batch of pair columns blocks until the sidecar's {@link + * org.apache.lucene.search.SearcherManager} is refreshed past it, so the freshly built pairs are + * visible to the caller that triggered the build. Default is {@code true}. + */ + public AuxIndexJoinConfig setBlockingRefresh(boolean blockingRefresh) { + this.blockingRefresh = blockingRefresh; + return this; + } + + /** Returns the current value set via {@link #setBlockingRefresh}. */ + public boolean getBlockingRefresh() { + return blockingRefresh; + } + + /** + * Whether loading the from-side leaves that feed a join build is parallelized across the caller's + * executor threads. When {@code true} (the default) each from-side segment is loaded on a + * separate executor thread; when {@code false} they are loaded sequentially on the calling + * thread. Set to {@code false} to bound the load to a single thread, e.g. when the executor is + * contended or to keep queries deterministic. + */ + public AuxIndexJoinConfig setUseFromSideThreads(boolean useFromSideThreads) { + this.useFromSideThreads = useFromSideThreads; + return this; + } + + /** Returns the current value set via {@link #setUseFromSideThreads}. */ + public boolean getUseFromSideThreads() { + return useFromSideThreads; + } + + /** + * How often {@link AuxIndexManager#onCreateWeight} actually samples searcher state for the + * dead-pair reaper; calls arriving sooner than this after the last accepted sample are skipped, + * since sampling is only a heuristic hint feeding the reap decision, not a correctness + * requirement. Default is one minute. Pass zero (or a non-positive value) to sample on every + * call. + */ + public AuxIndexJoinConfig setSweepSamplingInterval(long duration, TimeUnit unit) { + this.sweepSamplingIntervalNanos = unit.toNanos(duration); + return this; + } + + /** Returns the current value set via {@link #setSweepSamplingInterval}, in nanoseconds. */ + public long getSweepSamplingIntervalNanos() { + return sweepSamplingIntervalNanos; + } +} diff --git a/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinMergePolicy.java b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinMergePolicy.java new file mode 100644 index 000000000000..784b6b0d5c40 --- /dev/null +++ b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinMergePolicy.java @@ -0,0 +1,265 @@ +/* + * 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.solr.search.join.auxindexjoin; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.lucene.index.CodecReader; +import org.apache.lucene.index.FilterCodecReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.MergePolicy; +import org.apache.lucene.index.MergeTrigger; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentInfos; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.util.Bits; + +final class AuxIndexJoinMergePolicy extends MergePolicy { + @Override + public MergePolicy.MergeSpecification findMerges( + MergeTrigger mergeTrigger, SegmentInfos segmentInfos, MergeContext mergeContext) + throws IOException { + Set merging = mergeContext.getMergingSegments(); + MergeSpecification spec = null; + for (SegmentCommitInfo info : segmentInfos) { + if (merging.contains(info)) { + continue; + } + Set pairFieldNames = + JoinIndexUtils.pairFieldNames(JoinIndexUtils.readFieldInfos(info)); + if (!pairFieldNames.isEmpty() + && pendingPairRemovals.containsAll( + pairFieldNames)) { // todo sweep pending removals as well + if (spec == null) { + spec = new MergeSpecification(); + } + spec.add(new DropSegmentMerge(List.of(info))); + } + } + return spec; + } + + // counts merges that actually dropped a fully-dead segment; test-only observability, see + // droppedSegmentCount() + private final AtomicInteger droppedSegmentCount = new AtomicInteger(); + + /** + * A merge over a single dead segment whose contents are reported as fully deleted, so {@link + * IndexWriter} drops it instead of rewriting it -- see {@link #wrapForMerge}. Non-static so it + * can report back to the outer policy's {@link #droppedSegmentCount}. + */ + private final class DropSegmentMerge extends OneMerge { + DropSegmentMerge(List segments) { + super(segments); + } + + @Override + public CodecReader wrapForMerge(CodecReader reader) { + return new FilterCodecReader(reader) { + @Override + public CacheHelper getCoreCacheHelper() { + return reader.getCoreCacheHelper(); + } + + @Override + public CacheHelper getReaderCacheHelper() { + return null; // we are altering live docs + } + + @Override + public Bits getLiveDocs() { + return new Bits.MatchNoBits(reader.maxDoc()); + } + + @Override + public int numDocs() { + return 0; + } + }; + } + + @Override + public void mergeFinished(boolean success, boolean segmentDropped) throws IOException { + if (segmentDropped) { + droppedSegmentCount.incrementAndGet(); + } + super.mergeFinished(success, segmentDropped); + } + } + + /** Test-only: how many sidecar segments this policy has actually reaped so far. */ + int droppedSegmentCount() { + return droppedSegmentCount.get(); + } + + /** Test-only: how many dead pair field names are currently queued for the next reap. */ + int pendingPairRemovalsCount() { + return pendingPairRemovals.size(); + } + + @Override + public MergePolicy.MergeSpecification findForcedMerges( + SegmentInfos segmentInfos, + int maxSegmentCount, + Map segmentsToMerge, + MergeContext mergeContext) + throws IOException { + return null; + } + + @Override + public MergePolicy.MergeSpecification findForcedDeletesMerges( + SegmentInfos segmentInfos, MergeContext mergeContext) throws IOException { + return null; + } + + // caps how many distinct (from-searcher, to-searcher) pairs we remember snapshots for; a + // best-effort bound since this only anchors a heuristic reap, never correctness + private static final int MAX_TRACKED_SEARCHER_PAIRS = 256; + private final ConcurrentHashMap, Set> + lastNeededPairsBySearcherPair = new ConcurrentHashMap<>(); + private final ConcurrentLinkedQueue> trackedSearcherPairsOrder = + new ConcurrentLinkedQueue<>(); + + // pair field names seen in an earlier snapshot but missing from a later one for the same + // (from-searcher, to-searcher) pair -- i.e. no longer needed -- queued here for findMerges to + // reap; also size-capped, same reasoning + private static final int MAX_PENDING_PAIR_REMOVALS = 4096; + private final Set pendingPairRemovals = ConcurrentHashMap.newKeySet(); + private final ConcurrentLinkedQueue pendingPairRemovalsOrder = + new ConcurrentLinkedQueue<>(); + + // how often onCreateWeight actually bothers to sample searcher state; calls arriving sooner + // than this after the last accepted sample are skipped outright, since sampling is only a + // heuristic hint feeding findMerges' reap decision, not a correctness requirement. Zero (or + // negative) disables throttling entirely. Defaults to one minute; see #setSweepInterval. + private volatile long samplingIntervalNanos = TimeUnit.MINUTES.toNanos(1); + + // Long.MIN_VALUE marks "never sampled yet" so the very first call always goes through, + // regardless of what System.nanoTime()'s arbitrary origin happens to be. + private final AtomicLong nextSampleAtNanos = new AtomicLong(Long.MIN_VALUE); + + /** + * Configures how often {@link #onCreateWeight} actually samples searcher state for the dead-pair + * reaper; calls arriving sooner than this after the last accepted sample are skipped, since + * sampling is only a heuristic hint, not a correctness requirement. Defaults to one minute. Pass + * zero (or a non-positive value) to sample on every call. + */ + void setSweepInterval(long duration, TimeUnit unit) { + this.samplingIntervalNanos = unit.toNanos(duration); + } + + /** + * Approximates "has enough time passed since the last sample" with {@link System#nanoTime()} -- + * the JDK's cheapest monotonic timer, since it need not track wall-clock time -- gated by a + * single CAS so that under concurrent callers exactly one wins a given interval and the rest + * skip, without any lock. + */ + private boolean shouldSample() { + long interval = samplingIntervalNanos; + if (interval <= 0) { + return true; + } + long now = System.nanoTime(); + long next = nextSampleAtNanos.get(); + // subtraction (not direct comparison) so this stays correct across nanoTime() overflow, per + // its javadoc + if (next != Long.MIN_VALUE && now - next < 0) { + return false; + } + return nextSampleAtNanos.compareAndSet(next, now + interval); + } + + void onCreateWeight(Set neededPairs, IndexSearcher fromSearcher, IndexSearcher searcher) + throws IOException { + if (!shouldSample()) { + return; + } + Object fromKey = + JoinIndexUtils.directoryKey(JoinIndexUtils.directory(fromSearcher.getIndexReader())); + Object toKey = JoinIndexUtils.directoryKey(JoinIndexUtils.directory(searcher.getIndexReader())); + Map.Entry searcherKey = Map.entry(fromKey, toKey); + + Set currentSnapshot = Set.copyOf(neededPairs); + Set previousSnapshot = + AuxIndexJoinMergePolicy.putBounded( + lastNeededPairsBySearcherPair, + trackedSearcherPairsOrder, + searcherKey, + currentSnapshot, + MAX_TRACKED_SEARCHER_PAIRS); + if (previousSnapshot != null) { + for (String pairFieldName : previousSnapshot) { + if (!currentSnapshot.contains(pairFieldName)) { + AuxIndexJoinMergePolicy.addBounded( + pendingPairRemovals, + pendingPairRemovalsOrder, + pairFieldName, + MAX_PENDING_PAIR_REMOVALS); + } + } + } + } + + /** + * Puts {@code key} -> {@code value}, evicting the oldest key(s) once {@code map} exceeds {@code + * maxSize}. Approximate under races (an eviction can drop a key concurrently re-inserted, or the + * map can briefly exceed {@code maxSize}) -- acceptable since callers only use this as a soft cap + * on a best-effort cache. + */ + static V putBounded( + ConcurrentHashMap map, + ConcurrentLinkedQueue insertionOrder, + K key, + V value, + int maxSize) { + V previous = map.put(key, value); + if (previous == null) { + insertionOrder.add(key); + while (map.size() > maxSize) { + K oldest = insertionOrder.poll(); + if (oldest == null) { + break; + } + map.remove(oldest); + } + } + return previous; + } + + /** Same eviction policy as {@link AuxIndexJoinMergePolicy#putBounded}, for a plain set. */ + static void addBounded( + Set set, ConcurrentLinkedQueue insertionOrder, T value, int maxSize) { + if (set.add(value)) { + insertionOrder.add(value); + while (set.size() > maxSize) { + T oldest = insertionOrder.poll(); + if (oldest == null) { + break; + } + set.remove(oldest); + } + } + } +} diff --git a/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinQuery.java b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinQuery.java new file mode 100644 index 000000000000..297553b8f1c2 --- /dev/null +++ b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexJoinQuery.java @@ -0,0 +1,218 @@ +/* + * 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.solr.search.join.auxindexjoin; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.function.Predicate; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.internal.hppc.IntHashSet; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.Query; +import org.apache.lucene.search.QueryVisitor; +import org.apache.lucene.search.ScoreMode; +import org.apache.lucene.search.Weight; +import org.apache.solr.common.util.CollectionUtil; +import org.apache.solr.search.join.auxindexjoin.AuxIndexManager.JoinSegmentReference; +import org.jspecify.annotations.NonNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Joins the from-side index to the to-side index this query is executed against, resolving + * from-side docs matching {@code fromQuery} to to-side docs through the auxiliary join index + * managed by {@link AuxIndexManager}: there, each (from-segment, to-segment) pair owns a + * SORTED_NUMERIC column named by both sides' persistent keys, whose doc number is the from-side doc + * id and whose value is the matching to-side doc id. Pair columns missing from the join index are + * built on demand at weight creation, so no explicit build step exists; obtain instances via {@link + * AuxIndexManager#newJoinQuery}. Matches score a constant. + */ +class AuxIndexJoinQuery extends Query { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + final AuxIndexManager joinIndex; + final String fromField; + final Query fromQuery; + protected final IndexSearcher fromSearcher; + final String toField; + private final ExecutorService fromExecutorService; + + AuxIndexJoinQuery( + AuxIndexManager joinIndex, + String fromField, + Query fromQuery, + IndexSearcher fromSearcher, + String toField, + ExecutorService fromExecutorService) { + this.joinIndex = Objects.requireNonNull(joinIndex, "joinIndex"); + this.fromField = Objects.requireNonNull(fromField, "fromField"); + this.fromQuery = Objects.requireNonNull(fromQuery, "fromQuery"); + this.fromSearcher = Objects.requireNonNull(fromSearcher, "fromSearcher"); + this.toField = Objects.requireNonNull(toField, "toField"); + this.fromExecutorService = fromExecutorService; + } + + @SuppressWarnings("ReferenceEquality") + @Override + public Query rewrite(IndexSearcher indexSearcher) throws IOException { + // the from-side selection rewrites against the from-side searcher, not against the (to-side) + // searcher this query is executed with + Query rewrittenFrom = fromQuery.rewrite(fromSearcher); + if (rewrittenFrom != fromQuery) { // TODO check MatchNoDocs ? + return new AuxIndexJoinQuery( + joinIndex, fromField, rewrittenFrom, fromSearcher, toField, fromExecutorService); + } + return super.rewrite(indexSearcher); + } + + @Override + public Weight createWeight(IndexSearcher toSideSearcher, ScoreMode scoreMode, float boost) + throws IOException { + @NonNull Map neededPairs = + getRequiredColumNames(toSideSearcher); + + joinIndex.onCreateWeight(neededPairs.keySet(), fromSearcher, toSideSearcher); // ignoring fields + Predicate isNeeded = neededPairs::containsKey; + + Map existingJoinSegments; + IndexSearcher joinSearcher = this.joinIndex.acquire(); + try { + existingJoinSegments = JoinIndexUtils.extractExistingJoinColumns(joinSearcher, isNeeded); + } finally { + this.joinIndex.release(joinSearcher); + } + int pairsNeeded = neededPairs.size(); + neededPairs.keySet().removeAll(existingJoinSegments.keySet()); + IntHashSet fromOrdsToLoad = new IntHashSet(neededPairs.size()); + neededPairs.values().stream() + .mapToInt(AuxIndexManager.SegmentsTuple::fromLeafOrd) + .forEach(fromOrdsToLoad::add); + if (JoinIndexUtils.diagnosticsEnabled(log)) { + // pairsMissing > 0 on a repeat query means those pairs were never persisted by a previous + // run (writeBatch never captured them), so their from-segments' FK columns get reloaded + // here; pairsClaimed counts missing pairs another thread is building right now (claims are + // dropped once persisted), i.e. reloads that are pure waste + JoinIndexUtils.logDiagnostic( + log, + "AUXIJOIN evt=weight pairsNeeded={} pairsExisting={} pairsMissing={} pairsClaimed={}" + + " fkOrdsToLoad={} missingPairs={}", + pairsNeeded, + existingJoinSegments.size(), + neededPairs.size(), + joinIndex.countClaimedBuilds(neededPairs.keySet()), + fromOrdsToLoad.size(), + neededPairs.keySet()); + } + Map> fromFutures = loadFromSide(fromOrdsToLoad); + // TODO this might produce too many small tasks + return new JoinIndexWeight( + this, + joinSearcher, + existingJoinSegments, + toSideSearcher.getIndexReader(), + scoreMode, + boost, + fromFutures); + } + + @SuppressWarnings("unchecked") + private Map> loadFromSide(IntHashSet fromLeafsToLoad) + throws IOException { + LinkedHashMap> futuresByLeafOrds = + CollectionUtil.newLinkedHashMap(this.fromSearcher.getLeafContexts().size()); + final Weight fromWeight = + this.fromSearcher.createWeight(this.fromQuery, ScoreMode.COMPLETE_NO_SCORES, 1.0f); + + List fromLeafs = new ArrayList<>(this.fromSearcher.getLeafContexts()); + // heaviest first, the smallest last + fromLeafs.sort( + Comparator.comparing(ctx -> fromLeafsToLoad.contains(ctx.ord)) + .thenComparingInt(ctx -> ctx.reader().maxDoc()) + .reversed()); + + for (LeafReaderContext ctx : fromLeafs) { + futuresByLeafOrds.putLast( + ctx.ord, // the heaviest is submitted first, and accessed first as well + this.fromExecutorService.submit( + () -> + FromLeafJoinContext.heavyLoadFromLeaf( + fromWeight, fromField, ctx, fromLeafsToLoad.contains(ctx.ord)))); + } + + return futuresByLeafOrds; + } + + private @NonNull Map getRequiredColumNames( + IndexSearcher searcher) { + Map neededPairs = new HashMap<>(); + for (LeafReaderContext toContext : searcher.getIndexReader().leaves()) { + String toKey = JoinIndexUtils.getSideKey(toContext, toField); + for (LeafReaderContext fromCtx : fromSearcher.getLeafContexts()) { + String fromKey = JoinIndexUtils.getSideKey(fromCtx, fromField); + neededPairs.put( + fromKey + "_" + toKey, new AuxIndexManager.SegmentsTuple(fromCtx.ord, toContext.ord)); + } + } + return neededPairs; + } + + @Override + public String toString(String field) { + return "AuxIndexJoinQuery(" + fromField + " -> " + toField + ", from: " + fromQuery + ")"; + } + + @Override + public void visit(QueryVisitor visitor) { + visitor.visitLeaf(this); + } + + @Override + public boolean equals(Object other) { + return sameClassAs(other) && equalsTo((AuxIndexJoinQuery) other); + } + + private boolean equalsTo(AuxIndexJoinQuery other) { + // the join index and the from searcher compare by identity: a reopened from reader sees + // different ordinal spaces, so queries over different searcher instances must not be + // considered equal + return joinIndex == other.joinIndex + && fromSearcher == other.fromSearcher + && fromField.equals(other.fromField) + && fromQuery.equals(other.fromQuery) + && toField.equals(other.toField); + } + + @Override + public int hashCode() { + return Objects.hash( + classHash(), + System.identityHashCode(joinIndex), + fromField, + fromQuery, + System.identityHashCode(fromSearcher), + toField); + } +} diff --git a/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexManager.java b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexManager.java new file mode 100644 index 000000000000..fdc89164e142 --- /dev/null +++ b/solr/core/src/java/org/apache/solr/search/join/auxindexjoin/AuxIndexManager.java @@ -0,0 +1,262 @@ +/* + * 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.solr.search.join.auxindexjoin; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.apache.lucene.index.ConcurrentMergeScheduler; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; +import org.apache.lucene.index.MergeScheduler; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.Query; +import org.apache.lucene.search.SearcherManager; +import org.apache.lucene.store.Directory; +import org.apache.lucene.util.IOUtils; +import org.apache.solr.search.join.auxindexjoin.JoinIndexUtils.JoinColumnModel; + +/** + * The auxiliary join index: a self-maintaining sidecar persisting per (from-segment, to-segment) + * doc id mappings, so query-time joining reduces to bitset translation. It owns the sidecar's + * {@link IndexWriter} and {@link SearcherManager}; pair columns are built lazily when an {@link + * AuxIndexJoinQuery} first needs them, so users only construct an instance once, create queries + * with {@link #newJoinQuery} and search them with a bare to-side {@link IndexSearcher}: + * + *

+ * AuxIndexManager joinIndex = new AuxIndexManager(joinDir);   // once per process
+ * Query q = joinIndex.newJoinQuery(fromField, fromQuery, fromSearcher, toField);
+ * TopDocs hits = toSearcher.search(q, 10);
+ * ...
+ * joinIndex.close();                                   // app shutdown
+ * 
+ * + *

After either side reopens, the next query builds only the missing (from, to) segment pairs: + * pair columns are addressed by both sides' persistent segment keys, which survive reopens. Pair + * columns orphaned by merges are not reclaimed yet; see package's javadoc. + */ +public final class AuxIndexManager implements Closeable { + + private final IndexWriter writer; + private final SearcherManager manager; + private final JoinColumnIndexer pairBuilder = new JoinColumnIndexer(this); + + // package-private (not private): tests reach in directly to observe the reaper's state + final AuxIndexJoinMergePolicy mergePolicy; + private final MergeScheduler mergeScheduler; + private final JoinColumWriter writerDelegate; + private final boolean blockingRefresh; + private final boolean useFromSideThreads; + + /** A pair's (from-segment, to-segment) leaf ordinals. */ + record SegmentsTuple(int fromLeafOrd, int toLeafOrd) {} + + /** + * A pair column's address in the join index: the pair field name and the sidecar segment (name + * plus current leaf ordinal) carrying it -- enough to locate and open the column's real + * docvalues, or to check whether a pair already exists before deciding what still needs to be + * built. A resolved cell's edges are tracked separately, as a plain {@code DocEdges}, since they + * don't change as this reference is refreshed. + */ + record JoinSegmentReference( + String pairFieldName, String joinSegmentName, int joinSegmentLeafOrd) {} + + /** + * Opens a persistent auxiliary join index over the given directory, creating it if empty, using + * the default {@link AuxIndexJoinConfig}. The caller retains ownership of the directory: {@link + * #close()} does not close it. + */ + public AuxIndexManager(Directory directory) throws IOException { + this(directory, new AuxIndexJoinConfig()); + } + + /** + * Opens a persistent auxiliary join index over the given directory, creating it if empty, using + * the given {@link AuxIndexJoinConfig}. The caller retains ownership of the directory: {@link + * #close()} does not close it. + */ + public AuxIndexManager(Directory directory, AuxIndexJoinConfig config) throws IOException { + this(directory, config, new ConcurrentMergeScheduler()); + } + + /** + * Opens a persistent auxiliary join index over the given directory, creating it if empty, using + * the given {@link AuxIndexJoinConfig} and {@link MergeScheduler} in place of the default {@link + * ConcurrentMergeScheduler}. The caller retains ownership of the directory: {@link #close()} does + * not close it. + */ + public AuxIndexManager( + Directory directory, AuxIndexJoinConfig config, MergeScheduler mergeScheduler) + throws IOException { + this.mergeScheduler = mergeScheduler; + this.mergePolicy = new AuxIndexJoinMergePolicy(); + this.mergePolicy.setSweepInterval(config.getSweepSamplingIntervalNanos(), TimeUnit.NANOSECONDS); + this.writer = + new IndexWriter( + directory, + new IndexWriterConfig().setMergePolicy(mergePolicy).setMergeScheduler(mergeScheduler)); + this.manager = new SearcherManager(writer, null); + JoinColumWriter bulkWriter = new JoinColumnDocWriter(); // new AIJoinColumnWriter() + this.writerDelegate = + config.getSingleFieldPerSegment() + ? new SingleColumnBySegmentWriter(bulkWriter) + : bulkWriter; + this.blockingRefresh = config.getBlockingRefresh(); + this.useFromSideThreads = config.getUseFromSideThreads(); + } + + /** + * Creates a query joining the docs matching {@code fromQuery} in {@code fromSearcher}'s index to + * the index the returned query is executed against, through {@code fromField} = {@code toField} + * term equality. Missing pair columns are built into this join index on first execution. + * + * @deprecated use another constructor passing executor service + */ + @Deprecated + public Query newJoinQuery( + String fromField, Query fromQuery, IndexSearcher fromSearcher, String toField) { + return newJoinQuery(fromField, fromQuery, fromSearcher, toField, new DirectExecutorService()); + } + + public Query newJoinQuery( + String fromField, + Query fromQuery, + IndexSearcher fromSearcher, + String toField, + ExecutorService fromExecutor) { + if (fromExecutor == null || !useFromSideThreads) { + fromExecutor = new DirectExecutorService(); + } + return new AuxIndexJoinQuery(this, fromField, fromQuery, fromSearcher, toField, fromExecutor); + } + + /** + * How many of the given pair field names have a build currently in flight (claims are removed + * once persisted). Diagnostic only; see {@link JoinColumnIndexer#countClaimedBuilds}. + */ + int countClaimedBuilds(Set pairFieldNames) { + return pairBuilder.countClaimedBuilds(pairFieldNames); + } + + IndexSearcher acquire() throws IOException { + return manager.acquire(); + } + + void release(IndexSearcher searcher) throws IOException { + manager.release(searcher); + } + + /** + * Builds and persists the given missing pair columns, keyed by pair field name to their + * (from-segment, to-segment) leaf ordinals. Delegates to {@link + * JoinColumnIndexer#buildAndPersistJoinColumns}, which documents the claim/await dedup and the + * fresh-searcher double-check in detail. + * + * @param observedAbsentSearcher the join-index searcher in which the caller established that + * {@code missingPairs} are absent; possibly stale by now. Pass {@code null} when absence + * wasn't verified against a live searcher -- that forces the re-check scan. + * @return in memory data for just written segemts + */ + Map buildAndPersistJoinColumns( + Map missingPairs, + IndexReader fromReader, + IndexReader toReader, + String toField, + String traceCtxId, + IndexSearcher observedAbsentSearcher, + Map> fromColumnFutures) + throws IOException, ExecutionException, InterruptedException { + return pairBuilder.buildAndPersistJoinColumns( + missingPairs, + fromReader, + toReader, + toField, + traceCtxId, + observedAbsentSearcher, + fromColumnFutures); + } + + /** + * Serializes sidecar writes: one batch per commit keeps every batch at doc 0 of its own segment, + * preserving pair-column doc number == from-side doc id. Builders' futures are completed before + * this runs, so waiters consume the in-memory models without paying for the commit and refresh + * here -- a completed future does not mean the reader already exposes the column. + * + *

It should be plain simple synchronized. As alternatives + * + *