[elasticsearch] Add paimon_es_search for Spark, Flink and Java SDK by CrownChu · Pull Request #10777 · apache/paimon

CrownChu · 2026-10-23T11:05:12Z

Purpose

Add paimon_es_search over Paimon's existing Java Elasticsearch global indexes, with a shared request and result contract for Spark, Flink and the Java SDK. Today callers can use the vector/full-text entry points individually; this change also exposes a bounded Elasticsearch-style DSL, including text queries, Boolean filters and filtered KNN, with snapshot-consistent source fetching.

The branch is based directly on apache/paimon:master. It contains two commits:

Rename the paimon-eslib directory, Maven artifact and opt-in profile to paimon-elasticsearch. Keep the Java packages, es-index identifier, table options and persisted Java archive format compatible.
Add the shared request parser/planner/results, the Java Lucene query adapter, Spark table-valued function, distributed Flink procedure and Java SDK entry point. Extend the existing Java 11 CI lane to exercise all three entry points.

Usage:

Spark: SELECT * FROM paimon_es_search('catalog', '<request_json>').
Flink: CALL catalog.sys.paimon_es_search('<request_json>').
Java: EsIndexSearch.search(catalog, requestJson).

The Java integration reuses the existing published ESLib 1.0.7 and Lucene 9.12.0 dependencies. Index construction and persisted metadata are unchanged. The query adapter owns its archive readers independently and preserves stream cleanup when loading fails and is retried.

The initial contract supports one append-only data-evolution table, one ordered es-index field group, latest or explicit snapshots, text/scalar/bool queries and root KNN with an optional prefilter. Queries reject incomplete/stale index coverage, incompatible field mappings, deletion-vector snapshots and query authorization policies. Pagination uses a bounded global score/row-ID ordering; text scores use shard-local statistics. Aggregations, custom sort, multi-target search and tags are outside this initial contract.

See the search guide for request examples, result decoding, prerequisites and limits.

Tests

Local validation passed 101 unique functional tests on JDK 11 with Maven verify, including Checkstyle, Spotless, Enforcer and RAT. The CI routing/command suite also passed 30 tests. These are local results; upstream CI is reported separately by GitHub.

mvn -B -ntp -Pflink1,spark3,paimon-elasticsearch \
  -pl paimon-spark/paimon-spark-ut,paimon-flink/paimon-flink-common -am \
  -Dpaimon.es.java.tests -Dmaven.jar.forceCreation=true \
  '-Dtest=EsSearch*Test,ESIndex*Test,PaimonEsSearchProcedureITCase' \
  -DfailIfNoTests=false \
  -DwildcardSuites=org.apache.paimon.spark.catalyst.plans.logical.PaimonEsSearchQueryTest,org.apache.paimon.spark.query.PaimonEsSearchJavaTest \
  verify

Coverage includes Java archive compatibility, failed-load cleanup/retry, HNSW / INT8 HNSW / DiskBBQ, text/bool/filter-KNN, pagination, source reads, snapshot pinning, stale/incomplete index rejection, field renames, and Spark/Flink result parity with the Java SDK. Flink tests execute shard work through a local Flink runtime; Spark tests use local[2].

CI routing/command tests:

python3 -m unittest discover -s tools/ci -p 'test_*.py'
CI_DRY_RUN=true bash tools/ci/run-java-tests.sh eslib 2.12 11

Akash3121 · 2026-10-25T03:50:26Z


-    <artifactId>paimon-eslib</artifactId>
-    <name>Paimon : ESLib Index</name>
+    <artifactId>paimon-elasticsearch</artifactId>


[High] Renaming the Maven artifact/profile breaks existing users.
Renaming the directory is harmless, but changing the published artifact from org.apache.paimon:paimon-eslib to org.apache.paimon:paimon-elasticsearch (and removing the paimon-eslib profile) makes existing downstream builds and plugin deployment scripts fail immediately on upgrade. Keeping the Java packages and archive format compatible does not help those consumers because Maven can no longer resolve the old coordinate. Could we retain paimon-eslib as a deprecated relocation/shim artifact and keep the old profile as an alias for at least one release? Maven's relocation metadata or a small compatibility module depending on paimon-elasticsearch would preserve the upgrade path.

