jerry-024 commented on code in PR #8000: URL: https://github.com/apache/paimon/pull/8000#discussion_r3466257378
########## paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java: ########## @@ -0,0 +1,123 @@ +/* + * 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.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.GlobalIndexerFactory; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.DataField; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Factory for creating ES multi-index global indexers. Supports vector (DiskBBQ/HNSW/Native), + * fulltext (BM25), and scalar fields. + */ +public class ESIndexGlobalIndexerFactory implements GlobalIndexerFactory { + + public static final String IDENTIFIER = "es-index"; + + private static final String READ_SEARCH_THREADS_KEY = + "global-index.es-index.read-search-threads"; + private static final int DEFAULT_READ_SEARCH_THREADS = -1; + + private static volatile ExecutorService readSearchExecutor; + private static final Object LOCK = new Object(); + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public boolean supportsFullTextSearch() { Review Comment: The comment says FULLTEXT extra fields are supported, but the current full-text scan path only selects indexes whose queried text column is the primary `indexFieldId`. A hybrid index with vector as primary and text as an extra field will not produce `IndexFullTextSearchSplit`s for that text column, so full-text search on the extra field returns no indexed results. The scanner needs to include matching `extraFieldIds` as well, or this support claim should be narrowed to primary full-text fields only. ########## 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: This makes `es-index` fail for structured full-text queries. `FullTextReadImpl` can pass `Phrase`, `BooleanQuery`, `Boost`, and `MultiMatch` through to any backend that advertises `supportsFullTextSearch()`, but this reader throws for everything except `Match`. That means valid full-text reads against ES splits fail at runtime instead of being evaluated or falling back. Either implement these query shapes, avoid advertising support for queries this backend cannot serve, or make the planner/read path fall back before reaching this exception. ########## paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java: ########## @@ -0,0 +1,123 @@ +/* + * 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.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.GlobalIndexerFactory; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.DataField; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Factory for creating ES multi-index global indexers. Supports vector (DiskBBQ/HNSW/Native), + * fulltext (BM25), and scalar fields. + */ +public class ESIndexGlobalIndexerFactory implements GlobalIndexerFactory { + + public static final String IDENTIFIER = "es-index"; + + private static final String READ_SEARCH_THREADS_KEY = + "global-index.es-index.read-search-threads"; + private static final int DEFAULT_READ_SEARCH_THREADS = -1; + + private static volatile ExecutorService readSearchExecutor; + private static final Object LOCK = new Object(); + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public boolean supportsFullTextSearch() { + // The ES index is a hybrid backend: a FULLTEXT-typed column (whether it is the primary + // index field or an extra field of a multi-column index) is served by Lucene full-text + // search. Without this override FullTextScanImpl filters out every es-index file and the + // planner full-text path returns nothing. + return true; + } + + @Override + public GlobalIndexer create(DataField field, Options options) { + return new ESIndexGlobalIndexer( + java.util.Collections.singletonList(field), + options, + getOrCreateReadSearchExecutor(options)); + } + + @Override + public GlobalIndexer create( + DataField indexField, List<DataField> extraFields, Options options) { + List<DataField> fields; + if (extraFields == null || extraFields.isEmpty()) { + fields = java.util.Collections.singletonList(indexField); + } else { + fields = new java.util.ArrayList<>(extraFields.size() + 1); + fields.add(indexField); + fields.addAll(extraFields); + } + return new ESIndexGlobalIndexer(fields, options, getOrCreateReadSearchExecutor(options)); + } + + /** + * Returns the shared read/search thread pool. Default (unset or -1) creates a pool sized to + * CPU/2. Set to 0 to disable parallel search (returns null → serial only). + */ + private static ExecutorService getOrCreateReadSearchExecutor(Options options) { + int threads = options.getInteger(READ_SEARCH_THREADS_KEY, DEFAULT_READ_SEARCH_THREADS); + if (threads == 0) { + return null; + } Review Comment: This static executor is first-use-wins. Once one reader creates `readSearchExecutor`, later ES indexers with a different `global-index.es-index.read-search-threads` value reuse the old pool, and even a later `0` cannot disable async search if the pool already exists. Since this option is table/read configuration, the factory should either scope/cache executors by resolved thread count or make the first-use global behavior explicit and reject conflicting values. -- 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]
