CrownChu commented on code in PR #8000: URL: https://github.com/apache/paimon/pull/8000#discussion_r3471558785
########## paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java: ########## @@ -0,0 +1,797 @@ +/* + * 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.paimon.eslib.index; + +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexReader; +import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.ScoredGlobalIndexResult; +import org.apache.paimon.globalindex.io.GlobalIndexFileReader; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.FullTextSearch; +import org.apache.paimon.predicate.VectorSearch; +import org.apache.paimon.types.DataField; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import org.elasticsearch.eslib.api.ArchiveDataProvider; +import org.elasticsearch.eslib.api.ESIndexSearcher; +import org.elasticsearch.eslib.api.model.FieldIndexConfig; +import org.elasticsearch.eslib.api.model.FullTextParams; +import org.elasticsearch.eslib.api.model.IndexFilter; +import org.elasticsearch.eslib.api.model.ScalarPredicate; +import org.elasticsearch.eslib.api.model.SearchResult; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.StringJoiner; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.function.Supplier; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * ES multi-index reader using ESLib. Supports vector search, full-text search, and scalar filtering + * via Lucene-based indexes stored in archive format. + */ +public class ESIndexGlobalIndexReader implements GlobalIndexReader { + + private static final Logger LOG = LoggerFactory.getLogger(ESIndexGlobalIndexReader.class); + private static final String DEBUG_PROPERTY = "paimon.eslib.debug"; + private static final String HNSW_NUM_CANDIDATES_OPTION = "hnsw.num_candidates"; + private static final String HNSW_EF_SEARCH_OPTION = "hnsw.ef_search"; + private static final String NUM_CANDIDATES_OPTION = "num_candidates"; + private static final int DEFAULT_SAMPLE_LIMIT = 20; + + private final GlobalIndexFileReader fileReader; + private final List<GlobalIndexIOMeta> files; + private final List<DataField> fields; + private final ESIndexOptions indexOptions; + private final ExecutorService searchExecutor; + + private final List<SeekableInputStream> allStreams = new ArrayList<>(); + private final Object loadLock = new Object(); + private volatile ESIndexSearcher searcher; + private volatile boolean closed; + private volatile boolean loaded; + + public ESIndexGlobalIndexReader( + GlobalIndexFileReader fileReader, + List<GlobalIndexIOMeta> files, + List<DataField> fields, + ESIndexOptions indexOptions) { + this(fileReader, files, fields, indexOptions, null); + } + + public ESIndexGlobalIndexReader( + GlobalIndexFileReader fileReader, + List<GlobalIndexIOMeta> files, + List<DataField> fields, + ESIndexOptions indexOptions, + ExecutorService searchExecutor) { + checkArgument(files.size() == 1, "Expected exactly one ES index file per shard"); + this.fileReader = fileReader; + this.files = files; + this.fields = fields; + this.indexOptions = indexOptions; + this.searchExecutor = searchExecutor; + this.loaded = false; + this.closed = false; + } + + /** The read/search executor in use (null = serial); visible for tests. */ + ExecutorService searchExecutor() { + return searchExecutor; + } + + @Override + public CompletableFuture<Optional<ScoredGlobalIndexResult>> visitVectorSearch( + VectorSearch vectorSearch) { + return async( + () -> { + try { + ensureLoaded(); + float[] queryVector = vectorSearch.vector(); + int topK = vectorSearch.limit(); + int searchTopK = vectorSearchTopK(vectorSearch); + String fieldName = vectorSearch.fieldName(); + + long[] candidateIds = null; + RoaringNavigableMap64 includeRowIds = vectorSearch.includeRowIds(); + if (includeRowIds != null) { + candidateIds = toArray(includeRowIds); + } + + if (debugEnabled()) { + LOG.info( + "PAIMON_ESLIB_VECTOR_SEARCH_BEGIN field={} topK={} searchTopK={} queryDim={} querySample={} includeCount={} includeSample={} loaded={} indexFiles={} fields={}", + fieldName, + topK, + searchTopK, + queryVector == null ? -1 : queryVector.length, + sample(queryVector, DEFAULT_SAMPLE_LIMIT), + candidateIds == null ? -1 : candidateIds.length, + sample(candidateIds, DEFAULT_SAMPLE_LIMIT), + loaded, + files.size(), + fieldsSample()); + } + + SearchResult result = + searcher.vectorSearch( + fieldName, queryVector, searchTopK, candidateIds); + if (debugEnabled()) { + LOG.info( + "PAIMON_ESLIB_VECTOR_SEARCH_RESULT field={} topK={} searchTopK={} resultCount={} ids={} scores={}", + fieldName, + topK, + searchTopK, + result == null ? 0 : result.count, + result == null + ? "[]" + : sample( + result.ids, result.count, DEFAULT_SAMPLE_LIMIT), + result == null + ? "[]" + : sample( + result.scores, + result.count, + DEFAULT_SAMPLE_LIMIT)); + } + return toScoredResult(result); + } catch (IOException e) { + throw new RuntimeException("Vector search failed", e); + } + }); + } + + @Override + public CompletableFuture<Optional<ScoredGlobalIndexResult>> visitFullTextSearch( + FullTextSearch fullTextSearch) { + return async( + () -> { + try { + ensureLoaded(); + String fieldName = fullTextSearch.fieldName(); + FieldIndexConfig config = indexOptions.getConfig(fieldName); + if (config == null + || config.indexType() != FieldIndexConfig.IndexType.FULLTEXT) { + // This ES index carries the column, but not as a FULLTEXT field (a + // string column defaults to KEYWORD unless an analyzer is configured). + // Full-text search is index-only, so a non-FULLTEXT field yields an + // empty result here rather than letting the eslib searcher throw. + return Optional.empty(); + } + org.apache.paimon.predicate.FullTextQuery.Match match = + requireMatch(fullTextSearch); + int topK = fullTextSearch.limit(); + + SearchResult result = + searcher.fullTextSearch( + fieldName, match.query(), topK, toFullTextParams(match)); + return toScoredResult(result); + } catch (IOException e) { + throw new RuntimeException("Full-text search failed", e); + } + }); + } + + /** + * Returns the {@link org.apache.paimon.predicate.FullTextQuery.Match} of {@code + * fullTextSearch}, rejecting structured queries. Only {@code Match} maps onto the eslib + * single-field full-text API; Phrase/Boolean/Boost/MultiMatch are rejected rather than + * serialized to JSON and fed to the text parser, which would silently search for the literal + * JSON tokens. + */ + private static org.apache.paimon.predicate.FullTextQuery.Match requireMatch( + FullTextSearch fullTextSearch) { + org.apache.paimon.predicate.FullTextQuery ftq = fullTextSearch.query(); + if (ftq instanceof org.apache.paimon.predicate.FullTextQuery.Match) { + return (org.apache.paimon.predicate.FullTextQuery.Match) ftq; + } + throw new UnsupportedOperationException( Review Comment: Fixed in 9212f48b visitFullTextSearch now recursively maps the paimon FullTextQuery tree to an eslib FullTextQuerySpec and executes it via searcher.fullTextSearch(spec, topK). All shapes are implemented, not rejected: - Match → analyzed boolean of term/fuzzy queries (operator/boost/fuzziness honoured) - Phrase → QueryBuilder.createPhraseQuery (analyzer-aware, uses indexed positions) - BooleanQuery → Lucene BooleanQuery (must/should/must_not, recursive) - Boost → FunctionScoreQuery.boostByQuery (Lucene 9 replacement for BoostingQuery) - MultiMatch → expanded to a SHOULD-bool of per-column matches (per-column boost) The only remaining throw is a defensive guard for an unknown FullTextQuery subtype, which the framework never produces. This required eslib-core 1.0.2 (adds FullTextQuerySpec + the structured fullTextSearch overload + lucene-queries for the boost query). Covered by E2E tests (e.g. Phrase order semantics, operator AND vs OR, fuzziness). -- 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]