Addressed in aa86d32.

The implementation remains in paimon-elasticsearch. A deprecated paimon-eslib
relocation POM under paimon-elasticsearch/compat redirects the old Maven coordinate to
the new artifact at the same group/version. The old -Ppaimon-eslib profile is also
retained as an alias, and the Java ES CI lane includes the compatibility module.

Validated the new profile, the old profile, and both profiles together. A separate
consumer using the old coordinate compiled and ran against the new JAR and its transitive
Lucene dependency in a fresh local Maven repository.

The guide also clarifies that direct JAR-download scripts and source-directory selectors
must use the new name; Maven relocation only covers dependency resolution.

Akash3121 · 2026-10-25T03:53:45Z

+    <code>match_phrase</code>, <code>bool</code> and root-level <code>knn</code> with an optional
+    filter. The persisted field mappings and analyzers determine query semantics. Java numeric terms are typed, integer fields require integral values, and date bounds use epoch milliseconds.</td></tr>


[Med] documented date unit it incorrect and can silently return wrong results::
This unit is not correct for every temporal type. ESIndexGlobalIndexWriter.extractScalar stores DATE as epoch days, TIME as milliseconds-of-day, and only TIMESTAMP / TIMESTAMP_LTZ as epoch milliseconds; the query parser passes these numeric bounds directly to the Lucene LONG field. A client following this sentence and sending epoch milliseconds for a DATE column will therefore silently match the wrong range (usually nothing). Could we document the unit per logical type and update the parser's "dates use epoch millis" validation message accordingly? A focused DATE/TIME/TIMESTAMP test would also protect the contract.

Addressed in aa86d32.

The guide now documents the stored units explicitly: DATE uses days since 1970-01-01,
TIME uses milliseconds since midnight, and TIMESTAMP / TIMESTAMP_LTZ use Paimon's
epoch-millisecond representation.

The parser's validation message no longer incorrectly describes every temporal value as
epoch milliseconds. The writer encoding is unchanged.

Added four parameterized persisted writer/reader cases covering exact term and bounded
range queries for these types. All passed.

JingsongLi

The shared search entry has end-to-end value. I completed the code/caller review and ran the Java SDK, real local Spark and Flink search suites on JDK 11: 101 unique functional cases passed with Maven verify (Checkstyle, Spotless, Enforcer and RAT enabled). The 30 CI routing/command tests also passed.

Two additional correctness gaps are reproduced below. The relevance case was executed against this PR's exact JavaEsQueryParser and Lucene 9.12; the branch case used a real filesystem catalog, a tag-derived dev branch, persisted Java indexes and source fetching. The latter produces resolvedBranch=dev, emittedBranch=main, hits=4.

The upstream CI run also contains failures in existing Python native-backend tests and AppendTableITCase; those are separate from the two findings here.

JingsongLi · 2026-10-26T04:45:35Z

