Copilot commented on code in PR #4749: URL: https://github.com/apache/solr/pull/4749#discussion_r3813641879
########## solr/core/src/java/org/apache/solr/search/join/AIJoinQParserPlugin.java: ########## @@ -0,0 +1,230 @@ +/* + * 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 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.aijoin.AIJoinIndex; +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 AIJoinIndex} 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: + * + * <ul> + * <li>from - "foreign key" field name, collected while enumerating the subordinate query (the + * local parameter value). + * <li>fromIndex - optional core name to run the subordinate query against, when it differs from + * this core; cross-core joins are the reason {@link AIJoinIndex} exists in the first place, + * so this mirrors {@link ScoreJoinQParserPlugin}'s <code>fromIndex</code>, including + * SolrCloud alias/collection resolution via {@link ScoreJoinQParserPlugin#getCoreName}. + * <li>to - "primary key" field name looked up in this core's index. + * </ul> + * + * Example: {@code q={!aijoin from=manu_id_s to=id fromIndex=products}foo}. + * + * <p>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.aijoin.AIJoinQuery} 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}}. + * + * <p>One {@link AIJoinIndex} 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. This sidecar always belongs to the + * "to" side core -- the one this plugin is registered in. + * + * <p><b>Why this implements {@link QueryResponseWriter}:</b> {@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 <queryResponseWriter>}). 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 AIJoinQParserPlugin 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"; + + private static final String DEFAULT_DIR = "aijoin"; + + private String configuredDir = DEFAULT_DIR; + + private volatile AIJoinIndex joinIndex; + + @Override + public void init(NamedList<?> args) { + super.init(args); + if (args != null && args.get(DIR) != null) { + configuredDir = args.get(DIR).toString(); + } + } + + @Override + public void inform(SolrCore core) { + Path path = Path.of(configuredDir); + if (!path.isAbsolute()) { + path = Path.of(core.getDataDir()).resolve(path); + } Review Comment: Validate the configured sidecar path before opening it. As written, an absolute `dir` can point outside Solr's allowed roots, bypassing the path policy used by other filesystem features. Call `CoreContainer.assertPathAllowed` after resolving the relative default. ########## solr/core/src/java/org/apache/solr/search/join/AIJoinQParserPlugin.java: ########## @@ -0,0 +1,230 @@ +/* + * 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 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.aijoin.AIJoinIndex; +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 AIJoinIndex} 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: + * + * <ul> + * <li>from - "foreign key" field name, collected while enumerating the subordinate query (the + * local parameter value). + * <li>fromIndex - optional core name to run the subordinate query against, when it differs from + * this core; cross-core joins are the reason {@link AIJoinIndex} exists in the first place, + * so this mirrors {@link ScoreJoinQParserPlugin}'s <code>fromIndex</code>, including + * SolrCloud alias/collection resolution via {@link ScoreJoinQParserPlugin#getCoreName}. + * <li>to - "primary key" field name looked up in this core's index. + * </ul> + * + * Example: {@code q={!aijoin from=manu_id_s to=id fromIndex=products}foo}. + * + * <p>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.aijoin.AIJoinQuery} 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}}. + * + * <p>One {@link AIJoinIndex} 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. This sidecar always belongs to the + * "to" side core -- the one this plugin is registered in. + * + * <p><b>Why this implements {@link QueryResponseWriter}:</b> {@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 <queryResponseWriter>}). 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 AIJoinQParserPlugin 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"; + + private static final String DEFAULT_DIR = "aijoin"; + + private String configuredDir = DEFAULT_DIR; + + private volatile AIJoinIndex joinIndex; + + @Override + public void init(NamedList<?> args) { + super.init(args); + if (args != null && args.get(DIR) != null) { + configuredDir = args.get(DIR).toString(); + } + } + + @Override + public void inform(SolrCore core) { + Path path = Path.of(configuredDir); + if (!path.isAbsolute()) { + path = Path.of(core.getDataDir()).resolve(path); + } + final Directory directory; + try { + directory = + core.getDirectoryFactory() + .get(path.toString(), DirContext.DEFAULT, core.getSolrConfig().indexConfig.lockType); + joinIndex = new AIJoinIndex(directory); + } catch (IOException e) { + throw new SolrException( + SolrException.ErrorCode.SERVER_ERROR, "Failed to open AIJoinIndex at " + path, e); Review Comment: If `new AIJoinIndex(directory)` fails, the directory acquired just above is never released because the close hook is not registered yet. This leaks a `DirectoryFactory` reference on every failed core load/reload; release it in the failure path while preserving the original exception. ########## solr/core/src/java/org/apache/solr/search/join/aijoin/AIJoinQuery.java: ########## @@ -0,0 +1,252 @@ +/* + * 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.aijoin; + +import static org.apache.solr.search.join.aijoin.AIJoinUtil.cacheImpl; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.util.HashMap; +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.BulkScorer; +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.ScorerSupplier; +import org.apache.lucene.search.Weight; +import org.apache.solr.search.join.aijoin.AIJoinIndex.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 AIJoinIndex}: 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 + * AIJoinIndex#newJoinQuery}. Matches score a constant. + */ +class AIJoinQuery extends Query { + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + final AIJoinIndex joinIndex; + final String fromField; + final Query fromQuery; + protected final IndexSearcher fromSearcher; + final String toField; + private final ExecutorService fromExecutorService; + + AIJoinQuery( + AIJoinIndex 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; + } + + private AIJoinUtil.CacheAndCount computeDocIdSet(Weight fromWeight, LeafReaderContext ctx) + throws IOException { + // TODO figure out how to steal cached from side filters + // if (fromWeight!=null && fromWeight.getClass().getSimpleName().contains("Caching") ){ + // System.out.println("fromWeight is CachingWeight"); + // } + ScorerSupplier supplier = fromWeight.scorerSupplier(ctx); + if (supplier == null) { + return null; // NO matches ??? + } + // TODO handle already cached WeightWrapper + BulkScorer scorer = supplier.bulkScorer(); + return cacheImpl(scorer, ctx.reader().maxDoc(), ctx.reader().getLiveDocs()); + } + + @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 AIJoinQuery( + joinIndex, fromField, rewrittenFrom, fromSearcher, toField, fromExecutorService); + } + return super.rewrite(indexSearcher); + } + + @Override + public Weight createWeight(IndexSearcher toSideSearcher, ScoreMode scoreMode, float boost) + throws IOException { + @NonNull Map<String, AIJoinIndex.SegmentsTuple> neededPairs = + getRequiredColumNames(toSideSearcher); + + joinIndex.onCreateWeight(neededPairs.keySet(), fromSearcher, toSideSearcher); // ignoring fields + // + // DON'T write'em upfront + // + // WAS: + // joinIndex.ensureJoinSegments(neededPairs, fromSearcher, fromField, toSideSearcher, toField); + Predicate<String> isNeeded = neededPairs::containsKey; + + Map<String, JoinSegmentReference> existingJoinSegments; + IndexSearcher joinSearcher = this.joinIndex.acquire(); + try { + existingJoinSegments = AIJoinIndex.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(AIJoinIndex.SegmentsTuple::fromLeafOrd) + .forEach(fromOrdsToLoad::add); + if (AIJoinUtil.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 some build already claimed/completed in-process, + // i.e. reloads that are pure waste + AIJoinUtil.logDiagnostic( + log, + "AIJOIN evt=weight pairsNeeded={} pairsExisting={} pairsMissing={} pairsClaimed={}" + + " fkOrdsToLoad={} missingPairs={}", + pairsNeeded, + existingJoinSegments.size(), + neededPairs.size(), + joinIndex.countClaimedBuilds(neededPairs.keySet()), + fromOrdsToLoad.size(), + neededPairs.keySet()); + } + Future<FromLeafJoinContext>[] fromFutures = loadFromSide(fromOrdsToLoad); + // TODO this might produce too many small tasks + return new AIJoinWeight( + this, + joinSearcher, + existingJoinSegments, Review Comment: `joinSearcher` was released at line 124 but is still passed into the weight and later used as `maybeStaleJoinSearcher`. A concurrent `SearcherManager` refresh can close that reader before `scorerSupplier` accesses it, causing intermittent `AlreadyClosedException`/failed reference refreshes. Keep a valid acquisition for every use or make the weight retain only immutable addressing data that does not require the released searcher. ########## solr/core/src/java/org/apache/solr/search/join/aijoin/AIJoinIndex.java: ########## @@ -0,0 +1,447 @@ +/* + * 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.aijoin; + +import java.io.Closeable; +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.util.AbstractMap; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import org.apache.lucene.index.ConcurrentMergeScheduler; +import org.apache.lucene.index.FieldInfo; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; +import org.apache.lucene.index.LeafReaderContext; +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.aijoin.AIJoinUtil.JoinColumnModel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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 + * AIJoinQuery} 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}: + * + * <pre class="prettyprint"> + * AIJoinIndex joinIndex = new AIJoinIndex(joinDir); // once per process + * Query q = joinIndex.newJoinQuery(fromField, fromQuery, fromSearcher, toField); + * TopDocs hits = toSearcher.search(q, 10); + * ... + * joinIndex.close(); // app shutdown + * </pre> + * + * <p>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 {@code README.md} in this package. + */ +public final class AIJoinIndex implements Closeable { + + private final IndexWriter writer; + private final SearcherManager manager; + + /** + * Dedups concurrent builders per pair field name: the thread that installs the future writes the + * pair, others wait on it. Completed futures stay put so a builder that raced a not-yet-visible + * refresh cannot write a duplicate pair column. + */ + private final ConcurrentHashMap<String, CompletableFuture<Map.Entry<String, JoinColumnModel>>> + pairBuilds = new ConcurrentHashMap<>(); Review Comment: Completed futures remain here permanently and retain each `JoinColumnModel`, including its `int[maxDoc]` mapping. Since segment changes continually create new pair names, a long-running core accumulates heap proportional to every historical segment pair even after the sidecar reaper drops them. Retain only the in-flight deduplication state, or evict completed mappings once the refreshed index makes them visible. ########## solr/solr-ref-guide/modules/query-guide/pages/aijoin-query-parser.adoc: ########## @@ -0,0 +1,92 @@ += AIJoin Query Parser +// 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 + +The AIJoin (which stands for Auxiliary Index Join) Query Parser is similar to <<Join Query Parser,Join Query Parser>>, but uses a lazily written sidecar index for faster joins. Review Comment: This is an intra-document anchor, but `Join Query Parser` is defined in a different page, so the generated link on this standalone page is unresolved. Link to the existing page explicitly. ########## solr/core/src/java/org/apache/solr/search/join/aijoin/AIJoinUtil.java: ########## @@ -0,0 +1,580 @@ +/* + * 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.aijoin; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.regex.Pattern; +import org.apache.lucene.codecs.Codec; +import org.apache.lucene.codecs.FieldInfosFormat; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.DocValues; +import org.apache.lucene.index.FieldInfo; +import org.apache.lucene.index.FieldInfos; +import org.apache.lucene.index.FilterCodecReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.ParallelCompositeReader; +import org.apache.lucene.index.ParallelLeafReader; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.SortedNumericDocValues; +import org.apache.lucene.index.SortedSetDocValues; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.search.BulkScorer; +import org.apache.lucene.search.DocIdSet; +import org.apache.lucene.search.DocIdSetIterator; +import org.apache.lucene.search.DocIdStream; +import org.apache.lucene.search.LeafCollector; +import org.apache.lucene.search.Scorable; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.store.FilterDirectory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.util.Accountable; +import org.apache.lucene.util.BitDocIdSet; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.FixedBitSet; +import org.apache.lucene.util.RamUsageEstimator; +import org.apache.lucene.util.RoaringDocIdSet; +import org.apache.lucene.util.StringHelper; +import org.slf4j.Logger; +import org.slf4j.event.Level; + +/** + * Column-building and addressing helpers for the auxiliary join index managed by {@link + * AIJoinIndex}: for every (from-segment, to-segment) pair it produces a SORTED_NUMERIC column named + * {@link #pairFieldName}, whose doc number is the from-side doc id and whose value is the to-side + * doc id whose {@code toField} term equals the from doc's {@code fromField} term, plus two + * companion edges columns persisting the pair's {min, max} from-doc and to-doc bounds. + */ +final class AIJoinUtil { + + /** Suffix of the always-written column persisting a pair's {min, max} from-doc edges. */ + static final String FROM_EDGES_PREFIX = "fromDoc_edges_"; // TODO reduce to the singe letter + + /** Suffix of the always-written column persisting a pair's {min, max} to-doc edges. */ + static final String TO_EDGES_PREFIX = "toDoc_edges_"; + + /** main join colums for join index to_doc_num[from_docnum] */ + static final String TO_DOC_VAL_BY_FROM_DOCNUM = "join_toDoc_"; + + static final String TO_COUNT_PREFIX = "num_toDoc_"; + + private AIJoinUtil() {} + + /** + * Configurable level for the {@code AIJOIN evt=...} diagnostic logs; defaults to {@code INFO}, + * override with the {@code solr.aijoin.log.level} system property (or {@code + * SOLR_AIJOIN_LOG_LEVEL} env var). + */ + static final Level AIJOIN_LOG_LEVEL = Level.TRACE; + + /** Whether the AIJOIN diagnostic logs would emit at the configured level. */ + static boolean diagnosticsEnabled(Logger log) { + return log.isEnabledForLevel(AIJOIN_LOG_LEVEL); + } + + /** Emits an AIJOIN diagnostic line at the configured level. */ + static void logDiagnostic(Logger log, String message, Object... args) { + log.atLevel(AIJOIN_LOG_LEVEL).log(message, args); + } + + /** + * A pair's {min, max} from-doc and to-doc bounds and match count, common to both a pair freshly + * built on demand ({@link JoinColumnModel#edges()}) and one already persisted in the join index + * ({@link Edges}, loaded through {@link #loadEdges}), so code walking matches doesn't need to + * care which one backs it. + */ + interface DocEdges { + int[] fromDocEdges(); + + int[] toDocEdges(); + + /** this is rather doubtful */ + int toCount(); + } + + /** A self-contained {@link DocEdges} value, with no addressing information of its own. */ + record Edges(int[] fromDocEdges, int[] toDocEdges, int toCount) implements DocEdges {} + + /** + * The from-doc-to-to-doc map produced by {@link #computeDocMapping}, paired with its resolved + * {@link #edges()}. {@link #toDocByFromDoc()} mirrors the on-disk column's read API, so freshly + * built pairs (not yet flushed to the join index) and pairs loaded from the join index can be + * walked by the same code. + */ + static final class JoinColumnModel { + private final int[] toDocByFromDoc; + private final DocEdges edges; + + JoinColumnModel(int[] toDocByFromDoc, DocEdges edges) { + this.toDocByFromDoc = toDocByFromDoc; + this.edges = edges; + } + + /** + * Returns a fresh single-valued cursor over the from-doc -> to-doc map, positioned before doc + * 0. + */ + SortedNumericDocValues toDocByFromDoc() { + return new ArrayBackedSortedNumericDocValues(toDocByFromDoc); + } + + DocEdges edges() { + return edges; + } + } + + /** + * Adapts an int-array from-doc -> to-doc map (as produced by {@link #computeDocMapping}, {@code + * -1} meaning no value) to the {@link SortedNumericDocValues} read API, so it can be consumed the + * same way as the on-disk join column. Always single-valued until M:N pairs are supported. + */ + private static final class ArrayBackedSortedNumericDocValues extends SortedNumericDocValues { + private final int[] toDocByFromDoc; + private int doc = -1; + + ArrayBackedSortedNumericDocValues(int[] toDocByFromDoc) { + this.toDocByFromDoc = toDocByFromDoc; + } + + @Override + public long nextValue() { + return toDocByFromDoc[doc]; + } + + @Override + public int docValueCount() { + return 1; + } + + @Override + public boolean advanceExact(int target) { + doc = target; + return target < toDocByFromDoc.length && toDocByFromDoc[target] >= 0; + } + + @Override + public int docID() { + return doc; + } + + @Override + public int nextDoc() { + return advance(doc + 1); + } + + @Override + public int advance(int target) { + while (target < toDocByFromDoc.length && toDocByFromDoc[target] < 0) { + target++; + } + doc = target < toDocByFromDoc.length ? target : NO_MORE_DOCS; + return doc; + } + + @Override + public long cost() { + return toDocByFromDoc.length; + } + } + + /** + * Builds the join column for one (from-segment, to-segment) pair: resolves every from-side doc to + * its matching to-side doc id, along with the pair's from-doc and to-doc bounds. From-side terms + * are hashed by {@link ForeignKeyColumn}; each to-side term is looked up in that hash to map + * from-side ords to to-side ords. + * + * <p>Docs already deleted at build time are skipped, purely to avoid persisting entries nobody + * can ever match -- deletes are otherwise re-checked live at query time (from-side in {@code + * ToLeafJoinContext}, to-side by the searcher's own {@code acceptDocs}), since a pair's cached + * mapping outlives whatever gets deleted after it was built. + */ + static JoinColumnModel computeDocMapping( + LeafReaderContext toContext, String toField, ForeignKeyColumn fromSideData) + throws IOException { + assert fromSideData != null; + + long[] toOrdByFromOrd = new long[fromSideData.getFromValuesCount()]; + Arrays.fill(toOrdByFromOrd, -1L); + SortedSetDocValues toDV = DocValues.getSortedSet(toContext.reader(), toField); + Bits toLiveDocs = toContext.reader().getLiveDocs(); + TermsEnum toTerms = toDV.termsEnum(); + // resolve from-side ords to to-side ords: look each to-side term up in the from-side hash. + boolean termsAreDisjoint = true; + for (BytesRef term = toTerms.next(); term != null; term = toTerms.next()) { + int fromOrd = fromSideData.getFromTermOrdOrDashOne(term); + if (fromOrd != -1) { + toOrdByFromOrd[fromOrd] = (int) toTerms.ord(); + termsAreDisjoint = false; + } + } + // TODO: this degrades M:N joins to M:1. Both toDocByToOrd and toDocByFromDoc keep a single + // to-side doc per slot, so when several to docs share a term (non-unique toField) or a + // fromSideData + // doc is multi-valued with several matching terms, later assignments overwrite earlier ones + // and only the last match survives. The read side (AIJoinQuery) already consumes all + // docValueCount() values per doc, so only this writer needs to learn to emit multiple + // to docs per fromSideData doc. + if (!termsAreDisjoint) { + int[] toDocByToOrd = new int[Math.toIntExact(toDV.getValueCount())]; + Arrays.fill(toDocByToOrd, -1); + for (int toDoc = toDV.nextDoc(); + toDoc != DocIdSetIterator.NO_MORE_DOCS; + toDoc = toDV.nextDoc()) { + if (toLiveDocs != null && !toLiveDocs.get(toDoc)) { + continue; + } + for (int i = 0; i < toDV.docValueCount(); i++) { + long toOrd = toDV.nextOrd(); + toDocByToOrd[(int) toOrd] = toDoc; + // TODO we can apply toOrdByFromOrd right here + // and get toDocByFromOrd[] + } + } + + // resolve every fromSideData doc to its to-side doc. Docs without the field, or whose term + // has + // no to-side match, keep -1. + int[] toDocByFromDoc = fromSideData.cloneFromOrdByFromDoc(); + int minFromDoc = DocIdSetIterator.NO_MORE_DOCS; + int maxFromDoc = -1; + int minToDoc = DocIdSetIterator.NO_MORE_DOCS; + int maxToDoc = -1; + int toCount = 0; + // walk the array, mapping each fromSideData ord to its to-side doc in place. + for (int fromDoc = 0; fromDoc < toDocByFromDoc.length; fromDoc++) { + int fromOrd = toDocByFromDoc[fromDoc]; + if (fromOrd == -1) { + continue; + } + int toOrd = (int) toOrdByFromOrd[fromOrd]; + int toDoc = toOrd == -1 ? -1 : toDocByToOrd[toOrd]; + if (toDoc == -1) { + toDocByFromDoc[fromDoc] = -1; // wiping is crucial + continue; + } + toDocByFromDoc[fromDoc] = toDoc; + minFromDoc = Math.min(minFromDoc, fromDoc); + maxFromDoc = Math.max(maxFromDoc, fromDoc); + minToDoc = Math.min(minToDoc, toDoc); + maxToDoc = Math.max(maxToDoc, toDoc); + toCount++; + } + if (maxFromDoc < 0) { // tombstone - column is empty + // no fromSideData doc in this pair maps to any to doc: normalize both edges to the + // symmetric + // {-1, -1} sentinel. An asymmetric one (e.g. {NO_MORE_DOCS, -1}) doesn't round-trip + // through the join index's SORTED_NUMERIC edges column, which always returns its two + // values in ascending numeric order regardless of which was written as "min" -- so + // {NO_MORE_DOCS, -1} silently comes back as {-1, NO_MORE_DOCS} on the next read. + minFromDoc = -1; + minToDoc = -1; + maxToDoc = -1; + } + return new JoinColumnModel( + toDocByFromDoc, + new Edges(new int[] {minFromDoc, maxFromDoc}, new int[] {minToDoc, maxToDoc}, toCount)); + } else { // tombstone - column is empty, due to disjoint terms, perhaps one may optimize it + int[] minusones = new int[fromSideData.fromSideMaxDocs()]; + Arrays.fill(minusones, -1); + return new JoinColumnModel(minusones, new Edges(new int[] {-1, -1}, new int[] {-1, -1}, 0)); + } + } + + /** + * Reads a pair's persisted {@code {min, max}} edges (or {@code toCount}), all stored on doc 0 of + * the column -- the read-side counterpart of {@link AIJoinWriter}'s edges columns. + */ + static int[] loadEdges(LeafReaderContext joinContext, String edgesFieldName) throws IOException { + SortedNumericDocValues edgesDV = joinContext.reader().getSortedNumericDocValues(edgesFieldName); + assert edgesDV != null : "expected edges column to be present: " + edgesFieldName; + int zeroDoc = edgesDV.nextDoc(); + assert zeroDoc == 0 : "expected edges column to be fully materialized, but got doc " + zeroDoc; + int[] values = new int[edgesDV.docValueCount()]; + for (int i = 0; i < values.length; i++) { + values[i] = (int) edgesDV.nextValue(); + } + return values; + } + + /** + * The join index field name addressing the ordinal map of one (from-segment, to-segment) pair. + */ + static String pairFieldName( + LeafReaderContext fromContext, + String fromField, + LeafReaderContext toContext, + String toField) { + return getSideKey(fromContext, fromField) + "_" + getSideKey(toContext, toField); + } + + // Lucene puts no hard constraints on field names, but conservatively keep side keys usable as + // one by reducing them to identifier characters + private static final Pattern NON_IDENTIFIER = Pattern.compile("[^A-Za-z0-9_]"); + + /** + * Persistent identifier of one join side: the join field name, the immutable id the segment was + * created with (it survives reopens, growing deletes mask and reorderings of {@link + * IndexReader#leaves()}; a merge produces a new segment with a new id) and the docvalues + * generation of the join field. + */ + static String getSideKey(LeafReaderContext context, String field) { + byte[] segmentId = segmentReader(context.reader()).getSegmentInfo().info.getId(); + // dvGen starts at -1 and advances only when this particular field receives an in-place + // IndexWriter.updateDocValues update; deletes only bump delGen and leave it untouched. So the + // key is insensitive to deletes but changes when the join field's docvalues are updated. + long dvGen = context.reader().getFieldInfos().fieldInfo(field).getDocValuesGen(); + String key = field + ":" + StringHelper.idToString(segmentId) + ":" + dvGen; + // TODO this is dangerous, no one flip them back + return NON_IDENTIFIER.matcher(key).replaceAll("_"); Review Comment: Replacing every non-identifier character with `_` is not collision-safe: valid fields such as `field-a` and `field.a` produce the same side key for the same segment/generation. The second query can then reuse the first field's persisted pair column and return incorrect matches. Encode the field name reversibly (or include a collision-resistant digest) rather than normalizing distinct names to the same string. ########## solr/core/src/java/org/apache/solr/search/join/aijoin/AIJoinIndex.java: ########## @@ -0,0 +1,447 @@ +/* + * 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.aijoin; + +import java.io.Closeable; +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.util.AbstractMap; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import org.apache.lucene.index.ConcurrentMergeScheduler; +import org.apache.lucene.index.FieldInfo; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; +import org.apache.lucene.index.LeafReaderContext; +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.aijoin.AIJoinUtil.JoinColumnModel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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 + * AIJoinQuery} 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}: + * + * <pre class="prettyprint"> + * AIJoinIndex joinIndex = new AIJoinIndex(joinDir); // once per process + * Query q = joinIndex.newJoinQuery(fromField, fromQuery, fromSearcher, toField); + * TopDocs hits = toSearcher.search(q, 10); + * ... + * joinIndex.close(); // app shutdown + * </pre> + * + * <p>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 {@code README.md} in this package. + */ +public final class AIJoinIndex implements Closeable { + + private final IndexWriter writer; + private final SearcherManager manager; + + /** + * Dedups concurrent builders per pair field name: the thread that installs the future writes the + * pair, others wait on it. Completed futures stay put so a builder that raced a not-yet-visible + * refresh cannot write a duplicate pair column. + */ + private final ConcurrentHashMap<String, CompletableFuture<Map.Entry<String, JoinColumnModel>>> + pairBuilds = new ConcurrentHashMap<>(); + + // package-private (not private): tests reach in directly to observe the reaper's state + final AIJoinMergePolicy mergePolicy; + private final MergeScheduler mergeScheduler; + static final AIJoinWriter INSTANCE = new AIJoinDocWriter(); // new AIJoinColumnWriter(); + + private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + /** Why a build was triggered; reported as {@code cause=} on the {@code AIJOIN evt=build} line. */ + enum BuildCause { + /** {nolink #ensureJoinSegments}, i.e. eagerly at {@link AIJoinQuery#createWeight} time. */ + EAGER_CREATE_WEIGHT, + /** {@link ToLeafJoinContext}, i.e. lazily for a gap the eager pass did not cover. */ + LAZY_TO_SEGMENT; + + @Override + public String toString() { + return name().toLowerCase(Locale.ROOT).replace('_', '-'); + } + } + + /** 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) {} + + /** + * Scans {@code joinSearcher}'s leaves for every pair column whose field name satisfies {@code + * isNeeded}, returning where each one lives. Used both to seed a fresh {@link AIJoinWeight}'s + * view of already-built pairs, and by {@link ToLeafJoinContext} to relocate a pair whose cached + * segment reference no longer resolves. TODO subject for in-heap caching TODO commit's userdata + * might have a list of pairs with known segment ords and names + */ + static Map<String, JoinSegmentReference> extractExistingJoinColumns( + IndexSearcher joinSearcher, Predicate<String> isNeeded) { + Map<String, JoinSegmentReference> existingJoinSegments = + new HashMap<>(joinSearcher.getIndexReader().leaves().size()); + for (LeafReaderContext joinContext : joinSearcher.getIndexReader().leaves()) { + String segmentName = AIJoinUtil.segmentName(joinContext); + for (FieldInfo fieldInfo : joinContext.reader().getFieldInfos()) { + // pairs are detected by their toCount companion, which is written to doc 0 for every + // built pair; the join column itself is sparse and a tombstone pair (disjoint terms) + // never materializes it, so scanning for join columns kept re-reporting once-built + // tombstones as missing -- and re-triggering their from-side FK loads on every query + String splits[] = fieldInfo.name.split(AIJoinUtil.TO_COUNT_PREFIX); + if (splits.length == 2 && isNeeded.test(splits[1])) { + existingJoinSegments.computeIfAbsent( + splits[1], + fieldName -> new JoinSegmentReference(fieldName, segmentName, joinContext.ord)); + } + } + } + return existingJoinSegments; + } + + /** + * Opens a persistent auxiliary join index over the given directory, creating it if empty, using + * the default {@link AIJoinIndexConfig}. The caller retains ownership of the directory: {@link + * #close()} does not close it. + */ + public AIJoinIndex(Directory directory) throws IOException { + this(directory, new AIJoinIndexConfig()); + } + + /** + * Opens a persistent auxiliary join index over the given directory, creating it if empty, using + * the given {@link AIJoinIndexConfig}. The caller retains ownership of the directory: {@link + * #close()} does not close it. + */ + public AIJoinIndex(Directory directory, AIJoinIndexConfig 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 AIJoinIndexConfig} and {@link MergeScheduler} in place of the default {@link + * ConcurrentMergeScheduler}. The caller retains ownership of the directory: {@link #close()} does + * not close it. + */ + public AIJoinIndex(Directory directory, AIJoinIndexConfig config, MergeScheduler mergeScheduler) + throws IOException { + this.mergeScheduler = mergeScheduler; + this.mergePolicy = new AIJoinMergePolicy(); + this.mergePolicy.setSweepInterval(config.getSweepSamplingIntervalNanos(), TimeUnit.NANOSECONDS); Review Comment: Only the sweep interval is copied from `AIJoinIndexConfig`; `blockingRefresh` and `singleFieldPerSegment` are never read anywhere. Consequently `setBlockingRefresh(false)` still calls `maybeRefreshBlocking()`, and `setSingleFieldPerSegment(true)` still batches all fields, despite the public API promising otherwise. Implement both settings or remove them until supported. ########## solr/core/src/test/org/apache/solr/search/join/TestAIJoinQParserPlugin.java: ########## @@ -0,0 +1,172 @@ +/* + * 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.util.List; +import java.util.Map; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.client.solrj.embedded.EmbeddedSolrServer; +import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.core.SolrCore; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * Exercises {@link AIJoinQParserPlugin}, modeled on {@link TestScoreJoinQPNoScore}'s same-core + * {@code {!join}} coverage plus {@link org.apache.solr.TestCrossCoreJoin}'s cross-core setup, since + * cross-core joins -- not same-core -- are the reason {@link + * org.apache.solr.search.join.aijoin.AIJoinIndex} exists. + * + * <p>Every join here goes from the "many" side (an employee, single-valued FK) to the "few" side + * (that employee's department, a value unique per to-doc): {@code AIJoinUtil#computeDocMapping} is, + * per its own javadoc, "always single-valued until M:N pairs are supported" -- it keeps exactly one + * to-doc per from-doc. That's exact for this direction (each employee has exactly one department), + * but would silently drop matches for the reverse, one-department-to-many-employees direction, so + * this test doesn't exercise that one. + * + * <p>Join fields use the {@code *_s_dv} docValues companions that {@code schema-docValuesJoin.xml} + * copies {@code *_s} into, rather than the plain fields: {@link + * org.apache.solr.search.join.aijoin.AIJoinIndex} reads real per-segment {@link + * org.apache.lucene.index.SortedSetDocValues} directly, unlike {@link ScoreJoinQParserPlugin} (via + * {@link org.apache.lucene.search.join.JoinUtil}) which tolerates uninverted fields too. + */ +public class TestAIJoinQParserPlugin extends SolrTestCaseJ4 { Review Comment: New test suites should avoid the legacy `SolrTestCaseJ4` harness. Use `SolrTestCase` with `EmbeddedSolrServerTestRule` (or `SolrJettyTestRule` when HTTP is required) so this coverage follows the current test lifecycle and isolation model. ########## solr/core/src/java/org/apache/solr/search/join/aijoin/AIJoinUtil.java: ########## @@ -0,0 +1,580 @@ +/* + * 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.aijoin; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.regex.Pattern; +import org.apache.lucene.codecs.Codec; +import org.apache.lucene.codecs.FieldInfosFormat; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.DocValues; +import org.apache.lucene.index.FieldInfo; +import org.apache.lucene.index.FieldInfos; +import org.apache.lucene.index.FilterCodecReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.ParallelCompositeReader; +import org.apache.lucene.index.ParallelLeafReader; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.SortedNumericDocValues; +import org.apache.lucene.index.SortedSetDocValues; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.search.BulkScorer; +import org.apache.lucene.search.DocIdSet; +import org.apache.lucene.search.DocIdSetIterator; +import org.apache.lucene.search.DocIdStream; +import org.apache.lucene.search.LeafCollector; +import org.apache.lucene.search.Scorable; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.store.FilterDirectory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.util.Accountable; +import org.apache.lucene.util.BitDocIdSet; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.FixedBitSet; +import org.apache.lucene.util.RamUsageEstimator; +import org.apache.lucene.util.RoaringDocIdSet; +import org.apache.lucene.util.StringHelper; +import org.slf4j.Logger; +import org.slf4j.event.Level; + +/** + * Column-building and addressing helpers for the auxiliary join index managed by {@link + * AIJoinIndex}: for every (from-segment, to-segment) pair it produces a SORTED_NUMERIC column named + * {@link #pairFieldName}, whose doc number is the from-side doc id and whose value is the to-side + * doc id whose {@code toField} term equals the from doc's {@code fromField} term, plus two + * companion edges columns persisting the pair's {min, max} from-doc and to-doc bounds. + */ +final class AIJoinUtil { + + /** Suffix of the always-written column persisting a pair's {min, max} from-doc edges. */ + static final String FROM_EDGES_PREFIX = "fromDoc_edges_"; // TODO reduce to the singe letter + + /** Suffix of the always-written column persisting a pair's {min, max} to-doc edges. */ + static final String TO_EDGES_PREFIX = "toDoc_edges_"; + + /** main join colums for join index to_doc_num[from_docnum] */ + static final String TO_DOC_VAL_BY_FROM_DOCNUM = "join_toDoc_"; + + static final String TO_COUNT_PREFIX = "num_toDoc_"; + + private AIJoinUtil() {} + + /** + * Configurable level for the {@code AIJOIN evt=...} diagnostic logs; defaults to {@code INFO}, + * override with the {@code solr.aijoin.log.level} system property (or {@code + * SOLR_AIJOIN_LOG_LEVEL} env var). + */ + static final Level AIJOIN_LOG_LEVEL = Level.TRACE; Review Comment: The documented log-level configuration is not implemented: this constant hardcodes TRACE, so the stated INFO default and `solr.aijoin.log.level`/`SOLR_AIJOIN_LOG_LEVEL` overrides have no effect. Read the property through `EnvUtils` and parse it with an INFO default, or update the documentation if TRACE is intentionally fixed. ########## solr/core/src/java/org/apache/solr/search/join/aijoin/AIJoinUtil.java: ########## @@ -0,0 +1,580 @@ +/* + * 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.aijoin; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.regex.Pattern; +import org.apache.lucene.codecs.Codec; +import org.apache.lucene.codecs.FieldInfosFormat; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.DocValues; +import org.apache.lucene.index.FieldInfo; +import org.apache.lucene.index.FieldInfos; +import org.apache.lucene.index.FilterCodecReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.ParallelCompositeReader; +import org.apache.lucene.index.ParallelLeafReader; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.SortedNumericDocValues; +import org.apache.lucene.index.SortedSetDocValues; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.search.BulkScorer; +import org.apache.lucene.search.DocIdSet; +import org.apache.lucene.search.DocIdSetIterator; +import org.apache.lucene.search.DocIdStream; +import org.apache.lucene.search.LeafCollector; +import org.apache.lucene.search.Scorable; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.store.FilterDirectory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.util.Accountable; +import org.apache.lucene.util.BitDocIdSet; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.FixedBitSet; +import org.apache.lucene.util.RamUsageEstimator; +import org.apache.lucene.util.RoaringDocIdSet; +import org.apache.lucene.util.StringHelper; +import org.slf4j.Logger; +import org.slf4j.event.Level; + +/** + * Column-building and addressing helpers for the auxiliary join index managed by {@link + * AIJoinIndex}: for every (from-segment, to-segment) pair it produces a SORTED_NUMERIC column named + * {@link #pairFieldName}, whose doc number is the from-side doc id and whose value is the to-side + * doc id whose {@code toField} term equals the from doc's {@code fromField} term, plus two + * companion edges columns persisting the pair's {min, max} from-doc and to-doc bounds. + */ +final class AIJoinUtil { + + /** Suffix of the always-written column persisting a pair's {min, max} from-doc edges. */ + static final String FROM_EDGES_PREFIX = "fromDoc_edges_"; // TODO reduce to the singe letter + + /** Suffix of the always-written column persisting a pair's {min, max} to-doc edges. */ + static final String TO_EDGES_PREFIX = "toDoc_edges_"; + + /** main join colums for join index to_doc_num[from_docnum] */ + static final String TO_DOC_VAL_BY_FROM_DOCNUM = "join_toDoc_"; + + static final String TO_COUNT_PREFIX = "num_toDoc_"; + + private AIJoinUtil() {} + + /** + * Configurable level for the {@code AIJOIN evt=...} diagnostic logs; defaults to {@code INFO}, + * override with the {@code solr.aijoin.log.level} system property (or {@code + * SOLR_AIJOIN_LOG_LEVEL} env var). + */ + static final Level AIJOIN_LOG_LEVEL = Level.TRACE; + + /** Whether the AIJOIN diagnostic logs would emit at the configured level. */ + static boolean diagnosticsEnabled(Logger log) { + return log.isEnabledForLevel(AIJOIN_LOG_LEVEL); + } + + /** Emits an AIJOIN diagnostic line at the configured level. */ + static void logDiagnostic(Logger log, String message, Object... args) { + log.atLevel(AIJOIN_LOG_LEVEL).log(message, args); + } + + /** + * A pair's {min, max} from-doc and to-doc bounds and match count, common to both a pair freshly + * built on demand ({@link JoinColumnModel#edges()}) and one already persisted in the join index + * ({@link Edges}, loaded through {@link #loadEdges}), so code walking matches doesn't need to + * care which one backs it. + */ + interface DocEdges { + int[] fromDocEdges(); + + int[] toDocEdges(); + + /** this is rather doubtful */ + int toCount(); + } + + /** A self-contained {@link DocEdges} value, with no addressing information of its own. */ + record Edges(int[] fromDocEdges, int[] toDocEdges, int toCount) implements DocEdges {} + + /** + * The from-doc-to-to-doc map produced by {@link #computeDocMapping}, paired with its resolved + * {@link #edges()}. {@link #toDocByFromDoc()} mirrors the on-disk column's read API, so freshly + * built pairs (not yet flushed to the join index) and pairs loaded from the join index can be + * walked by the same code. + */ + static final class JoinColumnModel { + private final int[] toDocByFromDoc; + private final DocEdges edges; + + JoinColumnModel(int[] toDocByFromDoc, DocEdges edges) { + this.toDocByFromDoc = toDocByFromDoc; + this.edges = edges; + } + + /** + * Returns a fresh single-valued cursor over the from-doc -> to-doc map, positioned before doc + * 0. + */ + SortedNumericDocValues toDocByFromDoc() { + return new ArrayBackedSortedNumericDocValues(toDocByFromDoc); + } + + DocEdges edges() { + return edges; + } + } + + /** + * Adapts an int-array from-doc -> to-doc map (as produced by {@link #computeDocMapping}, {@code + * -1} meaning no value) to the {@link SortedNumericDocValues} read API, so it can be consumed the + * same way as the on-disk join column. Always single-valued until M:N pairs are supported. + */ + private static final class ArrayBackedSortedNumericDocValues extends SortedNumericDocValues { + private final int[] toDocByFromDoc; + private int doc = -1; + + ArrayBackedSortedNumericDocValues(int[] toDocByFromDoc) { + this.toDocByFromDoc = toDocByFromDoc; + } + + @Override + public long nextValue() { + return toDocByFromDoc[doc]; + } + + @Override + public int docValueCount() { + return 1; + } + + @Override + public boolean advanceExact(int target) { + doc = target; + return target < toDocByFromDoc.length && toDocByFromDoc[target] >= 0; + } + + @Override + public int docID() { + return doc; + } + + @Override + public int nextDoc() { + return advance(doc + 1); + } + + @Override + public int advance(int target) { + while (target < toDocByFromDoc.length && toDocByFromDoc[target] < 0) { + target++; + } + doc = target < toDocByFromDoc.length ? target : NO_MORE_DOCS; + return doc; + } + + @Override + public long cost() { + return toDocByFromDoc.length; + } + } + + /** + * Builds the join column for one (from-segment, to-segment) pair: resolves every from-side doc to + * its matching to-side doc id, along with the pair's from-doc and to-doc bounds. From-side terms + * are hashed by {@link ForeignKeyColumn}; each to-side term is looked up in that hash to map + * from-side ords to to-side ords. + * + * <p>Docs already deleted at build time are skipped, purely to avoid persisting entries nobody + * can ever match -- deletes are otherwise re-checked live at query time (from-side in {@code + * ToLeafJoinContext}, to-side by the searcher's own {@code acceptDocs}), since a pair's cached + * mapping outlives whatever gets deleted after it was built. + */ + static JoinColumnModel computeDocMapping( + LeafReaderContext toContext, String toField, ForeignKeyColumn fromSideData) + throws IOException { + assert fromSideData != null; + + long[] toOrdByFromOrd = new long[fromSideData.getFromValuesCount()]; + Arrays.fill(toOrdByFromOrd, -1L); + SortedSetDocValues toDV = DocValues.getSortedSet(toContext.reader(), toField); + Bits toLiveDocs = toContext.reader().getLiveDocs(); + TermsEnum toTerms = toDV.termsEnum(); + // resolve from-side ords to to-side ords: look each to-side term up in the from-side hash. + boolean termsAreDisjoint = true; + for (BytesRef term = toTerms.next(); term != null; term = toTerms.next()) { + int fromOrd = fromSideData.getFromTermOrdOrDashOne(term); + if (fromOrd != -1) { + toOrdByFromOrd[fromOrd] = (int) toTerms.ord(); + termsAreDisjoint = false; + } + } + // TODO: this degrades M:N joins to M:1. Both toDocByToOrd and toDocByFromDoc keep a single + // to-side doc per slot, so when several to docs share a term (non-unique toField) or a + // fromSideData + // doc is multi-valued with several matching terms, later assignments overwrite earlier ones + // and only the last match survives. The read side (AIJoinQuery) already consumes all + // docValueCount() values per doc, so only this writer needs to learn to emit multiple + // to docs per fromSideData doc. + if (!termsAreDisjoint) { + int[] toDocByToOrd = new int[Math.toIntExact(toDV.getValueCount())]; + Arrays.fill(toDocByToOrd, -1); + for (int toDoc = toDV.nextDoc(); + toDoc != DocIdSetIterator.NO_MORE_DOCS; + toDoc = toDV.nextDoc()) { + if (toLiveDocs != null && !toLiveDocs.get(toDoc)) { + continue; + } + for (int i = 0; i < toDV.docValueCount(); i++) { + long toOrd = toDV.nextOrd(); + toDocByToOrd[(int) toOrd] = toDoc; + // TODO we can apply toOrdByFromOrd right here + // and get toDocByFromOrd[] + } + } + + // resolve every fromSideData doc to its to-side doc. Docs without the field, or whose term + // has + // no to-side match, keep -1. + int[] toDocByFromDoc = fromSideData.cloneFromOrdByFromDoc(); + int minFromDoc = DocIdSetIterator.NO_MORE_DOCS; + int maxFromDoc = -1; + int minToDoc = DocIdSetIterator.NO_MORE_DOCS; + int maxToDoc = -1; + int toCount = 0; + // walk the array, mapping each fromSideData ord to its to-side doc in place. + for (int fromDoc = 0; fromDoc < toDocByFromDoc.length; fromDoc++) { + int fromOrd = toDocByFromDoc[fromDoc]; + if (fromOrd == -1) { + continue; + } + int toOrd = (int) toOrdByFromOrd[fromOrd]; + int toDoc = toOrd == -1 ? -1 : toDocByToOrd[toOrd]; + if (toDoc == -1) { + toDocByFromDoc[fromDoc] = -1; // wiping is crucial + continue; + } + toDocByFromDoc[fromDoc] = toDoc; + minFromDoc = Math.min(minFromDoc, fromDoc); + maxFromDoc = Math.max(maxFromDoc, fromDoc); + minToDoc = Math.min(minToDoc, toDoc); + maxToDoc = Math.max(maxToDoc, toDoc); + toCount++; + } + if (maxFromDoc < 0) { // tombstone - column is empty + // no fromSideData doc in this pair maps to any to doc: normalize both edges to the + // symmetric + // {-1, -1} sentinel. An asymmetric one (e.g. {NO_MORE_DOCS, -1}) doesn't round-trip + // through the join index's SORTED_NUMERIC edges column, which always returns its two + // values in ascending numeric order regardless of which was written as "min" -- so + // {NO_MORE_DOCS, -1} silently comes back as {-1, NO_MORE_DOCS} on the next read. + minFromDoc = -1; + minToDoc = -1; + maxToDoc = -1; + } + return new JoinColumnModel( + toDocByFromDoc, + new Edges(new int[] {minFromDoc, maxFromDoc}, new int[] {minToDoc, maxToDoc}, toCount)); + } else { // tombstone - column is empty, due to disjoint terms, perhaps one may optimize it + int[] minusones = new int[fromSideData.fromSideMaxDocs()]; + Arrays.fill(minusones, -1); + return new JoinColumnModel(minusones, new Edges(new int[] {-1, -1}, new int[] {-1, -1}, 0)); + } + } + + /** + * Reads a pair's persisted {@code {min, max}} edges (or {@code toCount}), all stored on doc 0 of + * the column -- the read-side counterpart of {@link AIJoinWriter}'s edges columns. + */ + static int[] loadEdges(LeafReaderContext joinContext, String edgesFieldName) throws IOException { + SortedNumericDocValues edgesDV = joinContext.reader().getSortedNumericDocValues(edgesFieldName); + assert edgesDV != null : "expected edges column to be present: " + edgesFieldName; + int zeroDoc = edgesDV.nextDoc(); + assert zeroDoc == 0 : "expected edges column to be fully materialized, but got doc " + zeroDoc; + int[] values = new int[edgesDV.docValueCount()]; + for (int i = 0; i < values.length; i++) { + values[i] = (int) edgesDV.nextValue(); + } + return values; + } + + /** + * The join index field name addressing the ordinal map of one (from-segment, to-segment) pair. + */ + static String pairFieldName( + LeafReaderContext fromContext, + String fromField, + LeafReaderContext toContext, + String toField) { + return getSideKey(fromContext, fromField) + "_" + getSideKey(toContext, toField); + } + + // Lucene puts no hard constraints on field names, but conservatively keep side keys usable as + // one by reducing them to identifier characters + private static final Pattern NON_IDENTIFIER = Pattern.compile("[^A-Za-z0-9_]"); + + /** + * Persistent identifier of one join side: the join field name, the immutable id the segment was + * created with (it survives reopens, growing deletes mask and reorderings of {@link + * IndexReader#leaves()}; a merge produces a new segment with a new id) and the docvalues + * generation of the join field. + */ + static String getSideKey(LeafReaderContext context, String field) { + byte[] segmentId = segmentReader(context.reader()).getSegmentInfo().info.getId(); + // dvGen starts at -1 and advances only when this particular field receives an in-place + // IndexWriter.updateDocValues update; deletes only bump delGen and leave it untouched. So the + // key is insensitive to deletes but changes when the join field's docvalues are updated. + long dvGen = context.reader().getFieldInfos().fieldInfo(field).getDocValuesGen(); Review Comment: A join field may legitimately be absent from one segment (all documents there lack the field). In that case `fieldInfo(field)` is null and every AIJoin query fails with an NPE while enumerating pairs instead of treating that segment as an empty join side. Use a stable absent-field generation and let the empty DocValues path produce a tombstone mapping. ########## solr/core/src/java/org/apache/solr/search/join/aijoin/AIJoinUtil.java: ########## @@ -0,0 +1,580 @@ +/* + * 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.aijoin; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.regex.Pattern; +import org.apache.lucene.codecs.Codec; +import org.apache.lucene.codecs.FieldInfosFormat; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.DocValues; +import org.apache.lucene.index.FieldInfo; +import org.apache.lucene.index.FieldInfos; +import org.apache.lucene.index.FilterCodecReader; +import org.apache.lucene.index.FilterLeafReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.LeafReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.ParallelCompositeReader; +import org.apache.lucene.index.ParallelLeafReader; +import org.apache.lucene.index.SegmentCommitInfo; +import org.apache.lucene.index.SegmentReader; +import org.apache.lucene.index.SortedNumericDocValues; +import org.apache.lucene.index.SortedSetDocValues; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.search.BulkScorer; +import org.apache.lucene.search.DocIdSet; +import org.apache.lucene.search.DocIdSetIterator; +import org.apache.lucene.search.DocIdStream; +import org.apache.lucene.search.LeafCollector; +import org.apache.lucene.search.Scorable; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.store.FilterDirectory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.util.Accountable; +import org.apache.lucene.util.BitDocIdSet; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.FixedBitSet; +import org.apache.lucene.util.RamUsageEstimator; +import org.apache.lucene.util.RoaringDocIdSet; +import org.apache.lucene.util.StringHelper; +import org.slf4j.Logger; +import org.slf4j.event.Level; + +/** + * Column-building and addressing helpers for the auxiliary join index managed by {@link + * AIJoinIndex}: for every (from-segment, to-segment) pair it produces a SORTED_NUMERIC column named + * {@link #pairFieldName}, whose doc number is the from-side doc id and whose value is the to-side + * doc id whose {@code toField} term equals the from doc's {@code fromField} term, plus two + * companion edges columns persisting the pair's {min, max} from-doc and to-doc bounds. + */ +final class AIJoinUtil { + + /** Suffix of the always-written column persisting a pair's {min, max} from-doc edges. */ + static final String FROM_EDGES_PREFIX = "fromDoc_edges_"; // TODO reduce to the singe letter + + /** Suffix of the always-written column persisting a pair's {min, max} to-doc edges. */ + static final String TO_EDGES_PREFIX = "toDoc_edges_"; + + /** main join colums for join index to_doc_num[from_docnum] */ + static final String TO_DOC_VAL_BY_FROM_DOCNUM = "join_toDoc_"; + + static final String TO_COUNT_PREFIX = "num_toDoc_"; + + private AIJoinUtil() {} + + /** + * Configurable level for the {@code AIJOIN evt=...} diagnostic logs; defaults to {@code INFO}, + * override with the {@code solr.aijoin.log.level} system property (or {@code + * SOLR_AIJOIN_LOG_LEVEL} env var). + */ + static final Level AIJOIN_LOG_LEVEL = Level.TRACE; + + /** Whether the AIJOIN diagnostic logs would emit at the configured level. */ + static boolean diagnosticsEnabled(Logger log) { + return log.isEnabledForLevel(AIJOIN_LOG_LEVEL); + } + + /** Emits an AIJOIN diagnostic line at the configured level. */ + static void logDiagnostic(Logger log, String message, Object... args) { + log.atLevel(AIJOIN_LOG_LEVEL).log(message, args); + } + + /** + * A pair's {min, max} from-doc and to-doc bounds and match count, common to both a pair freshly + * built on demand ({@link JoinColumnModel#edges()}) and one already persisted in the join index + * ({@link Edges}, loaded through {@link #loadEdges}), so code walking matches doesn't need to + * care which one backs it. + */ + interface DocEdges { + int[] fromDocEdges(); + + int[] toDocEdges(); + + /** this is rather doubtful */ + int toCount(); + } + + /** A self-contained {@link DocEdges} value, with no addressing information of its own. */ + record Edges(int[] fromDocEdges, int[] toDocEdges, int toCount) implements DocEdges {} + + /** + * The from-doc-to-to-doc map produced by {@link #computeDocMapping}, paired with its resolved + * {@link #edges()}. {@link #toDocByFromDoc()} mirrors the on-disk column's read API, so freshly + * built pairs (not yet flushed to the join index) and pairs loaded from the join index can be + * walked by the same code. + */ + static final class JoinColumnModel { + private final int[] toDocByFromDoc; + private final DocEdges edges; + + JoinColumnModel(int[] toDocByFromDoc, DocEdges edges) { + this.toDocByFromDoc = toDocByFromDoc; + this.edges = edges; + } + + /** + * Returns a fresh single-valued cursor over the from-doc -> to-doc map, positioned before doc + * 0. + */ + SortedNumericDocValues toDocByFromDoc() { + return new ArrayBackedSortedNumericDocValues(toDocByFromDoc); + } + + DocEdges edges() { + return edges; + } + } + + /** + * Adapts an int-array from-doc -> to-doc map (as produced by {@link #computeDocMapping}, {@code + * -1} meaning no value) to the {@link SortedNumericDocValues} read API, so it can be consumed the + * same way as the on-disk join column. Always single-valued until M:N pairs are supported. + */ + private static final class ArrayBackedSortedNumericDocValues extends SortedNumericDocValues { + private final int[] toDocByFromDoc; + private int doc = -1; + + ArrayBackedSortedNumericDocValues(int[] toDocByFromDoc) { + this.toDocByFromDoc = toDocByFromDoc; + } + + @Override + public long nextValue() { + return toDocByFromDoc[doc]; + } + + @Override + public int docValueCount() { + return 1; + } + + @Override + public boolean advanceExact(int target) { + doc = target; + return target < toDocByFromDoc.length && toDocByFromDoc[target] >= 0; + } + + @Override + public int docID() { + return doc; + } + + @Override + public int nextDoc() { + return advance(doc + 1); + } + + @Override + public int advance(int target) { + while (target < toDocByFromDoc.length && toDocByFromDoc[target] < 0) { + target++; + } + doc = target < toDocByFromDoc.length ? target : NO_MORE_DOCS; + return doc; + } + + @Override + public long cost() { + return toDocByFromDoc.length; + } + } + + /** + * Builds the join column for one (from-segment, to-segment) pair: resolves every from-side doc to + * its matching to-side doc id, along with the pair's from-doc and to-doc bounds. From-side terms + * are hashed by {@link ForeignKeyColumn}; each to-side term is looked up in that hash to map + * from-side ords to to-side ords. + * + * <p>Docs already deleted at build time are skipped, purely to avoid persisting entries nobody + * can ever match -- deletes are otherwise re-checked live at query time (from-side in {@code + * ToLeafJoinContext}, to-side by the searcher's own {@code acceptDocs}), since a pair's cached + * mapping outlives whatever gets deleted after it was built. + */ + static JoinColumnModel computeDocMapping( + LeafReaderContext toContext, String toField, ForeignKeyColumn fromSideData) + throws IOException { + assert fromSideData != null; + + long[] toOrdByFromOrd = new long[fromSideData.getFromValuesCount()]; + Arrays.fill(toOrdByFromOrd, -1L); + SortedSetDocValues toDV = DocValues.getSortedSet(toContext.reader(), toField); + Bits toLiveDocs = toContext.reader().getLiveDocs(); + TermsEnum toTerms = toDV.termsEnum(); + // resolve from-side ords to to-side ords: look each to-side term up in the from-side hash. + boolean termsAreDisjoint = true; + for (BytesRef term = toTerms.next(); term != null; term = toTerms.next()) { + int fromOrd = fromSideData.getFromTermOrdOrDashOne(term); + if (fromOrd != -1) { + toOrdByFromOrd[fromOrd] = (int) toTerms.ord(); + termsAreDisjoint = false; + } + } + // TODO: this degrades M:N joins to M:1. Both toDocByToOrd and toDocByFromDoc keep a single + // to-side doc per slot, so when several to docs share a term (non-unique toField) or a + // fromSideData + // doc is multi-valued with several matching terms, later assignments overwrite earlier ones + // and only the last match survives. The read side (AIJoinQuery) already consumes all + // docValueCount() values per doc, so only this writer needs to learn to emit multiple + // to docs per fromSideData doc. + if (!termsAreDisjoint) { + int[] toDocByToOrd = new int[Math.toIntExact(toDV.getValueCount())]; + Arrays.fill(toDocByToOrd, -1); + for (int toDoc = toDV.nextDoc(); + toDoc != DocIdSetIterator.NO_MORE_DOCS; + toDoc = toDV.nextDoc()) { + if (toLiveDocs != null && !toLiveDocs.get(toDoc)) { + continue; + } + for (int i = 0; i < toDV.docValueCount(); i++) { + long toOrd = toDV.nextOrd(); + toDocByToOrd[(int) toOrd] = toDoc; Review Comment: When multiple live `to` documents in one segment share a value, this assignment overwrites the earlier document, so AIJoin silently returns only the last one; the same duplicates in different segments return all documents. Results therefore depend on segment layout. Store all target doc IDs per ordinal, or reject non-unique `to` values consistently before building mappings. -- 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]
