Copilot commented on code in PR #16858: URL: https://github.com/apache/lucene/pull/16858#discussion_r4322623730
########## lucene/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/JoinIndices.java: ########## @@ -0,0 +1,179 @@ +/* + * 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.lucene.benchmark.byTask.tasks; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import org.apache.lucene.benchmark.byTask.PerfRunData; +import org.apache.lucene.benchmark.byTask.feeds.JoinDocMaker; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; +import org.apache.lucene.index.IndexWriterConfig.OpenMode; +import org.apache.lucene.index.MultiDocValues; +import org.apache.lucene.index.OrdinalMap; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.SearcherFactory; +import org.apache.lucene.search.SearcherManager; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.util.NamedThreadFactory; + +/** + * Writers and NRT searchers over the join indices at <b>join.dir</b>/{layout}/index (Default: + * work/join), shared by join tasks and closed with {@link PerfRunData}. Updates are never + * committed, so the indices stay intact for the next run. + * + * <p><b>join.searchThreads</b> - size of the executor shared by all searchers, i.e. segment + * parallel search on both join sides. Fixed per run, since searchers are only recreated on refresh. + * Default: 0, search segments sequentially. + */ +final class JoinIndices implements Closeable { + + /** + * Searcher that carries the global ordinals of {@link JoinDocMaker#FK} for the mixed index. They + * are built by the first query on each new searcher, so a cold search after refresh pays for + * them. + */ + static final class JoinSearcher extends IndexSearcher { + private final boolean ordinals; + private OrdinalMap ordinalMap; + private boolean built; + + JoinSearcher(IndexReader reader, Executor executor, boolean ordinals) { + super(reader, executor); + this.ordinals = ordinals; + } + + synchronized OrdinalMap ordinalMap() throws IOException { + if (ordinals && !built) { + if (MultiDocValues.getSortedValues(getIndexReader(), JoinDocMaker.FK) + instanceof MultiDocValues.MultiSortedDocValues multi) { + ordinalMap = multi.mapping; // a single segment needs no ordinal map + } + built = true; + } + return ordinalMap; + } + } + + record Index(Directory dir, IndexWriter writer, SearcherManager manager) {} + + private final Path root; + private final ExecutorService executor; + private final Map<String, Index> indices = new HashMap<>(); + + private JoinIndices(Path root, int searchThreads) { + this.root = root; + ThreadFactory named = new NamedThreadFactory("join-search"); + // daemon, since Benchmark never closes PerfRunData, and the pool must not keep the JVM alive + executor = + searchThreads > 0 + ? Executors.newFixedThreadPool( + searchThreads, + r -> { + Thread thread = named.newThread(r); + thread.setDaemon(true); + return thread; + }) + : null; + } + + static JoinIndices get(PerfRunData runData) { + synchronized (runData) { + JoinIndices indices = (JoinIndices) runData.getPerfObject(JoinIndices.class.getName()); + if (indices == null) { + indices = + new JoinIndices( + Path.of(runData.getConfig().get("join.dir", "work/join")), + runData.getConfig().get("join.searchThreads", 0)); + runData.setPerfObject(JoinIndices.class.getName(), indices); + } + return indices; + } + } + + synchronized Index index(String layout) throws IOException { + Index index = indices.get(layout); + if (index == null) { + Directory dir = FSDirectory.open(root.resolve(layout).resolve("index")); + IndexWriter writer = + new IndexWriter( + dir, new IndexWriterConfig().setOpenMode(OpenMode.APPEND).setCommitOnClose(false)); Review Comment: `IndexWriterConfig` generally requires an `Analyzer` via its constructor; `new IndexWriterConfig()` is likely not valid and will fail compilation. Please construct it with an analyzer available in this module (e.g., carry an Analyzer from `PerfRunData` into `JoinIndices`, or use a minimal analyzer appropriate for update-by-term usage), then apply `setOpenMode(OpenMode.APPEND)` and `setCommitOnClose(false)`. ########## lucene/benchmark/src/test/org/apache/lucene/benchmark/byTask/tasks/TestJoinSearchTask.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.lucene.benchmark.byTask.tasks; + +import java.util.Properties; +import org.apache.lucene.benchmark.BenchmarkTestCase; +import org.apache.lucene.benchmark.byTask.PerfRunData; +import org.apache.lucene.benchmark.byTask.feeds.JoinDocMaker; +import org.apache.lucene.benchmark.byTask.utils.Config; +import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.store.FSDirectory; + +public class TestJoinSearchTask extends BenchmarkTestCase { + + private static final int PRODUCTS = 300, SKUS_PER_PRODUCT = 4; + + public void testAlgorithmsAgreeBeforeAndAfterUpdates() throws Exception { + for (String layout : new String[] {"products", "skus", "mixed", "block"}) { + execBenchmark( + new String[] { + "doc.maker=" + JoinDocMaker.class.getName(), + "join.layout=" + layout, + "join.products=" + PRODUCTS, + "join.skusPerProduct=" + SKUS_PER_PRODUCT, + "work.dir=" + getWorkDirPath() + "/" + layout, + "directory=FSDirectory", + "max.buffered=" + (50 + random().nextInt(500)), + "CreateIndex", + "[ { AddJoinDocs > : * ] : 2", + "CloseIndex", + }); + } + assertNumDocs("products", PRODUCTS); + assertNumDocs("skus", PRODUCTS * SKUS_PER_PRODUCT); + assertNumDocs("mixed", PRODUCTS * (SKUS_PER_PRODUCT + 1)); + assertNumDocs("block", PRODUCTS * (SKUS_PER_PRODUCT + 1)); + + long base; + try (PerfRunData runData = runData("both")) { + base = assertAlgorithmsAgree(runData); + } + for (String side : new String[] {"products", "skus", "both"}) { + try (PerfRunData runData = runData(side)) { + for (String group : new String[] {"split", "mixed", "block"}) { + JoinUpdateTask update = new JoinUpdateTask(runData); + update.setParams(group); + for (int i = 0; i < 3; i++) { + update.runAndMaybeStats(false); + } + } + // deterministic: data, update and query sequences are seeded by join.seed; the randomized + // segment sizes and search threads don't affect hit counts + assertNotEquals(side, base, assertAlgorithmsAgree(runData)); Review Comment: This assertion makes the test depend on the total hit count changing after updates, which isn’t strictly guaranteed (it’s possible for the updated values to leave the aggregated counts unchanged by chance). A more deterministic way to validate that updates + refresh actually took effect is to assert an index-level signal post-update (e.g., the refreshed reader shows deletions via `numDeletedDocs()`/`maxDoc() != numDocs()`, or a known updated doc/term becomes visible), while keeping the core invariant that all algorithms agree. ########## lucene/benchmark/src/java/org/apache/lucene/benchmark/byTask/feeds/JoinDocMaker.java: ########## @@ -0,0 +1,226 @@ +/* + * 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.lucene.benchmark.byTask.feeds; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.SplittableRandom; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongUnaryOperator; +import org.apache.lucene.benchmark.byTask.utils.Config; +import org.apache.lucene.document.Document; +import org.apache.lucene.document.Field.Store; +import org.apache.lucene.document.LongPoint; +import org.apache.lucene.document.NumericDocValuesField; +import org.apache.lucene.document.SortedDocValuesField; +import org.apache.lucene.document.StringField; +import org.apache.lucene.util.BytesRef; + +/** + * Generates a products/skus 1:M dataset for {@link + * org.apache.lucene.benchmark.byTask.tasks.JoinSearchTask}: every sku refers to exactly one product + * by its primary key. Field values are a pure function of the document ordinal and its version (0 + * initially, bumped by {@link org.apache.lucene.benchmark.byTask.tasks.JoinUpdateTask}), so all + * layouts hold identical data. + * + * <p>Products carry the PK as indexed term {@link #ID} and as point {@link #ID_NUM}; skus carry + * their own {@link #ID} and the FK as {@link SortedDocValuesField} {@link #FK} and {@link + * NumericDocValuesField} {@link #FK_NUM}. Layouts, i.e. <b>join.layout</b>: + * + * <ul> + * <li><b>products</b>, <b>skus</b> - one side per index, for the term and numeric joins + * <li><b>mixed</b> - both sides as independent docs in one index; products carry their own id in + * {@link #FK}, as the global ordinals join requires + * <li><b>block</b> - each product's skus followed by the product, for the block join; every doc + * of a block carries the product id in {@link #PID}, to rewrite the block on update + * </ul> + * + * <p>Other config properties, see {@link Dataset}: <b>join.products</b> - Default: 1000; + * <b>join.skusPerProduct</b> - Default: 10; <b>join.seed</b> - Default: 42. + * + * <p>Index with {@code { AddJoinDocs } : *}; each call returns the next product group, and the + * {@link Dataset} goes into the commit user data. + */ +public class JoinDocMaker extends DocMaker { + + public static final String TYPE = "type"; + public static final String PRODUCT = "product"; + public static final String SKU = "sku"; + public static final String ID = "id"; + public static final String ID_NUM = "id_num"; + public static final String BRAND = "brand"; + public static final String FK = "fk"; + public static final String FK_NUM = "fk_num"; + public static final String PID = "pid"; + public static final String COLOR = "color"; + public static final String SIZE = "size"; + + public static final int BRANDS = 30; + public static final int COLORS = 20; + public static final int SIZES = 20; + + /** + * Size and seed of the generated data, which determine every document. Indexing records it in the + * commit user data, so updates regenerate documents consistently without repeating the config. + */ + public record Dataset(long products, int skusPerProduct, long seed) { + + private static final String PRODUCTS = "join.products"; + private static final String SKUS_PER_PRODUCT = "join.skusPerProduct"; + private static final String SEED = "join.seed"; + + public static Dataset fromConfig(Config config) { + return new Dataset( + config.get(PRODUCTS, 1000), config.get(SKUS_PER_PRODUCT, 10), config.get(SEED, 42)); + } + + /** Reads what {@link #toUserData()} wrote; fails if the index wasn't built by AddJoinDocs. */ + public static Dataset fromUserData(Iterable<Map.Entry<String, String>> userData) { + Map<String, String> data = new HashMap<>(); + userData.forEach(e -> data.put(e.getKey(), e.getValue())); + if (!data.containsKey(PRODUCTS)) { + throw new IllegalStateException("No join dataset in commit user data, see join-index.alg"); + } + return new Dataset( + Long.parseLong(data.get(PRODUCTS)), + Integer.parseInt(data.get(SKUS_PER_PRODUCT)), + Long.parseLong(data.get(SEED))); + } + + public Map<String, String> toUserData() { + return Map.of( + PRODUCTS, Long.toString(products), + SKUS_PER_PRODUCT, Integer.toString(skusPerProduct), + SEED, Long.toString(seed)); + } + + /** Product's skus followed by the product, i.e. the block join block. */ + public List<Document> block(long product, long version, LongUnaryOperator skuVersion) { + List<Document> block = skus("block", product, skuVersion); + block.add(product("block", product, version)); + return block; + } + + List<Document> skus(String layout, long product, LongUnaryOperator skuVersion) { + List<Document> skus = new ArrayList<>(skusPerProduct + 1); + for (long sku = product * skusPerProduct; sku < (product + 1) * skusPerProduct; sku++) { + skus.add(sku(layout, sku, skuVersion.applyAsLong(sku))); + } + return skus; + } + + public Document product(String layout, long product, long version) { + String id = productId(product); + Document doc = new Document(); + doc.add(new StringField(TYPE, PRODUCT, Store.NO)); + doc.add(new StringField(ID, id, Store.NO)); + doc.add(new LongPoint(ID_NUM, product)); + int brand = random(2 * product, version).nextInt(BRANDS); + doc.add(new StringField(BRAND, value(BRAND, brand), Store.NO)); + switch (layout) { + case "mixed" -> doc.add(new SortedDocValuesField(FK, new BytesRef(id))); + case "block" -> doc.add(new StringField(PID, id, Store.NO)); + default -> {} + } + return doc; + } + + public Document sku(String layout, long sku, long version) { + long product = sku / skusPerProduct; + String fk = productId(product); + SplittableRandom random = random(2 * sku + 1, version); + Document doc = new Document(); + doc.add(new StringField(TYPE, SKU, Store.NO)); + doc.add(new StringField(ID, skuId(sku), Store.NO)); + doc.add(new SortedDocValuesField(FK, new BytesRef(fk))); + doc.add(new NumericDocValuesField(FK_NUM, product)); + doc.add(new StringField(COLOR, value(COLOR, random.nextInt(COLORS)), Store.NO)); + doc.add(new StringField(SIZE, value(SIZE, random.nextInt(SIZES)), Store.NO)); + if (layout.equals("block")) { + doc.add(new StringField(PID, fk, Store.NO)); + } + return doc; + } + + private SplittableRandom random(long key, long version) { + return new SplittableRandom(seed + 31 * (31 * version + key)); + } + } + + private final AtomicLong next = new AtomicLong(); + private String layout; + private Dataset dataset; + + /** Keyword value of a generated attribute, e.g. {@code color7}. */ + public static String value(String field, int n) { + return field + n; + } + + public static String productId(long product) { + return "p" + product; + } + + public static String skuId(long sku) { + return "s" + sku; + } + + @Override + public void setConfig(Config config, ContentSource source) { + super.setConfig(config, source); + layout = config.get("join.layout", "mixed"); + dataset = Dataset.fromConfig(config); + if (!List.of("products", "skus", "mixed", "block").contains(layout)) { + throw new IllegalArgumentException("join.layout: " + layout); + } Review Comment: The thrown `IllegalArgumentException` doesn’t indicate the allowed values, which makes misconfiguration harder to diagnose. Consider including the expected set in the message (similar to the validation messages in `JoinSearchTask` / `JoinUpdateTask`). -- 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]