+            return new MatchAllDocsQuery();
+        }
+        if (positive == 0) {
+            result.add(new MatchAllDocsQuery(), BooleanClause.Occur.MUST);


[P2] Keep pure-negative bool clauses out of the relevance score

The synthetic match-all is added as MUST, so each pure-negative bool contributes 1.0 instead of acting only as an exclusion. This changes result membership after Top-K truncation. With row 0 (tag=x,n=1) and row 1 (tag=y,n=2), the query {"bool":{"should":[{"bool":{"must_not":{"term":{"tag":"x"}}}},{"bool":{"must_not":{"term":{"n":1}}}},{"term":{"n":1}}]}} ranks row 1 at 2.0 ahead of row 0 at 1.0. With size=1, the positive numeric match is lost. I reproduced this against the exact PR parser. Elasticsearch's fixNegativeQueryIfNeeded adds the match-all as FILTER; that produces row 0 at 1.0, then row 1 at 0.0. Please use FILTER here and add a nested-negative score/ranking regression.

Addressed in aa86d32.

The synthetic match-all clause now uses FILTER instead of MUST.

The persisted-index regression verifies that the example produces scores of 1.0 and 0.0
and retains row 0 for Top-1. It also verifies that a standalone pure-negative query scores
0.0. This regression failed before the fix with an actual score of 2.0.

A local Spark integration test additionally checks global Top-1/Top-2 ranking across
multiple index shards and verifies Java SDK result parity. All cases pass after the fix

JingsongLi · 2026-10-26T04:45:35Z

+            row[3] = queryId;
+            row[4] = request.target;
+            row[5] = request.database + "." + request.table;
+            row[6] = "main";


[P2] Reject non-main tables before emitting main-branch identities

The request validator checks only the separate branch member. "table":"T$branch_dev" is accepted and Identifier.create resolves it to the dev branch; the public prepare(FileStoreTable, request) overload also accepts a dev table. I reproduced a complete SDK search with a real indexed dev branch and source fetch: resolvedBranch=dev, emittedBranch=main, hits=4. Every returned hit and PAGE_END therefore identifies a dev snapshot/row as main, despite the documented main-only contract. Please reject non-main identifiers/resolved tables before planning (including the FileStoreTable overload), or explicitly support branches and emit the actual identity.

Addressed in aa86d32.

Search now validates the parsed identifier before catalog lookup and rejects a resolved
non-main table at the start of planning. The prepare(FileStoreTable, request) overload
also validates the request identifier.

Added tests using a tag-derived dev branch with persisted Java indexes for all three
cases:

A dev-qualified request through the catalog.

A dev table passed directly.

A dev-qualified request passed alongside a main table.

All three are rejected. Existing Spark/SDK tests also verify that every HIT and PAGE_END
record carries the main branch identity.

Local validation passed: 110 functional cases, 30 CI routing/command tests, and Maven
compatibility checks. Maven verify completed on JDK 11 with Checkstyle, Spotless,
Enforcer, and RAT enabled. These are local validation results, not a claim that all
upstream CI checks have passed.

…inates Use non-scoring filters for pure-negative bool queries and reject non-main identifiers and resolved tables before search planning. Document temporal query units and cover persisted archives, Spark top-K and SDK branch checks. Retain the paimon-eslib profile and a relocation POM under the renamed module. Include the compatibility artifact in the Java ES CI lane. Validation: 110 functional tests, 30 CI routing tests, all three profile combinations and an isolated Maven consumer compile/runtime check.

赴长风 added 2 commits October 23, 2026 01:19

[elasticsearch] Rename paimon-eslib module to paimon-elasticsearch

4a7ed62

[elasticsearch] Add Java ES DSL search for Spark, Flink and SDK

e60aac6

Akash3121 suggested changes Oct 25, 2026

View reviewed changes

JingsongLi requested changes Oct 26, 2026

View reviewed changes

Navigation Menu

Repository navigation

[elasticsearch] Add paimon_es_search for Spark, Flink and Java SDK - #10777

[elasticsearch] Add paimon_es_search for Spark, Flink and Java SDK#10777
CrownChu wants to merge 3 commits into
apache:masterfrom
CrownChu:codex/paimon-java-es-search-community-20261023

CrownChu commented Oct 23, 2026

Uh oh!

Akash3121 Oct 25, 2026

Uh oh!

CrownChu Oct 26, 2026

Uh oh!

Akash3121 Oct 25, 2026

Uh oh!

CrownChu Oct 26, 2026

Uh oh!

JingsongLi left a comment

Uh oh!

JingsongLi Oct 26, 2026

Uh oh!

CrownChu Oct 26, 2026

Uh oh!

JingsongLi Oct 26, 2026

Uh oh!

CrownChu Oct 26, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		<code>match_phrase</code>, <code>bool</code> and root-level <code>knn</code> with an optional
		filter. The persisted field mappings and analyzers determine query semantics. Java numeric terms are typed, integer fields require integral values, and date bounds use epoch milliseconds.</td></tr>

Conversation

CrownChu commented Oct 23, 2026

Purpose

Tests

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants