[global-index-eslib] integrates the Elasticsearch (Lucene) index engine into Paimon's global index system - #8000
Conversation
75ebc0d to
b4d2f23
Compare
CrownChu
changed the title
|
Thanks for the contribution. This is a large new global-index integration and the current CI status has many failing jobs, so I am holding off on approval for now. Please get the build/test matrix green first, then it will be easier to do a meaningful code review. |
addc1e8 to
caaa1f3
Compare
… archive Add the paimon-eslib module: an ESLib(Lucene)-backed GlobalIndexer that builds and searches a multi-field global index packed into a single archive file. - ESIndexGlobalIndexer / ESIndexGlobalIndexerFactory: 'es-index' indexer over a primary field + extra fields (multi-column), SPI-registered. - ESIndexGlobalIndexWriter: single-column write(Object,long) and multi-column write(long rowId, InternalRow) -> Lucene archive; finish() emits the offset-table meta; docId = relativeRowId (read maps _ROW_ID = rangeFrom + docId). - ESIndexGlobalIndexReader: async (CompletableFuture) vector / full-text / scalar search over the archive; DiskBBQ vector search via the eslib-core lucene9 reader. - ESIndexOptions / ESIndexBuilderFactory: per-field index config from table options. - depends on the published io.github.paimon.eslib:eslib-api/eslib-core:1.0.0 (lucene9) hosted at https://es-demo-test.oss-cn-hangzhou.aliyuncs.com/maven/. Validated by ESIndexGlobalIndexE2ETest (write -> read -> vector/DiskBBQ/full-text/scalar).
847b4ad to
3905513
Compare
paimon-eslib (ES global index): - reader: checkArgument(files.size()==1) invariant (matches Lumina/Tantivy); IS NULL -> Optional.empty() (raw-scan fallback, since null rows are not indexed); zero-hit filter returns an empty bitmap (not empty(), which would force a fallback); full-text uses the new FullTextQuery API. - writer: explicit null-row handling via builder.addNullDoc (keeps docId<->rowId dense); DATE/TIME->getInt, TIMESTAMP->getTimestamp(precision); drop diagnostic INFO logging. - pom: drop the stale eslib-api dependency (api classes are bundled in eslib-core). - test: nullValuesKeepDocIdRowIdAligned pins docId<->rowId alignment across a null gap. paimon-core/common (vector-read path): GlobalIndexEvaluator IS NOT NULL pruning, AbstractVectorRead/BatchVectorReadImpl/VectorReadImpl/VectorScanImpl eval + batch updates, DataFileRecordReader.
Remove the PAIMON_VECTOR_* debug instrumentation added to the vector read/scan path (AbstractVectorRead/VectorReadImpl/BatchVectorReadImpl fully reverted to the clean version — their only changes were debug; VectorScanImpl keeps the functional canServeScalarFilter change, debug stripped). Paimon keeps no ad-hoc debug logging. Fixes the 2 checkstyle 'System.getProperty' violations (paimon-core validate).
…pty docs
addNullDoc/padding now writes an empty doc (field absent) for every null row, so
notExists() = MUST_NOT FieldExistsQuery matches exactly the null rows. Switch visitIsNull
from the conservative Optional.empty() (raw-scan fallback) to dispatchFilter(notExists()):
IS NULL is now index-evaluated. Add isNullReturnsNullRowsViaIndex (IS NULL -> {3,7},
IS NOT NULL -> the other 8).
Restore DataFileRecordReader to apache/master behaviour: pass the selection-bearing FormatReaderContext straight to the format reader instead of stripping selection via withoutSelection(). For the columnar (Parquet/ORC) path the format-level selection and the reader-level iterator.selection() filter both key off the true returnedPosition(), so the double application is idempotent; the fork-only withoutSelection only disabled format-level row-skip (a perf regression) without fixing a real bug. Now identical to upstream.
1d5d801 to
f6b5076
Compare
Address review findings on the ES global index backend:
1. GlobalIndexEvaluator no longer drops "f IS NOT NULL" when the same
field also carries "f IS NULL". The previous prune treated every
non-IsNotNull leaf as a null-rejecting constraint, so
"f IS NULL AND f IS NOT NULL" collapsed from the empty set to the
null rows. Now only genuinely null-rejecting predicates (comparison /
match functions by arity, plus IS NaN) make IS NOT NULL redundant;
IS NULL never does. Added regression tests.
2. ESIndexGlobalIndexWriter.finish() closes the builder (Lucene
IndexWriter + Directory) and removes its temp directory on every
path via a finally block, including the docCount == 0 early return
and mid-build failures. GlobalIndexWriter has no separate close().
3. The LIKE / EndsWith / Contains visitors now escape Lucene wildcard
metacharacters. SQL LIKE has no escape char in paimon
(sqlToRegexLike(pattern, null)), so '*' and '?' are literal; the old
replace('%','*').replace('_','?') leaked them through as wildcards.
Added an E2E test asserting LIKE 'a*c' matches only the literal star
while '%' / '_' still behave as wildcards.
Three correctness gaps in the ES global-index full-text path: 1. ESIndexGlobalIndexerFactory now overrides supportsFullTextSearch() to return true. It previously inherited the default false, so FullTextScanImpl filtered out every es-index file and the planner full-text path returned nothing (TantivyFullTextGlobalIndexerFactory overrides it; the ES hybrid backend must too). 2. FullTextScanImpl now matches a searched text column against both the primary indexFieldId and the extraFieldIds of a GlobalIndexMeta, and groups each file under every matched text column. The canonical ES layout is "vector primary field + text extra field", so the text column id lives in extraFieldIds; the old indexFieldId-only logic both dropped those files and would NPE on idToColumn.get(indexFieldId). Single-column backends (extraFieldIds == null) are unaffected. 3. ESIndexGlobalIndexReader rejects structured full-text queries (Phrase/Boolean/Boost/MultiMatch) with a clear UnsupportedOperationException instead of serializing them to JSON and feeding the JSON to the plain-text query parser, which silently searched for the literal JSON tokens. Only Match is supported for now; structured-query translation is tracked as follow-up. Adds an E2E assertion that a Phrase query is rejected.
| GlobalIndexFileReader fileReader, | ||
| List<GlobalIndexIOMeta> files, | ||
| ExecutorService executor) { | ||
| return new ESIndexGlobalIndexReader(fileReader, files, fields, indexOptions, executor); |
There was a problem hiding this comment.
The configured read/search executor is dropped here. ESIndexGlobalIndexerFactory creates and passes the pool controlled by global-index.es-index.read-search-threads, but ESIndexGlobalIndexer only stores it and then constructs the reader with the caller-provided executor instead. As a result the new option is silently ignored: setting it to 0 does not disable async/searcher execution, and setting a custom size does not affect the ESLib searcher.load(..., searchExecutor) or the reader async path. Could we pass the stored searchExecutor into ESIndexGlobalIndexReader here (or otherwise merge the two executors deliberately) and add a small test so the option is exercised?
There was a problem hiding this comment.
fixed in 4f39eaf.
createReader now passes the factory-resolved searchExecutor straight to ESIndexGlobalIndexReader instead of the caller-provided executor, so global-index.es-index.read-search-threads is authoritative:
- 0 → factory returns null → reader runs serially and searcher.load(..., null) / the async path are both disabled;
- N (>0) → an N-thread pool is used;
- unset / -1 → the default CPU/2 pool.
The caller-supplied executor is intentionally not substituted when the configured one is null, otherwise 0 could not actually disable async execution (the previous behavior silently fell back to the caller's pool).
Added ESIndexGlobalIndexerExecutorTest to exercise the option: - the configured executor (not the caller's) reaches the reader;
- read-search-threads=0 leaves the reader serial (null executor).
…rimary assumption - ESIndexOptions/ESIndexGlobalIndexWriter now map an ARRAY scalar column to a multi-value scalar index: ARRAY<TINYINT|SMALLINT|INTEGER> -> INT, ARRAY<BIGINT> -> LONG, ARRAY<CHAR|VARCHAR> -> KEYWORD. The element values are extracted as primitive arrays / String[] and fed through ScalarFieldHandler's multi-value path (IntPoint/SortedNumeric per element), so term and IN filters match any row containing the value. Unsupported element types are rejected explicitly. Adds an E2E test for ARRAY<BIGINT> labels (term + IN). - Document in VectorScanImpl that the vector search only routes to a file whose PRIMARY field is the vector column (the canonical ES hybrid layout is vector-as-primary). A vector carried as an extra field would fall back to a brute-force RawVectorSearchSplit rather than the ANN index.
… read pool 1. ESIndexGlobalIndexReader.dispatchFilter now returns Optional.empty() for FULLTEXT fields. A FULLTEXT column is indexed as analyzer-produced tokens, so ordinary SQL predicates (=, <>, <, >, IN, LIKE, IS NULL, ...) evaluated against those tokens do not match the raw column value and would return a wrong bitmap that prunes rows incorrectly. They now fall back to raw scan; only visitFullTextSearch serves FULLTEXT fields (mirrors the Tantivy backend). Adds E2E assertions for =, LIKE, > on a FULLTEXT field. 2. ESIndexGlobalIndexer.createReader now prefers the ES-configured read/search pool (global-index.es-index.read-search-threads) over the caller-supplied executor, falling back to the caller's only when the option is 0 (pool == null). Previously the configured pool was built but never used.
matchQueryText() previously forwarded only FullTextQuery.Match.query() to the eslib plain-text searcher, silently dropping the rest of the Match: operator=AND ran as the parser's default OR (result set too large), and boost / fuzziness / maxExpansions / prefixLength had no effect (wrong scoring / matching). It now serves only a default Match (operator=OR, boost=1.0, no fuzziness, maxExpansions=50, prefixLength=0) and throws UnsupportedOperationException for any non-default parameter, mirroring the existing rejection of structured queries. Wiring these parameters into eslib/Lucene is follow-up work. Adds an E2E assertion that a Match with operator=AND is rejected.
ESIndexGlobalIndexReader now maps a paimon FullTextQuery.Match onto the
new eslib FullTextParams and calls the param-aware
fullTextSearch(field, text, topK, params), so operator / boost /
fuzziness / maxExpansions / prefixLength are honoured by the underlying
Lucene query instead of being rejected. Structured queries (Phrase /
Boolean / Boost / MultiMatch) are still rejected explicitly. Requires
the updated eslib-core jar (FullTextParams + 4-arg fullTextSearch).
Updates the E2E test: AND('even document') matches the 10 docs with both
tokens while OR matches all 20, and fuzzy 'evon'~1 matches the 10 'even'
docs while the exact term matches none.
ESIndexGlobalIndexReader.visitFullTextSearch now checks that the target field is configured as FULLTEXT before querying. FullTextScanImpl selects an ES index whenever the field id is the primary or an extra field, but a string column defaults to KEYWORD (FULLTEXT only when an analyzer is configured). Issuing full-text search on such a field previously let the eslib searcher throw IllegalArgumentException; it now returns Optional.empty() so the engine falls back to raw scan. Adds an E2E assertion that full-text search on the KEYWORD 'category' field of a hybrid index returns empty instead of throwing.
createReader now passes the factory-configured executor straight to the reader instead of falling back to the caller-supplied executor. This makes global-index.es-index.read-search-threads authoritative: 0 leaves the reader serial (null -> async disabled), N uses an N-thread pool, and the default uses the CPU/2 pool. Previously a value of 0 silently fell back to the caller's executor, so async could not be disabled. Adds ESIndexGlobalIndexerExecutorTest asserting the configured executor (and null for 0) reaches the reader rather than the caller's.
Restore DataFileRecordReader to apache/master: pass the selection-bearing FormatReaderContext straight to the format reader. The withoutSelection strip had slipped back into the tree and was re-committed by accident; it is intentionally removed again. Now identical to upstream.
paimon-spark-3.5 declared a top-level compile dependency on paimon-eslib, but paimon-eslib is a Java-11 module built only under the paimon-eslib profile and is not published to the default snapshot repo, so the default CI reactor failed to resolve paimon-eslib:1.5-SNAPSHOT. The Spark module has no compile-time reference to it (the dependency only feeds the shaded jar to bundle the ES index backend), so the dependency is moved into a paimon-eslib profile. Default builds resolve cleanly; -Ppaimon-eslib still bundles paimon-eslib. The shade <include> is a no-op when the dependency is absent.
| 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; |
There was a problem hiding this comment.
This class is not present in the declared io.github.paimon.eslib:eslib-core:1.0.0:lucene9 artifact, so the new module does not compile. With JDK 11 and the profile enabled, mvn -pl paimon-eslib -Ppaimon-eslib,fast-build -DfailIfNoTests=false test fails with cannot find symbol: class FullTextParams; jar tf eslib-core-1.0.0-lucene9.jar only contains FieldIndexConfig, IndexFilter, ScalarPredicate, SearchResult, etc. Please either publish/use an eslib version that contains FullTextParams and the matching fullTextSearch overload, or adapt this code to the API available in 1.0.0 so the optional module can compile.
There was a problem hiding this comment.
Thanks — root cause was that the earlier change reused version 1.0.0 for eslib-core, so consumers (and CI) kept resolving a stale cached 1.0.0 jar that predates FullTextParams. Republishing over the same version is unreliable:
Maven caches release versions in ~/.m2 and never re-downloads them.
Fixed by cutting a new version instead of overwriting:
- Bumped io.github.paimon.eslib:eslib-core / eslib-simdvec to 1.0.1, which now contains FullTextParams and the matching fullTextSearch(String, String, int, FullTextParams) overload.
- Published 1.0.1 (jar + pom + checksums) to the OSS repo; verified the artifact contains FullTextParams (jar tf … | grep FullTextParams → present, sha1 4e2e2be3…).
- Updated paimon-eslib/pom.xml to depend on 1.0.1 for both eslib-core and eslib-simdvec.
- Removed the dead eslib-releases GitHub-raw repository entry (it was empty / 404), so the module now resolves from a single, populated source.
Verified locally with the profile enabled:
mvn -pl paimon-eslib -Ppaimon-eslib,fast-build -DfailIfNoTests=false test → compiles and passes (9 tests).
A new version number forces every environment (CI, local .m2) to pull the fresh jar, so the cannot find symbol: FullTextParams failure no longer reproduces.
(Commits: paimon 3fcbefb bump to 1.0.1, 2badffe drop dead repo; es-paimon-lib d0de40e release 1.0.1.)
…y field) The extraFieldIds matching added earlier let full-text search target a text column carried as an extra field of a multi-column index. That contradicts apache master's intentional design — full-text search requires the text column to be the index's primary field — which is enforced by FullTextSearchBuilderTest.testFullTextSearchRequiresText- ColumnAsPrimaryField (the test failed in CI). Restore the master indexFieldId-only matching. The ES factory's supportsFullTextSearch() override and the reader-side FULLTEXT-field guard are unaffected.
@leaves12138 @jerry-024 @JingsongLi
|
| <module>paimon-eslib</module> | ||
| </modules> | ||
| <activation> | ||
| <jdk>11</jdk> |
There was a problem hiding this comment.
This only auto-activates the module on JDK 11, not on later JDKs. I verified mvn help:active-profiles -DskipTests locally: with JAVA_HOME set to JDK 11 the paimon-eslib profile is active, but with JDK 17 or JDK 21 it is not, so the new module is silently skipped in those builds despite the comment saying JDK 11+. Please use a range such as [11,) (and keep the Spark-side profile aligned) so Java 17/21 builds also compile and test paimon-eslib.
There was a problem hiding this comment.
Thanks. 11 is a prefix match that only fires on JDK 11, so 17/21 builds silently skipped the module. Changed both the root paimon-eslib profile and the paimon-spark-3.5 paimon-eslib profile to activate on
[11,) — the Spark-side one had no at all, so it now auto-activates in sync with the root, and JDK 8 still skips both. Verified with mvn help:active-profiles that paimon-eslib is now active on JDK 11 and 17. Fixed in c2e1f85.
| .add(indexFile); | ||
| for (int columnId : matchedTextColumnIds(meta, textColumnIds)) { | ||
| String columnName = checkNotNull(idToColumn.get(columnId)); | ||
| if (!identity.equals(chosenByColumn.get(columnName))) { |
There was a problem hiding this comment.
This selection is global for the whole column, so it can drop valid ranges from the non-chosen identity. For example, suppose a dedicated full-text index on content covers rows [0, 9], and an ES/vector index with content in extraFieldIds covers rows [10, 19]. chooseIndexPerColumn picks the dedicated identity, this branch skips the ES file for [10, 19], and the raw fallback below still builds GlobalIndexCoverage from allIndexFiles; because coverage counts extraFieldIds, [10, 19] is considered indexed and is not added as a raw range. A full-text query then misses matching rows in [10, 19]. Please choose the serving identity per column/range, or compute raw coverage from the same files that are actually emitted as searchable splits.
There was a problem hiding this comment.
thanks. The selection was per-column, so a range covered only by the non-chosen identity was dropped while the raw-coverage fallback (computed from all index files) still counted it as indexed, leaving those rows unsearchable. Switched to selecting the serving identity per (column, row range) in chooseIndexPerColumnAndRange, so each range's covering index is emitted as its own split. This also required building the reader per split in FullTextReadImpl, since a column can now legitimately carry splits from multiple identities (reusing the first split's config would misread the others). Added a regression test that places a dedicated full-text index over rows [0,1] and an extra-field index over [2,3] and asserts a query matches rows in both ranges. Fixed in c2e1f85.
| case DOUBLE: | ||
| return ScalarFieldType.DOUBLE; | ||
| default: | ||
| return ScalarFieldType.KEYWORD; |
There was a problem hiding this comment.
This fallback makes unsupported scalar types fail later in the writer. For example, a table can include a BOOLEAN or DECIMAL companion column in an es-index; ESIndexOptions maps it to ScalarFieldType.KEYWORD, but ESIndexGlobalIndexWriter.extractScalar(...) falls through to row.getString(pos).toString() for those type roots, and GenericRow.getString casts the stored value to BinaryString. The index build then throws a ClassCastException instead of either indexing the value correctly or rejecting the schema up front. Please either add type-specific extraction/support for the non-numeric scalar roots you want to allow, or throw an IllegalArgumentException here like mapArrayScalarType does for unsupported array element types.
There was a problem hiding this comment.
Fixed by failing fast, mirroring mapArrayScalarType: mapScalarType now maps CHAR/VARCHAR to KEYWORD explicitly and throws IllegalArgumentException for unsupported scalar roots (BOOLEAN, DECIMAL, ...), instead of defaulting them to KEYWORD and blowing up later in ESIndexGlobalIndexWriter.extractScalar with a ClassCastException. Added a test asserting a BOOLEAN companion column is rejected up front at option-parse time. Fixed in c2e1f85.
- FullTextScanImpl/FullTextReadImpl: choose the serving index identity per (column, row range) instead of per whole column, so a range covered only by a non-preferred index is no longer dropped; build the reader per split so a column served by several identities reads each split with its own config. - ESIndexOptions.mapScalarType: reject unsupported scalar roots (BOOLEAN, DECIMAL, ...) up front, mirroring mapArrayScalarType, instead of mapping them to KEYWORD and failing with a ClassCastException at index build time. - paimon-eslib and paimon-spark-3.5 profiles: activate on <jdk>[11,)</jdk> so Java 17/21 builds also compile and test paimon-eslib, not only JDK 11.
master replaced predicate.FullTextQuery (the Match/Phrase/Boolean/Boost query tree) with a flat FullTextSearch(fieldName, jsonQuery, limit) and renamed paimon-tantivy to paimon-full-text. Reconcile the eslib global-index integration that was built on the old query tree: - FullTextSearchBuilderTest: query via withQuery(field, matchQuery(..)) - ESIndexGlobalIndexReader: parse the JSON DSL (match / match_phrase / boolean / boost) into FullTextQuerySpec, reading operator / boost / fuzziness / max_expansions / prefix_length from the match node, instead of mapping the removed FullTextQuery tree - ESIndexGlobalIndexE2ETest: build queries as JSON DSL strings paimon-core FullTextSearchBuilderTest (20) and eslib E2E (8) pass.
A String column with no analyzer/type configured now defaults to FULLTEXT (standard analyzer) with a keyword sub-field, instead of KEYWORD. This closes a full-text coverage gap: a text column carried only as an extra field of a hybrid index would default to KEYWORD, be counted as full-text coverage during planning, yet no-match at read time, silently skipping those rows. Exact filters keep exact semantics by routing to the keyword sub-field; opt out of full-text with type=keyword.
| } | ||
| return textColumnIds.contains(globalIndex.indexFieldId()) | ||
| return !matchedTextColumnIds(globalIndex, textColumnIds).isEmpty() | ||
| && supportsFullTextSearch(entry.indexFile().indexType()); |
There was a problem hiding this comment.
Blocking: this still treats every es-index file as full-text coverage at the index-type level. The default STRING change fixes the implicit case, but a user can still explicitly configure global-index.es-index.fields.title.type=keyword. In that valid layout, FullTextScanImpl includes the ES file because supportsFullTextSearch() is true and the title field is covered by indexFieldId/extraFieldIds; GlobalIndexCoverage then marks that row range as indexed, so no RawFullTextSearchSplit is emitted. At read time ESIndexGlobalIndexReader no-matches the KEYWORD field for full-text search, so matching rows in that range are silently skipped. Please make the coverage decision field-capability-aware, or exclude explicit keyword fields from full-text coverage, and add a regression for an es-index range where the searched text column is configured as type=keyword.
There was a problem hiding this comment.
Thanks for pointing this out. This has been addressed in 3b2f57a without adding a field-level capability API to GlobalIndexerFactory.
We no longer allow a STRING field in es-index to be keyword-only.type=keyword now selects KEYWORD as the primary representation, while ESIndexOptions automatically adds a <field>.fulltext FULLTEXT multi-field.
The writer indexes the same value into both fields, and ESIndexGlobalIndexReader routes full-text queries on a KEYWORD-primary field to <field>.fulltext.
Therefore, an es-index file covering a text column is now genuinely capable of serving full-text search, so the existing index-type-level coverage decision no longer suppresses a required raw fallback.
Add complementary text multi-fields, align the module with master, and keep the ESLib integration self-contained without changes to Spark, paimon-common, or paimon-core.
1060ca5 to
48035a5
Compare
|
ESLib's SQL LIKE conversion does not follow the default slash escape, resulting in incorrect matching for 'admin \ _%'; The supplementary regression test failed stably (expected 1, actual 0). All 22 tests have passed * *. Additionally, there is a potential concurrent resource leakage window between 'ensueLoaded()' and 'close()' |
…ex-support-multi-eslib
|
Thanks — both issues have now been fixed.
The branch has also been upgraded to ESLib 1.0.4 in Local verification:
回复 executor 的旧 review 线程： Sorry for missing the reply on this thread. This was fixed in
Therefore the executor configuration is no longer first-use-wins. The executor regression tests also cover configured/caller executor separation, serial mode, and the nested-executor deadlock guard. |
…ex-support-multi-eslib
|
+1 |
4 participants
Summary
This PR integrates the Elasticsearch (Lucene) index engine into Paimon's global index system via the paimon-eslib module, enabling Paimon tables to directly leverage the ES engine's vector search (DiskBBQ / HNSW) and scalar/text
filtering capabilities. Key changes:
Details
Paimon Integration with the ES Index Engine
Paimon's global index is bridged to the ES (Lucene) engine through paimon-eslib: on the write side, Flink uses this engine to build vector/scalar indexes (producing ESLib archive files); on the query side, ES's paimon-store mounts
and reuses the same engine. This PR completes the query-side engine's parallel search, concurrency safety, and filter operators.
Parallel Search
The search thread pool lifecycle is owned by the factory layer (ESIndexGlobalIndexerFactory), lazily initialized and shared across all readers. Configurable via global-index.es-index.read-search-threads:
The executor is injected through the full chain: Factory → Indexer → Reader → ESIndexSearcher → Lucene Codec (via the SearchExecutorHolder ThreadLocal bridge, to work around Lucene SPI's no-arg constructor constraint).
Scalar Filter
ESIndexGlobalIndexReader now implements all visitor methods, dispatching to ESLib's unified filter API:
Dependency Publishing
ESLib jars are published to a GitHub raw repository (CrownChu/es-paimon-lib-releases) with full Maven metadata and checksums. paimon-eslib/pom.xml declares the repository so CI can resolve dependencies without manual local
installation.