You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/14 06:16:11 UTC
(paimon) branch master updated: [global-index-eslib] integrates the Elasticsearch (Lucene) index engine into Paimon's global index system (#8000)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d37e503dc1 [global-index-eslib] integrates the Elasticsearch (Lucene) index engine into Paimon's global index system (#8000)
d37e503dc1 is described below
commit d37e503dc19019a9f1bfd2b349d6a2fae245e0c4
Author: CrownChu <49...@users.noreply.github.com>
AuthorDate: Tue Jul 14 14:16:05 2026 +0800
[global-index-eslib] integrates the Elasticsearch (Lucene) index engine into Paimon's global index system (#8000)
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:
- Parallel cluster search: Inject a shared ExecutorService from
ESIndexGlobalIndexerFactory for DiskBBQ search, replacing per-reader
thread pool creation
- Concurrent close safety: Add a volatile closed flag with
checkNotClosed() to fix a race condition in the reader lifecycle
- Full scalar/text filtering: Implement all GlobalIndexReader visitor
methods (visitEqual, visitLessThan, visitStartsWith, visitLike,
visitIsNull, etc.), dispatching to ESLib's unified IndexFilter /
ScalarPredicate API
- Dependency coordinates: Change the eslib dependency groupId from
org.elasticsearch to io.github.crownchu, version from 1.0.0-SNAPSHOT to
1.0.0
- Public dependency repository: Add a GitHub-hosted Maven repository for
public CI dependency resolution
---
paimon-eslib/README.md | 216 +++
paimon-eslib/pom.xml | 169 ++
.../paimon/eslib/index/ESIndexBuilderFactory.java | 41 +
.../apache/paimon/eslib/index/ESIndexFileMeta.java | 382 ++++
.../eslib/index/ESIndexGlobalIndexReader.java | 1889 ++++++++++++++++++
.../eslib/index/ESIndexGlobalIndexWriter.java | 629 ++++++
.../paimon/eslib/index/ESIndexGlobalIndexer.java | 62 +
.../eslib/index/ESIndexGlobalIndexerFactory.java | 64 +
.../apache/paimon/eslib/index/ESIndexOptions.java | 623 ++++++
....apache.paimon.globalindex.GlobalIndexerFactory | 16 +
.../paimon/eslib/index/ESIndexFileMetaTest.java | 365 ++++
.../index/ESIndexFullTextQueryParserTest.java | 126 ++
.../eslib/index/ESIndexGlobalIndexE2ETest.java | 2000 ++++++++++++++++++++
.../index/ESIndexGlobalIndexReaderDebugTest.java | 82 +
.../index/ESIndexGlobalIndexerExecutorTest.java | 92 +
.../paimon/eslib/index/ESIndexOptionsTest.java | 403 ++++
pom.xml | 15 +
17 files changed, 7174 insertions(+)
diff --git a/paimon-eslib/README.md b/paimon-eslib/README.md
new file mode 100644
index 0000000000..d64ef34a49
--- /dev/null
+++ b/paimon-eslib/README.md
@@ -0,0 +1,216 @@
+<!--
+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.
+-->
+
+# paimon-eslib: ES Index (Multi-Column Global Index)
+
+`paimon-eslib` provides the `es-index` global index: a **Lucene-backed, multi-column, multi-modal**
+index. A single `es-index` file indexes a **primary column** (typically a vector) together with
+optional **companion columns** — full-text, keyword, scalar, and date — so vector search and
+full-text search can be served from the same index, and companion columns stay available for scoring
+and filtering inside one index definition.
+
+Unlike the single-purpose vector and full-text global indexes — each of which indexes one column with
+one dedicated engine — `es-index` groups a primary column plus optional companion columns into one
+index. The primary column determines the main modality (usually a vector); companion columns are
+indexed according to their data type and per-field options.
+
+> **Plugin required.** `es-index` is provided by this `paimon-eslib` module. Make sure `paimon-eslib`
+> is on the classpath of your Spark or Flink job (and of any reader) before building or querying an
+> `es-index`. The module and its ESLib/Lucene dependencies require Java 11 or newer; the root Maven
+> build intentionally skips `paimon-eslib` when it runs on JDK 8.
+
+See the general [Global Index](../docs/docs/multimodal-table/global-index.mdx) documentation for the
+required Data Evolution table properties, coverage/freshness behavior, and shared build options.
+
+## How columns map to index types
+
+For each indexed column, `es-index` decides the sub-index type from the column data type, unless you
+override it with `fields.<field>.type`:
+
+| Column data type | Default sub-index | Override with `type` |
+|---|---|---|
+| `VECTOR<FLOAT>` / `ARRAY<FLOAT>` | Vector (ANN) | `vector` |
+| `STRING` | Full-text primary + keyword sub-field | `fulltext` or `keyword` |
+| `TIMESTAMP` / `DATE` | Long scalar | `date` |
+| supported numeric scalars | Scalar | — |
+
+Text columns always provide both capabilities. A `fulltext` primary field gets a `<column>.keyword`
+exact-match sub-field; a `keyword` primary field gets a `<column>.fulltext` analyzed sub-field.
+
+## Prerequisites
+
+Create a Data Evolution table with the required properties:
+
+```sql
+CREATE TABLE my_table (
+ id INT,
+ embedding ARRAY<FLOAT>,
+ content STRING,
+ category STRING,
+ price INT
+) TBLPROPERTIES (
+ 'bucket' = '-1',
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true',
+ 'global-index.enabled' = 'true'
+);
+```
+
+## Build ES Index
+
+Build the index with `sys.create_global_index` and `index_type => 'es-index'`. To index several
+columns in one `es-index`, pass a **comma-separated** `index_column` list: the **first** column is
+the primary column, the rest are companion columns. Per-column behavior is configured through options
+under the `global-index.es-index.` prefix.
+
+### Spark SQL
+
+```sql
+-- Single vector column with an HNSW index
+CALL sys.create_global_index(
+ table => 'db.my_table',
+ index_column => 'embedding',
+ index_type => 'es-index',
+ options => 'global-index.es-index.fields.embedding.algorithm=hnsw,global-index.es-index.fields.embedding.dimension=768,global-index.es-index.fields.embedding.metric=cosine'
+);
+
+-- Multi-column: vector (primary) + full-text + keyword + scalar in one index.
+-- 'content' becomes full-text because an analyzer is set; 'category' stays keyword; 'price' scalar.
+CALL sys.create_global_index(
+ table => 'db.my_table',
+ index_column => 'embedding,content,category,price',
+ index_type => 'es-index',
+ options => 'global-index.es-index.fields.embedding.algorithm=diskbbq,global-index.es-index.fields.embedding.dimension=768,global-index.es-index.fields.embedding.metric=cosine,global-index.es-index.fields.content.analyzer=standard,global-index.es-index.fields.category.type=keyword'
+);
+```
+
+### Flink SQL
+
+```sql
+-- Single vector column with an HNSW index
+CALL sys.create_global_index(
+ `table` => 'db.my_table',
+ index_column => 'embedding',
+ index_type => 'es-index',
+ options => 'global-index.es-index.fields.embedding.algorithm=hnsw,global-index.es-index.fields.embedding.dimension=768,global-index.es-index.fields.embedding.metric=cosine'
+);
+
+-- Multi-column: vector (primary) + full-text + keyword + scalar in one index.
+CALL sys.create_global_index(
+ `table` => 'db.my_table',
+ index_column => 'embedding,content,category,price',
+ index_type => 'es-index',
+ options => 'global-index.es-index.fields.embedding.algorithm=diskbbq,global-index.es-index.fields.embedding.dimension=768,global-index.es-index.fields.embedding.metric=cosine,global-index.es-index.fields.content.analyzer=standard,global-index.es-index.fields.category.type=keyword'
+);
+```
+
+You can also set the same options in `TBLPROPERTIES` at table-creation time instead of passing them
+per build. Options are resolved with field-level keys taking precedence over index-type-level keys:
+`global-index.es-index.fields.<field>.<key>` overrides `global-index.es-index.<key>`.
+
+## Options
+
+Index-type-level keys use the `global-index.es-index.<key>` prefix and apply to every column of that
+kind; field-level keys use `global-index.es-index.fields.<field>.<key>` and override them for one
+column.
+
+| Key | Applies to | Default | Description |
+|---|---|---|---|
+| `type` | any column | inferred from data type | Force the sub-index type: `vector`, `fulltext`, `keyword`, `date`. |
+| `algorithm` | vector | `hnsw` | Vector algorithm: `hnsw` or `diskbbq`. `native` is not supported by paimon-eslib. |
+| `dimension` | vector | inferred for `VECTOR<FLOAT>` | Vector dimension (1–4096). Required for `ARRAY<FLOAT>`; taken from the type for `VECTOR<FLOAT>`. |
+| `metric` | vector | `cosine` | Distance metric (for example `cosine`, `l2`, `dot_product`). Use the same metric at build and query time. |
+| `m` | vector (`hnsw`) | engine default | HNSW graph out-degree; valid range: 1–512. |
+| `ef_construction` | vector (`hnsw`) | engine default | HNSW construction search width; valid range: 1–3200. |
+| `vectors_per_cluster` | vector (`diskbbq`) | engine default | Target vectors per cluster for DiskBBQ; valid range: 64–65536. |
+| `centroids_per_parent_cluster` | vector (`diskbbq`) | engine default | Number of child centroids per parent cluster for DiskBBQ; valid range: 2–384. |
+| `analyzer` | string | `standard` | Text analyzer used by the full-text primary field or `.fulltext` sub-field: `standard`, `whitespace`, `simple`, or `keyword`. |
+
+## Query
+
+`es-index` implements the same search API as the other global indexes, so querying does not depend on
+which engine built the index — only the build step differs.
+
+### Spark SQL
+
+```sql
+-- Vector search: top-5 nearest neighbors on the primary vector column
+SELECT * FROM vector_search('my_table', 'embedding', array(1.0f, 2.0f, 3.0f), 5);
+
+-- Full-text search on a companion full-text column, using the JSON query DSL
+SELECT * FROM full_text_search(
+ 'my_table',
+ 'content',
+ '{"match":{"query":"paimon lake format"}}',
+ 10
+);
+```
+
+Vector routes and full-text routes over an `es-index` can also be combined with the
+`hybrid_search(...)` table-valued function.
+
+### Flink SQL
+
+Flink exposes vector search as a `CALL` procedure. The procedure returns JSON-serialized rows as
+strings.
+
+```sql
+-- Vector search: top-5 nearest neighbors on the primary vector column
+CALL sys.vector_search(
+ `table` => 'db.my_table',
+ vector_column => 'embedding',
+ query_vector => '1.0,2.0,3.0',
+ top_k => 5
+);
+
+-- Optionally restrict the returned columns
+CALL sys.vector_search(
+ `table` => 'db.my_table',
+ vector_column => 'embedding',
+ query_vector => '1.0,2.0,3.0',
+ top_k => 5,
+ projection => 'id,content'
+);
+```
+
+> **Note.** Flink SQL currently exposes vector search only. For full-text search and hybrid search
+> over an `es-index`, use Spark SQL (the `full_text_search` / `hybrid_search` table-valued functions)
+> or the Java API (`Table.newFullTextSearchBuilder()` / `Table.newHybridSearchBuilder()`).
+
+## Drop ES Index
+
+### Spark SQL
+
+```sql
+CALL sys.drop_global_index(
+ table => 'db.my_table',
+ index_column => 'embedding',
+ index_type => 'es-index'
+);
+```
+
+### Flink SQL
+
+```sql
+CALL sys.drop_global_index(
+ `table` => 'db.my_table',
+ index_column => 'embedding',
+ index_type => 'es-index'
+);
+```
diff --git a/paimon-eslib/pom.xml b/paimon-eslib/pom.xml
new file mode 100644
index 0000000000..8b9ea3ce80
--- /dev/null
+++ b/paimon-eslib/pom.xml
@@ -0,0 +1,169 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>paimon-parent</artifactId>
+ <groupId>org.apache.paimon</groupId>
+ <version>2.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>paimon-eslib</artifactId>
+ <name>Paimon : ESLib Index</name>
+
+ <properties>
+ <!-- eslib-api / eslib-core are compiled for Java 11 (Lucene 9.12); compile this module
+ at Java 11 instead of the project default (1.8) so their class files are readable. -->
+ <target.java.version>11</target.java.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- ESLib dependencies. eslib-core bundles the api + api.model classes (eslib-api was merged
+ into eslib-core), so no separate eslib-api artifact is needed. -->
+ <dependency>
+ <groupId>io.github.paimon.eslib</groupId>
+ <artifactId>eslib-core</artifactId>
+ <version>1.0.6</version>
+ <classifier>lucene9</classifier>
+ <!-- Paimon supplies its own ArchiveDataProvider over FileIO; the standalone
+ OSSRangeIndexInput and Aliyun SDK are not used in this module. -->
+ <exclusions>
+ <exclusion>
+ <groupId>com.aliyun.oss</groupId>
+ <artifactId>aliyun-sdk-oss</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>io.github.paimon.eslib</groupId>
+ <artifactId>eslib-simdvec</artifactId>
+ <version>1.0.6</version>
+ </dependency>
+ <!-- eslib-native (Havenask JNI) intentionally omitted: it is a JDK 21 module (Phase 3,
+ not implemented) and is not referenced by paimon-eslib. Re-add when NATIVE algorithm
+ support lands. -->
+
+ <!-- Lucene (same version as eslib) -->
+ <dependency>
+ <groupId>org.apache.lucene</groupId>
+ <artifactId>lucene-core</artifactId>
+ <version>9.12.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.lucene</groupId>
+ <artifactId>lucene-analysis-common</artifactId>
+ <version>9.12.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>1.7.36</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Test -->
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-common</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <!-- Provides TestLoggerExtension, auto-registered via the JUnit extension SPI that the
+ paimon-common test-jar declares; without it surefire fails to load the extension. -->
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <repositories>
+ <repository>
+ <id>eslib-oss</id>
+ <url>https://paimon-es-public-bucket.oss-cn-beijing.aliyuncs.com/maven/</url>
+ <releases><enabled>true</enabled></releases>
+ <snapshots><enabled>false</enabled></snapshots>
+ </repository>
+ </repositories>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-eslib</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <includes>
+ <include>io.github.paimon.eslib:eslib-core</include>
+ <include>io.github.paimon.eslib:eslib-simdvec</include>
+ <include>org.apache.lucene:*</include>
+ </includes>
+ </artifactSet>
+ <relocations>
+ <relocation>
+ <pattern>org.apache.lucene</pattern>
+ <shadedPattern>org.apache.paimon.shade.lucene912</shadedPattern>
+ </relocation>
+ </relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>module-info.class</exclude>
+ <exclude>META-INF/MANIFEST.MF</exclude>
+ <exclude>META-INF/versions/**</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexBuilderFactory.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexBuilderFactory.java
new file mode 100644
index 0000000000..6104da5df9
--- /dev/null
+++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexBuilderFactory.java
@@ -0,0 +1,41 @@
+/*
+ * 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.elasticsearch.eslib.api.ESIndexBuilder;
+import org.elasticsearch.eslib.api.ESIndexSearcher;
+import org.elasticsearch.eslib.api.model.FieldIndexConfig;
+import org.elasticsearch.eslib.builder.DefaultESIndexBuilder;
+import org.elasticsearch.eslib.searcher.DefaultESIndexSearcher;
+
+import java.io.IOException;
+import java.util.Map;
+
+/** Factory to create ESIndexBuilder/ESIndexSearcher from eslib-core. */
+public class ESIndexBuilderFactory {
+
+ public static ESIndexBuilder create(Map<String, FieldIndexConfig> fieldConfigs)
+ throws IOException {
+ return new DefaultESIndexBuilder(fieldConfigs);
+ }
+
+ public static ESIndexSearcher createSearcher() {
+ return new DefaultESIndexSearcher();
+ }
+}
diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexFileMeta.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexFileMeta.java
new file mode 100644
index 0000000000..eb14b11195
--- /dev/null
+++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexFileMeta.java
@@ -0,0 +1,382 @@
+/*
+ * 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.elasticsearch.eslib.api.model.BuiltinAnalyzer;
+import org.elasticsearch.eslib.api.model.FieldIndexConfig;
+import org.elasticsearch.eslib.api.model.ScalarFieldType;
+import org.elasticsearch.eslib.api.model.VectorAlgorithm;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Versioned metadata for an archived ESLib index. */
+final class ESIndexFileMeta {
+
+ private static final int MAGIC = 0x45534D31; // "ESM1"
+ private static final int VERSION = 2;
+ private static final int CONFIG_ONLY_VERSION = 1;
+
+ private ESIndexFileMeta() {}
+
+ static byte[] write(
+ File[] files,
+ List<String> indexedFieldNames,
+ List<String> indexedFieldTypes,
+ Map<String, FieldIndexConfig> fieldConfigs)
+ throws IOException {
+ if (files.length == 0) {
+ throw new IOException("Cannot write es-index metadata without Lucene files");
+ }
+ if (indexedFieldNames.isEmpty()) {
+ throw new IOException("Cannot write es-index metadata without indexed fields");
+ }
+ if (indexedFieldTypes.size() != indexedFieldNames.size()) {
+ throw new IOException(
+ "ES index field name/type count mismatch: names="
+ + indexedFieldNames.size()
+ + ", types="
+ + indexedFieldTypes.size());
+ }
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeInt(MAGIC);
+ out.writeInt(VERSION);
+ out.writeInt(indexedFieldNames.size());
+ for (int i = 0; i < indexedFieldNames.size(); i++) {
+ String fieldName = indexedFieldNames.get(i);
+ if (!fieldConfigs.containsKey(fieldName)) {
+ throw new IOException(
+ "Missing primary field config in es-index metadata: " + fieldName);
+ }
+ writeString(out, fieldName);
+ writeString(out, indexedFieldTypes.get(i));
+ }
+ out.writeInt(fieldConfigs.size());
+ for (Map.Entry<String, FieldIndexConfig> entry : fieldConfigs.entrySet()) {
+ FieldIndexConfig config = entry.getValue();
+ writeString(out, entry.getKey());
+ writeString(out, config.indexType().name());
+ writeNullableEnum(out, config.algorithm());
+ out.writeInt(config.dimension());
+ writeNullableString(out, config.metric());
+ writeNullableEnum(out, config.analyzer());
+ writeNullableEnum(out, config.scalarType());
+ out.writeInt(config.algorithmParams().size());
+ for (Map.Entry<String, String> parameter : config.algorithmParams().entrySet()) {
+ writeString(out, parameter.getKey());
+ writeString(out, parameter.getValue());
+ }
+ }
+
+ writeOffsets(out, files);
+ out.flush();
+ return bytes.toByteArray();
+ }
+
+ static Parsed read(byte[] metadata) throws IOException {
+ if (metadata == null || metadata.length == 0) {
+ return new Parsed(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ false);
+ }
+
+ DataInputStream in = new DataInputStream(new ByteArrayInputStream(metadata));
+ int first = in.readInt();
+ if (first != MAGIC) {
+ Map<String, long[]> offsets = readOffsets(in, first);
+ requireFullyConsumed(in);
+ return new Parsed(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ offsets,
+ false);
+ }
+
+ int version = in.readInt();
+ if (version != CONFIG_ONLY_VERSION && version != VERSION) {
+ throw new IOException("Unsupported es-index metadata version: " + version);
+ }
+
+ List<String> indexedFieldNames = new ArrayList<>();
+ List<String> indexedFieldTypes = new ArrayList<>();
+ if (version >= VERSION) {
+ int indexedFieldCount = readCount(in, "indexed field");
+ for (int i = 0; i < indexedFieldCount; i++) {
+ String fieldName = readString(in);
+ if (indexedFieldNames.contains(fieldName)) {
+ throw new IOException(
+ "Duplicate indexed field in es-index metadata: " + fieldName);
+ }
+ indexedFieldNames.add(fieldName);
+ indexedFieldTypes.add(readString(in));
+ }
+ }
+
+ int configCount = readCount(in, "field config");
+ Map<String, FieldIndexConfig> configs = new LinkedHashMap<>();
+ for (int i = 0; i < configCount; i++) {
+ String fieldName = readString(in);
+ FieldIndexConfig.IndexType indexType =
+ parseEnum(FieldIndexConfig.IndexType.class, readString(in), "index type");
+ VectorAlgorithm algorithm = readNullableEnum(in, VectorAlgorithm.class, "algorithm");
+ int dimension = in.readInt();
+ String metric = readNullableString(in);
+ BuiltinAnalyzer analyzer = readNullableEnum(in, BuiltinAnalyzer.class, "analyzer");
+ ScalarFieldType scalarType = readNullableEnum(in, ScalarFieldType.class, "scalar type");
+
+ int parameterCount = readCount(in, "algorithm parameter");
+ Map<String, String> parameters = new LinkedHashMap<>();
+ for (int j = 0; j < parameterCount; j++) {
+ String parameterName = readString(in);
+ String parameterValue = readString(in);
+ if (parameters.put(parameterName, parameterValue) != null) {
+ throw new IOException(
+ "Duplicate algorithm parameter in es-index metadata for field '"
+ + fieldName
+ + "': "
+ + parameterName);
+ }
+ }
+
+ final FieldIndexConfig config;
+ try {
+ config =
+ FieldIndexConfig.builder(fieldName, indexType)
+ .algorithm(algorithm)
+ .dimension(dimension)
+ .metric(metric)
+ .analyzer(analyzer)
+ .scalarType(scalarType)
+ .algorithmParams(parameters)
+ .build();
+ } catch (IllegalArgumentException | NullPointerException e) {
+ throw new IOException(
+ "Invalid field config in es-index metadata for field '" + fieldName + "'",
+ e);
+ }
+ if (configs.put(fieldName, config) != null) {
+ throw new IOException("Duplicate field config in es-index metadata: " + fieldName);
+ }
+ }
+
+ int fileCount = readCount(in, "file");
+ Map<String, long[]> offsets = readOffsets(in, fileCount);
+ requireFullyConsumed(in);
+ if (version == CONFIG_ONLY_VERSION) {
+ indexedFieldNames = inferIndexedFieldNames(configs);
+ }
+ for (String fieldName : indexedFieldNames) {
+ if (!configs.containsKey(fieldName)) {
+ throw new IOException(
+ "Missing primary field config in es-index metadata: " + fieldName);
+ }
+ }
+ return new Parsed(indexedFieldNames, indexedFieldTypes, configs, offsets, true);
+ }
+
+ /**
+ * Version 1 did not store the ordered primary-field list. Its writer emitted each primary
+ * config immediately followed by its generated multi-fields, so recover that order for indexes
+ * created before the explicit mapping was added.
+ */
+ private static List<String> inferIndexedFieldNames(Map<String, FieldIndexConfig> fieldConfigs) {
+ List<Map.Entry<String, FieldIndexConfig>> entries =
+ new ArrayList<>(fieldConfigs.entrySet());
+ List<String> fields = new ArrayList<>();
+ for (int i = 0; i < entries.size(); ) {
+ Map.Entry<String, FieldIndexConfig> primary = entries.get(i++);
+ String fieldName = primary.getKey();
+ fields.add(fieldName);
+ if (primary.getValue().indexType() == FieldIndexConfig.IndexType.FULLTEXT
+ && i < entries.size()
+ && entries.get(i)
+ .getKey()
+ .equals(fieldName + ESIndexOptions.KEYWORD_SUBFIELD_SUFFIX)) {
+ i++;
+ } else if (primary.getValue().indexType() == FieldIndexConfig.IndexType.KEYWORD
+ && i < entries.size()
+ && entries.get(i)
+ .getKey()
+ .equals(fieldName + ESIndexOptions.FULLTEXT_SUBFIELD_SUFFIX)) {
+ i++;
+ }
+ if (i < entries.size()
+ && entries.get(i)
+ .getKey()
+ .equals(fieldName + ESIndexOptions.ARRAY_PRESENCE_SUBFIELD_SUFFIX)) {
+ i++;
+ }
+ }
+ return fields;
+ }
+
+ private static void writeOffsets(DataOutputStream out, File[] files) throws IOException {
+ out.writeInt(files.length);
+ long offset = 4;
+ for (File file : files) {
+ byte[] name = file.getName().getBytes(StandardCharsets.UTF_8);
+ long length = file.length();
+ long dataOffset = offset + 4 + name.length + 8;
+ out.writeInt(name.length);
+ out.write(name);
+ out.writeLong(dataOffset);
+ out.writeLong(length);
+ offset = dataOffset + length;
+ }
+ }
+
+ private static Map<String, long[]> readOffsets(DataInputStream in, int fileCount)
+ throws IOException {
+ if (fileCount < 0) {
+ throw new IOException("Negative file count in es-index metadata: " + fileCount);
+ }
+ Map<String, long[]> offsets = new LinkedHashMap<>();
+ for (int i = 0; i < fileCount; i++) {
+ String name = readString(in);
+ long offset = in.readLong();
+ long length = in.readLong();
+ if (offset < 0 || length < 0 || offset > Long.MAX_VALUE - length) {
+ throw new IOException("Invalid file offset/length in es-index metadata: " + name);
+ }
+ if (offsets.put(name, new long[] {offset, length}) != null) {
+ throw new IOException("Duplicate file offset in es-index metadata: " + name);
+ }
+ }
+ return offsets;
+ }
+
+ private static void requireFullyConsumed(DataInputStream in) throws IOException {
+ if (in.available() != 0) {
+ throw new IOException("Trailing bytes in es-index metadata: " + in.available());
+ }
+ }
+
+ private static int readCount(DataInputStream in, String description) throws IOException {
+ int count = in.readInt();
+ if (count < 0) {
+ throw new IOException(
+ "Negative " + description + " count in es-index metadata: " + count);
+ }
+ return count;
+ }
+
+ private static void writeString(DataOutputStream out, String value) throws IOException {
+ byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
+
+ private static String readString(DataInputStream in) throws IOException {
+ int length = in.readInt();
+ if (length < 0 || length > in.available()) {
+ throw new EOFException("Invalid string length in es-index metadata: " + length);
+ }
+ byte[] bytes = new byte[length];
+ in.readFully(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+
+ private static void writeNullableString(DataOutputStream out, String value) throws IOException {
+ out.writeBoolean(value != null);
+ if (value != null) {
+ writeString(out, value);
+ }
+ }
+
+ private static String readNullableString(DataInputStream in) throws IOException {
+ return in.readBoolean() ? readString(in) : null;
+ }
+
+ private static void writeNullableEnum(DataOutputStream out, Enum<?> value) throws IOException {
+ writeNullableString(out, value == null ? null : value.name());
+ }
+
+ private static <E extends Enum<E>> E readNullableEnum(
+ DataInputStream in, Class<E> enumClass, String description) throws IOException {
+ String value = readNullableString(in);
+ return value == null ? null : parseEnum(enumClass, value, description);
+ }
+
+ private static <E extends Enum<E>> E parseEnum(
+ Class<E> enumClass, String value, String description) throws IOException {
+ try {
+ return Enum.valueOf(enumClass, value);
+ } catch (IllegalArgumentException e) {
+ throw new IOException("Unknown " + description + " in es-index metadata: " + value, e);
+ }
+ }
+
+ static final class Parsed {
+ private final List<String> indexedFieldNames;
+ private final List<String> indexedFieldTypes;
+ private final Map<String, FieldIndexConfig> fieldConfigs;
+ private final Map<String, long[]> fileOffsets;
+ private final boolean hasFieldConfigs;
+
+ private Parsed(
+ List<String> indexedFieldNames,
+ List<String> indexedFieldTypes,
+ Map<String, FieldIndexConfig> fieldConfigs,
+ Map<String, long[]> fileOffsets,
+ boolean hasFieldConfigs) {
+ this.indexedFieldNames = indexedFieldNames;
+ this.indexedFieldTypes = indexedFieldTypes;
+ this.fieldConfigs = fieldConfigs;
+ this.fileOffsets = fileOffsets;
+ this.hasFieldConfigs = hasFieldConfigs;
+ }
+
+ List<String> indexedFieldNames() {
+ return indexedFieldNames;
+ }
+
+ List<String> indexedFieldTypes() {
+ return indexedFieldTypes;
+ }
+
+ Map<String, FieldIndexConfig> fieldConfigs() {
+ return fieldConfigs;
+ }
+
+ Map<String, long[]> fileOffsets() {
+ return fileOffsets;
+ }
+
+ boolean hasFieldConfigs() {
+ return hasFieldConfigs;
+ }
+ }
+}
diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java
new file mode 100644
index 0000000000..f95a269eb1
--- /dev/null
+++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java
@@ -0,0 +1,1889 @@
+/*
+ * 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.data.Timestamp;
+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.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.VectorType;
+import org.apache.paimon.utils.JsonSerdeUtil;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+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.FullTextQuerySpec;
+import org.elasticsearch.eslib.api.model.IndexFilter;
+import org.elasticsearch.eslib.api.model.ScalarFieldType;
+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.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.StringJoiner;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
+
+import static org.apache.paimon.predicate.CompareUtils.compareLiteral;
+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 static final int MAX_FULL_TEXT_QUERY_DEPTH = 64;
+
+ private final GlobalIndexFileReader fileReader;
+ private final List<GlobalIndexIOMeta> files;
+ private final List<DataField> fields;
+ private final Map<String, String> logicalToPhysicalFields;
+ private final Set<String> incompatibleLogicalFields;
+ private final ESIndexOptions indexOptions;
+ private final Map<String, long[]> fileOffsets;
+ private final boolean hasPersistedFieldConfigs;
+ // Paimon's caller-owned executor schedules the outer CompletableFuture. The standalone ES940
+ // reader currently scores clusters serially and does not accept a per-reader scoring pool.
+ private final ExecutorService queryExecutor;
+
+ // Identity semantics are intentional: distinct streams may wrap the same path and must each be
+ // tracked until their own provider closes.
+ private final Set<TrackedArchiveDataProvider> openProviders =
+ Collections.newSetFromMap(new IdentityHashMap<>());
+ // Queries hold the read lock for their complete load/search operation. close() takes the write
+ // lock, so it cannot close or replace resources that an in-flight query is still using.
+ private final ReentrantReadWriteLock lifecycleLock = new ReentrantReadWriteLock();
+ // Serializes the one-time lazy load while still allowing already-loaded queries to run in
+ // parallel under lifecycleLock's read lock.
+ 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 queryExecutor) {
+ Objects.requireNonNull(files, "files");
+ checkArgument(files.size() == 1, "Expected exactly one ES index file per shard");
+ this.fileReader = Objects.requireNonNull(fileReader, "fileReader");
+ this.files = Collections.unmodifiableList(new ArrayList<>(files));
+ this.fields =
+ Collections.unmodifiableList(
+ new ArrayList<>(Objects.requireNonNull(fields, "fields")));
+ ESIndexOptions fallbackIndexOptions = Objects.requireNonNull(indexOptions, "indexOptions");
+ try {
+ GlobalIndexIOMeta archiveMeta =
+ Objects.requireNonNull(this.files.get(0), "index file metadata");
+ ESIndexFileMeta.Parsed parsed = ESIndexFileMeta.read(archiveMeta.metadata());
+ this.fileOffsets = parsed.fileOffsets();
+ validateArchiveOffsets(archiveMeta, fileOffsets);
+ this.hasPersistedFieldConfigs = parsed.hasFieldConfigs();
+ this.indexOptions =
+ parsed.hasFieldConfigs()
+ ? ESIndexOptions.fromFieldConfigs(parsed.fieldConfigs())
+ : fallbackIndexOptions;
+ List<String> physicalFields =
+ parsed.hasFieldConfigs()
+ ? parsed.indexedFieldNames()
+ : currentFieldNames(this.fields);
+ this.logicalToPhysicalFields =
+ createFieldMapping(this.fields, physicalFields, this.indexOptions);
+ this.incompatibleLogicalFields =
+ findIncompatibleFields(
+ this.fields,
+ physicalFields,
+ parsed.indexedFieldTypes(),
+ this.indexOptions,
+ parsed.hasFieldConfigs());
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Invalid es-index metadata", e);
+ }
+ this.queryExecutor = queryExecutor;
+ this.loaded = false;
+ this.closed = false;
+ }
+
+ private static List<String> currentFieldNames(List<DataField> fields) throws IOException {
+ List<String> names = new ArrayList<>(fields.size());
+ for (DataField field : fields) {
+ if (field == null) {
+ throw new IOException("Null current field in es-index reader");
+ }
+ names.add(field.name());
+ }
+ return names;
+ }
+
+ private static Map<String, String> createFieldMapping(
+ List<DataField> currentFields, List<String> physicalFields, ESIndexOptions indexOptions)
+ throws IOException {
+ if (physicalFields.size() != currentFields.size()) {
+ throw new IOException(
+ "ES index field count does not match current schema: indexed="
+ + physicalFields.size()
+ + ", current="
+ + currentFields.size());
+ }
+ Map<String, String> mapping = new HashMap<>();
+ Set<String> physicalNames = new HashSet<>();
+ for (int i = 0; i < currentFields.size(); i++) {
+ DataField current = currentFields.get(i);
+ if (current == null) {
+ throw new IOException("Null current field in es-index reader at position " + i);
+ }
+ String physical = physicalFields.get(i);
+ if (physical == null || physical.isEmpty()) {
+ throw new IOException(
+ "Empty indexed field name in es-index metadata at position " + i);
+ }
+ if (indexOptions.getConfig(physical) == null) {
+ throw new IOException(
+ "Missing primary field config in es-index metadata: " + physical);
+ }
+ if (mapping.put(current.name(), physical) != null) {
+ throw new IOException(
+ "Duplicate current field in es-index reader: " + current.name());
+ }
+ if (!physicalNames.add(physical)) {
+ throw new IOException("Duplicate indexed field in es-index metadata: " + physical);
+ }
+ }
+ return Collections.unmodifiableMap(mapping);
+ }
+
+ private String physicalField(String logicalField) {
+ return logicalToPhysicalFields.get(logicalField);
+ }
+
+ private FieldIndexConfig logicalFieldConfig(String logicalField) {
+ if (incompatibleLogicalFields.contains(logicalField)) {
+ return null;
+ }
+ String physicalField = physicalField(logicalField);
+ return physicalField == null ? null : indexOptions.getConfig(physicalField);
+ }
+
+ private static Set<String> findIncompatibleFields(
+ List<DataField> currentFields,
+ List<String> physicalFields,
+ List<String> indexedFieldTypes,
+ ESIndexOptions indexOptions,
+ boolean hasPersistedFieldConfigs)
+ throws IOException {
+ if (!indexedFieldTypes.isEmpty() && indexedFieldTypes.size() != currentFields.size()) {
+ throw new IOException(
+ "ES index field type count does not match current schema: indexed="
+ + indexedFieldTypes.size()
+ + ", current="
+ + currentFields.size());
+ }
+ if (!hasPersistedFieldConfigs) {
+ return Collections.emptySet();
+ }
+
+ Set<String> incompatible = new HashSet<>();
+ for (int i = 0; i < currentFields.size(); i++) {
+ DataField current = currentFields.get(i);
+ boolean compatible;
+ if (!indexedFieldTypes.isEmpty()) {
+ compatible =
+ current.type().copy(true).asSQLString().equals(indexedFieldTypes.get(i));
+ } else {
+ compatible =
+ legacyTypeCompatible(
+ current.type(), indexOptions.getConfig(physicalFields.get(i)));
+ }
+ if (!compatible) {
+ incompatible.add(current.name());
+ }
+ }
+ return incompatible.isEmpty()
+ ? Collections.emptySet()
+ : Collections.unmodifiableSet(incompatible);
+ }
+
+ private static boolean legacyTypeCompatible(DataType type, FieldIndexConfig config) {
+ if (config == null) {
+ return false;
+ }
+ switch (config.indexType()) {
+ case VECTOR:
+ if (type instanceof VectorType) {
+ VectorType vectorType = (VectorType) type;
+ return vectorType.getElementType().getTypeRoot() == DataTypeRoot.FLOAT
+ && vectorType.getLength() == config.dimension();
+ }
+ return type instanceof ArrayType
+ && ((ArrayType) type).getElementType().getTypeRoot() == DataTypeRoot.FLOAT;
+ case FULLTEXT:
+ case KEYWORD:
+ return type.getTypeRoot() == DataTypeRoot.CHAR
+ || type.getTypeRoot() == DataTypeRoot.VARCHAR;
+ case SCALAR:
+ case DATE:
+ return expectedScalarType(type) == config.scalarType();
+ default:
+ return false;
+ }
+ }
+
+ private static ScalarFieldType expectedScalarType(DataType type) {
+ DataType valueType = type instanceof ArrayType ? ((ArrayType) type).getElementType() : type;
+ switch (valueType.getTypeRoot()) {
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ return ScalarFieldType.INT;
+ case BIGINT:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return ScalarFieldType.LONG;
+ case FLOAT:
+ return ScalarFieldType.FLOAT;
+ case DOUBLE:
+ return ScalarFieldType.DOUBLE;
+ case CHAR:
+ case VARCHAR:
+ return ScalarFieldType.KEYWORD;
+ default:
+ return null;
+ }
+ }
+
+ private static void validateArchiveOffsets(
+ GlobalIndexIOMeta archiveMeta, Map<String, long[]> fileOffsets) throws IOException {
+ if (archiveMeta.fileSize() < 0) {
+ throw new IOException("Negative es-index archive size: " + archiveMeta.fileSize());
+ }
+ if (fileOffsets.isEmpty()) {
+ throw new IOException("ES index metadata contains no Lucene files");
+ }
+ for (Map.Entry<String, long[]> entry : fileOffsets.entrySet()) {
+ long[] range = entry.getValue();
+ if (range == null
+ || range.length != 2
+ || range[0] < 0
+ || range[1] < 0
+ || range[0] > archiveMeta.fileSize() - range[1]) {
+ throw new IOException(
+ "File range exceeds es-index archive size for "
+ + entry.getKey()
+ + ": archiveSize="
+ + archiveMeta.fileSize()
+ + ", range="
+ + sample(range, range == null ? 0 : range.length, 2));
+ }
+ }
+ }
+
+ ExecutorService queryExecutor() {
+ return queryExecutor;
+ }
+
+ int openStreamCount() {
+ synchronized (openProviders) {
+ return openProviders.size();
+ }
+ }
+
+ @Override
+ public CompletableFuture<Optional<ScoredGlobalIndexResult>> visitVectorSearch(
+ VectorSearch vectorSearch) {
+ Objects.requireNonNull(vectorSearch, "vectorSearch");
+ return async(
+ () -> {
+ try {
+ // Extract params before vector-search: query vector, field, result count
+ // (topK), candidate
+ float[] queryVector = vectorSearch.vector();
+ int topK = vectorSearch.limit();
+ if (topK <= 0) {
+ throw new IllegalArgumentException(
+ "Vector search limit must be positive; got: " + topK);
+ }
+ int searchTopK = vectorSearchTopK(vectorSearch);
+ String fieldName = vectorSearch.fieldName();
+ String physicalField = physicalField(fieldName);
+ if (physicalField == null) {
+ return Optional.empty();
+ }
+ FieldIndexConfig config = indexOptions.getConfig(physicalField);
+ if (config == null
+ || config.indexType() != FieldIndexConfig.IndexType.VECTOR) {
+ if (incompatibleLogicalFields.contains(fieldName)) {
+ throw incompatibleSearchSchema(fieldName, "vector");
+ }
+ return Optional.empty();
+ }
+ if (incompatibleLogicalFields.contains(fieldName)
+ && !vectorTypeCompatible(fieldName, config)) {
+ throw incompatibleSearchSchema(fieldName, "vector");
+ }
+ if (queryVector == null || queryVector.length != config.dimension()) {
+ throw new IllegalArgumentException(
+ "Vector query for field '"
+ + fieldName
+ + "' expects dimension "
+ + config.dimension()
+ + " but received "
+ + (queryVector == null ? "null" : queryVector.length)
+ + ".");
+ }
+ ensureLoaded();
+
+ RoaringNavigableMap64 includeRowIds = vectorSearch.includeRowIds();
+ long[] candidateIds = includeRowIds == null ? null : 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(
+ physicalField, 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, topK);
+ } catch (IOException e) {
+ throw new RuntimeException("Vector search failed", e);
+ }
+ });
+ }
+
+ @Override
+ public CompletableFuture<Optional<ScoredGlobalIndexResult>> visitFullTextSearch(
+ FullTextSearch fullTextSearch) {
+ Objects.requireNonNull(fullTextSearch, "fullTextSearch");
+ return async(
+ () -> {
+ try {
+ if (fullTextSearch.limit() <= 0) {
+ throw new IllegalArgumentException(
+ "Full-text search limit must be positive; got: "
+ + fullTextSearch.limit());
+ }
+ String physicalField = physicalField(fullTextSearch.fieldName());
+ String searchField =
+ physicalField == null
+ ? null
+ : indexOptions.fullTextSearchField(physicalField);
+ if (physicalField != null
+ && incompatibleLogicalFields.contains(fullTextSearch.fieldName())
+ && (searchField == null
+ || !isCurrentTextField(fullTextSearch.fieldName()))) {
+ throw incompatibleSearchSchema(fullTextSearch.fieldName(), "full-text");
+ }
+ if (searchField == null) {
+ return Optional.empty();
+ }
+ // Parse the JSON DSL query into an eslib FullTextQuerySpec and run it. A
+ // FULLTEXT primary field is searched directly; a KEYWORD primary field is
+ // searched through its analyzed .fulltext multi-field.
+ FullTextQuerySpec spec = parseSpec(searchField, fullTextSearch.query());
+ ensureLoaded();
+ SearchResult result =
+ fullTextSearch(
+ spec,
+ fullTextSearch.limit(),
+ fullTextSearch.includeRowIds());
+ return toScoredResult(result, fullTextSearch.limit());
+ } catch (IOException e) {
+ throw new RuntimeException("Full-text search failed", e);
+ }
+ });
+ }
+
+ private boolean vectorTypeCompatible(String logicalField, FieldIndexConfig config) {
+ DataType currentType = currentFieldType(logicalField);
+ if (currentType instanceof VectorType) {
+ VectorType vectorType = (VectorType) currentType;
+ return vectorType.getElementType().getTypeRoot() == DataTypeRoot.FLOAT
+ && vectorType.getLength() == config.dimension();
+ }
+ return currentType instanceof ArrayType
+ && ((ArrayType) currentType).getElementType().getTypeRoot() == DataTypeRoot.FLOAT;
+ }
+
+ private boolean isCurrentTextField(String logicalField) {
+ DataType currentType = currentFieldType(logicalField);
+ return currentType != null
+ && (currentType.getTypeRoot() == DataTypeRoot.CHAR
+ || currentType.getTypeRoot() == DataTypeRoot.VARCHAR);
+ }
+
+ private DataType currentFieldType(String logicalField) {
+ for (DataField field : fields) {
+ if (field.name().equals(logicalField)) {
+ return field.type();
+ }
+ }
+ return null;
+ }
+
+ private static IllegalStateException incompatibleSearchSchema(
+ String logicalField, String searchType) {
+ return new IllegalStateException(
+ "The persisted es-index for field '"
+ + logicalField
+ + "' is incompatible with the current schema and cannot safely answer "
+ + searchType
+ + " search. Rebuild the es-index after the schema change.");
+ }
+
+ /**
+ * Applies {@link FullTextSearch#includeRowIds()} without changing top-K semantics. ESLib does
+ * not expose a full-text candidate-filter argument, so search progressively over-fetches until
+ * it has the requested number of included hits or has exhausted all matching documents. The
+ * table path normally supplies the whole shard row count as {@code limit}, making this a single
+ * search; the loop also preserves the direct reader API for smaller limits.
+ */
+ private SearchResult fullTextSearch(
+ FullTextQuerySpec spec, int limit, RoaringNavigableMap64 includeRowIds)
+ throws IOException {
+ if (includeRowIds == null) {
+ return searcher.fullTextSearch(spec, limit);
+ }
+ if (includeRowIds.isEmpty()) {
+ return emptySearchResult();
+ }
+
+ int searchLimit = limit;
+ while (true) {
+ SearchResult unfiltered = searcher.fullTextSearch(spec, searchLimit);
+ SearchResult filtered = filterSearchResult(unfiltered, includeRowIds, limit);
+ if (filtered.count >= limit
+ || filtered.count >= includeRowIds.getLongCardinality()
+ || unfiltered == null
+ || unfiltered.count < searchLimit
+ || searchLimit == Integer.MAX_VALUE) {
+ return filtered;
+ }
+ searchLimit =
+ searchLimit >= Integer.MAX_VALUE / 2 ? Integer.MAX_VALUE : searchLimit * 2;
+ }
+ }
+
+ private static SearchResult filterSearchResult(
+ SearchResult result, RoaringNavigableMap64 includeRowIds, int limit) {
+ if (result == null || result.count == 0) {
+ return emptySearchResult();
+ }
+ int capacity = Math.min(limit, result.count);
+ long[] ids = new long[capacity];
+ float[] scores = new float[capacity];
+ int count = 0;
+ for (int i = 0; i < result.count && count < limit; i++) {
+ if (includeRowIds.contains(result.ids[i])) {
+ ids[count] = result.ids[i];
+ scores[count] = result.scores[i];
+ count++;
+ }
+ }
+ if (count == 0) {
+ return emptySearchResult();
+ }
+ return new SearchResult(Arrays.copyOf(ids, count), Arrays.copyOf(scores, count), count);
+ }
+
+ private static SearchResult emptySearchResult() {
+ return new SearchResult(new long[0], new float[0], 0);
+ }
+
+ /**
+ * Parses the {@link FullTextSearch#query()} JSON DSL into an eslib {@link FullTextQuerySpec}.
+ */
+ static FullTextQuerySpec parseSpec(String field, String json) {
+ try {
+ return parseSpec(field, JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(json), 0);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Invalid full-text query JSON: " + json, e);
+ }
+ }
+
+ /**
+ * Maps the JSON query DSL ({@code {"match":...}}, {@code {"match_phrase":...}}, {@code
+ * {"boolean":...}}, {@code {"boost":...}}) onto an eslib spec. The field comes from {@link
+ * FullTextSearch#fieldName()} (the flat model carries a single field); the query semantics come
+ * from the JSON. The caller resolves the logical Paimon field to its physical FULLTEXT field.
+ */
+ private static FullTextQuerySpec parseSpec(String field, JsonNode q, int depth) {
+ if (depth >= MAX_FULL_TEXT_QUERY_DEPTH) {
+ throw new IllegalArgumentException(
+ "Full-text query exceeds the maximum nesting depth of "
+ + MAX_FULL_TEXT_QUERY_DEPTH);
+ }
+ requireObject(q, "full-text query");
+ requireOnlyFields(
+ q,
+ "full-text query",
+ "match",
+ "match_phrase",
+ "phrase",
+ "boost",
+ "boolean",
+ "bool");
+ int queryTypeCount =
+ present(q, "match")
+ + present(q, "match_phrase")
+ + present(q, "phrase")
+ + present(q, "boost")
+ + present(q, "boolean")
+ + present(q, "bool");
+ if (queryTypeCount != 1) {
+ throw new IllegalArgumentException(
+ "Full-text query must contain exactly one supported query type: " + q);
+ }
+ if (q.has("match")) {
+ JsonNode m = requireObject(q.get("match"), "match query");
+ requireOnlyFields(
+ m,
+ "match query",
+ "query",
+ "terms",
+ "operator",
+ "boost",
+ "fuzziness",
+ "max_expansions",
+ "maxExpansions",
+ "prefix_length",
+ "prefixLength");
+ return new FullTextQuerySpec.Match(field, queryText(m), parseParams(m));
+ } else if (q.has("match_phrase") || q.has("phrase")) {
+ JsonNode p =
+ requireObject(
+ q.has("match_phrase") ? q.get("match_phrase") : q.get("phrase"),
+ "phrase query");
+ requireOnlyFields(p, "phrase query", "query", "terms", "slop");
+ int slop = p.has("slop") ? intValue(p.get("slop"), "slop") : 0;
+ return new FullTextQuerySpec.Phrase(field, queryText(p), slop);
+ } else if (q.has("boost")) {
+ JsonNode b = requireObject(q.get("boost"), "boost query");
+ requireOnlyFields(
+ b, "boost query", "positive", "negative", "negative_boost", "negativeBoost");
+ rejectAliasPair(b, "negative_boost", "negativeBoost");
+ float negativeBoost = 0.5f;
+ if (b.has("negative_boost")) {
+ negativeBoost = floatValue(b.get("negative_boost"), "negative_boost");
+ } else if (b.has("negativeBoost")) {
+ negativeBoost = floatValue(b.get("negativeBoost"), "negativeBoost");
+ }
+ return new FullTextQuerySpec.Boost(
+ parseSpec(field, required(b, "positive"), depth + 1),
+ parseSpec(field, required(b, "negative"), depth + 1),
+ negativeBoost);
+ } else if (q.has("boolean") || q.has("bool")) {
+ List<FullTextQuerySpec> must = new ArrayList<>();
+ List<FullTextQuerySpec> should = new ArrayList<>();
+ List<FullTextQuerySpec> mustNot = new ArrayList<>();
+ JsonNode booleanNode =
+ requireObject(
+ q.has("boolean") ? q.get("boolean") : q.get("bool"), "boolean query");
+ requireOnlyFields(
+ booleanNode,
+ "boolean query",
+ "must",
+ "should",
+ "must_not",
+ "mustNot",
+ "queries");
+ rejectAliasPair(booleanNode, "must_not", "mustNot");
+ addBooleanClauses(field, booleanNode.get("must"), must, depth + 1);
+ addBooleanClauses(field, booleanNode.get("should"), should, depth + 1);
+ JsonNode namedMustNot =
+ booleanNode.has("must_not")
+ ? booleanNode.get("must_not")
+ : booleanNode.get("mustNot");
+ addBooleanClauses(field, namedMustNot, mustNot, depth + 1);
+
+ // Keep the original tuple/object representation for compatibility.
+ JsonNode clauses = booleanNode.get("queries");
+ if (clauses != null && !clauses.isNull()) {
+ if (!clauses.isArray()) {
+ throw new IllegalArgumentException(
+ "Boolean 'queries' must be an array: " + clauses);
+ }
+ for (JsonNode clause : clauses) {
+ String occur;
+ JsonNode child;
+ if (clause.isArray()) {
+ if (clause.size() != 2 || !clause.get(0).isTextual()) {
+ throw new IllegalArgumentException(
+ "Boolean tuple clause must be [occur, query]: " + clause);
+ }
+ occur = clause.get(0).textValue();
+ child = clause.get(1);
+ } else {
+ JsonNode clauseObject = requireObject(clause, "boolean clause");
+ requireOnlyFields(clauseObject, "boolean clause", "occur", "query");
+ JsonNode occurNode = clauseObject.get("occur");
+ if (occurNode != null && !occurNode.isTextual()) {
+ throw new IllegalArgumentException(
+ "Boolean clause 'occur' must be a string: " + occurNode);
+ }
+ occur = occurNode == null ? "Should" : occurNode.textValue();
+ child = required(clauseObject, "query");
+ }
+ FullTextQuerySpec spec = parseSpec(field, child, depth + 1);
+ String normalizedOccur =
+ occur.replace("_", "").replace("-", "").toLowerCase(Locale.ROOT);
+ if ("must".equals(normalizedOccur)) {
+ must.add(spec);
+ } else if ("mustnot".equals(normalizedOccur)) {
+ mustNot.add(spec);
+ } else if ("should".equals(normalizedOccur)) {
+ should.add(spec);
+ } else {
+ throw new IllegalArgumentException(
+ "Boolean occur must be MUST, SHOULD, or MUST_NOT; got: " + occur);
+ }
+ }
+ }
+ return new FullTextQuerySpec.Bool(must, should, mustNot);
+ }
+ throw new UnsupportedOperationException("Unsupported full-text query JSON: " + q);
+ }
+
+ private static void addBooleanClauses(
+ String field, JsonNode node, List<FullTextQuerySpec> destination, int depth) {
+ if (node == null || node.isNull()) {
+ return;
+ }
+ if (node.isArray()) {
+ for (JsonNode child : node) {
+ destination.add(parseSpec(field, child, depth));
+ }
+ } else {
+ destination.add(parseSpec(field, node, depth));
+ }
+ }
+
+ private static String queryText(JsonNode node) {
+ if (node.has("query") && node.has("terms")) {
+ throw new IllegalArgumentException(
+ "Full-text query cannot contain both 'query' and 'terms': " + node);
+ }
+ JsonNode text = node.has("query") ? node.get("query") : node.get("terms");
+ if (text == null || !text.isTextual()) {
+ throw new IllegalArgumentException(
+ "Full-text query requires a string 'query' or 'terms': " + node);
+ }
+ return text.textValue();
+ }
+
+ /**
+ * Reads the ES-style match params ({@code operator} / {@code boost} / {@code fuzziness} /
+ * {@code max_expansions} / {@code prefix_length}) from a match node into {@link
+ * FullTextParams}; the eslib searcher applies them to the Lucene query. Absent params keep
+ * their defaults.
+ */
+ private static FullTextParams parseParams(JsonNode match) {
+ FullTextParams.Operator operator = FullTextParams.Operator.OR;
+ if (match.has("operator")) {
+ String value = match.get("operator").asText();
+ if ("and".equalsIgnoreCase(value)) {
+ operator = FullTextParams.Operator.AND;
+ } else if (!"or".equalsIgnoreCase(value)) {
+ throw new IllegalArgumentException(
+ "Full-text match operator must be AND or OR; got: " + value);
+ }
+ }
+ float boost = match.has("boost") ? floatValue(match.get("boost"), "boost") : 1.0f;
+ int fuzziness = 0;
+ if (match.has("fuzziness")) {
+ JsonNode value = match.get("fuzziness");
+ if (value.isTextual() && "auto".equalsIgnoreCase(value.asText())) {
+ fuzziness = FullTextParams.AUTO_FUZZINESS;
+ } else if (value.isTextual()) {
+ try {
+ fuzziness = Integer.parseInt(value.asText());
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "fuzziness must be 0, 1, 2, or AUTO; got: " + value.asText(), e);
+ }
+ } else {
+ fuzziness = intValue(value, "fuzziness");
+ }
+ }
+ int maxExpansions = intParam(match, "max_expansions", "maxExpansions", 50);
+ int prefixLength = intParam(match, "prefix_length", "prefixLength", 0);
+ return new FullTextParams(operator, boost, fuzziness, maxExpansions, prefixLength);
+ }
+
+ private static int intParam(
+ JsonNode node, String snakeCaseName, String camelCaseName, int defaultValue) {
+ rejectAliasPair(node, snakeCaseName, camelCaseName);
+ if (node.has(snakeCaseName)) {
+ return intValue(node.get(snakeCaseName), snakeCaseName);
+ }
+ return node.has(camelCaseName)
+ ? intValue(node.get(camelCaseName), camelCaseName)
+ : defaultValue;
+ }
+
+ private static int present(JsonNode node, String field) {
+ return node.has(field) ? 1 : 0;
+ }
+
+ private static JsonNode requireObject(JsonNode node, String description) {
+ if (node == null || !node.isObject()) {
+ throw new IllegalArgumentException(description + " must be a JSON object: " + node);
+ }
+ return node;
+ }
+
+ private static void requireOnlyFields(
+ JsonNode node, String description, String... allowedFields) {
+ Set<String> allowed = new HashSet<>(Arrays.asList(allowedFields));
+ Iterator<String> fields = node.fieldNames();
+ while (fields.hasNext()) {
+ String field = fields.next();
+ if (!allowed.contains(field)) {
+ throw new IllegalArgumentException(
+ "Unknown field '" + field + "' in " + description + ": " + node);
+ }
+ }
+ }
+
+ private static void rejectAliasPair(JsonNode node, String first, String second) {
+ if (node.has(first) && node.has(second)) {
+ throw new IllegalArgumentException(
+ "Full-text query cannot contain both '"
+ + first
+ + "' and '"
+ + second
+ + "': "
+ + node);
+ }
+ }
+
+ private static JsonNode required(JsonNode node, String field) {
+ JsonNode value = node.get(field);
+ if (value == null || value.isNull()) {
+ throw new IllegalArgumentException("Missing full-text query field: " + field);
+ }
+ return value;
+ }
+
+ private static int intValue(JsonNode value, String name) {
+ if (value != null && value.isIntegralNumber() && value.canConvertToInt()) {
+ return value.intValue();
+ }
+ if (value != null && value.isTextual()) {
+ try {
+ return Integer.parseInt(value.textValue());
+ } catch (NumberFormatException ignored) {
+ // Report the same parameter-specific error below.
+ }
+ }
+ throw new IllegalArgumentException(name + " must be an integer; got: " + value);
+ }
+
+ private static float floatValue(JsonNode value, String name) {
+ if (value == null || (!value.isNumber() && !value.isTextual())) {
+ throw new IllegalArgumentException(name + " must be a number; got: " + value);
+ }
+ try {
+ return Float.parseFloat(value.asText());
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(name + " must be a number; got: " + value, e);
+ }
+ }
+
+ /**
+ * Run {@code body} on Paimon's caller-owned query executor when one is available, otherwise
+ * execute it inline.
+ */
+ private <T> CompletableFuture<T> async(Supplier<T> body) {
+ Supplier<T> guardedBody =
+ () -> {
+ Lock readLock = lifecycleLock.readLock();
+ readLock.lock();
+ try {
+ return body.get();
+ } finally {
+ readLock.unlock();
+ }
+ };
+ if (queryExecutor != null) {
+ try {
+ return CompletableFuture.supplyAsync(guardedBody, queryExecutor);
+ } catch (Throwable failure) {
+ CompletableFuture<T> future = new CompletableFuture<>();
+ future.completeExceptionally(failure);
+ return future;
+ }
+ }
+ CompletableFuture<T> future = new CompletableFuture<>();
+ try {
+ future.complete(guardedBody.get());
+ } catch (Throwable failure) {
+ future.completeExceptionally(failure);
+ }
+ return future;
+ }
+
+ static int vectorSearchTopK(VectorSearch vectorSearch) {
+ int limit = vectorSearch.limit();
+ Map<String, String> options = vectorSearch.options();
+ int candidates = positiveIntOption(options, HNSW_NUM_CANDIDATES_OPTION);
+ if (candidates <= 0) {
+ candidates = positiveIntOption(options, NUM_CANDIDATES_OPTION);
+ }
+ if (candidates <= 0) {
+ candidates = positiveIntOption(options, HNSW_EF_SEARCH_OPTION);
+ }
+ return candidates <= 0 ? limit : Math.max(limit, candidates);
+ }
+
+ private static int positiveIntOption(Map<String, String> options, String key) {
+ String value = options.get(key);
+ if (value == null || value.trim().isEmpty()) {
+ return -1;
+ }
+ try {
+ int parsed = Integer.parseInt(value.trim());
+ if (parsed <= 0) {
+ throw new IllegalArgumentException(
+ "Vector search option '" + key + "' must be positive; got: " + value);
+ }
+ return parsed;
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Vector search option '" + key + "' must be an integer; got: " + value, e);
+ }
+ }
+
+ private Optional<ScoredGlobalIndexResult> toScoredResult(SearchResult result, int limit) {
+ if (result == null || result.count == 0) {
+ return Optional.empty();
+ }
+
+ int count = Math.min(result.count, limit);
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ Map<Long, Float> scoreMap = new HashMap<>(count);
+ for (int i = 0; i < count; i++) {
+ long id = result.ids[i];
+ bitmap.add(id);
+ scoreMap.put(id, result.scores[i]);
+ }
+
+ return Optional.of(ScoredGlobalIndexResult.create(bitmap, scoreMap::get));
+ }
+
+ private void checkNotClosed() throws IOException {
+ if (closed) {
+ throw new IOException("Reader already closed");
+ }
+ }
+
+ private void ensureLoaded() throws IOException {
+ checkNotClosed();
+ if (loaded) {
+ return;
+ }
+ synchronized (loadLock) {
+ // close() may have won the lifecycle write lock while this operation was waiting to
+ // start. Recheck under the load lock before opening any stream or creating a searcher.
+ checkNotClosed();
+ if (loaded) {
+ return;
+ }
+
+ if (files.isEmpty()) {
+ throw new IOException("No index files to load");
+ }
+
+ long loadStartNanos = System.nanoTime();
+ GlobalIndexIOMeta meta = files.get(0);
+ byte[] metaBytes = meta.metadata();
+
+ if (debugEnabled()) {
+ LOG.info(
+ "PAIMON_ESLIB_READER_LOAD files={} firstFile={} firstFileSize={} metadataBytes={} offsets={} offsetSample={} fieldConfigs={} fields={}",
+ files.size(),
+ meta.filePath(),
+ meta.fileSize(),
+ metaBytes == null ? 0 : metaBytes.length,
+ fileOffsets.size(),
+ offsetsSample(fileOffsets),
+ fieldConfigsSample(),
+ fieldsSample());
+ }
+
+ // Offset/length + fork() reader over the Paimon stream of the global-index archive
+ // (*.index)
+ // dataProvider bridge Lucene's per-file reads into offset/length reads over the packed
+ // archive
+ // ESLib's ArchiveIndexInput pools provider forks across clones so concurrent reads
+ // remain independent without opening one stream for every query-scoped clone.
+ ArchiveDataProvider dataProvider = null;
+ ESIndexSearcher candidateSearcher = null;
+ try {
+ // Mount the packed archive as a Lucene index, then open the reader/searcher. Keep
+ // it local until load succeeds so a retry cannot overwrite a partially loaded
+ // searcher and leak its directory/providers.
+ dataProvider = createProvider(meta);
+ candidateSearcher = ESIndexBuilderFactory.createSearcher();
+ candidateSearcher.load(dataProvider, fileOffsets, indexOptions.getFieldConfigs());
+ } catch (IOException | RuntimeException | Error failure) {
+ Throwable cleanupFailure = cleanupFailedLoad(candidateSearcher, dataProvider);
+ if (cleanupFailure != null) {
+ addSuppressed(failure, cleanupFailure);
+ }
+ throw failure;
+ }
+ searcher = candidateSearcher;
+ loaded = true;
+ if (debugEnabled()) {
+ LOG.info(
+ "PAIMON_ESLIB_READER_LOADED files={} firstFile={} elapsedMs={} searcher={}",
+ files.size(),
+ meta.filePath(),
+ (System.nanoTime() - loadStartNanos) / 1_000_000L,
+ searcher == null ? "null" : searcher.getClass().getName());
+ }
+ }
+ }
+
+ private ArchiveDataProvider createProvider(GlobalIndexIOMeta meta) throws IOException {
+ SeekableInputStream inputStream = fileReader.getInputStream(meta);
+ if (inputStream == null) {
+ throw new IOException(
+ "Global index file reader returned a null stream for " + meta.filePath());
+ }
+ TrackedArchiveDataProvider provider = new TrackedArchiveDataProvider(meta, inputStream);
+ synchronized (openProviders) {
+ openProviders.add(provider);
+ }
+ return provider;
+ }
+
+ private final class TrackedArchiveDataProvider implements ArchiveDataProvider {
+ private final GlobalIndexIOMeta meta;
+ private final SeekableInputStream inputStream;
+ private final AtomicBoolean providerClosed = new AtomicBoolean();
+
+ private TrackedArchiveDataProvider(
+ GlobalIndexIOMeta meta, SeekableInputStream inputStream) {
+ this.meta = meta;
+ this.inputStream = inputStream;
+ }
+
+ @Override
+ public byte[] readRange(long offset, int length) throws IOException {
+ byte[] buf = new byte[length];
+ inputStream.seek(offset);
+ int read = 0;
+ while (read < length) {
+ int n = inputStream.read(buf, read, length - read);
+ if (n < 0) {
+ throw new IOException("Unexpected EOF at offset " + (offset + read));
+ }
+ if (n == 0) {
+ int single = inputStream.read();
+ if (single < 0) {
+ throw new IOException("Unexpected EOF at offset " + (offset + read));
+ }
+ buf[read++] = (byte) single;
+ } else {
+ read += n;
+ }
+ }
+ return buf;
+ }
+
+ @Override
+ public ArchiveDataProvider fork() throws IOException {
+ return createProvider(meta);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!providerClosed.compareAndSet(false, true)) {
+ return;
+ }
+ boolean closeSucceeded = false;
+ try {
+ inputStream.close();
+ closeSucceeded = true;
+ } finally {
+ // Retain a provider whose close failed so reader close / failed-load cleanup can
+ // make one final best-effort attempt.
+ if (closeSucceeded) {
+ synchronized (openProviders) {
+ openProviders.remove(this);
+ }
+ } else {
+ // Allow failed-load cleanup or the reader's final close to retry.
+ providerClosed.set(false);
+ }
+ }
+ }
+ }
+
+ private Throwable cleanupFailedLoad(
+ ESIndexSearcher candidateSearcher, ArchiveDataProvider dataProvider) {
+ Throwable failure = null;
+ if (candidateSearcher != null) {
+ try {
+ candidateSearcher.close();
+ } catch (Throwable e) {
+ failure = e;
+ }
+ }
+ if (dataProvider != null) {
+ try {
+ dataProvider.close();
+ } catch (Throwable e) {
+ failure = mergeFailure(failure, e);
+ }
+ }
+ return mergeFailure(failure, closeTrackedProviders());
+ }
+
+ private Throwable closeTrackedProviders() {
+ List<TrackedArchiveDataProvider> providers;
+ synchronized (openProviders) {
+ providers = new ArrayList<>(openProviders);
+ }
+
+ Throwable failure = null;
+ for (TrackedArchiveDataProvider provider : providers) {
+ try {
+ provider.close();
+ } catch (Throwable e) {
+ failure = mergeFailure(failure, e);
+ }
+ }
+ return failure;
+ }
+
+ private static Throwable mergeFailure(Throwable failure, Throwable next) {
+ if (next == null) {
+ return failure;
+ }
+ if (failure == null) {
+ return next;
+ }
+ addSuppressed(failure, next);
+ return failure;
+ }
+
+ private static void addSuppressed(Throwable failure, Throwable suppressed) {
+ if (failure != suppressed) {
+ failure.addSuppressed(suppressed);
+ }
+ }
+
+ private static long[] toArray(RoaringNavigableMap64 bitmap) {
+ long cardinality = bitmap.getLongCardinality();
+ if (cardinality > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException(
+ "Candidate row-id bitmap is too large to materialize: " + cardinality);
+ }
+ long[] arr = new long[(int) cardinality];
+ int i = 0;
+ for (long id : bitmap) {
+ arr[i++] = id;
+ }
+ return arr;
+ }
+
+ private static boolean debugEnabled() {
+ return Boolean.parseBoolean(System.getProperties().getProperty(DEBUG_PROPERTY, "false"));
+ }
+
+ static String sample(long[] values, int limit) {
+ return values == null ? "[]" : sample(values, values.length, limit);
+ }
+
+ static String sample(long[] values, int count, int limit) {
+ if (values == null || count <= 0 || limit <= 0) {
+ return "[]";
+ }
+ int actual = Math.min(Math.min(values.length, count), limit);
+ StringJoiner joiner = new StringJoiner(", ", "[", "]");
+ for (int i = 0; i < actual; i++) {
+ joiner.add(String.valueOf(values[i]));
+ }
+ if (count > actual) {
+ joiner.add("...(+" + (count - actual) + ")");
+ }
+ return joiner.toString();
+ }
+
+ static String sample(float[] values, int limit) {
+ return values == null ? "[]" : sample(values, values.length, limit);
+ }
+
+ static String sample(float[] values, int count, int limit) {
+ if (values == null || count <= 0 || limit <= 0) {
+ return "[]";
+ }
+ int actual = Math.min(Math.min(values.length, count), limit);
+ StringJoiner joiner = new StringJoiner(", ", "[", "]");
+ for (int i = 0; i < actual; i++) {
+ joiner.add(Float.toString(values[i]));
+ }
+ if (count > actual) {
+ joiner.add("...(+" + (count - actual) + ")");
+ }
+ return joiner.toString();
+ }
+
+ private String fieldsSample() {
+ StringJoiner joiner = new StringJoiner(", ", "[", "]");
+ for (DataField field : fields) {
+ joiner.add(field.name() + "#" + field.id());
+ }
+ return joiner.toString();
+ }
+
+ private String fieldConfigsSample() {
+ Map<String, FieldIndexConfig> configs = indexOptions.getFieldConfigs();
+ if (configs == null || configs.isEmpty()) {
+ return "[]";
+ }
+ StringJoiner joiner = new StringJoiner(", ", "[", "]");
+ for (Map.Entry<String, FieldIndexConfig> entry : configs.entrySet()) {
+ FieldIndexConfig config = entry.getValue();
+ joiner.add(
+ entry.getKey()
+ + "="
+ + config.indexType()
+ + "/"
+ + config.algorithm()
+ + "/"
+ + config.algorithmParams());
+ }
+ return joiner.toString();
+ }
+
+ private static String offsetsSample(Map<String, long[]> fileOffsets) {
+ if (fileOffsets == null || fileOffsets.isEmpty()) {
+ return "[]";
+ }
+ StringJoiner joiner = new StringJoiner(", ", "[", "]");
+ int seen = 0;
+ for (Map.Entry<String, long[]> entry : fileOffsets.entrySet()) {
+ if (seen++ >= DEFAULT_SAMPLE_LIMIT) {
+ joiner.add("...(+" + (fileOffsets.size() - DEFAULT_SAMPLE_LIMIT) + ")");
+ break;
+ }
+ long[] range = entry.getValue();
+ joiner.add(entry.getKey() + "=" + sample(range, range == null ? 0 : range.length, 2));
+ }
+ return joiner.toString();
+ }
+
+ @Override
+ public void close() throws IOException {
+ Lock writeLock = lifecycleLock.writeLock();
+ writeLock.lock();
+ try {
+ // Prevent new queries on the first attempt, but keep cleanup retryable. A seekable
+ // stream may fail to close transiently; TrackedArchiveDataProvider then remains in
+ // openProviders with its close guard reset so a later close() can retry it.
+ closed = true;
+
+ Throwable failure = null;
+ ESIndexSearcher currentSearcher = searcher;
+ if (currentSearcher != null) {
+ try {
+ currentSearcher.close();
+ searcher = null;
+ } catch (Throwable e) {
+ failure = e;
+ }
+ }
+
+ failure = mergeFailure(failure, closeTrackedProviders());
+ loaded = false;
+ if (failure != null) {
+ rethrowCloseFailure(failure);
+ }
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ private static void rethrowCloseFailure(Throwable failure) throws IOException {
+ if (failure instanceof IOException) {
+ throw (IOException) failure;
+ }
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw new IOException("Failed to close ES index reader", failure);
+ }
+
+ // =================== unified filter dispatch =====================
+
+ private Optional<GlobalIndexResult> executeFilter(String fieldName, IndexFilter filter) {
+ try {
+ ensureLoaded();
+ if (debugEnabled()) {
+ FieldIndexConfig config = indexOptions.getConfig(fieldName);
+ LOG.info(
+ "PAIMON_ESLIB_FILTER_BEGIN field={} filter={} config={} fields={}",
+ fieldName,
+ filterSummary(filter),
+ config == null ? "null" : config.indexType() + "/" + config.scalarType(),
+ fieldsSample());
+ }
+ long[] ids = searcher.filter(fieldName, filter);
+ if (debugEnabled()) {
+ LOG.info(
+ "PAIMON_ESLIB_FILTER_RESULT field={} filter={} resultCount={} ids={}",
+ fieldName,
+ filterSummary(filter),
+ ids == null ? 0 : ids.length,
+ sample(ids, DEFAULT_SAMPLE_LIMIT));
+ }
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ if (ids != null) {
+ for (long id : ids) {
+ bitmap.add(id);
+ }
+ }
+ // The field IS indexed (dispatchFilter returned empty() already when it is not), so a
+ // zero-hit result is a genuine "0 rows match" — return an empty bitmap so the index
+ // prunes
+ // everything. Returning Optional.empty() here would mean "index can't evaluate" and
+ // force a
+ // raw-scan fallback (wrong semantics + loses the index's selectivity).
+ return Optional.of(GlobalIndexResult.create(bitmap));
+ } catch (IOException e) {
+ throw new RuntimeException("Filter failed on field: " + fieldName, e);
+ }
+ }
+
+ private Optional<GlobalIndexResult> conservativeAllRows(String logicalField) {
+ if (physicalField(logicalField) == null) {
+ return Optional.empty();
+ }
+ try {
+ ensureLoaded();
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ for (long id : searcher.allRowIds()) {
+ bitmap.add(id);
+ }
+ return Optional.of(GlobalIndexResult.create(bitmap));
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to read conservative ES index rows for field: " + logicalField, e);
+ }
+ }
+
+ private boolean canEvaluateFilter(FieldRef fieldRef, IndexFilter filter) {
+ String physicalField = physicalField(fieldRef.name());
+ FieldIndexConfig config = logicalFieldConfig(fieldRef.name());
+ if (physicalField == null || config == null) {
+ return false;
+ }
+ if (filter.filterType() == IndexFilter.FilterType.EXISTS
+ && fieldRef.type() instanceof ArrayType) {
+ return hasPersistedFieldConfigs
+ && indexOptions.arrayPresenceField(physicalField) != null;
+ }
+ if (filter.filterType() != IndexFilter.FilterType.EXISTS
+ && losesTimestampPrecision(fieldRef.type())) {
+ return false;
+ }
+ switch (config.indexType()) {
+ case FULLTEXT:
+ return indexOptions.keywordSubField(physicalField) != null
+ && !isKeywordRangeFilter(filter);
+ case KEYWORD:
+ return !isKeywordRangeFilter(filter);
+ case SCALAR:
+ case DATE:
+ return filter.filterType() == IndexFilter.FilterType.SCALAR
+ || filter.filterType() == IndexFilter.FilterType.EXISTS;
+ case VECTOR:
+ return filter.filterType() == IndexFilter.FilterType.EXISTS;
+ default:
+ return false;
+ }
+ }
+
+ private Optional<GlobalIndexResult> dispatchFilter(FieldRef fieldRef, IndexFilter filter) {
+ String physicalField = physicalField(fieldRef.name());
+ FieldIndexConfig config = logicalFieldConfig(fieldRef.name());
+ if (!canEvaluateFilter(fieldRef, filter)) {
+ return conservativeAllRows(fieldRef.name());
+ }
+ if (filter.filterType() == IndexFilter.FilterType.EXISTS
+ && fieldRef.type() instanceof ArrayType) {
+ String presenceField = indexOptions.arrayPresenceField(physicalField);
+ try {
+ return executeFilter(presenceField, filter);
+ } catch (UnsupportedOperationException | IllegalArgumentException e) {
+ return conservativeAllRows(fieldRef.name());
+ }
+ }
+ if (config.indexType() == FieldIndexConfig.IndexType.FULLTEXT) {
+ // A FULLTEXT field is indexed as analyzer-produced tokens, which are not equivalent to
+ // the raw column value, so ordinary predicates (=, <>, <, >, IN, LIKE, IS NULL, ...)
+ // cannot run on it directly. Route them to the keyword multi-field sub-field
+ // (content.keyword) when present — it holds the exact value and serves these correctly.
+ // Without the sub-field, conservatively retain every row so the caller can evaluate
+ // the predicate. Full-text search (visitFullTextSearch) still targets the analyzed
+ // primary field.
+ String subField = indexOptions.keywordSubField(physicalField);
+ try {
+ return executeFilter(subField, filter);
+ } catch (UnsupportedOperationException | IllegalArgumentException e) {
+ return conservativeAllRows(fieldRef.name());
+ }
+ }
+ try {
+ return executeFilter(physicalField, filter);
+ } catch (UnsupportedOperationException | IllegalArgumentException e) {
+ return conservativeAllRows(fieldRef.name());
+ }
+ }
+
+ private static boolean losesTimestampPrecision(DataType type) {
+ if (type instanceof TimestampType) {
+ return ((TimestampType) type).getPrecision() > 3;
+ }
+ return type instanceof LocalZonedTimestampType
+ && ((LocalZonedTimestampType) type).getPrecision() > 3;
+ }
+
+ private static boolean isKeywordRangeFilter(IndexFilter filter) {
+ if (filter.filterType() != IndexFilter.FilterType.SCALAR) {
+ return false;
+ }
+ ScalarPredicate.Op op = ((IndexFilter.ScalarFilter) filter).predicate().op();
+ return op == ScalarPredicate.Op.LESS_THAN
+ || op == ScalarPredicate.Op.LESS_OR_EQUAL
+ || op == ScalarPredicate.Op.GREATER_THAN
+ || op == ScalarPredicate.Op.GREATER_OR_EQUAL
+ || op == ScalarPredicate.Op.RANGE;
+ }
+
+ /**
+ * Evaluates a SQL negation as exists(field) AND NOT matchingFilter. Keeping the complement at
+ * the Paimon layer works for both scalar and keyword fields and preserves SQL null semantics.
+ */
+ private Optional<GlobalIndexResult> dispatchExistingRowsNotMatching(
+ FieldRef fieldRef, IndexFilter matchingFilter) {
+ if (!canEvaluateFilter(fieldRef, IndexFilter.exists())
+ || !canEvaluateFilter(fieldRef, matchingFilter)) {
+ return conservativeAllRows(fieldRef.name());
+ }
+ Optional<GlobalIndexResult> existing = dispatchFilter(fieldRef, IndexFilter.exists());
+ if (existing.isEmpty()) {
+ return Optional.empty();
+ }
+ Optional<GlobalIndexResult> matching = dispatchFilter(fieldRef, matchingFilter);
+ if (matching.isEmpty()) {
+ return Optional.empty();
+ }
+
+ RoaringNavigableMap64 result = new RoaringNavigableMap64();
+ result.or(existing.get().results());
+ result.andNot(matching.get().results());
+ return Optional.of(GlobalIndexResult.create(result));
+ }
+
+ private IndexFilter exactValueFilter(FieldRef fieldRef, Object literal) {
+ FieldIndexConfig config = logicalFieldConfig(fieldRef.name());
+ if (config != null
+ && (config.indexType() == FieldIndexConfig.IndexType.KEYWORD
+ || config.indexType() == FieldIndexConfig.IndexType.FULLTEXT)) {
+ return IndexFilter.text(IndexFilter.TextFilter.TextOp.TERM, str(literal));
+ }
+ return IndexFilter.scalar(ScalarPredicate.eq(indexedScalarLiteral(fieldRef, literal)));
+ }
+
+ private static List<Object> indexedScalarLiterals(FieldRef fieldRef, List<Object> literals) {
+ if (literals == null) {
+ return null;
+ }
+ List<Object> indexed = new ArrayList<>(literals.size());
+ for (Object literal : literals) {
+ indexed.add(indexedScalarLiteral(fieldRef, literal));
+ }
+ return indexed;
+ }
+
+ private static Object indexedScalarLiteral(FieldRef fieldRef, Object literal) {
+ if (literal == null || fieldRef.type() == null) {
+ return literal;
+ }
+ switch (fieldRef.type().getTypeRoot()) {
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ return literal instanceof Number ? ((Number) literal).longValue() : literal;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ if (literal instanceof Timestamp) {
+ return ((Timestamp) literal).getMillisecond();
+ }
+ if (literal instanceof java.sql.Timestamp) {
+ return ((java.sql.Timestamp) literal).getTime();
+ }
+ return literal instanceof Number ? ((Number) literal).longValue() : literal;
+ default:
+ return literal;
+ }
+ }
+
+ private static String filterSummary(IndexFilter filter) {
+ if (filter == null) {
+ return "null";
+ }
+ switch (filter.filterType()) {
+ case TEXT:
+ IndexFilter.TextFilter text = (IndexFilter.TextFilter) filter;
+ return "TEXT/" + text.op() + "/" + text.value();
+ case SCALAR:
+ return "SCALAR/" + ((IndexFilter.ScalarFilter) filter).predicate().op();
+ case EXISTS:
+ return "EXISTS/" + ((IndexFilter.ExistsFilter) filter).mustExist();
+ case GEO:
+ return "GEO";
+ default:
+ return filter.filterType().name();
+ }
+ }
+
+ // =================== scalar / keyword comparison visitors =====================
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () -> {
+ FieldIndexConfig config = logicalFieldConfig(fieldRef.name());
+ if (config == null) {
+ return conservativeAllRows(fieldRef.name());
+ }
+ FieldIndexConfig.IndexType type = config.indexType();
+ if (type == FieldIndexConfig.IndexType.KEYWORD
+ || type == FieldIndexConfig.IndexType.FULLTEXT) {
+ return dispatchFilter(
+ fieldRef,
+ IndexFilter.text(IndexFilter.TextFilter.TextOp.TERM, str(literal)));
+ }
+ return dispatchFilter(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.eq(indexedScalarLiteral(fieldRef, literal))));
+ });
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchExistingRowsNotMatching(
+ fieldRef, exactValueFilter(fieldRef, literal)));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.lt(
+ indexedScalarLiteral(fieldRef, literal)))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.lte(
+ indexedScalarLiteral(fieldRef, literal)))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.gt(
+ indexedScalarLiteral(fieldRef, literal)))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.gte(
+ indexedScalarLiteral(fieldRef, literal)))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ if (from == null || to == null || isReversedRange(fieldRef, from, to)) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.range(
+ indexedScalarLiteral(fieldRef, from),
+ indexedScalarLiteral(fieldRef, to)))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ if (from == null || to == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () -> {
+ if (isReversedRange(fieldRef, from, to)) {
+ return dispatchFilter(fieldRef, IndexFilter.exists());
+ }
+ return dispatchExistingRowsNotMatching(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.range(
+ indexedScalarLiteral(fieldRef, from),
+ indexedScalarLiteral(fieldRef, to))));
+ });
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
+ FieldRef fieldRef, List<Object> literals) {
+ List<Object> nonNullLiterals = withoutNullLiterals(literals);
+ if (nonNullLiterals.isEmpty()) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.in(
+ indexedScalarLiterals(
+ fieldRef, nonNullLiterals)))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
+ FieldRef fieldRef, List<Object> literals) {
+ if (literals == null || literals.contains(null)) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchExistingRowsNotMatching(
+ fieldRef,
+ IndexFilter.scalar(
+ ScalarPredicate.in(
+ indexedScalarLiterals(fieldRef, literals)))));
+ }
+
+ // =================== text pattern visitors (keyword / fulltext) =====================
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.text(
+ IndexFilter.TextFilter.TextOp.PREFIX, str(literal))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.text(
+ IndexFilter.TextFilter.TextOp.WILDCARD,
+ "*" + escapeWildcardLiteral(str(literal)))));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.text(
+ IndexFilter.TextFilter.TextOp.WILDCARD,
+ "*" + escapeWildcardLiteral(str(literal)) + "*")));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
+ FieldRef fieldRef, Object literal) {
+ if (literal == null) {
+ return emptyFilterFuture();
+ }
+ return async(
+ () ->
+ dispatchFilter(
+ fieldRef,
+ IndexFilter.text(
+ IndexFilter.TextFilter.TextOp.WILDCARD,
+ sqlLikeToWildcard(str(literal)))));
+ }
+
+ /**
+ * Escape the Lucene {@link org.apache.lucene.search.WildcardQuery} metacharacters ({@code *},
+ * {@code ?} and the escape char {@code \}) in a literal so it is matched verbatim. StartsWith /
+ * EndsWith / Contains literals carry no wildcards, so any such character in them is data and
+ * must not be reinterpreted as a wildcard.
+ */
+ private static String escapeWildcardLiteral(String s) {
+ StringBuilder sb = new StringBuilder(s.length() + 4);
+ for (int i = 0; i < s.length(); i++) {
+ char c = s.charAt(i);
+ if (c == '*' || c == '?' || c == '\\') {
+ sb.append('\\');
+ }
+ sb.append(c);
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Translate a SQL LIKE pattern into a Lucene wildcard pattern. Paimon's default SQL LIKE escape
+ * is {@code \}: escaped {@code _}, {@code %}, and {@code \} are literals, while unescaped
+ * {@code %} and {@code _} are the only SQL wildcards. Lucene's own wildcard metacharacters are
+ * escaped whenever they represent literal input.
+ */
+ private static String sqlLikeToWildcard(String sql) {
+ StringBuilder sb = new StringBuilder(sql.length() + 4);
+ for (int i = 0; i < sql.length(); i++) {
+ char c = sql.charAt(i);
+ switch (c) {
+ case '\\':
+ if (i == sql.length() - 1) {
+ throw invalidLikeEscapeSequence(sql, i);
+ }
+ char escaped = sql.charAt(++i);
+ if (escaped != '_' && escaped != '%' && escaped != '\\') {
+ throw invalidLikeEscapeSequence(sql, i - 1);
+ }
+ if (escaped == '\\') {
+ sb.append('\\');
+ }
+ sb.append(escaped);
+ break;
+ case '%':
+ sb.append('*');
+ break;
+ case '_':
+ sb.append('?');
+ break;
+ case '*':
+ case '?':
+ sb.append('\\').append(c);
+ break;
+ default:
+ sb.append(c);
+ }
+ }
+ return sb.toString();
+ }
+
+ private static RuntimeException invalidLikeEscapeSequence(String pattern, int position) {
+ return new RuntimeException("Invalid escape sequence '" + pattern + "', " + position);
+ }
+
+ // =================== null checks =====================
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIsNotNull(FieldRef fieldRef) {
+ return async(() -> dispatchFilter(fieldRef, IndexFilter.exists()));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIsNull(FieldRef fieldRef) {
+ // The writer registers an empty doc for every null row (addNullDoc / flushPendingDocs
+ // padding), with the field absent. So notExists() = MUST_NOT FieldExistsQuery matches
+ // exactly
+ // those rows and IS NULL is index-evaluable (no raw-scan fallback needed).
+ return async(() -> dispatchFilter(fieldRef, IndexFilter.notExists()));
+ }
+
+ // =================== helpers =====================
+
+ private static CompletableFuture<Optional<GlobalIndexResult>> emptyFilterFuture() {
+ return CompletableFuture.completedFuture(
+ Optional.of(GlobalIndexResult.create(new RoaringNavigableMap64())));
+ }
+
+ private static boolean isReversedRange(FieldRef fieldRef, Object from, Object to) {
+ return fieldRef.type() != null && compareLiteral(fieldRef.type(), from, to) > 0;
+ }
+
+ private static List<Object> withoutNullLiterals(List<Object> literals) {
+ if (literals == null || literals.isEmpty()) {
+ return Collections.emptyList();
+ }
+ List<Object> result = new ArrayList<>(literals.size());
+ for (Object literal : literals) {
+ if (literal != null) {
+ result.add(literal);
+ }
+ }
+ return result;
+ }
+
+ private static String str(Object literal) {
+ return literal.toString();
+ }
+}
diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexWriter.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexWriter.java
new file mode 100644
index 0000000000..1de6847e43
--- /dev/null
+++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexWriter.java
@@ -0,0 +1,629 @@
+/*
+ * 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.data.InternalArray;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.VectorType;
+import org.apache.paimon.utils.FileIOUtils;
+
+import org.elasticsearch.eslib.api.ESIndexBuilder;
+import org.elasticsearch.eslib.api.model.FieldIndexConfig;
+
+import java.io.Closeable;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Multi-column writer that builds ESLib (Lucene-based) indexes. Accepts InternalRow with all
+ * indexed fields and produces a single archive file.
+ */
+public class ESIndexGlobalIndexWriter
+ implements GlobalIndexSingleColumnWriter, GlobalIndexMultiColumnWriter, Closeable {
+
+ private static final String FILE_NAME_PREFIX = "es-index";
+
+ private final GlobalIndexFileWriter fileWriter;
+ private final List<DataField> fields;
+ private final ESIndexOptions indexOptions;
+ private final ESIndexBuilder builder;
+ private long docCount;
+ private Throwable writeFailure;
+ private boolean finished;
+ private boolean closed;
+
+ public ESIndexGlobalIndexWriter(
+ GlobalIndexFileWriter fileWriter, List<DataField> fields, ESIndexOptions indexOptions)
+ throws IOException {
+ this.fileWriter = Objects.requireNonNull(fileWriter, "fileWriter");
+ if (fields == null || fields.isEmpty()) {
+ throw new IllegalArgumentException("ES index requires at least one field");
+ }
+ this.indexOptions = Objects.requireNonNull(indexOptions, "indexOptions");
+ List<DataField> fieldsCopy = new ArrayList<>(fields.size());
+ for (DataField field : fields) {
+ DataField nonNullField = Objects.requireNonNull(field, "field");
+ if (indexOptions.getConfig(nonNullField.name()) == null) {
+ throw new IllegalArgumentException(
+ "Missing es-index configuration for field '" + nonNullField.name() + "'");
+ }
+ fieldsCopy.add(nonNullField);
+ }
+ this.fields = Collections.unmodifiableList(fieldsCopy);
+ this.builder = ESIndexBuilderFactory.create(indexOptions.getFieldConfigs());
+ this.docCount = 0;
+ this.finished = false;
+ this.closed = false;
+ }
+
+ @Override
+ public void write(Object fieldData, long relativeRowId) {
+ checkWritable();
+ try {
+ // Use the builder-supplied shard-relative row id as the Lucene docId (read side maps
+ // _ROW_ID = rangeFrom + docId). finishDocument pads any skipped positions, so the
+ // logical row count is the highest finished id plus one rather than the number of
+ // write() calls.
+ long docId = relativeRowId;
+ DataField field = fields.get(0);
+ FieldIndexConfig config = indexOptions.getConfig(field.name());
+ if (fieldData == null) {
+ // Null value: explicitly occupy this row's slot with an empty doc so
+ // docId<->rowId stays dense; the absent field marks it null.
+ // flushPendingDocs
+ // is the back-stop.
+ builder.addNullDoc(docId);
+ builder.finishDocument(docId);
+ docCount = docId + 1;
+ return;
+ }
+ writeArrayPresence(field, docId);
+ writeSingleField(fieldData, field, config, docId);
+ builder.finishDocument(docId);
+ docCount = docId + 1;
+ } catch (IOException e) {
+ RuntimeException failure =
+ new RuntimeException("Failed to write document to ES index", e);
+ writeFailure = failure;
+ throw failure;
+ } catch (RuntimeException | Error e) {
+ writeFailure = e;
+ throw e;
+ }
+ }
+
+ @Override
+ public void write(long rowId, InternalRow row) {
+ writeRow(rowId, row);
+ }
+
+ public void writeRow(long rowId, InternalRow row) {
+ checkWritable();
+ try {
+ // Use the shard-relative rowId supplied by the builder as the Lucene docId; the read
+ // side reconstructs the absolute id as `_ROW_ID = rangeFrom + docId`. Count every row.
+ long docId = rowId;
+ if (row == null) {
+ builder.addNullDoc(docId);
+ builder.finishDocument(docId);
+ docCount = docId + 1;
+ return;
+ }
+ boolean wroteAny = false;
+ for (int i = 0; i < fields.size(); i++) {
+ if (row.isNullAt(i)) {
+ continue;
+ }
+ DataField field = fields.get(i);
+ FieldIndexConfig config = indexOptions.getConfig(field.name());
+ writeArrayPresence(field, docId);
+ writeField(row, i, field, config, docId);
+ wroteAny = true;
+ }
+ if (!wroteAny) {
+ // Every indexed field is null for this row: explicitly occupy its slot with an
+ // empty
+ // doc so docId<->rowId stays dense (incl. trailing all-null rows). flushPendingDocs
+ // is the back-stop.
+ builder.addNullDoc(docId);
+ }
+ builder.finishDocument(docId);
+ docCount = docId + 1;
+ } catch (IOException e) {
+ RuntimeException failure =
+ new RuntimeException("Failed to write document to ES index", e);
+ writeFailure = failure;
+ throw failure;
+ } catch (RuntimeException | Error e) {
+ writeFailure = e;
+ throw e;
+ }
+ }
+
+ private void writeSingleField(
+ Object fieldData, DataField field, FieldIndexConfig config, long docId)
+ throws IOException {
+ switch (config.indexType()) {
+ case VECTOR:
+ float[] vector = null;
+ if (fieldData instanceof float[]) {
+ vector = (float[]) fieldData;
+ } else if (fieldData instanceof InternalArray) {
+ vector = materializeVector((InternalArray) fieldData, field.name());
+ }
+ if (vector != null) {
+ validateVectorDimension(field.name(), vector, config);
+ builder.addVector(field.name(), docId, vector);
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported vector value for field '"
+ + field.name()
+ + "': "
+ + fieldData.getClass().getName());
+ }
+ break;
+ case FULLTEXT:
+ String singleText = fieldData.toString();
+ builder.addTextField(field.name(), docId, singleText);
+ writeKeywordSubField(field.name(), docId, singleText);
+ break;
+ case KEYWORD:
+ String singleKeyword = fieldData.toString();
+ builder.addScalarField(
+ field.name(),
+ docId,
+ singleKeyword,
+ org.elasticsearch.eslib.api.model.ScalarFieldType.KEYWORD);
+ writeFullTextSubField(field.name(), docId, singleKeyword);
+ break;
+ case SCALAR:
+ case DATE:
+ Object value = extractScalar(fieldData, field.type());
+ if (value != null) {
+ builder.addScalarField(field.name(), docId, value, config.scalarType());
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported es-index type "
+ + config.indexType()
+ + " for field '"
+ + field.name()
+ + "'.");
+ }
+ }
+
+ private void writeField(
+ InternalRow row, int pos, DataField field, FieldIndexConfig config, long docId)
+ throws IOException {
+ switch (config.indexType()) {
+ case VECTOR:
+ float[] vector = extractVector(row, pos, field);
+ if (vector != null) {
+ validateVectorDimension(field.name(), vector, config);
+ builder.addVector(field.name(), docId, vector);
+ }
+ break;
+ case FULLTEXT:
+ String text = row.getString(pos).toString();
+ builder.addTextField(field.name(), docId, text);
+ writeKeywordSubField(field.name(), docId, text);
+ break;
+ case KEYWORD:
+ String keyword = row.getString(pos).toString();
+ builder.addScalarField(
+ field.name(),
+ docId,
+ keyword,
+ org.elasticsearch.eslib.api.model.ScalarFieldType.KEYWORD);
+ writeFullTextSubField(field.name(), docId, keyword);
+ break;
+ case SCALAR:
+ case DATE:
+ Object value = extractScalar(row, pos, field.type());
+ if (value != null) {
+ builder.addScalarField(field.name(), docId, value, config.scalarType());
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported es-index type "
+ + config.indexType()
+ + " for field '"
+ + field.name()
+ + "'.");
+ }
+ }
+
+ private void writeArrayPresence(DataField field, long docId) throws IOException {
+ if (!(field.type() instanceof ArrayType)) {
+ return;
+ }
+ String presenceField = indexOptions.arrayPresenceField(field.name());
+ if (presenceField != null) {
+ builder.addScalarField(
+ presenceField, docId, 1, org.elasticsearch.eslib.api.model.ScalarFieldType.INT);
+ }
+ }
+
+ private static void validateVectorDimension(
+ String fieldName, float[] vector, FieldIndexConfig config) {
+ if (vector.length != config.dimension()) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' expects dimension "
+ + config.dimension()
+ + " but received "
+ + vector.length
+ + ".");
+ }
+ }
+
+ /**
+ * Multi-field: when a FULLTEXT column has a keyword sub-field configured, write the raw value
+ * into {@code <field>.keyword} as an exact keyword term so exact filters can use it while
+ * full-text match uses the analyzed primary field. No-op when the sub-field is disabled.
+ */
+ private void writeKeywordSubField(String fieldName, long docId, String value)
+ throws IOException {
+ String subField = indexOptions.keywordSubField(fieldName);
+ if (subField != null) {
+ builder.addScalarField(
+ subField,
+ docId,
+ value,
+ org.elasticsearch.eslib.api.model.ScalarFieldType.KEYWORD);
+ }
+ }
+
+ /** Writes the analyzed multi-field paired with a KEYWORD primary field. */
+ private void writeFullTextSubField(String fieldName, long docId, String value)
+ throws IOException {
+ String subField = indexOptions.fullTextSubField(fieldName);
+ if (subField != null) {
+ builder.addTextField(subField, docId, value);
+ }
+ }
+
+ private float[] extractVector(InternalRow row, int pos, DataField field) {
+ InternalArray vector =
+ field.type() instanceof VectorType ? row.getVector(pos) : row.getArray(pos);
+ return materializeVector(vector, field.name());
+ }
+
+ private static float[] materializeVector(InternalArray vector, String fieldName) {
+ float[] values = new float[vector.size()];
+ for (int i = 0; i < vector.size(); i++) {
+ if (vector.isNullAt(i)) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' must not contain null elements; found null at position "
+ + i
+ + ".");
+ }
+ values[i] = vector.getFloat(i);
+ }
+ return values;
+ }
+
+ private Object extractScalar(InternalRow row, int pos, DataType type) {
+ switch (type.getTypeRoot()) {
+ case ARRAY:
+ return extractScalarArray(row.getArray(pos), (ArrayType) type);
+ case INTEGER:
+ return row.getInt(pos);
+ case SMALLINT:
+ return (int) row.getShort(pos);
+ case TINYINT:
+ return (int) row.getByte(pos);
+ case BIGINT:
+ return row.getLong(pos);
+ case FLOAT:
+ return row.getFloat(pos);
+ case DOUBLE:
+ return row.getDouble(pos);
+ case CHAR:
+ case VARCHAR:
+ return row.getString(pos).toString();
+ case DATE:
+ // DATE is stored as an int day-count; TIME as an int millis-of-day.
+ case TIME_WITHOUT_TIME_ZONE:
+ return (long) row.getInt(pos);
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return row.getTimestamp(pos, ((TimestampType) type).getPrecision())
+ .getMillisecond();
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return row.getTimestamp(pos, ((LocalZonedTimestampType) type).getPrecision())
+ .getMillisecond();
+ default:
+ return row.getString(pos).toString();
+ }
+ }
+
+ private Object extractScalar(Object fieldData, DataType type) {
+ if (type instanceof ArrayType && fieldData instanceof InternalArray) {
+ return extractScalarArray((InternalArray) fieldData, (ArrayType) type);
+ }
+ switch (type.getTypeRoot()) {
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ return fieldData instanceof Number ? ((Number) fieldData).longValue() : fieldData;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ if (fieldData instanceof Timestamp) {
+ return ((Timestamp) fieldData).getMillisecond();
+ }
+ if (fieldData instanceof java.sql.Timestamp) {
+ return ((java.sql.Timestamp) fieldData).getTime();
+ }
+ return fieldData instanceof Number ? ((Number) fieldData).longValue() : fieldData;
+ default:
+ break;
+ }
+ return fieldData;
+ }
+
+ private Object extractScalarArray(InternalArray array, ArrayType type) {
+ Object[] values = new Object[array.size()];
+ switch (type.getElementType().getTypeRoot()) {
+ case TINYINT:
+ for (int i = 0; i < array.size(); i++) {
+ values[i] = array.isNullAt(i) ? null : (int) array.getByte(i);
+ }
+ return values;
+ case SMALLINT:
+ for (int i = 0; i < array.size(); i++) {
+ values[i] = array.isNullAt(i) ? null : (int) array.getShort(i);
+ }
+ return values;
+ case INTEGER:
+ for (int i = 0; i < array.size(); i++) {
+ values[i] = array.isNullAt(i) ? null : array.getInt(i);
+ }
+ return values;
+ case BIGINT:
+ for (int i = 0; i < array.size(); i++) {
+ values[i] = array.isNullAt(i) ? null : array.getLong(i);
+ }
+ return values;
+ case CHAR:
+ case VARCHAR:
+ for (int i = 0; i < array.size(); i++) {
+ values[i] = array.isNullAt(i) ? null : array.getString(i).toString();
+ }
+ return values;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported scalar array element type for es-index: "
+ + type.getElementType());
+ }
+ }
+
+ @Override
+ public List<ResultEntry> finish() {
+ checkNotFinished();
+ finished = true;
+ Throwable failure = null;
+ try {
+ if (writeFailure != null) {
+ throw new IllegalStateException(
+ "Cannot finish ES index after a document write failed", writeFailure);
+ }
+ if (docCount == 0) {
+ // Nothing was indexed, but the builder already eagerly created a temp directory and
+ // an open Lucene IndexWriter in its constructor; the finally block below releases
+ // them. GlobalIndexWriter has no separate close(), so finish() must do the cleanup.
+ return Collections.emptyList();
+ }
+
+ builder.build();
+ Path outputDir = builder.getOutputDir();
+
+ // Snapshot the file list ONCE so the archive layout and the offset table in meta agree
+ // exactly (listFiles() order must not differ between packing and meta building).
+ java.io.File[] segFiles = outputDir.toFile().listFiles();
+ if (segFiles == null) {
+ segFiles = new java.io.File[0];
+ }
+
+ byte[] meta = buildMeta(segFiles);
+
+ String fileName = fileWriter.newFileName(FILE_NAME_PREFIX);
+ // Stream the archive directly to the output stream — DO NOT materialise the entire
+ // archive in memory. For real DiskBBQ output (1M × 768-dim vectors ≈ 3 GB per shard)
+ // a ByteArrayOutputStream-backed approach hits both heap and DirectByteBuffer limits
+ // (java.nio.Bits.reserveMemory OOM at Files.readAllBytes).
+ try (PositionOutputStream out = fileWriter.newOutputStream(fileName)) {
+ packDirectoryStream(segFiles, out);
+ out.flush();
+ }
+
+ return Collections.singletonList(new ResultEntry(fileName, docCount, meta));
+ } catch (IOException e) {
+ RuntimeException wrapped = new RuntimeException("Failed to finish ES index build", e);
+ failure = wrapped;
+ throw wrapped;
+ } catch (RuntimeException | Error e) {
+ failure = e;
+ throw e;
+ } finally {
+ try {
+ close();
+ } catch (Throwable cleanupFailure) {
+ if (failure != null) {
+ failure.addSuppressed(cleanupFailure);
+ } else {
+ rethrowCleanupFailure(cleanupFailure);
+ }
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ finished = true;
+ if (closed) {
+ return;
+ }
+ Throwable failure = null;
+ try {
+ builder.close();
+ } catch (Throwable e) {
+ failure = e;
+ }
+ try {
+ FileIOUtils.deleteDirectory(builder.getOutputDir().toFile());
+ } catch (Throwable e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ if (failure != null) {
+ if (failure instanceof IOException) {
+ throw (IOException) failure;
+ }
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw new IOException("Failed to clean up ES index builder", failure);
+ }
+ closed = true;
+ }
+
+ private static void rethrowCleanupFailure(Throwable failure) {
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw new RuntimeException("Failed to clean up ES index builder", failure);
+ }
+
+ private void checkNotFinished() {
+ if (finished) {
+ throw new IllegalStateException("ES index writer is already finished");
+ }
+ }
+
+ private void checkWritable() {
+ checkNotFinished();
+ if (writeFailure != null) {
+ throw new IllegalStateException(
+ "ES index writer cannot accept rows after a document write failed",
+ writeFailure);
+ }
+ }
+
+ /**
+ * Pack the given files into a single archive, streaming directly to {@code out}. Layout is
+ * interleaved per file (big-endian): [4-byte file count] then for each file: [4-byte name
+ * len][name bytes][8-byte data len][data bytes]. {@link #buildMeta} computes offsets against
+ * this exact layout.
+ *
+ * <p>Streams each file via a small buffered copy so total memory usage is O(buffer), not
+ * O(archive size). Returns the total number of bytes written.
+ */
+ private long packDirectoryStream(java.io.File[] segFiles, java.io.OutputStream out)
+ throws IOException {
+ // Wrap in BufferedOutputStream so DataOutputStream's small writes don't translate
+ // to a flood of underlying OSS writes; the buffer is fixed (64 KB) regardless of file size.
+ java.io.BufferedOutputStream bufOut = new java.io.BufferedOutputStream(out, 64 * 1024);
+ DataOutputStream dos = new DataOutputStream(bufOut);
+
+ long written = 0;
+ dos.writeInt(segFiles.length);
+ written += 4;
+
+ byte[] copyBuf = new byte[64 * 1024];
+ for (java.io.File file : segFiles) {
+ byte[] nameBytes = file.getName().getBytes(StandardCharsets.UTF_8);
+ long fileLen = file.length();
+ dos.writeInt(nameBytes.length);
+ dos.write(nameBytes);
+ dos.writeLong(fileLen);
+ written += 4 + nameBytes.length + 8;
+
+ long copied = 0;
+ try (java.io.InputStream in = Files.newInputStream(file.toPath())) {
+ int n;
+ while ((n = in.read(copyBuf)) > 0) {
+ dos.write(copyBuf, 0, n);
+ copied += n;
+ }
+ }
+ if (copied != fileLen) {
+ throw new IOException(
+ "Short read packing "
+ + file.getName()
+ + ": expected "
+ + fileLen
+ + " bytes but got "
+ + copied);
+ }
+ written += fileLen;
+ }
+ dos.flush();
+ return written;
+ }
+
+ /**
+ * Build the versioned metadata stored in ResultEntry. It persists the resolved per-field index
+ * configuration followed by the archive file offset table, so readers use the same mapping that
+ * built the index. {@link ESIndexFileMeta} also accepts the legacy offset-only format.
+ *
+ * <p>The offsets MUST match {@link #packDirectoryStream}'s interleaved layout: each file's data
+ * begins right after its own [nameLen][name][dataLen] header.
+ */
+ private byte[] buildMeta(java.io.File[] segFiles) throws IOException {
+ List<String> indexedFieldNames = new ArrayList<>(fields.size());
+ List<String> indexedFieldTypes = new ArrayList<>(fields.size());
+ for (DataField field : fields) {
+ indexedFieldNames.add(field.name());
+ indexedFieldTypes.add(field.type().copy(true).asSQLString());
+ }
+ return ESIndexFileMeta.write(
+ segFiles, indexedFieldNames, indexedFieldTypes, indexOptions.getFieldConfigs());
+ }
+}
diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java
new file mode 100644
index 0000000000..293e33a595
--- /dev/null
+++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java
@@ -0,0 +1,62 @@
+/*
+ * 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.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * ES multi-index global indexer using ESLib. Builds Lucene-based indexes supporting vector,
+ * fulltext, and scalar fields.
+ */
+public class ESIndexGlobalIndexer implements GlobalIndexer {
+
+ private final List<DataField> fields;
+ private final ESIndexOptions indexOptions;
+
+ public ESIndexGlobalIndexer(List<DataField> fields, Options options) {
+ this.fields = Collections.unmodifiableList(new ArrayList<>(fields));
+ this.indexOptions = new ESIndexOptions(this.fields, options);
+ }
+
+ @Override
+ public GlobalIndexWriter createWriter(GlobalIndexFileWriter fileWriter) throws IOException {
+ return new ESIndexGlobalIndexWriter(fileWriter, fields, indexOptions);
+ }
+
+ @Override
+ public GlobalIndexReader createReader(
+ GlobalIndexFileReader fileReader,
+ List<GlobalIndexIOMeta> files,
+ ExecutorService executor) {
+ return new ESIndexGlobalIndexReader(fileReader, files, fields, indexOptions, executor);
+ }
+}
diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java
new file mode 100644
index 0000000000..5e520b7e14
--- /dev/null
+++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java
@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+/**
+ * Factory for creating ES multi-index global indexers. Supports vector (DiskBBQ/HNSW), fulltext
+ * (BM25), and scalar fields.
+ */
+public class ESIndexGlobalIndexerFactory implements GlobalIndexerFactory {
+
+ public static final String IDENTIFIER = "es-index";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public boolean supportsFullTextSearch() {
+ return true;
+ }
+
+ @Override
+ public GlobalIndexer create(DataField field, Options options) {
+ return new ESIndexGlobalIndexer(java.util.Collections.singletonList(field), 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);
+ }
+}
diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexOptions.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexOptions.java
new file mode 100644
index 0000000000..530e5fbe00
--- /dev/null
+++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexOptions.java
@@ -0,0 +1,623 @@
+/*
+ * 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.options.Options;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.VectorType;
+
+import org.apache.lucene.codecs.lucene99.Lucene99HnswVectorsFormat;
+import org.elasticsearch.eslib.api.model.BuiltinAnalyzer;
+import org.elasticsearch.eslib.api.model.FieldIndexConfig;
+import org.elasticsearch.eslib.api.model.ScalarFieldType;
+import org.elasticsearch.eslib.api.model.VectorAlgorithm;
+import org.elasticsearch.eslib.diskbbq.es94.ES940DiskBBQVectorsFormat;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Parses Paimon Options into ESLib FieldIndexConfig for each field.
+ *
+ * <p>Keys are read from the table options using the persisted {@code global-index.es-index.*}
+ * convention (same keys the table schema stores and the ES mount reads). A field-level key {@code
+ * global-index.es-index.fields.<field>.<key>} takes precedence over the index-type-level key {@code
+ * global-index.es-index.<key>}.
+ *
+ * <pre>
+ * global-index.es-index.metric = cosine # index-type-level default
+ * global-index.es-index.fields.vector_field.algorithm = diskbbq # field-level override
+ * global-index.es-index.fields.vector_field.dimension = 128
+ * global-index.es-index.fields.text_field.analyzer = standard
+ * global-index.es-index.fields.id_field.type = keyword
+ * </pre>
+ */
+public class ESIndexOptions {
+
+ private static final int MAX_VECTOR_DIMENSION = 4096;
+
+ /**
+ * Index-type-level key prefix (e.g. {@code global-index.es-index.metric}); overridden by
+ * field-level keys. Aligned with the persisted table-property / ES-mount convention ({@code
+ * global-index.<indexType>.*}) so build, read and mount all read the same keys. The {@code
+ * es-index} part still comes from {@link ESIndexGlobalIndexerFactory#IDENTIFIER}.
+ */
+ private static final String INDEX_TYPE_PREFIX =
+ "global-index." + ESIndexGlobalIndexerFactory.IDENTIFIER + ".";
+
+ /** Field-level key prefix, e.g. {@code global-index.es-index.fields.<field>.algorithm}. */
+ private static final String FIELDS_PREFIX = INDEX_TYPE_PREFIX + "fields.";
+
+ /** Suffix of the keyword multi-field sub-field of a FULLTEXT column. */
+ public static final String KEYWORD_SUBFIELD_SUFFIX = ".keyword";
+
+ /** Suffix of the full-text multi-field sub-field of a KEYWORD column. */
+ public static final String FULLTEXT_SUBFIELD_SUFFIX = ".fulltext";
+
+ /** Internal field used to distinguish a non-null empty array from a null array. */
+ static final String ARRAY_PRESENCE_SUBFIELD_SUFFIX = ".__paimon_array_present";
+
+ private final Map<String, FieldIndexConfig> fieldConfigs;
+
+ public ESIndexOptions(List<DataField> fields, Options options) {
+ this.fieldConfigs = new LinkedHashMap<>();
+ Set<String> logicalFieldNames = new HashSet<>();
+ for (DataField field : fields) {
+ if (!logicalFieldNames.add(field.name())) {
+ throw new IllegalArgumentException(
+ "Duplicate field name in es-index: " + field.name());
+ }
+ }
+ for (DataField field : fields) {
+ FieldIndexConfig config = parseFieldConfig(field, options);
+ fieldConfigs.put(field.name(), config);
+ // Text fields always support both analyzed search and exact matching. The configured
+ // type selects the primary field; the complementary capability is stored in a
+ // multi-field sub-field.
+ if (isTextType(field.type())
+ && config.indexType() == FieldIndexConfig.IndexType.FULLTEXT) {
+ String subField = field.name() + KEYWORD_SUBFIELD_SUFFIX;
+ putGeneratedFieldConfig(
+ logicalFieldNames,
+ subField,
+ FieldIndexConfig.builder(subField, FieldIndexConfig.IndexType.KEYWORD)
+ .scalarType(ScalarFieldType.KEYWORD)
+ .build());
+ } else if (isTextType(field.type())
+ && config.indexType() == FieldIndexConfig.IndexType.KEYWORD) {
+ String subField = field.name() + FULLTEXT_SUBFIELD_SUFFIX;
+ String analyzer = resolve(options, field.name(), "analyzer", "standard");
+ putGeneratedFieldConfig(
+ logicalFieldNames,
+ subField,
+ FieldIndexConfig.builder(subField, FieldIndexConfig.IndexType.FULLTEXT)
+ .analyzer(parseAnalyzer(field.name(), analyzer))
+ .build());
+ }
+ if (field.type() instanceof ArrayType) {
+ String presenceField = field.name() + ARRAY_PRESENCE_SUBFIELD_SUFFIX;
+ putGeneratedFieldConfig(
+ logicalFieldNames,
+ presenceField,
+ FieldIndexConfig.builder(presenceField, FieldIndexConfig.IndexType.SCALAR)
+ .scalarType(ScalarFieldType.INT)
+ .build());
+ }
+ }
+ }
+
+ private void putGeneratedFieldConfig(
+ Set<String> logicalFieldNames,
+ String generatedFieldName,
+ FieldIndexConfig generatedConfig) {
+ if (logicalFieldNames.contains(generatedFieldName)
+ || fieldConfigs.putIfAbsent(generatedFieldName, generatedConfig) != null) {
+ throw new IllegalArgumentException(
+ "Field name conflicts with reserved es-index generated field: "
+ + generatedFieldName);
+ }
+ }
+
+ private ESIndexOptions(Map<String, FieldIndexConfig> fieldConfigs) {
+ this.fieldConfigs = new LinkedHashMap<>(fieldConfigs);
+ }
+
+ /** Recreates the exact build-time configuration persisted in the index metadata. */
+ static ESIndexOptions fromFieldConfigs(Map<String, FieldIndexConfig> fieldConfigs) {
+ return new ESIndexOptions(fieldConfigs);
+ }
+
+ public Map<String, FieldIndexConfig> getFieldConfigs() {
+ return Collections.unmodifiableMap(fieldConfigs);
+ }
+
+ public FieldIndexConfig getConfig(String fieldName) {
+ return fieldConfigs.get(fieldName);
+ }
+
+ /** Returns the keyword multi-field sub-field name for {@code fieldName} if one exists. */
+ public String keywordSubField(String fieldName) {
+ String subField = fieldName + KEYWORD_SUBFIELD_SUFFIX;
+ return fieldConfigs.containsKey(subField) ? subField : null;
+ }
+
+ /** Returns the full-text multi-field sub-field name for {@code fieldName} if one exists. */
+ public String fullTextSubField(String fieldName) {
+ String subField = fieldName + FULLTEXT_SUBFIELD_SUFFIX;
+ return fieldConfigs.containsKey(subField) ? subField : null;
+ }
+
+ /** Returns the internal presence field for an array, or {@code null} for legacy indexes. */
+ public String arrayPresenceField(String fieldName) {
+ String subField = fieldName + ARRAY_PRESENCE_SUBFIELD_SUFFIX;
+ return fieldConfigs.containsKey(subField) ? subField : null;
+ }
+
+ /** Returns the physical FULLTEXT field used to search the logical Paimon field. */
+ public String fullTextSearchField(String fieldName) {
+ FieldIndexConfig config = fieldConfigs.get(fieldName);
+ if (config != null && config.indexType() == FieldIndexConfig.IndexType.FULLTEXT) {
+ return fieldName;
+ }
+ return fullTextSubField(fieldName);
+ }
+
+ /**
+ * Resolves a config value for {@code fieldName}/{@code key}: the field-level key {@code
+ * fields.<field>.<key>} takes precedence over the index-type-level key {@code es-index.<key>};
+ * returns {@code defaultValue} when neither is set.
+ */
+ private static String resolve(
+ Options options, String fieldName, String key, String defaultValue) {
+ String value = resolveField(options, fieldName, key);
+ if (value == null) {
+ value = options.getString(INDEX_TYPE_PREFIX + key, null);
+ }
+ return value == null ? defaultValue : value;
+ }
+
+ private static String resolveField(Options options, String fieldName, String key) {
+ return options.getString(FIELDS_PREFIX + fieldName + "." + key, null);
+ }
+
+ private FieldIndexConfig parseFieldConfig(DataField field, Options options) {
+ DataType dataType = field.type();
+
+ // Explicit type override takes priority
+ String explicitType = resolve(options, field.name(), "type", null);
+ if (explicitType != null) {
+ return parseExplicitType(field.name(), explicitType, dataType, options);
+ }
+
+ if (isVectorType(dataType)) {
+ return parseVectorConfig(field.name(), dataType, options);
+ } else if (isTextType(dataType)) {
+ // String fields default to FULLTEXT (analyzed with the standard analyzer) so full-text
+ // search always works on a text column; a keyword sub-field (<field>.keyword) is added
+ // below for exact filters — the ES text/keyword multi-field, so both capabilities are
+ // available. An explicit analyzer overrides the default. Configuring type=keyword
+ // makes the exact field primary and adds a FULLTEXT sub-field instead.
+ String analyzer = resolve(options, field.name(), "analyzer", "standard");
+ return FieldIndexConfig.builder(field.name(), FieldIndexConfig.IndexType.FULLTEXT)
+ .analyzer(parseAnalyzer(field.name(), analyzer))
+ .build();
+ } else if (isTemporalType(dataType)) {
+ // ESLib does not implement DATE scalar filters. Store DATE/TIMESTAMP as a
+ // long scalar instead (DATE = epoch day, TIMESTAMP = epoch millis) so predicate
+ // filtering uses the supported LONG query path.
+ return FieldIndexConfig.builder(field.name(), FieldIndexConfig.IndexType.SCALAR)
+ .scalarType(ScalarFieldType.LONG)
+ .build();
+ } else {
+ return FieldIndexConfig.builder(field.name(), FieldIndexConfig.IndexType.SCALAR)
+ .scalarType(mapScalarType(dataType))
+ .build();
+ }
+ }
+
+ private FieldIndexConfig parseExplicitType(
+ String fieldName, String typeName, DataType dataType, Options options) {
+ switch (typeName.toLowerCase(Locale.ROOT)) {
+ case "fulltext":
+ requireTextType(fieldName, typeName, dataType);
+ String analyzer = resolve(options, fieldName, "analyzer", "standard");
+ return FieldIndexConfig.builder(fieldName, FieldIndexConfig.IndexType.FULLTEXT)
+ .analyzer(parseAnalyzer(fieldName, analyzer))
+ .build();
+ case "keyword":
+ requireTextType(fieldName, typeName, dataType);
+ return FieldIndexConfig.builder(fieldName, FieldIndexConfig.IndexType.KEYWORD)
+ .scalarType(ScalarFieldType.KEYWORD)
+ .build();
+ case "geo_point":
+ throw new IllegalArgumentException(
+ "Explicit es-index type 'geo_point' is not supported for field '"
+ + fieldName
+ + "'.");
+ case "date":
+ if (!isTemporalType(dataType)) {
+ throw incompatibleType(fieldName, typeName, dataType);
+ }
+ return FieldIndexConfig.builder(fieldName, FieldIndexConfig.IndexType.SCALAR)
+ .scalarType(ScalarFieldType.LONG)
+ .build();
+ case "vector":
+ if (!isVectorType(dataType)) {
+ throw incompatibleType(fieldName, typeName, dataType);
+ }
+ return parseVectorConfig(fieldName, dataType, options);
+ default:
+ throw new IllegalArgumentException(
+ "Unknown es-index type '"
+ + typeName
+ + "' for field '"
+ + fieldName
+ + "'. Supported values: vector, fulltext, keyword, date.");
+ }
+ }
+
+ private FieldIndexConfig parseVectorConfig(
+ String fieldName, DataType dataType, Options options) {
+ String algorithm = resolve(options, fieldName, "algorithm", "hnsw");
+ VectorAlgorithm vectorAlgorithm = VectorAlgorithm.fromName(algorithm);
+ if (vectorAlgorithm == VectorAlgorithm.NATIVE) {
+ throw new IllegalArgumentException(
+ "Vector algorithm 'native' is not supported by paimon-eslib; "
+ + "use 'hnsw' or 'diskbbq'");
+ }
+
+ String dimStr = resolve(options, fieldName, "dimension", null);
+ int inferredDimension = inferDimension(dataType);
+ int dimension;
+ if (dimStr == null) {
+ if (inferredDimension <= 0) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' requires a positive dimension; ARRAY<FLOAT> has no fixed "
+ + "dimension in its data type.");
+ }
+ dimension = inferredDimension;
+ } else {
+ dimension = parsePositiveInteger(fieldName, "dimension", dimStr);
+ if (inferredDimension > 0 && dimension != inferredDimension) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' dimension "
+ + dimension
+ + " does not match the VECTOR type dimension "
+ + inferredDimension
+ + ".");
+ }
+ }
+ if (dimension > MAX_VECTOR_DIMENSION) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' dimension "
+ + dimension
+ + " exceeds the maximum supported dimension "
+ + MAX_VECTOR_DIMENSION
+ + ".");
+ }
+ String metric = validateMetric(fieldName, resolve(options, fieldName, "metric", "cosine"));
+
+ Map<String, String> params = new LinkedHashMap<>();
+ // Table-level algorithm parameters are defaults only for fields using that algorithm. A
+ // table can contain HNSW and DiskBBQ fields at the same time, so a global HNSW parameter
+ // must not make a DiskBBQ field invalid (and vice versa). Field-level parameters remain
+ // strict because they unambiguously target this field.
+ String mStr =
+ vectorAlgorithm == VectorAlgorithm.HNSW
+ ? resolve(options, fieldName, "m", null)
+ : resolveField(options, fieldName, "m");
+ String efStr =
+ vectorAlgorithm == VectorAlgorithm.HNSW
+ ? resolve(options, fieldName, "ef_construction", null)
+ : resolveField(options, fieldName, "ef_construction");
+ if (vectorAlgorithm == VectorAlgorithm.HNSW) {
+ putBoundedPositiveIntegerParameter(
+ params, fieldName, "m", mStr, Lucene99HnswVectorsFormat.MAXIMUM_MAX_CONN);
+ putBoundedPositiveIntegerParameter(
+ params,
+ fieldName,
+ "ef_construction",
+ efStr,
+ Lucene99HnswVectorsFormat.MAXIMUM_BEAM_WIDTH);
+ } else if (mStr != null || efStr != null) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' configures m/ef_construction but algorithm is "
+ + algorithm
+ + "; these parameters require algorithm=hnsw.");
+ }
+ String vpcStr =
+ vectorAlgorithm == VectorAlgorithm.DISKBBQ
+ ? resolve(options, fieldName, "vectors_per_cluster", null)
+ : resolveField(options, fieldName, "vectors_per_cluster");
+ if (vectorAlgorithm == VectorAlgorithm.DISKBBQ) {
+ putBoundedPositiveIntegerParameter(
+ params, fieldName, "vectors_per_cluster", vpcStr, 64, 1 << 16);
+ } else if (vpcStr != null) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' configures vectors_per_cluster but algorithm is "
+ + algorithm
+ + "; this parameter requires algorithm=diskbbq.");
+ }
+ String cpcStr =
+ vectorAlgorithm == VectorAlgorithm.DISKBBQ
+ ? resolve(options, fieldName, "centroids_per_parent_cluster", null)
+ : resolveField(options, fieldName, "centroids_per_parent_cluster");
+ if (vectorAlgorithm == VectorAlgorithm.DISKBBQ) {
+ putBoundedPositiveIntegerParameter(
+ params,
+ fieldName,
+ "centroids_per_parent_cluster",
+ cpcStr,
+ ES940DiskBBQVectorsFormat.MIN_CENTROIDS_PER_PARENT_CLUSTER,
+ ES940DiskBBQVectorsFormat.MAX_CENTROIDS_PER_PARENT_CLUSTER);
+ } else if (cpcStr != null) {
+ throw new IllegalArgumentException(
+ "Vector field '"
+ + fieldName
+ + "' configures centroids_per_parent_cluster but algorithm is "
+ + algorithm
+ + "; this parameter requires algorithm=diskbbq.");
+ }
+
+ return FieldIndexConfig.builder(fieldName, FieldIndexConfig.IndexType.VECTOR)
+ .algorithm(vectorAlgorithm)
+ .dimension(dimension)
+ .metric(metric)
+ .algorithmParams(params)
+ .build();
+ }
+
+ private static BuiltinAnalyzer parseAnalyzer(String fieldName, String analyzerName) {
+ BuiltinAnalyzer analyzer = BuiltinAnalyzer.fromName(analyzerName);
+ if (analyzer == BuiltinAnalyzer.IK_SMART || analyzer == BuiltinAnalyzer.IK_MAX_WORD) {
+ throw new IllegalArgumentException(
+ "Analyzer '"
+ + analyzerName
+ + "' is not supported by paimon-eslib for field '"
+ + fieldName
+ + "'; use standard, whitespace, simple, or keyword.");
+ }
+ return analyzer;
+ }
+
+ private static void putBoundedPositiveIntegerParameter(
+ Map<String, String> params,
+ String fieldName,
+ String key,
+ String rawValue,
+ int maximum) {
+ if (rawValue == null) {
+ return;
+ }
+ final int value;
+ try {
+ value = Integer.parseInt(rawValue);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid HNSW parameter '"
+ + key
+ + "' for vector field '"
+ + fieldName
+ + "': must be an integer between 1 and "
+ + maximum
+ + ".",
+ e);
+ }
+ if (value <= 0 || value > maximum) {
+ throw new IllegalArgumentException(
+ "Invalid HNSW parameter '"
+ + key
+ + "' for vector field '"
+ + fieldName
+ + "': must be an integer between 1 and "
+ + maximum
+ + ".");
+ }
+ params.put(key, String.valueOf(value));
+ }
+
+ private static void putBoundedPositiveIntegerParameter(
+ Map<String, String> params,
+ String fieldName,
+ String key,
+ String rawValue,
+ int minimum,
+ int maximum) {
+ if (rawValue == null) {
+ return;
+ }
+ final int value;
+ try {
+ value = Integer.parseInt(rawValue);
+ } catch (NumberFormatException e) {
+ throw invalidIntegerParameter(fieldName, key, minimum, maximum, e);
+ }
+ if (value < minimum || value > maximum) {
+ throw invalidIntegerParameter(fieldName, key, minimum, maximum, null);
+ }
+ params.put(key, String.valueOf(value));
+ }
+
+ private static IllegalArgumentException invalidIntegerParameter(
+ String fieldName, String key, int minimum, int maximum, Throwable cause) {
+ String message =
+ "Invalid parameter '"
+ + key
+ + "' for vector field '"
+ + fieldName
+ + "': must be an integer between "
+ + minimum
+ + " and "
+ + maximum
+ + ".";
+ return cause == null
+ ? new IllegalArgumentException(message)
+ : new IllegalArgumentException(message, cause);
+ }
+
+ private static int parsePositiveInteger(String fieldName, String key, String rawValue) {
+ final int value;
+ try {
+ value = Integer.parseInt(rawValue);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " for vector field '" + fieldName + "': " + rawValue, e);
+ }
+ if (value <= 0) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " for vector field '" + fieldName + "': must be positive.");
+ }
+ return value;
+ }
+
+ private static String validateMetric(String fieldName, String metric) {
+ String normalized = metric.toLowerCase(Locale.ROOT);
+ switch (normalized) {
+ case "cosine":
+ case "l2":
+ case "euclidean":
+ case "dot_product":
+ case "dp":
+ case "inner_product":
+ case "mip":
+ case "maximum_inner_product":
+ return normalized;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown vector metric '"
+ + metric
+ + "' for field '"
+ + fieldName
+ + "'. Supported values: cosine, l2, euclidean, dot_product, "
+ + "inner_product, mip, maximum_inner_product.");
+ }
+ }
+
+ private static void requireTextType(String fieldName, String typeName, DataType dataType) {
+ if (!isTextType(dataType)) {
+ throw incompatibleType(fieldName, typeName, dataType);
+ }
+ }
+
+ private static IllegalArgumentException incompatibleType(
+ String fieldName, String typeName, DataType dataType) {
+ return new IllegalArgumentException(
+ "Explicit es-index type '"
+ + typeName
+ + "' is incompatible with field '"
+ + fieldName
+ + "' of type "
+ + dataType
+ + ".");
+ }
+
+ private static ScalarFieldType mapScalarType(DataType type) {
+ if (type instanceof ArrayType) {
+ return mapArrayScalarType((ArrayType) type);
+ }
+ switch (type.getTypeRoot()) {
+ case INTEGER:
+ case SMALLINT:
+ case TINYINT:
+ return ScalarFieldType.INT;
+ case BIGINT:
+ return ScalarFieldType.LONG;
+ case FLOAT:
+ return ScalarFieldType.FLOAT;
+ case DOUBLE:
+ return ScalarFieldType.DOUBLE;
+ case CHAR:
+ case VARCHAR:
+ return ScalarFieldType.KEYWORD;
+ default:
+ // Fail fast on scalar roots the writer cannot extract (e.g. BOOLEAN, DECIMAL,
+ // BINARY). Mapping them to KEYWORD would defer the failure to build time, where
+ // ESIndexGlobalIndexWriter.extractScalar falls through to getString() and throws a
+ // ClassCastException. Mirror mapArrayScalarType, which rejects unsupported types.
+ throw new IllegalArgumentException("Unsupported scalar type for es-index: " + type);
+ }
+ }
+
+ private static ScalarFieldType mapArrayScalarType(ArrayType type) {
+ DataType elementType = type.getElementType();
+ switch (elementType.getTypeRoot()) {
+ case INTEGER:
+ case SMALLINT:
+ case TINYINT:
+ return ScalarFieldType.INT;
+ case BIGINT:
+ return ScalarFieldType.LONG;
+ case CHAR:
+ case VARCHAR:
+ return ScalarFieldType.KEYWORD;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported scalar array element type for es-index: " + elementType);
+ }
+ }
+
+ private static boolean isVectorType(DataType type) {
+ if (type instanceof VectorType) {
+ return ((VectorType) type).getElementType().getTypeRoot() == DataTypeRoot.FLOAT;
+ }
+ if (type instanceof ArrayType) {
+ DataType elementType = ((ArrayType) type).getElementType();
+ return elementType.getTypeRoot() == DataTypeRoot.FLOAT;
+ }
+ return false;
+ }
+
+ private static boolean isTextType(DataType type) {
+ return type.getTypeRoot() == DataTypeRoot.VARCHAR
+ || type.getTypeRoot() == DataTypeRoot.CHAR;
+ }
+
+ private static boolean isTemporalType(DataType type) {
+ DataTypeRoot root = type.getTypeRoot();
+ return root == DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE
+ || root == DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE
+ || root == DataTypeRoot.DATE
+ || root == DataTypeRoot.TIME_WITHOUT_TIME_ZONE;
+ }
+
+ private static int inferDimension(DataType type) {
+ if (type instanceof VectorType) {
+ return ((VectorType) type).getLength();
+ }
+ return 0;
+ }
+}
diff --git a/paimon-eslib/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory b/paimon-eslib/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
new file mode 100644
index 0000000000..c184d03fff
--- /dev/null
+++ b/paimon-eslib/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.paimon.eslib.index.ESIndexGlobalIndexerFactory
diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexFileMetaTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexFileMetaTest.java
new file mode 100644
index 0000000000..5c17587def
--- /dev/null
+++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexFileMetaTest.java
@@ -0,0 +1,365 @@
+/*
+ * 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.GlobalIndexIOMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.VectorSearch;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.elasticsearch.eslib.api.model.BuiltinAnalyzer;
+import org.elasticsearch.eslib.api.model.FieldIndexConfig;
+import org.elasticsearch.eslib.api.model.ScalarFieldType;
+import org.elasticsearch.eslib.api.model.VectorAlgorithm;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ESIndexFileMetaTest {
+
+ @Test
+ void rejectsMetadataWithoutLuceneFiles() {
+ assertThatThrownBy(
+ () ->
+ ESIndexFileMeta.write(
+ new java.io.File[0],
+ List.of("id"),
+ List.of("INT"),
+ java.util.Collections.emptyMap()))
+ .isInstanceOf(java.io.IOException.class)
+ .hasMessageContaining("without Lucene files");
+ }
+
+ @Test
+ void roundTripsBuildConfigurationAndArchiveOffsets(@TempDir Path tempDir) throws Exception {
+ Path first = tempDir.resolve("segments_1");
+ Path second = tempDir.resolve("_0.cfs");
+ Files.write(first, new byte[] {1, 2, 3});
+ Files.write(second, new byte[] {4, 5});
+
+ Map<String, String> parameters = new LinkedHashMap<>();
+ parameters.put("m", "32");
+ parameters.put("ef_construction", "200");
+ Map<String, FieldIndexConfig> configs = new LinkedHashMap<>();
+ configs.put(
+ "embedding",
+ FieldIndexConfig.builder("embedding", FieldIndexConfig.IndexType.VECTOR)
+ .algorithm(VectorAlgorithm.HNSW)
+ .dimension(4)
+ .metric("cosine")
+ .algorithmParams(parameters)
+ .build());
+
+ ESIndexFileMeta.Parsed parsed =
+ ESIndexFileMeta.read(
+ ESIndexFileMeta.write(
+ new java.io.File[] {first.toFile(), second.toFile()},
+ List.of("embedding"),
+ List.of("ARRAY<FLOAT>"),
+ configs));
+
+ assertThat(parsed.hasFieldConfigs()).isTrue();
+ assertThat(parsed.indexedFieldNames()).containsExactly("embedding");
+ assertThat(parsed.indexedFieldTypes()).containsExactly("ARRAY<FLOAT>");
+ FieldIndexConfig vector = parsed.fieldConfigs().get("embedding");
+ assertThat(vector.algorithm()).isEqualTo(VectorAlgorithm.HNSW);
+ assertThat(vector.dimension()).isEqualTo(4);
+ assertThat(vector.metric()).isEqualTo("cosine");
+ assertThat(vector.algorithmParams()).containsAllEntriesOf(parameters);
+
+ long firstOffset = 4L + 4L + "segments_1".getBytes(StandardCharsets.UTF_8).length + 8L;
+ assertThat(parsed.fileOffsets().get("segments_1")).containsExactly(firstOffset, 3L);
+ long secondOffset =
+ firstOffset + 3L + 4L + "_0.cfs".getBytes(StandardCharsets.UTF_8).length + 8L;
+ assertThat(parsed.fileOffsets().get("_0.cfs")).containsExactly(secondOffset, 2L);
+ }
+
+ @Test
+ void readsLegacyOffsetOnlyMetadata() throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ byte[] name = "segments_1".getBytes(StandardCharsets.UTF_8);
+ out.writeInt(1);
+ out.writeInt(name.length);
+ out.write(name);
+ out.writeLong(17L);
+ out.writeLong(23L);
+ out.flush();
+
+ ESIndexFileMeta.Parsed parsed = ESIndexFileMeta.read(bytes.toByteArray());
+ assertThat(parsed.hasFieldConfigs()).isFalse();
+ assertThat(parsed.indexedFieldTypes()).isEmpty();
+ assertThat(parsed.fieldConfigs()).isEmpty();
+ assertThat(parsed.fileOffsets().get("segments_1")).containsExactly(17L, 23L);
+ }
+
+ @Test
+ void readsVersionOnePrimaryFieldOrder() throws Exception {
+ Map<String, FieldIndexConfig> configs = new LinkedHashMap<>();
+ configs.put(
+ "title",
+ FieldIndexConfig.builder("title", FieldIndexConfig.IndexType.FULLTEXT)
+ .analyzer(BuiltinAnalyzer.STANDARD)
+ .build());
+ configs.put(
+ "title.keyword",
+ FieldIndexConfig.builder("title.keyword", FieldIndexConfig.IndexType.KEYWORD)
+ .scalarType(ScalarFieldType.KEYWORD)
+ .build());
+ configs.put(
+ "price",
+ FieldIndexConfig.builder("price", FieldIndexConfig.IndexType.SCALAR)
+ .scalarType(ScalarFieldType.INT)
+ .build());
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeInt(0x45534D31); // ESM1
+ out.writeInt(1);
+ out.writeInt(configs.size());
+ for (Map.Entry<String, FieldIndexConfig> entry : configs.entrySet()) {
+ writeConfig(out, entry.getKey(), entry.getValue());
+ }
+ out.writeInt(1);
+ writeString(out, "segments_1");
+ out.writeLong(17L);
+ out.writeLong(23L);
+ out.flush();
+
+ ESIndexFileMeta.Parsed parsed = ESIndexFileMeta.read(bytes.toByteArray());
+ assertThat(parsed.indexedFieldNames()).containsExactly("title", "price");
+ assertThat(parsed.indexedFieldTypes()).isEmpty();
+ assertThat(parsed.fileOffsets().get("segments_1")).containsExactly(17L, 23L);
+ }
+
+ @Test
+ void versionOneVectorMetadataRejectsChangedFixedDimension() throws Exception {
+ FieldIndexConfig config =
+ FieldIndexConfig.builder("embedding", FieldIndexConfig.IndexType.VECTOR)
+ .algorithm(VectorAlgorithm.HNSW)
+ .dimension(2)
+ .metric("cosine")
+ .build();
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeInt(0x45534D31); // ESM1
+ out.writeInt(1);
+ out.writeInt(1);
+ writeConfig(out, "embedding", config);
+ out.writeInt(1);
+ writeString(out, "segments_1");
+ out.writeLong(0L);
+ out.writeLong(1L);
+ out.flush();
+
+ List<DataField> currentFields =
+ List.of(new DataField(0, "embedding", DataTypes.VECTOR(3, DataTypes.FLOAT())));
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ meta -> null,
+ List.of(
+ new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path("/tmp/legacy.index"),
+ 1L,
+ bytes.toByteArray())),
+ currentFields,
+ new ESIndexOptions(currentFields, new Options()));
+ try {
+ assertThatThrownBy(
+ () ->
+ reader.visitVectorSearch(
+ new VectorSearch(
+ new float[] {1.0f, 2.0f, 3.0f},
+ 1,
+ "embedding"))
+ .join())
+ .hasCauseInstanceOf(IllegalStateException.class)
+ .hasRootCauseMessage(
+ "The persisted es-index for field 'embedding' is incompatible with "
+ + "the current schema and cannot safely answer vector search. "
+ + "Rebuild the es-index after the schema change.");
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void rejectsDuplicateFilesInLegacyMetadata() throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ byte[] name = "segments_1".getBytes(StandardCharsets.UTF_8);
+ out.writeInt(2);
+ for (int i = 0; i < 2; i++) {
+ out.writeInt(name.length);
+ out.write(name);
+ out.writeLong(17L + i);
+ out.writeLong(23L);
+ }
+ out.flush();
+
+ assertThatThrownBy(() -> ESIndexFileMeta.read(bytes.toByteArray()))
+ .isInstanceOf(java.io.IOException.class)
+ .hasMessageContaining("Duplicate file offset")
+ .hasMessageContaining("segments_1");
+ }
+
+ @Test
+ void rejectsDuplicateAlgorithmParameters() throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeInt(0x45534D31); // ESM1
+ out.writeInt(1);
+ out.writeInt(1);
+ writeString(out, "embedding");
+ writeString(out, FieldIndexConfig.IndexType.VECTOR.name());
+ out.writeBoolean(true);
+ writeString(out, VectorAlgorithm.HNSW.name());
+ out.writeInt(4);
+ out.writeBoolean(true);
+ writeString(out, "cosine");
+ out.writeBoolean(false); // analyzer
+ out.writeBoolean(false); // scalar type
+ out.writeInt(2);
+ for (String value : new String[] {"16", "32"}) {
+ writeString(out, "m");
+ writeString(out, value);
+ }
+ out.writeInt(0); // files
+ out.flush();
+
+ assertThatThrownBy(() -> ESIndexFileMeta.read(bytes.toByteArray()))
+ .isInstanceOf(java.io.IOException.class)
+ .hasMessageContaining("Duplicate algorithm parameter")
+ .hasMessageContaining("embedding")
+ .hasMessageContaining("m");
+ }
+
+ @Test
+ void rejectsTrailingMetadataBytes() throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeInt(0);
+ out.writeByte(1);
+ out.flush();
+
+ assertThatThrownBy(() -> ESIndexFileMeta.read(bytes.toByteArray()))
+ .isInstanceOf(java.io.IOException.class)
+ .hasMessageContaining("Trailing bytes");
+ }
+
+ @Test
+ void readerRejectsFileRangesOutsideArchive() throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ writeString(out, "segments_1");
+ out.writeLong(4L);
+ out.writeLong(10L);
+ out.flush();
+ byte[] offsetEntry = bytes.toByteArray();
+ bytes.reset();
+ out = new DataOutputStream(bytes);
+ out.writeInt(1);
+ out.write(offsetEntry);
+ out.flush();
+
+ List<DataField> fields = List.of(new DataField(0, "id", DataTypes.INT()));
+ GlobalIndexIOMeta archive =
+ new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path("/tmp/invalid.index"),
+ 5L,
+ bytes.toByteArray());
+ assertThatThrownBy(
+ () ->
+ new ESIndexGlobalIndexReader(
+ meta -> null,
+ List.of(archive),
+ fields,
+ new ESIndexOptions(fields, new Options())))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid es-index metadata")
+ .hasRootCauseMessage(
+ "File range exceeds es-index archive size for segments_1: "
+ + "archiveSize=5, range=[4, 10]");
+ }
+
+ @Test
+ void readerRejectsMetadataWithoutLuceneFiles() {
+ List<DataField> fields = List.of(new DataField(0, "id", DataTypes.INT()));
+ GlobalIndexIOMeta archive =
+ new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path("/tmp/empty.index"),
+ 4L,
+ new byte[] {0, 0, 0, 0});
+
+ assertThatThrownBy(
+ () ->
+ new ESIndexGlobalIndexReader(
+ meta -> null,
+ List.of(archive),
+ fields,
+ new ESIndexOptions(fields, new Options())))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasRootCauseMessage("ES index metadata contains no Lucene files");
+ }
+
+ private static void writeString(DataOutputStream out, String value) throws Exception {
+ byte[] encoded = value.getBytes(StandardCharsets.UTF_8);
+ out.writeInt(encoded.length);
+ out.write(encoded);
+ }
+
+ private static void writeConfig(DataOutputStream out, String fieldName, FieldIndexConfig config)
+ throws Exception {
+ writeString(out, fieldName);
+ writeString(out, config.indexType().name());
+ writeNullableEnum(out, config.algorithm());
+ out.writeInt(config.dimension());
+ writeNullableString(out, config.metric());
+ writeNullableEnum(out, config.analyzer());
+ writeNullableEnum(out, config.scalarType());
+ out.writeInt(config.algorithmParams().size());
+ for (Map.Entry<String, String> parameter : config.algorithmParams().entrySet()) {
+ writeString(out, parameter.getKey());
+ writeString(out, parameter.getValue());
+ }
+ }
+
+ private static void writeNullableEnum(DataOutputStream out, Enum<?> value) throws Exception {
+ writeNullableString(out, value == null ? null : value.name());
+ }
+
+ private static void writeNullableString(DataOutputStream out, String value) throws Exception {
+ out.writeBoolean(value != null);
+ if (value != null) {
+ writeString(out, value);
+ }
+ }
+}
diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexFullTextQueryParserTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexFullTextQueryParserTest.java
new file mode 100644
index 0000000000..adb86557bc
--- /dev/null
+++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexFullTextQueryParserTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.elasticsearch.eslib.api.model.FullTextParams;
+import org.elasticsearch.eslib.api.model.FullTextQuerySpec;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ESIndexFullTextQueryParserTest {
+
+ @Test
+ void parsesCaseInsensitiveBooleanOperatorsAndRejectsTypos() {
+ FullTextQuerySpec.Match and =
+ (FullTextQuerySpec.Match)
+ ESIndexGlobalIndexReader.parseSpec(
+ "text", "{\"match\":{\"query\":\"a b\",\"operator\":\"AnD\"}}");
+ assertThat(and.params().operator()).isEqualTo(FullTextParams.Operator.AND);
+
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"match\":{\"query\":\"a b\",\"operator\":\"all\"}}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("AND or OR")
+ .hasMessageContaining("all");
+ }
+
+ @Test
+ void rejectsMalformedNumericAndBooleanParameters() {
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"match\":{\"query\":\"a\",\"max_expansions\":\"many\"}}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("max_expansions must be an integer");
+
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"bool\":{\"queries\":[[\"sometimes\",{\"match\":{\"query\":\"a\"}}]]}}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("MUST, SHOULD, or MUST_NOT");
+ }
+
+ @Test
+ void rejectsMissingBoostChildrenAndAmbiguousQueryTypes() {
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"boost\":{\"positive\":{\"match\":{\"query\":\"a\"}}}}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("negative");
+
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"match\":{\"query\":\"a\"},\"phrase\":{\"query\":\"a\"}}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("exactly one");
+ }
+
+ @Test
+ void rejectsUnknownFieldsAndDuplicateAliases() {
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"bool\":{\"mustt\":[{\"match\":{\"query\":\"a\"}}]}}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unknown field 'mustt'")
+ .hasMessageContaining("boolean query");
+
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"match\":{\"query\":\"a\",\"max_expansions\":10,\"maxExpansions\":20}}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("both 'max_expansions' and 'maxExpansions'");
+
+ assertThatThrownBy(
+ () ->
+ ESIndexGlobalIndexReader.parseSpec(
+ "text",
+ "{\"match\":{\"query\":\"a\"},\"unexpected\":true}"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unknown field 'unexpected'");
+ }
+
+ @Test
+ void rejectsExcessivelyNestedQueries() {
+ String query = "{\"match\":{\"query\":\"a\"}}";
+ for (int i = 0; i < 64; i++) {
+ query = "{\"bool\":{\"must\":[" + query + "]}}";
+ }
+ String deeplyNestedQuery = query;
+
+ assertThatThrownBy(() -> ESIndexGlobalIndexReader.parseSpec("text", deeplyNestedQuery))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maximum nesting depth");
+ }
+}
diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexE2ETest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexE2ETest.java
new file mode 100644
index 0000000000..27ec952689
--- /dev/null
+++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexE2ETest.java
@@ -0,0 +1,2000 @@
+/*
+ * 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.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.columnar.ColumnarArray;
+import org.apache.paimon.data.columnar.heap.HeapFloatVector;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.SeekableInputStreamWrapper;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.OffsetGlobalIndexReader;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.globalindex.UnionGlobalIndexReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.options.Options;
+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.types.DataTypes;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * End-to-end test of the P5 ↔ P4 integration: {@link ESIndexGlobalIndexWriter} feeds Paimon rows to
+ * {@code DefaultESIndexBuilder} (loaded reflectively), packs the Lucene segment files into an
+ * archive, and {@link ESIndexGlobalIndexReader} reads that archive back through {@code
+ * DefaultESIndexSearcher} (also reflective) to serve vector and full-text queries.
+ *
+ * <p>This exercises the real archive write/offset-table/read roundtrip and the eslib-core
+ * reflection loading, without Flink or OSS.
+ */
+class ESIndexGlobalIndexE2ETest {
+
+ /** {@link GlobalIndexFileWriter} backed by a local directory. */
+ private static final class LocalDirWriter implements GlobalIndexFileWriter {
+ private final java.nio.file.Path dir;
+ private final LocalFileIO fio = LocalFileIO.create();
+
+ LocalDirWriter(java.nio.file.Path dir) {
+ this.dir = dir;
+ }
+
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + "-" + UUID.randomUUID() + ".index";
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) throws IOException {
+ return fio.newOutputStream(
+ new org.apache.paimon.fs.Path(dir.resolve(fileName).toString()), true);
+ }
+ }
+
+ /** {@link GlobalIndexFileReader} backed by local files. */
+ private static final class LocalFileReader implements GlobalIndexFileReader {
+ private final LocalFileIO fio = LocalFileIO.create();
+
+ @Override
+ public SeekableInputStream getInputStream(GlobalIndexIOMeta meta) throws IOException {
+ return fio.newInputStream(meta.filePath());
+ }
+ }
+
+ /** Blocks the first lazy-load stream open and counts every stream's lifecycle. */
+ private static final class BlockingCountingFileReader implements GlobalIndexFileReader {
+ private final LocalFileIO fio = LocalFileIO.create();
+ private final CountDownLatch firstOpenStarted = new CountDownLatch(1);
+ private final CountDownLatch allowOpen = new CountDownLatch(1);
+ private final AtomicInteger openCount = new AtomicInteger();
+ private final AtomicInteger closeCount = new AtomicInteger();
+
+ @Override
+ public SeekableInputStream getInputStream(GlobalIndexIOMeta meta) throws IOException {
+ firstOpenStarted.countDown();
+ try {
+ allowOpen.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while waiting to open index stream", e);
+ }
+
+ openCount.incrementAndGet();
+ return new SeekableInputStreamWrapper(fio.newInputStream(meta.filePath())) {
+ private final AtomicBoolean streamClosed = new AtomicBoolean();
+
+ @Override
+ public void close() throws IOException {
+ if (streamClosed.compareAndSet(false, true)) {
+ try {
+ super.close();
+ } finally {
+ closeCount.incrementAndGet();
+ }
+ }
+ }
+ };
+ }
+
+ boolean awaitFirstOpen() throws InterruptedException {
+ return firstOpenStarted.await(10, TimeUnit.SECONDS);
+ }
+
+ void releaseOpen() {
+ allowOpen.countDown();
+ }
+
+ int openCount() {
+ return openCount.get();
+ }
+
+ int closeCount() {
+ return closeCount.get();
+ }
+ }
+
+ /** Can make stream-close attempts fail until the test explicitly allows retry cleanup. */
+ private static final class TransientCloseFailureFileReader implements GlobalIndexFileReader {
+ private final LocalFileIO fio = LocalFileIO.create();
+ private final AtomicBoolean failCloses = new AtomicBoolean();
+
+ @Override
+ public SeekableInputStream getInputStream(GlobalIndexIOMeta meta) throws IOException {
+ return new SeekableInputStreamWrapper(fio.newInputStream(meta.filePath())) {
+ private final AtomicBoolean streamClosed = new AtomicBoolean();
+
+ @Override
+ public void close() throws IOException {
+ if (streamClosed.get()) {
+ return;
+ }
+ if (failCloses.get()) {
+ throw new IOException("transient close failure");
+ }
+ if (streamClosed.compareAndSet(false, true)) {
+ super.close();
+ }
+ }
+ };
+ }
+
+ void setCloseFailures(boolean fail) {
+ failCloses.set(fail);
+ }
+ }
+
+ @Test
+ void closeWithoutFinishReleasesBuilderAndPreventsFurtherUse(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields = List.of(new DataField(0, "title", DataTypes.STRING()));
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(
+ new LocalDirWriter(tmp), fields, new ESIndexOptions(fields, new Options()));
+
+ writer.write(BinaryString.fromString("uncommitted"), 0);
+ writer.close();
+ writer.close();
+
+ assertThrows(IllegalStateException.class, writer::finish);
+ assertThrows(
+ IllegalStateException.class,
+ () -> writer.write(BinaryString.fromString("late"), 1));
+ try (java.util.stream.Stream<java.nio.file.Path> filesInArchive = Files.list(tmp)) {
+ assertEquals(0L, filesInArchive.count(), "close must not publish a partial archive");
+ }
+ }
+
+ @Test
+ void failedWriteCannotPublishPartialIndex(@TempDir java.nio.file.Path tmp) throws IOException {
+ List<DataField> fields =
+ List.of(
+ new DataField(
+ 0, "embedding", DataTypes.ARRAY(DataTypes.FLOAT().notNull())));
+ Map<String, String> opt = new HashMap<>();
+ opt.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ opt.put("global-index.es-index.fields.embedding.dimension", "2");
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(
+ new LocalDirWriter(tmp),
+ fields,
+ new ESIndexOptions(fields, Options.fromMap(opt)));
+
+ writer.write(new float[] {1.0f, 2.0f}, 0);
+ IllegalArgumentException writeFailure =
+ assertThrows(
+ IllegalArgumentException.class, () -> writer.write(new float[] {3.0f}, 1));
+ IllegalStateException retryFailure =
+ assertThrows(
+ IllegalStateException.class,
+ () -> writer.write(new float[] {4.0f, 5.0f}, 2));
+ assertEquals(writeFailure, retryFailure.getCause());
+
+ IllegalStateException finishFailure =
+ assertThrows(IllegalStateException.class, writer::finish);
+ assertEquals(writeFailure, finishFailure.getCause());
+ try (java.util.stream.Stream<java.nio.file.Path> filesInArchive = Files.list(tmp)) {
+ assertEquals(
+ 0L,
+ filesInArchive.count(),
+ "finish must not publish an index after a partial document write");
+ }
+ }
+
+ @Test
+ void vectorNullElementsAreRejectedBeforeIndexing(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields =
+ List.of(new DataField(0, "embedding", DataTypes.ARRAY(DataTypes.FLOAT())));
+ Map<String, String> opt = new HashMap<>();
+ opt.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ opt.put("global-index.es-index.fields.embedding.dimension", "2");
+
+ HeapFloatVector elements = new HeapFloatVector(2);
+ elements.setFloat(0, 1.0f);
+ elements.setNullAt(1);
+ ColumnarArray vector = new ColumnarArray(elements, 0, 2);
+
+ try (ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(
+ new LocalDirWriter(tmp),
+ fields,
+ new ESIndexOptions(fields, Options.fromMap(opt)))) {
+ IllegalArgumentException failure =
+ assertThrows(IllegalArgumentException.class, () -> writer.write(vector, 0));
+ assertTrue(failure.getMessage().contains("must not contain null elements"));
+ }
+ }
+
+ @Test
+ void writeArchiveThenReadAndSearch(@TempDir java.nio.file.Path tmp) throws IOException {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "embedding", DataTypes.ARRAY(DataTypes.FLOAT())),
+ new DataField(1, "title", DataTypes.STRING()),
+ new DataField(2, "category", DataTypes.STRING()),
+ new DataField(3, "price", DataTypes.INT()));
+
+ Map<String, String> opt = new HashMap<>();
+ opt.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ opt.put("global-index.es-index.fields.embedding.dimension", "4");
+ opt.put("global-index.es-index.fields.embedding.metric", "l2");
+ opt.put("global-index.es-index.fields.title.analyzer", "standard");
+ // category uses KEYWORD as its primary field; ESIndexOptions adds category.fulltext so the
+ // same logical field can also serve analyzed search.
+ opt.put("global-index.es-index.fields.category.type", "keyword");
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(opt));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive");
+ Files.createDirectories(archiveDir);
+
+ // --- write 20 rows → archive ---
+ LocalDirWriter fileWriter = new LocalDirWriter(archiveDir);
+ ESIndexGlobalIndexWriter writer = new ESIndexGlobalIndexWriter(fileWriter, fields, options);
+ for (int i = 0; i < 20; i++) {
+ GenericRow row =
+ GenericRow.of(
+ new GenericArray(new float[] {i, i, i, i}),
+ BinaryString.fromString(
+ "document " + i + (i % 2 == 0 ? " even" : " odd")),
+ BinaryString.fromString(i % 2 == 0 ? "even" : "odd"),
+ i * 10);
+ writer.write(i, row);
+ }
+ List<ResultEntry> entries = writer.finish();
+ assertEquals(1, entries.size(), "one archive produced");
+ ResultEntry entry = entries.get(0);
+ assertEquals(20L, entry.rowCount(), "row count");
+ assertNotNull(entry.meta(), "offset-table meta present");
+ assertThrows(IllegalStateException.class, writer::finish);
+ assertThrows(IllegalStateException.class, () -> writer.write(new float[] {1, 1, 1, 1}, 20));
+
+ // --- reconstruct reader over the archive ---
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ long fileSize = Files.size(archiveDir.resolve(entry.fileName()));
+ GlobalIndexIOMeta ioMeta = new GlobalIndexIOMeta(filePath, fileSize, entry.meta());
+
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+
+ // --- vector search: nearest to origin is row 0 (vector [0,0,0,0]) ---
+ Optional<ScoredGlobalIndexResult> vr =
+ reader.visitVectorSearch(new VectorSearch(new float[] {0, 0, 0, 0}, 5, "embedding"))
+ .join();
+ assertTrue(vr.isPresent(), "vector search returns a result");
+ RoaringNavigableMap64 vrows = vr.get().results();
+ assertEquals(5, vrows.getIntCardinality(), "k=5 results");
+ assertTrue(contains(vrows, 0L), "row 0 (origin) recalled by HNSW");
+ Optional<ScoredGlobalIndexResult> overFetched =
+ reader.visitVectorSearch(
+ new VectorSearch(
+ new float[] {0, 0, 0, 0},
+ 3,
+ "embedding",
+ Map.of("hnsw.num_candidates", "10")))
+ .join();
+ assertTrue(overFetched.isPresent());
+ assertEquals(
+ 3,
+ overFetched.get().results().getIntCardinality(),
+ "num_candidates must not change the requested result limit");
+ CompletionException invalidDimension =
+ assertThrows(
+ CompletionException.class,
+ () ->
+ reader.visitVectorSearch(
+ new VectorSearch(
+ new float[] {0, 0, 0}, 5, "embedding"))
+ .join(),
+ "query vectors must match the persisted build dimension");
+ assertTrue(invalidDimension.getCause() instanceof IllegalArgumentException);
+
+ // --- full-text search: 10 rows contain "even" ---
+ Optional<ScoredGlobalIndexResult> ft =
+ reader.visitFullTextSearch(new FullTextSearch("title", matchQuery("even"), 50))
+ .join();
+ assertTrue(ft.isPresent(), "full-text search returns a result");
+ assertEquals(10, ft.get().results().getIntCardinality(), "10 even docs");
+
+ // --- Phrase query honours token order: titles are "document {i} even/odd". "document 0"
+ // is a consecutive phrase only in row 0; "even document" never occurs in that order. ---
+ Optional<ScoredGlobalIndexResult> phraseHit =
+ reader.visitFullTextSearch(
+ new FullTextSearch("title", phraseQuery("document 0"), 50))
+ .join();
+ assertTrue(phraseHit.isPresent(), "phrase 'document 0' matches");
+ assertEquals(
+ 1, phraseHit.get().results().getIntCardinality(), "only row 0 has 'document 0'");
+ assertTrue(contains(phraseHit.get().results(), 0L), "phrase 'document 0' is row 0");
+ assertTrue(
+ reader.visitFullTextSearch(
+ new FullTextSearch("title", phraseQuery("even document"), 50))
+ .join()
+ .isEmpty(),
+ "phrase 'even document' never occurs in that order");
+
+ // --- operator is honoured: every doc has "document", only 10 have "even" ---
+ // default OR("even document") matches all 20; AND("even document") matches only the 10
+ // even docs that contain both tokens.
+ Optional<ScoredGlobalIndexResult> orFt =
+ reader.visitFullTextSearch(
+ new FullTextSearch("title", matchQuery("even document"), 50))
+ .join();
+ assertEquals(
+ 20,
+ orFt.get().results().getIntCardinality(),
+ "OR match on 'even document' hits every doc (all contain 'document')");
+ Optional<ScoredGlobalIndexResult> andFt =
+ reader.visitFullTextSearch(
+ new FullTextSearch("title", matchQuery("even document", "and"), 50))
+ .join();
+ assertEquals(
+ 10,
+ andFt.get().results().getIntCardinality(),
+ "AND match on 'even document' hits only docs with both tokens");
+
+ // --- fuzziness is honoured: "evon" is one edit from "even" ---
+ Optional<ScoredGlobalIndexResult> exactTypo =
+ reader.visitFullTextSearch(new FullTextSearch("title", matchQuery("evon"), 50))
+ .join();
+ assertTrue(exactTypo.isEmpty(), "exact 'evon' matches nothing");
+ Optional<ScoredGlobalIndexResult> fuzzyTypo =
+ reader.visitFullTextSearch(
+ new FullTextSearch("title", fuzzyMatchQuery("evon", 1), 50))
+ .join();
+ assertEquals(
+ 10,
+ fuzzyTypo.get().results().getIntCardinality(),
+ "fuzzy 'evon'~1 matches the 10 'even' docs");
+
+ // --- category is KEYWORD-primary, but full-text search routes to category.fulltext. ---
+ Optional<ScoredGlobalIndexResult> categoryFt =
+ reader.visitFullTextSearch(new FullTextSearch("category", matchQuery("even"), 50))
+ .join();
+ assertTrue(categoryFt.isPresent(), "KEYWORD field has a FULLTEXT multi-field");
+ assertEquals(
+ 10,
+ categoryFt.get().results().getIntCardinality(),
+ "category.fulltext matches 10 even rows");
+
+ // --- multi-field: title is FULLTEXT, so ordinary predicates route to the keyword
+ // sub-field title.keyword (written by default) and evaluate exactly on the raw value ---
+ FieldRef titleRef = new FieldRef(1, "title", DataTypes.STRING());
+ Optional<GlobalIndexResult> titleEq = reader.visitEqual(titleRef, "document 0 even").join();
+ assertTrue(titleEq.isPresent(), "= on FULLTEXT routes to keyword sub-field");
+ assertEquals(
+ 1, titleEq.get().results().getIntCardinality(), "exact title matches only row 0");
+ assertTrue(contains(titleEq.get().results(), 0L), "exact title 'document 0 even' is row 0");
+ assertEquals(
+ 20,
+ reader.visitLike(titleRef, "doc%").join().get().results().getIntCardinality(),
+ "LIKE 'doc%' on title.keyword matches all 20 'document ...' titles");
+ // A range predicate on a string keyword sub-field is unsupported. Return the complete
+ // shard as a conservative result so a union with differently configured shards cannot
+ // drop rows; Paimon's residual predicate performs the actual comparison.
+ Optional<GlobalIndexResult> unsupportedTitleRange =
+ reader.visitGreaterThan(titleRef, "abc").join();
+ assertTrue(unsupportedTitleRange.isPresent());
+ assertEquals(20, unsupportedTitleRange.get().results().getIntCardinality());
+
+ // --- scalar filter: price >= 100 means rows 10..19 (price = i*10) ---
+ FieldRef priceRef = new FieldRef(3, "price", DataTypes.INT());
+ Optional<GlobalIndexResult> sf = reader.visitGreaterOrEqual(priceRef, 100).join();
+ assertTrue(sf.isPresent(), "scalar filter returns a result");
+ assertEquals(10, sf.get().results().getIntCardinality(), "10 rows with price >= 100");
+
+ // --- scalar filter: price == 50 → row 5 only ---
+ Optional<GlobalIndexResult> eq = reader.visitEqual(priceRef, 50).join();
+ assertTrue(eq.isPresent(), "scalar eq filter returns a result");
+ assertEquals(1, eq.get().results().getIntCardinality(), "1 row with price == 50");
+ assertTrue(contains(eq.get().results(), 5L), "row 5 has price=50");
+
+ // --- scalar negative filters are implemented in the reader as exists AND NOT(eq/in). ---
+ Optional<GlobalIndexResult> ne = reader.visitNotEqual(priceRef, 50).join();
+ assertTrue(ne.isPresent(), "scalar <> filter returns a result");
+ assertEquals(19, ne.get().results().getIntCardinality(), "all rows except price=50");
+ assertTrue(!contains(ne.get().results(), 5L), "row 5 must be excluded by price<>50");
+
+ Optional<GlobalIndexResult> notIn =
+ reader.visitNotIn(priceRef, Arrays.asList(50, 60)).join();
+ assertTrue(notIn.isPresent(), "scalar NOT IN filter returns a result");
+ assertEquals(
+ 18, notIn.get().results().getIntCardinality(), "all rows except price in (50, 60)");
+ assertTrue(!contains(notIn.get().results(), 5L) && !contains(notIn.get().results(), 6L));
+
+ // --- keyword filter: category == "even" → 10 rows ---
+ FieldRef categoryRef = new FieldRef(2, "category", DataTypes.STRING());
+ Optional<GlobalIndexResult> kw = reader.visitEqual(categoryRef, "even").join();
+ assertTrue(kw.isPresent(), "keyword eq filter returns a result");
+ assertEquals(10, kw.get().results().getIntCardinality(), "10 even category rows");
+
+ // --- keyword prefix: category startsWith "ev" → 10 rows with "even" ---
+ Optional<GlobalIndexResult> sw = reader.visitStartsWith(categoryRef, "ev").join();
+ assertTrue(sw.isPresent(), "startsWith filter returns a result");
+ assertEquals(10, sw.get().results().getIntCardinality(), "10 rows start with 'ev'");
+
+ // --- keyword contains: category contains "dd" → 10 rows with "odd" ---
+ Optional<GlobalIndexResult> ct = reader.visitContains(categoryRef, "dd").join();
+ assertTrue(ct.isPresent(), "contains filter returns a result");
+ assertEquals(10, ct.get().results().getIntCardinality(), "10 rows contain 'dd'");
+
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ void readsIndexAfterIndexedColumnsAreRenamed(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> buildFields =
+ Arrays.asList(
+ new DataField(0, "embedding", DataTypes.ARRAY(DataTypes.FLOAT())),
+ new DataField(1, "title", DataTypes.STRING()),
+ new DataField(2, "category", DataTypes.STRING()),
+ new DataField(3, "price", DataTypes.INT()));
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ optionMap.put("global-index.es-index.fields.embedding.dimension", "2");
+ optionMap.put("global-index.es-index.fields.embedding.metric", "l2");
+ optionMap.put("global-index.es-index.fields.category.type", "keyword");
+ ESIndexOptions buildOptions = new ESIndexOptions(buildFields, Options.fromMap(optionMap));
+
+ java.nio.file.Path archiveDir = tmp.resolve("renamed-fields");
+ Files.createDirectories(archiveDir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(
+ new LocalDirWriter(archiveDir), buildFields, buildOptions);
+ writer.write(
+ 0,
+ GenericRow.of(
+ new GenericArray(new float[] {0, 0}),
+ BinaryString.fromString("Apache Paimon"),
+ BinaryString.fromString("docs"),
+ 10));
+ writer.write(
+ 1,
+ GenericRow.of(
+ new GenericArray(new float[] {10, 10}),
+ BinaryString.fromString("Other engine"),
+ BinaryString.fromString("other"),
+ 20));
+ ResultEntry entry = writer.finish().get(0);
+
+ List<DataField> currentFields =
+ Arrays.asList(
+ new DataField(0, "vector", DataTypes.ARRAY(DataTypes.FLOAT())),
+ new DataField(1, "headline", DataTypes.STRING()),
+ new DataField(2, "kind", DataTypes.STRING()),
+ new DataField(3, "cost", DataTypes.INT()));
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(),
+ List.of(ioMeta(archiveDir, entry)),
+ currentFields,
+ buildOptions);
+ try {
+ Optional<ScoredGlobalIndexResult> vector =
+ reader.visitVectorSearch(new VectorSearch(new float[] {0, 0}, 1, "vector"))
+ .join();
+ assertTrue(vector.isPresent());
+ assertTrue(contains(vector.get().results(), 0L));
+
+ Optional<ScoredGlobalIndexResult> fullText =
+ reader.visitFullTextSearch(
+ new FullTextSearch("headline", matchQuery("paimon"), 10))
+ .join();
+ assertTrue(fullText.isPresent());
+ assertTrue(contains(fullText.get().results(), 0L));
+
+ Optional<GlobalIndexResult> keyword =
+ reader.visitEqual(new FieldRef(2, "kind", DataTypes.STRING()), "docs").join();
+ assertTrue(keyword.isPresent());
+ assertTrue(contains(keyword.get().results(), 0L));
+
+ Optional<GlobalIndexResult> scalar =
+ reader.visitEqual(new FieldRef(3, "cost", DataTypes.INT()), 20).join();
+ assertTrue(scalar.isPresent());
+ assertTrue(contains(scalar.get().results(), 1L));
+ } finally {
+ reader.close();
+ }
+
+ List<DataField> widenedFields =
+ Arrays.asList(
+ currentFields.get(0),
+ currentFields.get(1),
+ currentFields.get(2),
+ new DataField(3, "cost", DataTypes.BIGINT()));
+ ESIndexGlobalIndexReader widenedReader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(),
+ List.of(ioMeta(archiveDir, entry)),
+ widenedFields,
+ buildOptions);
+ try {
+ Optional<GlobalIndexResult> widenedFallback =
+ widenedReader
+ .visitLessThan(
+ new FieldRef(3, "cost", DataTypes.BIGINT()), 3_000_000_000L)
+ .join();
+ assertTrue(widenedFallback.isPresent());
+ assertEquals(
+ 2,
+ widenedFallback.get().results().getIntCardinality(),
+ "a widened field must keep every old INT-indexed row");
+ } finally {
+ widenedReader.close();
+ }
+
+ List<DataField> resizedVectorFields =
+ Arrays.asList(
+ new DataField(0, "vector", DataTypes.VECTOR(3, DataTypes.FLOAT())),
+ currentFields.get(1),
+ currentFields.get(2),
+ currentFields.get(3));
+ ESIndexGlobalIndexReader resizedVectorReader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(),
+ List.of(ioMeta(archiveDir, entry)),
+ resizedVectorFields,
+ buildOptions);
+ try {
+ CompletionException failure =
+ assertThrows(
+ CompletionException.class,
+ () ->
+ resizedVectorReader
+ .visitVectorSearch(
+ new VectorSearch(
+ new float[] {0, 0, 0}, 1, "vector"))
+ .join());
+ assertTrue(failure.getCause() instanceof IllegalStateException);
+ assertTrue(failure.getCause().getMessage().contains("Rebuild the es-index"));
+ } finally {
+ resizedVectorReader.close();
+ }
+ }
+
+ @Test
+ void mixedSchemaShardsDoNotDropRowsFromAnOlderScalarType(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> oldFields = List.of(new DataField(0, "old_value", DataTypes.INT()));
+ ESIndexOptions oldOptions = new ESIndexOptions(oldFields, new Options());
+ java.nio.file.Path oldDir = tmp.resolve("old-int");
+ Files.createDirectories(oldDir);
+ ESIndexGlobalIndexWriter oldWriter =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(oldDir), oldFields, oldOptions);
+ oldWriter.write(1, 0);
+ oldWriter.write(100, 1);
+ ResultEntry oldEntry = oldWriter.finish().get(0);
+
+ List<DataField> currentFields = List.of(new DataField(0, "value", DataTypes.BIGINT()));
+ ESIndexOptions currentOptions = new ESIndexOptions(currentFields, new Options());
+ java.nio.file.Path currentDir = tmp.resolve("current-bigint");
+ Files.createDirectories(currentDir);
+ ESIndexGlobalIndexWriter currentWriter =
+ new ESIndexGlobalIndexWriter(
+ new LocalDirWriter(currentDir), currentFields, currentOptions);
+ currentWriter.write(3_000_000_000L, 0);
+ currentWriter.write(5L, 1);
+ ResultEntry currentEntry = currentWriter.finish().get(0);
+
+ ESIndexGlobalIndexReader oldReader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(),
+ List.of(ioMeta(oldDir, oldEntry)),
+ currentFields,
+ currentOptions);
+ ESIndexGlobalIndexReader currentReader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(),
+ List.of(ioMeta(currentDir, currentEntry)),
+ currentFields,
+ currentOptions);
+ UnionGlobalIndexReader union =
+ new UnionGlobalIndexReader(
+ Arrays.asList(
+ new OffsetGlobalIndexReader(oldReader, 0, 1),
+ new OffsetGlobalIndexReader(currentReader, 2, 3)));
+ try {
+ Optional<GlobalIndexResult> candidates =
+ union.visitLessThan(
+ new FieldRef(0, "value", DataTypes.BIGINT()), 3_000_000_000L)
+ .join();
+ assertTrue(candidates.isPresent());
+ assertEquals(3, candidates.get().results().getIntCardinality());
+ assertTrue(contains(candidates.get().results(), 0L));
+ assertTrue(contains(candidates.get().results(), 1L));
+ assertTrue(contains(candidates.get().results(), 3L));
+ } finally {
+ union.close();
+ }
+ }
+
+ @Test
+ void textMultiFieldServesBothMatchAndExact(@TempDir java.nio.file.Path tmp) throws IOException {
+ // Whichever text representation is primary, the complementary multi-field is written so
+ // both analyzed search and exact predicates are always available.
+ List<DataField> fields = Arrays.asList(new DataField(0, "t", DataTypes.STRING()));
+ String[] docs = {"Apache Paimon", "vector search", "Apache Paimon"};
+
+ for (String primaryType : new String[] {"fulltext", "keyword"}) {
+ Map<String, String> opt = new HashMap<>();
+ opt.put("global-index.es-index.fields.t.type", primaryType);
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(opt));
+
+ java.nio.file.Path dir = tmp.resolve("mf-" + primaryType);
+ Files.createDirectories(dir);
+ LocalDirWriter fw = new LocalDirWriter(dir);
+ ESIndexGlobalIndexWriter w = new ESIndexGlobalIndexWriter(fw, fields, options);
+ for (int i = 0; i < docs.length; i++) {
+ w.write(BinaryString.fromString(docs[i]), i);
+ }
+ ResultEntry entry = w.finish().get(0);
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path(dir.resolve(entry.fileName()).toString()),
+ Files.size(dir.resolve(entry.fileName())),
+ entry.meta());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+ FieldRef tRef = new FieldRef(0, "t", DataTypes.STRING());
+
+ // FULLTEXT-primary searches t; KEYWORD-primary searches t.fulltext.
+ Optional<ScoredGlobalIndexResult> m =
+ reader.visitFullTextSearch(new FullTextSearch("t", matchQuery("paimon"), 50))
+ .join();
+ assertTrue(m.isPresent(), "match works for " + primaryType + " primary");
+ assertEquals(2, m.get().results().getIntCardinality(), "two docs contain 'paimon'");
+
+ // FULLTEXT-primary routes exact matching to t.keyword; KEYWORD-primary uses t.
+ Optional<GlobalIndexResult> eq = reader.visitEqual(tRef, "Apache Paimon").join();
+ assertTrue(eq.isPresent(), "exact = works for " + primaryType + " primary");
+ assertEquals(
+ 2, eq.get().results().getIntCardinality(), "two rows equal 'Apache Paimon'");
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Test
+ void fullTextSearchHonorsIncludeRowIdsBeforeTopK(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields = Arrays.asList(new DataField(0, "t", DataTypes.STRING()));
+ ESIndexOptions options = new ESIndexOptions(fields, new Options());
+ java.nio.file.Path dir = tmp.resolve("fulltext-live-rows");
+ Files.createDirectories(dir);
+
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(dir), fields, options);
+ writer.write(BinaryString.fromString("paimon paimon paimon paimon"), 0);
+ writer.write(BinaryString.fromString("paimon paimon"), 1);
+ writer.write(BinaryString.fromString("paimon"), 2);
+ ResultEntry entry = writer.finish().get(0);
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path(dir.resolve(entry.fileName()).toString()),
+ Files.size(dir.resolve(entry.fileName())),
+ entry.meta());
+
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+ try {
+ FullTextSearch unfiltered = new FullTextSearch("t", matchQuery("paimon"), 1);
+ Optional<ScoredGlobalIndexResult> top = reader.visitFullTextSearch(unfiltered).join();
+ assertTrue(top.isPresent());
+ assertTrue(contains(top.get().results(), 0L), "row 0 is the unfiltered top hit");
+
+ RoaringNavigableMap64 includeRowIds = new RoaringNavigableMap64();
+ includeRowIds.add(2L);
+ FullTextSearch filtered =
+ new FullTextSearch("t", matchQuery("paimon"), 1)
+ .withIncludeRowIds(includeRowIds);
+ Optional<ScoredGlobalIndexResult> result = reader.visitFullTextSearch(filtered).join();
+ assertTrue(result.isPresent(), "the included lower-ranked row must still be found");
+ assertEquals(1, result.get().results().getIntCardinality());
+ assertTrue(
+ contains(result.get().results(), 2L),
+ "includeRowIds must be applied before selecting the effective top-K");
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void nullLiteralsFollowSqlThreeValuedLogicAndKeywordRangeFallsBack(
+ @TempDir java.nio.file.Path tmp) throws IOException {
+ List<DataField> fields = Arrays.asList(new DataField(0, "k", DataTypes.STRING()));
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put("global-index.es-index.fields.k.type", "keyword");
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(optionMap));
+ java.nio.file.Path dir = tmp.resolve("null-literal-semantics");
+ Files.createDirectories(dir);
+
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(dir), fields, options);
+ writer.write(BinaryString.fromString(""), 0);
+ writer.write(BinaryString.fromString("a"), 1);
+ writer.write(null, 2);
+ writer.write(BinaryString.fromString("b"), 3);
+ ResultEntry entry = writer.finish().get(0);
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path(dir.resolve(entry.fileName()).toString()),
+ Files.size(dir.resolve(entry.fileName())),
+ entry.meta());
+
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+ FieldRef kRef = new FieldRef(0, "k", DataTypes.STRING());
+ try {
+ Optional<GlobalIndexResult> equalNull = reader.visitEqual(kRef, null).join();
+ assertTrue(equalNull.isPresent());
+ assertTrue(
+ equalNull.get().results().isEmpty(),
+ "k = NULL must not match the real empty-string row");
+
+ Optional<GlobalIndexResult> notEqualNull = reader.visitNotEqual(kRef, null).join();
+ assertTrue(notEqualNull.isPresent());
+ assertTrue(notEqualNull.get().results().isEmpty(), "k <> NULL is UNKNOWN for all rows");
+
+ Optional<GlobalIndexResult> in = reader.visitIn(kRef, Arrays.asList(null, "a")).join();
+ assertTrue(in.isPresent());
+ assertEquals(1, in.get().results().getIntCardinality());
+ assertTrue(contains(in.get().results(), 1L), "NULL is ignored in a positive IN list");
+
+ Optional<GlobalIndexResult> onlyNullIn =
+ reader.visitIn(kRef, Arrays.asList(null, null)).join();
+ assertTrue(onlyNullIn.isPresent());
+ assertTrue(onlyNullIn.get().results().isEmpty());
+
+ Optional<GlobalIndexResult> notInWithNull =
+ reader.visitNotIn(kRef, Arrays.asList("a", null)).join();
+ assertTrue(notInWithNull.isPresent());
+ assertTrue(
+ notInWithNull.get().results().isEmpty(),
+ "NOT IN containing NULL is UNKNOWN for every row");
+
+ Optional<GlobalIndexResult> nullRange = reader.visitGreaterThan(kRef, null).join();
+ assertTrue(nullRange.isPresent());
+ assertTrue(nullRange.get().results().isEmpty());
+
+ Optional<GlobalIndexResult> keywordRange = reader.visitGreaterThan(kRef, "a").join();
+ assertTrue(keywordRange.isPresent());
+ assertEquals(
+ 4,
+ keywordRange.get().results().getIntCardinality(),
+ "unsupported KEYWORD ordering must conservatively retain the whole shard");
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void arrayLongLabelsRoundTripAsMultiValueScalar(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.BIGINT()),
+ new DataField(1, "labels", DataTypes.ARRAY(DataTypes.BIGINT())));
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(new HashMap<>()));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-labels");
+ Files.createDirectories(archiveDir);
+
+ LocalDirWriter fileWriter = new LocalDirWriter(archiveDir);
+ ESIndexGlobalIndexWriter writer = new ESIndexGlobalIndexWriter(fileWriter, fields, options);
+ writer.write(0, GenericRow.of(100L, new GenericArray(new long[] {3L, 7L})));
+ writer.write(1, GenericRow.of(101L, new GenericArray(new long[] {5L})));
+ writer.write(2, GenericRow.of(102L, new GenericArray(new long[] {7L, 11L})));
+
+ List<ResultEntry> entries = writer.finish();
+ assertEquals(1, entries.size(), "one archive produced");
+ ResultEntry entry = entries.get(0);
+ assertEquals(3L, entry.rowCount(), "row count");
+
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ long fileSize = Files.size(archiveDir.resolve(entry.fileName()));
+ GlobalIndexIOMeta ioMeta = new GlobalIndexIOMeta(filePath, fileSize, entry.meta());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+
+ FieldRef labelsRef = new FieldRef(1, "labels", DataTypes.ARRAY(DataTypes.BIGINT()));
+
+ Optional<GlobalIndexResult> hit7 = reader.visitEqual(labelsRef, 7L).join();
+ assertTrue(hit7.isPresent(), "labels term filter returns a result");
+ RoaringNavigableMap64 rows7 = hit7.get().results();
+ assertEquals(2, rows7.getIntCardinality(), "two rows contain label 7");
+ assertTrue(contains(rows7, 0L), "row 0 contains label 7");
+ assertTrue(contains(rows7, 2L), "row 2 contains label 7");
+
+ Optional<GlobalIndexResult> hitIn =
+ reader.visitIn(labelsRef, Arrays.asList(5L, 11L)).join();
+ assertTrue(hitIn.isPresent(), "labels IN filter returns a result");
+ RoaringNavigableMap64 rowsIn = hitIn.get().results();
+ assertEquals(2, rowsIn.getIntCardinality(), "two rows contain labels in (5, 11)");
+ assertTrue(contains(rowsIn, 1L), "row 1 contains label 5");
+ assertTrue(contains(rowsIn, 2L), "row 2 contains label 11");
+
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ void nullValuesKeepDocIdRowIdAligned(@TempDir java.nio.file.Path tmp) throws IOException {
+ // Single-column keyword index where row 3's value is NULL. The writer skips null and
+ // registers
+ // an empty doc for that slot (addNullDoc), with flushPendingDocs as the back-stop; both
+ // stamp
+ // _ROW_ID=docId so the null gap is occupied. This guards docId<->rowId alignment: an exact
+ // term lookup must resolve to the row's OWN id. A regression in padding / _ROW_ID stamping
+ // would shift every row after the gap by one (e.g. "row7" would come back as rowId 6).
+ List<DataField> fields = Arrays.asList(new DataField(0, "k", DataTypes.STRING()));
+ // No analyzer on "k" -> KEYWORD (exact-term) index.
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(new HashMap<>()));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-null");
+ Files.createDirectories(archiveDir);
+ LocalDirWriter fileWriter = new LocalDirWriter(archiveDir);
+ ESIndexGlobalIndexWriter writer = new ESIndexGlobalIndexWriter(fileWriter, fields, options);
+
+ int n = 10;
+ int nullRow = 3;
+ for (int i = 0; i < n; i++) {
+ writer.write(i == nullRow ? null : BinaryString.fromString("row" + i), i);
+ }
+ List<ResultEntry> entries = writer.finish();
+ assertEquals(1, entries.size(), "one archive produced");
+ ResultEntry entry = entries.get(0);
+ assertEquals((long) n, entry.rowCount(), "row count includes the null row");
+
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ long fileSize = Files.size(archiveDir.resolve(entry.fileName()));
+ GlobalIndexIOMeta ioMeta = new GlobalIndexIOMeta(filePath, fileSize, entry.meta());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+
+ FieldRef kRef = new FieldRef(0, "k", DataTypes.STRING());
+ for (int i = 0; i < n; i++) {
+ if (i == nullRow) {
+ continue;
+ }
+ Optional<GlobalIndexResult> r = reader.visitEqual(kRef, "row" + i).join();
+ assertTrue(r.isPresent(), "term lookup returns for row" + i);
+ RoaringNavigableMap64 rows = r.get().results();
+ assertEquals(1, rows.getIntCardinality(), "exactly one match for row" + i);
+ assertTrue(contains(rows, (long) i), "term 'row" + i + "' must map to rowId " + i);
+ }
+
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ void skippedRowIdsAreCountedAndRemainSearchableAsNull(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields = List.of(new DataField(0, "k", DataTypes.STRING()));
+ ESIndexOptions options = new ESIndexOptions(fields, new Options());
+ java.nio.file.Path archiveDir = tmp.resolve("archive-skipped-row");
+ Files.createDirectories(archiveDir);
+
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, options);
+ writer.write(BinaryString.fromString("row0"), 0);
+ writer.write(BinaryString.fromString("row2"), 2);
+ ResultEntry entry = writer.finish().get(0);
+ assertEquals(3L, entry.rowCount(), "the padded row-id position is a logical row");
+
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta(archiveDir, entry)), fields, options);
+ try {
+ Optional<GlobalIndexResult> nullRows =
+ reader.visitIsNull(new FieldRef(0, "k", DataTypes.STRING())).join();
+ assertTrue(nullRows.isPresent());
+ assertEquals(1, nullRows.get().results().getIntCardinality());
+ assertTrue(contains(nullRows.get().results(), 1L));
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void likeHonorsSqlEscapeAndLuceneLiterals(@TempDir java.nio.file.Path tmp) throws IOException {
+ // Paimon's default SQL LIKE escape is '\': \_ and \% are literals, while unescaped '%' /
+ // '_'
+ // are wildcards. '*' / '?' are ordinary SQL characters and must also be escaped before the
+ // pattern reaches Lucene's WildcardQuery.
+ List<DataField> fields = Arrays.asList(new DataField(0, "k", DataTypes.STRING()));
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put("global-index.es-index.fields.k.type", "keyword");
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(optionMap));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-like");
+ Files.createDirectories(archiveDir);
+ LocalDirWriter fileWriter = new LocalDirWriter(archiveDir);
+ ESIndexGlobalIndexWriter writer = new ESIndexGlobalIndexWriter(fileWriter, fields, options);
+
+ // Lucene metacharacter cases plus SQL backslash-escape cases.
+ writer.write(BinaryString.fromString("abc"), 0);
+ writer.write(BinaryString.fromString("a*c"), 1);
+ writer.write(BinaryString.fromString("axc"), 2);
+ writer.write(BinaryString.fromString("admin_001"), 3);
+ writer.write(BinaryString.fromString("adminX001"), 4);
+ writer.write(BinaryString.fromString("admin%done"), 5);
+ writer.write(BinaryString.fromString("admin\\path"), 6);
+ List<ResultEntry> entries = writer.finish();
+ ResultEntry entry = entries.get(0);
+
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ long fileSize = Files.size(archiveDir.resolve(entry.fileName()));
+ GlobalIndexIOMeta ioMeta = new GlobalIndexIOMeta(filePath, fileSize, entry.meta());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+
+ FieldRef kRef = new FieldRef(0, "k", DataTypes.STRING());
+
+ // LIKE 'a*c': '*' is literal -> only the exact "a*c" (row 1).
+ RoaringNavigableMap64 starHits = reader.visitLike(kRef, "a*c").join().get().results();
+ assertEquals(1, starHits.getIntCardinality(), "LIKE 'a*c' matches only the literal star");
+ assertTrue(contains(starHits, 1L), "LIKE 'a*c' matches row1 \"a*c\"");
+
+ // LIKE 'a_c': '_' = exactly one char -> all three 3-char a?c values.
+ RoaringNavigableMap64 underHits = reader.visitLike(kRef, "a_c").join().get().results();
+ assertEquals(3, underHits.getIntCardinality(), "LIKE 'a_c' matches all a?c rows");
+
+ // LIKE 'a%c': '%' = any -> all three.
+ RoaringNavigableMap64 pctHits = reader.visitLike(kRef, "a%c").join().get().results();
+ assertEquals(3, pctHits.getIntCardinality(), "LIKE 'a%c' matches all a*c rows");
+
+ // LIKE 'admin\_%': '\_' is a literal underscore and '%' remains a wildcard.
+ RoaringNavigableMap64 escapedUnderHits =
+ reader.visitLike(kRef, "admin\\_%").join().get().results();
+ assertEquals(
+ 1,
+ escapedUnderHits.getIntCardinality(),
+ "escaped underscore matches only the literal underscore");
+ assertTrue(contains(escapedUnderHits, 3L), "escaped underscore matches admin_001");
+
+ // LIKE 'admin\%%': '\%' is a literal percent followed by a wildcard percent.
+ RoaringNavigableMap64 escapedPctHits =
+ reader.visitLike(kRef, "admin\\%%").join().get().results();
+ assertEquals(
+ 1,
+ escapedPctHits.getIntCardinality(),
+ "escaped percent matches only the literal percent");
+ assertTrue(contains(escapedPctHits, 5L), "escaped percent matches admin%done");
+
+ // LIKE 'admin\\%': '\\' is one literal backslash followed by a wildcard percent.
+ RoaringNavigableMap64 escapedSlashHits =
+ reader.visitLike(kRef, "admin\\\\%").join().get().results();
+ assertEquals(
+ 1,
+ escapedSlashHits.getIntCardinality(),
+ "escaped backslash matches one literal backslash");
+ assertTrue(contains(escapedSlashHits, 6L), "escaped backslash matches admin\\path");
+
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ void closeWaitsForInFlightLoadAndClosesEveryStream(@TempDir java.nio.file.Path tmp)
+ throws Exception {
+ List<DataField> fields = Arrays.asList(new DataField(0, "k", DataTypes.STRING()));
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put("global-index.es-index.fields.k.type", "keyword");
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(optionMap));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-concurrent-close");
+ Files.createDirectories(archiveDir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, options);
+ writer.write(BinaryString.fromString("admin_001"), 0);
+ ResultEntry entry = writer.finish().get(0);
+
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ filePath, Files.size(archiveDir.resolve(entry.fileName())), entry.meta());
+ BlockingCountingFileReader fileReader = new BlockingCountingFileReader();
+ ExecutorService queryExecutor = Executors.newSingleThreadExecutor();
+ ExecutorService closeExecutor = Executors.newSingleThreadExecutor();
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ fileReader, List.of(ioMeta), fields, options, queryExecutor);
+
+ try {
+ FieldRef kRef = new FieldRef(0, "k", DataTypes.STRING());
+ CompletableFuture<Optional<GlobalIndexResult>> searchFuture =
+ reader.visitEqual(kRef, "admin_001");
+ assertTrue(fileReader.awaitFirstOpen(), "lazy load reached the blocking stream open");
+
+ CountDownLatch closeStarted = new CountDownLatch(1);
+ Future<?> closeFuture =
+ closeExecutor.submit(
+ () -> {
+ closeStarted.countDown();
+ reader.close();
+ return null;
+ });
+ assertTrue(closeStarted.await(10, TimeUnit.SECONDS), "close task started");
+ assertThrows(
+ TimeoutException.class,
+ () -> closeFuture.get(200, TimeUnit.MILLISECONDS),
+ "close must wait for the in-flight load/search operation");
+
+ fileReader.releaseOpen();
+ Optional<GlobalIndexResult> result = searchFuture.get(10, TimeUnit.SECONDS);
+ assertTrue(result.isPresent(), "in-flight search completes before close");
+ assertEquals(1, result.get().results().getIntCardinality(), "one exact match");
+ closeFuture.get(10, TimeUnit.SECONDS);
+ assertEquals(
+ fileReader.openCount(),
+ fileReader.closeCount(),
+ "close releases every stream opened during lazy loading");
+ } finally {
+ fileReader.releaseOpen();
+ reader.close();
+ queryExecutor.shutdownNow();
+ closeExecutor.shutdownNow();
+ }
+ }
+
+ @Test
+ void failedLazyLoadClosesStreamsBeforeRetry(@TempDir java.nio.file.Path tmp) throws Exception {
+ List<DataField> fields = Arrays.asList(new DataField(0, "k", DataTypes.STRING()));
+ Map<String, String> optionMap = new HashMap<>();
+ optionMap.put("global-index.es-index.fields.k.type", "keyword");
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(optionMap));
+
+ java.nio.file.Path dir = tmp.resolve("failed-lazy-load");
+ Files.createDirectories(dir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(dir), fields, options);
+ writer.write(BinaryString.fromString("x"), 0);
+ ResultEntry entry = writer.finish().get(0);
+ java.nio.file.Path badArchive = dir.resolve(entry.fileName());
+ long declaredArchiveSize = Files.size(badArchive);
+ // Keep the valid offset metadata, but truncate the archive so failure happens after the
+ // root provider (and potentially Lucene forks) have actually been opened.
+ Files.write(badArchive, new byte[] {1, 2, 3, 4});
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path(badArchive.toString()),
+ declaredArchiveSize,
+ entry.meta());
+ BlockingCountingFileReader fileReader = new BlockingCountingFileReader();
+ fileReader.releaseOpen();
+ ExecutorService queryExecutor = Executors.newSingleThreadExecutor();
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ fileReader, List.of(ioMeta), fields, options, queryExecutor);
+ FieldRef kRef = new FieldRef(0, "k", DataTypes.STRING());
+
+ try {
+ for (int attempt = 0; attempt < 2; attempt++) {
+ int opensBeforeAttempt = fileReader.openCount();
+ assertThrows(
+ RuntimeException.class,
+ () -> reader.visitEqual(kRef, "x").join(),
+ "invalid archive load must fail");
+ assertTrue(
+ fileReader.openCount() > opensBeforeAttempt,
+ "each failed attempt must open the archive provider");
+ assertEquals(
+ fileReader.openCount(),
+ fileReader.closeCount(),
+ "failed load must close every stream before a retry");
+ assertEquals(0, reader.openStreamCount(), "failed load retains no tracked stream");
+ }
+ } finally {
+ reader.close();
+ queryExecutor.shutdownNow();
+ }
+ }
+
+ @Test
+ void isNullReturnsNullRowsViaIndex(@TempDir java.nio.file.Path tmp) throws IOException {
+ // Rows 3 and 7 have a NULL value. addNullDoc writes an empty doc (field absent) for them,
+ // so
+ // IS NULL must be index-evaluable (present, NOT Optional.empty()) and return exactly {3,7};
+ // IS NOT NULL returns the other 8.
+ List<DataField> fields = Arrays.asList(new DataField(0, "k", DataTypes.STRING()));
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(new HashMap<>()));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-isnull");
+ Files.createDirectories(archiveDir);
+ LocalDirWriter fileWriter = new LocalDirWriter(archiveDir);
+ ESIndexGlobalIndexWriter writer = new ESIndexGlobalIndexWriter(fileWriter, fields, options);
+
+ int n = 10;
+ Set<Integer> nullRows = new HashSet<>(Arrays.asList(3, 7));
+ for (int i = 0; i < n; i++) {
+ writer.write(nullRows.contains(i) ? null : BinaryString.fromString("row" + i), i);
+ }
+ List<ResultEntry> entries = writer.finish();
+ ResultEntry entry = entries.get(0);
+
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ long fileSize = Files.size(archiveDir.resolve(entry.fileName()));
+ GlobalIndexIOMeta ioMeta = new GlobalIndexIOMeta(filePath, fileSize, entry.meta());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+
+ FieldRef kRef = new FieldRef(0, "k", DataTypes.STRING());
+
+ Optional<GlobalIndexResult> isNull = reader.visitIsNull(kRef).join();
+ assertTrue(isNull.isPresent(), "IS NULL is index-evaluable (not a raw-scan fallback)");
+ RoaringNavigableMap64 nullHits = isNull.get().results();
+ assertEquals(2, nullHits.getIntCardinality(), "two null rows");
+ assertTrue(contains(nullHits, 3L) && contains(nullHits, 7L), "rows 3 and 7 are null");
+
+ Optional<GlobalIndexResult> isNotNull = reader.visitIsNotNull(kRef).join();
+ assertTrue(isNotNull.isPresent(), "IS NOT NULL is index-evaluable");
+ RoaringNavigableMap64 nn = isNotNull.get().results();
+ assertEquals(8, nn.getIntCardinality(), "eight non-null rows");
+ assertTrue(!contains(nn, 3L) && !contains(nn, 7L), "null rows excluded from IS NOT NULL");
+
+ Optional<GlobalIndexResult> notRow1 = reader.visitNotEqual(kRef, "row1").join();
+ assertTrue(notRow1.isPresent(), "<> on keyword is index-evaluable");
+ RoaringNavigableMap64 notRow1Hits = notRow1.get().results();
+ assertEquals(7, notRow1Hits.getIntCardinality(), "non-null rows except row1");
+ assertTrue(
+ !contains(notRow1Hits, 1L)
+ && !contains(notRow1Hits, 3L)
+ && !contains(notRow1Hits, 7L),
+ "<> excludes the matching row and null rows");
+
+ Optional<GlobalIndexResult> notRows1And2 =
+ reader.visitNotIn(kRef, Arrays.asList("row1", "row2")).join();
+ assertTrue(notRows1And2.isPresent(), "NOT IN on keyword is index-evaluable");
+ assertEquals(
+ 6,
+ notRows1And2.get().results().getIntCardinality(),
+ "non-null rows except row1/row2");
+
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ void exclusiveNumericExtremaDoNotOverflow(@TempDir java.nio.file.Path tmp) throws IOException {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "i", DataTypes.INT()),
+ new DataField(1, "l", DataTypes.BIGINT()),
+ new DataField(2, "f", DataTypes.FLOAT()),
+ new DataField(3, "d", DataTypes.DOUBLE()));
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(new HashMap<>()));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-numeric-extrema");
+ Files.createDirectories(archiveDir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, options);
+ writer.write(
+ 0,
+ GenericRow.of(
+ Integer.MIN_VALUE,
+ Long.MIN_VALUE,
+ Float.NEGATIVE_INFINITY,
+ Double.NEGATIVE_INFINITY));
+ writer.write(1, GenericRow.of(0, 0L, 0.0F, 0.0D));
+ writer.write(
+ 2,
+ GenericRow.of(
+ Integer.MAX_VALUE,
+ Long.MAX_VALUE,
+ Float.POSITIVE_INFINITY,
+ Double.POSITIVE_INFINITY));
+ ResultEntry entry = writer.finish().get(0);
+
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ filePath, Files.size(archiveDir.resolve(entry.fileName())), entry.meta());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), Arrays.asList(ioMeta), fields, options);
+
+ FieldRef intRef = new FieldRef(0, "i", DataTypes.INT());
+ FieldRef longRef = new FieldRef(1, "l", DataTypes.BIGINT());
+ FieldRef floatRef = new FieldRef(2, "f", DataTypes.FLOAT());
+ FieldRef doubleRef = new FieldRef(3, "d", DataTypes.DOUBLE());
+ try {
+ assertEmptyIndexResult(
+ reader.visitLessThan(intRef, Integer.MIN_VALUE).join(), "INT < MIN_VALUE");
+ assertEmptyIndexResult(
+ reader.visitGreaterThan(intRef, Integer.MAX_VALUE).join(), "INT > MAX_VALUE");
+ assertEmptyIndexResult(
+ reader.visitLessThan(longRef, Long.MIN_VALUE).join(), "LONG < MIN_VALUE");
+ assertEmptyIndexResult(
+ reader.visitGreaterThan(longRef, Long.MAX_VALUE).join(), "LONG > MAX_VALUE");
+ assertEmptyIndexResult(
+ reader.visitLessThan(floatRef, Float.NEGATIVE_INFINITY).join(),
+ "FLOAT < -Infinity");
+ assertEmptyIndexResult(
+ reader.visitGreaterThan(floatRef, Float.POSITIVE_INFINITY).join(),
+ "FLOAT > +Infinity");
+ assertEmptyIndexResult(
+ reader.visitLessThan(doubleRef, Double.NEGATIVE_INFINITY).join(),
+ "DOUBLE < -Infinity");
+ assertEmptyIndexResult(
+ reader.visitGreaterThan(doubleRef, Double.POSITIVE_INFINITY).join(),
+ "DOUBLE > +Infinity");
+
+ assertEquals(
+ 1,
+ reader.visitLessOrEqual(intRef, Integer.MIN_VALUE)
+ .join()
+ .get()
+ .results()
+ .getIntCardinality());
+ assertEquals(
+ 1,
+ reader.visitGreaterOrEqual(doubleRef, Double.POSITIVE_INFINITY)
+ .join()
+ .get()
+ .results()
+ .getIntCardinality());
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void temporalPredicatesUseLongScalarPath(@TempDir java.nio.file.Path tmp) throws IOException {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "d", DataTypes.DATE()),
+ new DataField(1, "ts", DataTypes.TIMESTAMP(3)));
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(new HashMap<>()));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-temporal");
+ Files.createDirectories(archiveDir);
+ LocalDirWriter fileWriter = new LocalDirWriter(archiveDir);
+ ESIndexGlobalIndexWriter writer = new ESIndexGlobalIndexWriter(fileWriter, fields, options);
+ writer.write(0, GenericRow.of(1, Timestamp.fromEpochMillis(1000L)));
+ writer.write(1, GenericRow.of(2, Timestamp.fromEpochMillis(2000L)));
+ writer.write(2, GenericRow.of(3, Timestamp.fromEpochMillis(3000L)));
+ writer.write(3, GenericRow.of(null, null));
+ ResultEntry entry = writer.finish().get(0);
+
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ filePath, Files.size(archiveDir.resolve(entry.fileName())), entry.meta());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, options);
+
+ FieldRef dateRef = new FieldRef(0, "d", DataTypes.DATE());
+ Optional<GlobalIndexResult> dateGe2 = reader.visitGreaterOrEqual(dateRef, 2).join();
+ assertTrue(dateGe2.isPresent(), "DATE >= literal is index-evaluable");
+ assertEquals(2, dateGe2.get().results().getIntCardinality(), "two rows with d>=2");
+ assertTrue(contains(dateGe2.get().results(), 1L) && contains(dateGe2.get().results(), 2L));
+
+ FieldRef tsRef = new FieldRef(1, "ts", DataTypes.TIMESTAMP(3));
+ Optional<GlobalIndexResult> tsLt3s =
+ reader.visitLessThan(tsRef, Timestamp.fromEpochMillis(3000L)).join();
+ assertTrue(tsLt3s.isPresent(), "TIMESTAMP < literal is index-evaluable");
+ assertEquals(2, tsLt3s.get().results().getIntCardinality(), "two rows before 3000ms");
+ assertTrue(contains(tsLt3s.get().results(), 0L) && contains(tsLt3s.get().results(), 1L));
+
+ Optional<GlobalIndexResult> tsNe2s =
+ reader.visitNotEqual(tsRef, Timestamp.fromEpochMillis(2000L)).join();
+ assertTrue(tsNe2s.isPresent(), "TIMESTAMP <> literal is index-evaluable");
+ assertEquals(2, tsNe2s.get().results().getIntCardinality(), "two rows except 2000ms");
+ assertTrue(contains(tsNe2s.get().results(), 0L) && contains(tsNe2s.get().results(), 2L));
+
+ Optional<GlobalIndexResult> dateBetween = reader.visitBetween(dateRef, 2, 3).join();
+ assertTrue(dateBetween.isPresent(), "DATE BETWEEN is index-evaluable");
+ assertEquals(2, dateBetween.get().results().getIntCardinality());
+ assertTrue(
+ contains(dateBetween.get().results(), 1L)
+ && contains(dateBetween.get().results(), 2L));
+
+ Optional<GlobalIndexResult> dateNotBetween = reader.visitNotBetween(dateRef, 2, 3).join();
+ assertTrue(dateNotBetween.isPresent(), "DATE NOT BETWEEN is index-evaluable");
+ assertEquals(1, dateNotBetween.get().results().getIntCardinality());
+ assertTrue(contains(dateNotBetween.get().results(), 0L), "null rows must stay excluded");
+
+ Optional<GlobalIndexResult> reversedBetween = reader.visitBetween(dateRef, 3, 1).join();
+ assertTrue(reversedBetween.isPresent());
+ assertTrue(reversedBetween.get().results().isEmpty(), "reversed BETWEEN matches no row");
+
+ Optional<GlobalIndexResult> reversedNotBetween =
+ reader.visitNotBetween(dateRef, 3, 1).join();
+ assertTrue(reversedNotBetween.isPresent());
+ assertEquals(
+ 3,
+ reversedNotBetween.get().results().getIntCardinality(),
+ "reversed NOT BETWEEN matches every non-null row");
+
+ assertTrue(reader.visitBetween(dateRef, null, 3).join().get().results().isEmpty());
+ assertTrue(reader.visitNotBetween(dateRef, 1, null).join().get().results().isEmpty());
+
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ void buildTimeFieldConfigurationIsReadFromIndexMetadata(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields = Arrays.asList(new DataField(0, "category", DataTypes.STRING()));
+ Map<String, String> buildOptionMap = new HashMap<>();
+ buildOptionMap.put("global-index.es-index.fields.category.type", "keyword");
+ ESIndexOptions buildOptions = new ESIndexOptions(fields, Options.fromMap(buildOptionMap));
+
+ java.nio.file.Path archiveDir = tmp.resolve("persisted-config");
+ Files.createDirectories(archiveDir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, buildOptions);
+ writer.write(BinaryString.fromString("admin"), 0);
+ writer.write(BinaryString.fromString("viewer"), 1);
+ ResultEntry entry = writer.finish().get(0);
+
+ GlobalIndexIOMeta ioMeta = ioMeta(archiveDir, entry);
+ // Deliberately use the current/default FULLTEXT options. The reader must instead use the
+ // KEYWORD-primary layout persisted when this particular index was built.
+ ESIndexOptions currentTableOptions = new ESIndexOptions(fields, new Options());
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta), fields, currentTableOptions);
+ try {
+ FieldRef category = new FieldRef(0, "category", DataTypes.STRING());
+ Optional<GlobalIndexResult> result = reader.visitEqual(category, "admin").join();
+ assertTrue(result.isPresent());
+ assertEquals(1, result.get().results().getIntCardinality());
+ assertTrue(contains(result.get().results(), 0L));
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void tinySmallAndEmptyArrayNullSemanticsAreIndexed(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "tiny", DataTypes.TINYINT()),
+ new DataField(1, "small", DataTypes.SMALLINT()),
+ new DataField(2, "labels", DataTypes.ARRAY(DataTypes.INT())));
+ ESIndexOptions options = new ESIndexOptions(fields, new Options());
+
+ java.nio.file.Path archiveDir = tmp.resolve("small-types-empty-array");
+ Files.createDirectories(archiveDir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, options);
+ writer.write(0, GenericRow.of((byte) 1, (short) 2, new GenericArray(new int[0])));
+ writer.write(1, GenericRow.of((byte) 2, (short) 3, null));
+ writer.write(2, GenericRow.of((byte) 3, (short) 4, new GenericArray(new int[] {7})));
+ writer.write(
+ 3, GenericRow.of((byte) 4, (short) 5, new GenericArray(new Object[] {null, 8})));
+ ResultEntry entry = writer.finish().get(0);
+
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta(archiveDir, entry)), fields, options);
+ try {
+ Optional<GlobalIndexResult> tiny =
+ reader.visitEqual(new FieldRef(0, "tiny", DataTypes.TINYINT()), (byte) 1)
+ .join();
+ assertTrue(tiny.isPresent());
+ assertTrue(contains(tiny.get().results(), 0L));
+
+ Optional<GlobalIndexResult> small =
+ reader.visitEqual(new FieldRef(1, "small", DataTypes.SMALLINT()), (short) 3)
+ .join();
+ assertTrue(small.isPresent());
+ assertTrue(contains(small.get().results(), 1L));
+
+ FieldRef labels = new FieldRef(2, "labels", DataTypes.ARRAY(DataTypes.INT()));
+ Optional<GlobalIndexResult> isNull = reader.visitIsNull(labels).join();
+ assertTrue(isNull.isPresent());
+ assertEquals(1, isNull.get().results().getIntCardinality());
+ assertTrue(contains(isNull.get().results(), 1L));
+
+ Optional<GlobalIndexResult> isNotNull = reader.visitIsNotNull(labels).join();
+ assertTrue(isNotNull.isPresent());
+ assertEquals(3, isNotNull.get().results().getIntCardinality());
+ assertTrue(contains(isNotNull.get().results(), 0L));
+ assertTrue(contains(isNotNull.get().results(), 2L));
+ assertTrue(contains(isNotNull.get().results(), 3L));
+
+ Optional<GlobalIndexResult> nullElementIsNotZero = reader.visitEqual(labels, 0).join();
+ assertTrue(nullElementIsNotZero.isPresent());
+ assertEquals(0, nullElementIsNotZero.get().results().getIntCardinality());
+
+ Optional<GlobalIndexResult> nonNullElement = reader.visitEqual(labels, 8).join();
+ assertTrue(nonNullElement.isPresent());
+ assertEquals(1, nonNullElement.get().results().getIntCardinality());
+ assertTrue(contains(nonNullElement.get().results(), 3L));
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void closeCanRetryTransientStreamFailures(@TempDir java.nio.file.Path tmp) throws Exception {
+ List<DataField> fields = Arrays.asList(new DataField(0, "id", DataTypes.INT()));
+ ESIndexOptions options = new ESIndexOptions(fields, new Options());
+ java.nio.file.Path archiveDir = tmp.resolve("retry-close");
+ Files.createDirectories(archiveDir);
+
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, options);
+ writer.write(1, 0);
+ ResultEntry entry = writer.finish().get(0);
+
+ TransientCloseFailureFileReader fileReader = new TransientCloseFailureFileReader();
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ fileReader, List.of(ioMeta(archiveDir, entry)), fields, options);
+ assertTrue(reader.visitEqual(new FieldRef(0, "id", DataTypes.INT()), 1).join().isPresent());
+
+ fileReader.setCloseFailures(true);
+ assertThrows(IOException.class, reader::close);
+ assertTrue(reader.openStreamCount() > 0, "the failed provider remains registered");
+ fileReader.setCloseFailures(false);
+ reader.close();
+ assertEquals(0, reader.openStreamCount());
+ }
+
+ @Test
+ void highPrecisionTimestampPredicatesFallBackToRawScan(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields = Arrays.asList(new DataField(0, "ts", DataTypes.TIMESTAMP(6)));
+ ESIndexOptions options = new ESIndexOptions(fields, new Options());
+ java.nio.file.Path archiveDir = tmp.resolve("timestamp-micros");
+ Files.createDirectories(archiveDir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, options);
+ writer.write(Timestamp.fromEpochMillis(1_000L, 1_000), 0);
+ ResultEntry entry = writer.finish().get(0);
+
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta(archiveDir, entry)), fields, options);
+ try {
+ FieldRef ts = new FieldRef(0, "ts", DataTypes.TIMESTAMP(6));
+ Optional<GlobalIndexResult> fallback =
+ reader.visitEqual(ts, Timestamp.fromEpochMillis(1_000L, 2_000)).join();
+ assertTrue(fallback.isPresent());
+ assertEquals(
+ 1,
+ fallback.get().results().getIntCardinality(),
+ "a lossy millisecond index must conservatively retain every row");
+ assertTrue(reader.visitIsNotNull(ts).join().isPresent());
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void standardBooleanDslAndCamelCaseFuzzyParamsAreApplied(@TempDir java.nio.file.Path tmp)
+ throws IOException {
+ List<DataField> fields = Arrays.asList(new DataField(0, "text", DataTypes.STRING()));
+ ESIndexOptions options = new ESIndexOptions(fields, new Options());
+ java.nio.file.Path archiveDir = tmp.resolve("boolean-dsl");
+ Files.createDirectories(archiveDir);
+ ESIndexGlobalIndexWriter writer =
+ new ESIndexGlobalIndexWriter(new LocalDirWriter(archiveDir), fields, options);
+ writer.write(BinaryString.fromString("administrator active"), 0);
+ writer.write(BinaryString.fromString("administrator disabled"), 1);
+ writer.write(BinaryString.fromString("viewer active"), 2);
+ ResultEntry entry = writer.finish().get(0);
+
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ new LocalFileReader(), List.of(ioMeta(archiveDir, entry)), fields, options);
+ String query =
+ "{\"boolean\":{\"must\":[{\"match\":{\"query\":\"administrater\","
+ + "\"fuzziness\":\"AUTO\",\"maxExpansions\":20,\"prefixLength\":0}}],"
+ + "\"must_not\":[{\"match\":{\"query\":\"disabled\"}}]}}";
+ try {
+ Optional<ScoredGlobalIndexResult> result =
+ reader.visitFullTextSearch(new FullTextSearch("text", query, 10)).join();
+ assertTrue(result.isPresent());
+ assertEquals(1, result.get().results().getIntCardinality());
+ assertTrue(contains(result.get().results(), 0L));
+
+ Optional<ScoredGlobalIndexResult> emptyBool =
+ reader.visitFullTextSearch(new FullTextSearch("text", "{\"bool\":{}}", 10))
+ .join();
+ assertTrue(emptyBool.isPresent());
+ assertEquals(3, emptyBool.get().results().getIntCardinality());
+
+ String pureNegativeQuery =
+ "{\"bool\":{\"must_not\":[{\"match\":{\"query\":\"disabled\"}}]}}";
+ Optional<ScoredGlobalIndexResult> pureNegative =
+ reader.visitFullTextSearch(new FullTextSearch("text", pureNegativeQuery, 10))
+ .join();
+ assertTrue(pureNegative.isPresent());
+ assertEquals(2, pureNegative.get().results().getIntCardinality());
+ assertTrue(contains(pureNegative.get().results(), 0L));
+ assertTrue(contains(pureNegative.get().results(), 2L));
+ } finally {
+ reader.close();
+ }
+ }
+
+ private static GlobalIndexIOMeta ioMeta(java.nio.file.Path archiveDir, ResultEntry entry)
+ throws IOException {
+ return new GlobalIndexIOMeta(
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString()),
+ Files.size(archiveDir.resolve(entry.fileName())),
+ entry.meta());
+ }
+
+ /**
+ * Verifies the DiskBBQ vector codec write path through {@link ESIndexGlobalIndexWriter}.
+ *
+ * <p>Distinct from {@link #writeArchiveThenReadAndSearch} (which uses {@code algorithm=hnsw}),
+ * this test sets {@code algorithm=diskbbq} so the writer should pick {@code
+ * ES940DiskBBQVectorsFormat} via {@code PaimonLucene9Codec.getKnnVectorsFormatForField},
+ * causing {@code IVFVectorsWriter#flush} (in practice {@code mergeOneFieldIVF} via {@code
+ * IndexWriter.forceMerge(1)}) to emit the IVF triplet ({@code .cenivf} / {@code .clivf} /
+ * {@code .mivf}) into the archive instead of the HNSW triplet ({@code .vec} / {@code .vex} /
+ * {@code .vem}).
+ *
+ * <p>Assertions cover the writer path end-to-end (archive contents, scalar/keyword/numeric
+ * filters round-tripping through the DiskBBQ-built segment), plus a deterministic vector-search
+ * assertion whose nearest results must all come from the queried cluster.
+ */
+ @Test
+ void writeDiskBBQArchiveThenReadAndSearch(@TempDir java.nio.file.Path tmp) throws Exception {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "embedding", DataTypes.ARRAY(DataTypes.FLOAT())),
+ new DataField(1, "category", DataTypes.STRING()),
+ new DataField(2, "price", DataTypes.INT()));
+
+ Map<String, String> opt = new HashMap<>();
+ opt.put("global-index.es-index.fields.embedding.algorithm", "diskbbq");
+ opt.put("global-index.es-index.fields.embedding.dimension", "32");
+ opt.put("global-index.es-index.fields.embedding.metric", "l2");
+ // 64 is the MIN_VECTORS_PER_CLUSTER allowed by ES940DiskBBQVectorsFormat — picking the
+ // minimum lets a small fixture still produce multiple IVF clusters.
+ opt.put("global-index.es-index.fields.embedding.vectors_per_cluster", "64");
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(opt));
+
+ java.nio.file.Path archiveDir = tmp.resolve("archive-diskbbq");
+ Files.createDirectories(archiveDir);
+
+ // 4 well-separated cluster centers in 32 dims — matches the standalone
+ // DiskBBQVectorIndexTests dim and keeps clustering deterministic.
+ int dim = 32;
+ int clusters = 4;
+ int perCluster = 250; // 1000 rows total — well above vectors_per_cluster=64
+ float[][] centers = new float[clusters][dim];
+ for (int c = 0; c < clusters; c++) {
+ centers[c][c * 8] = 100f; // distinct spike per cluster
+ }
+
+ LocalDirWriter fileWriter = new LocalDirWriter(archiveDir);
+ ESIndexGlobalIndexWriter writer = new ESIndexGlobalIndexWriter(fileWriter, fields, options);
+ for (int c = 0; c < clusters; c++) {
+ for (int j = 0; j < perCluster; j++) {
+ long rowId = (long) c * perCluster + j;
+ float[] v = new float[dim];
+ for (int d = 0; d < dim; d++) {
+ v[d] = centers[c][d] + ((j % 7) - 3) * 0.001f; // tiny deterministic jitter
+ }
+ GenericRow row =
+ GenericRow.of(
+ new GenericArray(v), BinaryString.fromString("c" + c), (int) rowId);
+ writer.write(rowId, row);
+ }
+ }
+ List<ResultEntry> entries = writer.finish();
+ assertEquals(1, entries.size(), "one diskbbq archive produced");
+ ResultEntry entry = entries.get(0);
+ assertEquals((long) clusters * perCluster, entry.rowCount(), "row count");
+ assertNotNull(entry.meta(), "offset-table meta present");
+
+ // --- writer-side check: the archive must contain the DiskBBQ IVF triplet, not HNSW
+ // vector files.
+ Set<String> exts = new HashSet<>();
+ for (String name : ESIndexFileMeta.read(entry.meta()).fileOffsets().keySet()) {
+ int dot = name.lastIndexOf('.');
+ if (dot >= 0) {
+ exts.add(name.substring(dot + 1));
+ }
+ }
+ assertTrue(
+ exts.contains("cenivf") && exts.contains("clivf") && exts.contains("mivf"),
+ "archive must contain DiskBBQ IVF triplet (.cenivf/.clivf/.mivf), got: " + exts);
+ // .vex (HNSW graph) and .vem (HNSW meta) are HNSW-specific; .vec / .vemf belong to the
+ // underlying Lucene99FlatVectorsFormat that DiskBBQ also uses as raw-vector backing
+ // storage, so they may legitimately appear with DiskBBQ too.
+ assertTrue(
+ exts.stream().noneMatch(e -> e.equals("vex") || e.equals("vem")),
+ "archive must not contain HNSW-specific files (.vex/.vem), got: " + exts);
+
+ // --- reconstruct reader over the archive ---
+ org.apache.paimon.fs.Path filePath =
+ new org.apache.paimon.fs.Path(archiveDir.resolve(entry.fileName()).toString());
+ long fileSize = Files.size(archiveDir.resolve(entry.fileName()));
+ GlobalIndexIOMeta ioMeta = new GlobalIndexIOMeta(filePath, fileSize, entry.meta());
+
+ BlockingCountingFileReader fileReader = new BlockingCountingFileReader();
+ fileReader.releaseOpen();
+ ExecutorService queryExecutor = Executors.newSingleThreadExecutor();
+ ESIndexGlobalIndexReader reader =
+ new ESIndexGlobalIndexReader(
+ fileReader, List.of(ioMeta), fields, options, queryExecutor);
+
+ try {
+ // --- scalar filters round-trip through a DiskBBQ-built segment ---
+ FieldRef categoryRef = new FieldRef(1, "category", DataTypes.STRING());
+ Optional<GlobalIndexResult> kw = reader.visitEqual(categoryRef, "c1").join();
+ assertTrue(kw.isPresent(), "keyword filter on diskbbq index returns a result");
+ assertEquals(
+ perCluster, kw.get().results().getIntCardinality(), "perCluster rows in c1");
+
+ // --- numeric filter: price in [perCluster, 2*perCluster) → exactly cluster 1's rows
+ // ---
+ FieldRef priceRef = new FieldRef(2, "price", DataTypes.INT());
+ Optional<GlobalIndexResult> sf =
+ reader.visitGreaterOrEqual(priceRef, perCluster).join();
+ Optional<GlobalIndexResult> sfMax =
+ reader.visitLessThan(priceRef, perCluster * 2).join();
+ assertTrue(sf.isPresent() && sfMax.isPresent(), "numeric filters return results");
+ RoaringNavigableMap64 range =
+ RoaringNavigableMap64.and(sf.get().results(), sfMax.get().results());
+ assertEquals(perCluster, range.getIntCardinality(), "perCluster rows in [pc, 2*pc)");
+
+ // --- vector search via DiskBBQ: top-5 from cluster-0 center must all be cluster 0. ---
+ Optional<ScoredGlobalIndexResult> vr =
+ reader.visitVectorSearch(new VectorSearch(centers[0], 5, "embedding"))
+ .get(30, TimeUnit.SECONDS);
+ assertTrue(vr.isPresent(), "diskbbq vector search returns a result");
+ RoaringNavigableMap64 vrows = vr.get().results();
+ assertEquals(5, vrows.getIntCardinality(), "k=5 results");
+ for (long rid : vrows) {
+ assertTrue(
+ rid < perCluster,
+ "row " + rid + " must be from cluster 0 (rowId < " + perCluster + ")");
+ }
+
+ Optional<ScoredGlobalIndexResult> broadResult =
+ reader.visitVectorSearch(new VectorSearch(centers[0], 256, "embedding"))
+ .get(30, TimeUnit.SECONDS);
+ assertTrue(broadResult.isPresent(), "broad DiskBBQ search returns a result");
+ assertEquals(
+ fileReader.openCount() - fileReader.closeCount(),
+ reader.openStreamCount(),
+ "providers whose streams closed during search must unregister immediately");
+
+ // Query-scoped IndexInput clones are not guaranteed to be closed by Lucene. ESLib's
+ // ArchiveIndexInput must reuse provider forks bounded by peak read concurrency instead
+ // of retaining one newly forked stream per rescore query until reader.close().
+ int opensAfterWarmup = fileReader.openCount();
+ int liveStreamsAfterWarmup = reader.openStreamCount();
+ for (int repeat = 0; repeat < 3; repeat++) {
+ Optional<ScoredGlobalIndexResult> repeatedResult =
+ reader.visitVectorSearch(new VectorSearch(centers[0], 256, "embedding"))
+ .get(30, TimeUnit.SECONDS);
+ assertTrue(repeatedResult.isPresent());
+ }
+ assertEquals(
+ opensAfterWarmup,
+ fileReader.openCount(),
+ "repeated DiskBBQ rescore queries must reuse provider streams");
+ assertEquals(
+ liveStreamsAfterWarmup,
+ reader.openStreamCount(),
+ "repeated DiskBBQ queries must not retain additional live streams");
+ } finally {
+ reader.close();
+ queryExecutor.shutdownNow();
+ }
+ assertEquals(
+ fileReader.openCount(),
+ fileReader.closeCount(),
+ "reader close must release every persistent DiskBBQ stream");
+ }
+
+ /**
+ * Bypass-archive regression test for the DiskBBQ read path. Writes 1000 random normalised
+ * vectors directly to an {@code FSDirectory} via raw Lucene + {@code PaimonLucene912Codec}
+ * ({@code algorithm=diskbbq}) — skipping paimon-eslib's archive layer entirely — and reads them
+ * back through {@code KnnFloatVectorQuery}.
+ *
+ * <p>Locks in two fixes in eslib-core {@code lucene9/DiskBBQVectorsReader}:
+ *
+ * <ul>
+ * <li>{@code search()} now calls {@code knnCollector.incVisitedCount(scoredDocs)} —
+ * previously {@code TopDocs.totalHits} stayed at 0 even when candidates were collected,
+ * making {@code KnnFloatVectorQuery} treat the leaf result as empty.
+ * <li>{@code visitCluster()} now gates {@code collector.collect(...)} on a non-NaN score (and
+ * {@code minCompetitiveSimilarity()}) — previously NaN scores poisoned the collector's
+ * priority queue.
+ * </ul>
+ */
+ @Test
+ void diskBBQVectorSearchOnFSDirectory(@TempDir java.nio.file.Path tmp) throws IOException {
+ int dim = 32;
+ int numVectors = 1000;
+
+ // Match the standalone DiskBBQ test: random normalized vectors + DOT_PRODUCT. Sparse
+ // one-hot vectors with EUCLIDEAN are covered separately in eslib-core.
+ java.util.Random rng = new java.util.Random(42);
+ float[][] vectors = new float[numVectors][];
+ for (int i = 0; i < numVectors; i++) {
+ float[] v = new float[dim];
+ float norm = 0;
+ for (int d = 0; d < dim; d++) {
+ v[d] = rng.nextFloat() - 0.5f;
+ norm += v[d] * v[d];
+ }
+ norm = (float) Math.sqrt(norm);
+ for (int d = 0; d < dim; d++) {
+ v[d] /= norm;
+ }
+ vectors[i] = v;
+ }
+ float[] query = vectors[0]; // query is the first vector — its own row must be top-1
+
+ Map<String, org.elasticsearch.eslib.api.model.FieldIndexConfig> fieldConfigs =
+ new HashMap<>();
+ Map<String, String> params = new HashMap<>();
+ params.put("vectors_per_cluster", "64");
+ fieldConfigs.put(
+ "embedding",
+ org.elasticsearch.eslib.api.model.FieldIndexConfig.builder(
+ "embedding",
+ org.elasticsearch.eslib.api.model.FieldIndexConfig.IndexType.VECTOR)
+ .algorithm(org.elasticsearch.eslib.api.model.VectorAlgorithm.DISKBBQ)
+ .dimension(dim)
+ .metric("dot_product")
+ .algorithmParams(params)
+ .build());
+
+ java.nio.file.Path idxDir = tmp.resolve("fs-diskbbq");
+ Files.createDirectories(idxDir);
+
+ // --- write via raw Lucene + PaimonLucene9Codec (the codec paimon-eslib uses;
+ // also the only codec registered in Lucene SPI, so DirectoryReader.open finds it back). ---
+ org.apache.lucene.codecs.Codec codec =
+ new org.elasticsearch.eslib.adapter.lucene9.PaimonLucene9Codec(fieldConfigs);
+ org.apache.lucene.index.IndexWriterConfig iwc =
+ new org.apache.lucene.index.IndexWriterConfig(
+ new org.apache.lucene.analysis.core.WhitespaceAnalyzer());
+ iwc.setCodec(codec);
+ iwc.setUseCompoundFile(false);
+ try (org.apache.lucene.store.FSDirectory dir =
+ org.apache.lucene.store.FSDirectory.open(idxDir);
+ org.apache.lucene.index.IndexWriter w =
+ new org.apache.lucene.index.IndexWriter(dir, iwc)) {
+ for (int i = 0; i < numVectors; i++) {
+ org.apache.lucene.document.Document doc = new org.apache.lucene.document.Document();
+ doc.add(
+ new org.apache.lucene.document.KnnFloatVectorField(
+ "embedding",
+ vectors[i],
+ org.apache.lucene.index.VectorSimilarityFunction.DOT_PRODUCT));
+ w.addDocument(doc);
+ }
+ w.forceMerge(1);
+ w.commit();
+ }
+
+ // --- list produced files for visibility ---
+ Set<String> exts = new HashSet<>();
+ try (java.util.stream.Stream<java.nio.file.Path> walk = Files.list(idxDir)) {
+ walk.forEach(
+ p -> {
+ String n = p.getFileName().toString();
+ int dot = n.lastIndexOf('.');
+ if (dot >= 0) {
+ exts.add(n.substring(dot + 1));
+ }
+ });
+ }
+ assertTrue(
+ exts.contains("cenivf") && exts.contains("clivf") && exts.contains("mivf"),
+ "FSDirectory must hold DiskBBQ IVF triplet, got: " + exts);
+
+ // --- read back via raw Lucene KnnFloatVectorQuery ---
+ try (org.apache.lucene.store.FSDirectory dir =
+ org.apache.lucene.store.FSDirectory.open(idxDir);
+ org.apache.lucene.index.DirectoryReader r =
+ org.apache.lucene.index.DirectoryReader.open(dir)) {
+ assertEquals(numVectors, r.numDocs(), "all docs flushed");
+ assertEquals(1, r.leaves().size(), "single segment after forceMerge(1)");
+
+ // Sanity: PerFieldKnnVectorsFormat correctly recorded a DiskBBQ vectors format for this
+ // field (version-agnostic: ES920 / ES940 / ...). Reads it back from FieldInfo attrs.
+ org.apache.lucene.index.LeafReader leaf = r.leaves().get(0).reader();
+ org.apache.lucene.index.FieldInfo fi = leaf.getFieldInfos().fieldInfo("embedding");
+ assertTrue(
+ fi.attributes()
+ .get("PerFieldKnnVectorsFormat.format")
+ .contains("DiskBBQVectorsFormat"),
+ "DiskBBQ format must be dispatched at read time");
+
+ // Direct leaf-level KNN — bypasses IndexSearcher / KnnFloatVectorQuery rewrite.
+ // Lucene 9.12 returns via the collector (void method).
+ org.apache.lucene.search.TopKnnCollector leafCollector =
+ new org.apache.lucene.search.TopKnnCollector(5, Integer.MAX_VALUE);
+ leaf.searchNearestVectors("embedding", query, leafCollector, null);
+ org.apache.lucene.search.TopDocs leafTop = leafCollector.topDocs();
+
+ // Document the bug: DiskBBQVectorsReader.search DOES collect documents …
+ assertTrue(
+ leafTop.scoreDocs.length > 0,
+ "DiskBBQVectorsReader.search should collect candidates");
+ for (org.apache.lucene.search.ScoreDoc sd : leafTop.scoreDocs) {
+ assertTrue(
+ Float.isFinite(sd.score),
+ "score must be finite (NaN guard / simdvec stats fix), got: " + sd.score);
+ }
+ assertTrue(
+ leafTop.totalHits.value > 0,
+ "TopDocs.totalHits.value must be > 0 (incVisitedCount fix)");
+
+ // The wrapping KnnFloatVectorQuery path must now return the requested k hits, with
+ // the query vector itself (doc 0) among them.
+ // KnnFloatVectorQuery on top of the leaf must now produce k=5 finite-scored hits
+ // (before the fixes it produced MatchNoDocsQuery because the FloatVectorValues stub
+ // reported size==0, which short-circuited KnnFloatVectorQuery.approximateSearch).
+ org.apache.lucene.search.IndexSearcher s =
+ new org.apache.lucene.search.IndexSearcher(r);
+ org.apache.lucene.search.KnnFloatVectorQuery q =
+ new org.apache.lucene.search.KnnFloatVectorQuery("embedding", query, 5);
+ org.apache.lucene.search.TopDocs td = s.search(q, 5);
+ assertEquals(5, td.scoreDocs.length, "KnnFloatVectorQuery returns k=5 hits");
+ for (org.apache.lucene.search.ScoreDoc sd : td.scoreDocs) {
+ assertTrue(Float.isFinite(sd.score), "score must be finite, got: " + sd.score);
+ assertTrue(
+ sd.score > 0, "score must be positive for DOT_PRODUCT, got: " + sd.score);
+ }
+ // Recall (whether the query vector itself appears in top-5) is intentionally NOT
+ // asserted: DiskBBQ is an approximate IVF method whose recall depends on nprobe and
+ // clustering quality; with vpc=64 + 1000 docs we visit only 4/~16 clusters by default.
+ // The bug-fix verification is "finite scores + non-zero hits" — see Javadoc.
+ }
+ }
+
+ private static String matchQuery(String terms) {
+ return "{\"match\":{\"query\":\"" + terms + "\"}}";
+ }
+
+ private static String matchQuery(String terms, String operator) {
+ return "{\"match\":{\"query\":\"" + terms + "\",\"operator\":\"" + operator + "\"}}";
+ }
+
+ private static String fuzzyMatchQuery(String terms, int fuzziness) {
+ return "{\"match\":{\"query\":\"" + terms + "\",\"fuzziness\":" + fuzziness + "}}";
+ }
+
+ private static String phraseQuery(String terms) {
+ return "{\"match_phrase\":{\"query\":\"" + terms + "\"}}";
+ }
+
+ private static void assertEmptyIndexResult(
+ Optional<GlobalIndexResult> result, String predicate) {
+ assertTrue(result.isPresent(), predicate + " is index-evaluable");
+ assertTrue(result.get().results().isEmpty(), predicate + " must match no row");
+ }
+
+ private static boolean contains(RoaringNavigableMap64 bitmap, long id) {
+ for (long v : bitmap) {
+ if (v == id) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReaderDebugTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReaderDebugTest.java
new file mode 100644
index 0000000000..19ef7aabd5
--- /dev/null
+++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReaderDebugTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.predicate.VectorSearch;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class ESIndexGlobalIndexReaderDebugTest {
+
+ @Test
+ void samplesLongArraysWithTruncation() {
+ assertEquals(
+ "[11, 22, 33, ...(+2)]",
+ ESIndexGlobalIndexReader.sample(new long[] {11L, 22L, 33L, 44L, 55L}, 5, 3));
+ }
+
+ @Test
+ void samplesFloatArraysWithTruncation() {
+ assertEquals(
+ "[0.1, 0.2, 0.3, ...(+1)]",
+ ESIndexGlobalIndexReader.sample(new float[] {0.1f, 0.2f, 0.3f, 0.4f}, 4, 3));
+ }
+
+ @Test
+ void resolvesVectorSearchTopKFromOptions() {
+ assertEquals(
+ 1000,
+ ESIndexGlobalIndexReader.vectorSearchTopK(
+ new VectorSearch(
+ new float[] {1.0f},
+ 10,
+ "embedding",
+ Map.of("hnsw.num_candidates", "1000"))));
+ assertEquals(
+ 64,
+ ESIndexGlobalIndexReader.vectorSearchTopK(
+ new VectorSearch(
+ new float[] {1.0f},
+ 10,
+ "embedding",
+ Map.of("hnsw.ef_search", "64"))));
+ assertEquals(
+ 10,
+ ESIndexGlobalIndexReader.vectorSearchTopK(
+ new VectorSearch(
+ new float[] {1.0f},
+ 10,
+ "embedding",
+ Map.of("hnsw.num_candidates", "5"))));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ ESIndexGlobalIndexReader.vectorSearchTopK(
+ new VectorSearch(
+ new float[] {1.0f},
+ 10,
+ "embedding",
+ Map.of("hnsw.num_candidates", "bad"))));
+ }
+}
diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerExecutorTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerExecutorTest.java
new file mode 100644
index 0000000000..952ea8b093
--- /dev/null
+++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerExecutorTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.Path;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.FullTextSearch;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Verifies that Paimon's caller-owned executor is used for the outer reader future. */
+class ESIndexGlobalIndexerExecutorTest {
+
+ private static final List<DataField> FIELDS =
+ Arrays.asList(new DataField(0, "k", DataTypes.STRING()));
+
+ private static List<GlobalIndexIOMeta> oneFile() throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream out = new DataOutputStream(bytes)) {
+ out.writeInt(1);
+ byte[] name = "segments_1".getBytes(java.nio.charset.StandardCharsets.UTF_8);
+ out.writeInt(name.length);
+ out.write(name);
+ out.writeLong(0L);
+ out.writeLong(0L);
+ }
+ return Arrays.asList(
+ new GlobalIndexIOMeta(new Path("/tmp/none.index"), 0L, bytes.toByteArray()));
+ }
+
+ @Test
+ void callerExecutorIsPassedToReader() throws IOException {
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ try {
+ ESIndexGlobalIndexer indexer = new ESIndexGlobalIndexer(FIELDS, new Options());
+ GlobalIndexReader reader = indexer.createReader(meta -> null, oneFile(), caller);
+ assertSame(
+ caller,
+ ((ESIndexGlobalIndexReader) reader).queryExecutor(),
+ "outer reader futures must use Paimon's caller executor");
+ } finally {
+ caller.shutdownNow();
+ }
+ }
+
+ @Test
+ void inlineValidationFailuresCompleteTheReturnedFutureExceptionally() throws Exception {
+ ESIndexGlobalIndexer indexer = new ESIndexGlobalIndexer(FIELDS, new Options());
+ ESIndexGlobalIndexReader reader =
+ (ESIndexGlobalIndexReader) indexer.createReader(meta -> null, oneFile(), null);
+ try {
+ CompletableFuture<?> future =
+ reader.visitFullTextSearch(new FullTextSearch("k", "{not-json", 1));
+ assertThrows(CompletionException.class, future::join);
+ } finally {
+ reader.close();
+ }
+ }
+}
diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexOptionsTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexOptionsTest.java
new file mode 100644
index 0000000000..eb1690bb50
--- /dev/null
+++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexOptionsTest.java
@@ -0,0 +1,403 @@
+/*
+ * 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.options.Options;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.lucene.codecs.KnnVectorsFormat;
+import org.apache.lucene.codecs.perfield.PerFieldKnnVectorsFormat;
+import org.elasticsearch.eslib.adapter.PaimonHnswVectorsFormat;
+import org.elasticsearch.eslib.adapter.lucene9.PaimonLucene9Codec;
+import org.elasticsearch.eslib.api.model.FieldIndexConfig;
+import org.elasticsearch.eslib.api.model.ScalarFieldType;
+import org.elasticsearch.eslib.api.model.VectorAlgorithm;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests option-key resolution in {@link ESIndexOptions}. */
+class ESIndexOptionsTest {
+
+ private static final List<DataField> FIELDS =
+ Arrays.asList(
+ new DataField(0, "embedding", DataTypes.VECTOR(128, DataTypes.FLOAT())),
+ new DataField(1, "title", DataTypes.STRING()));
+
+ @Test
+ void fieldLevelKeysAreRead() {
+ Map<String, String> m = new HashMap<>();
+ m.put("global-index.es-index.fields.embedding.algorithm", "diskbbq");
+ m.put("global-index.es-index.fields.embedding.dimension", "128");
+ m.put("global-index.es-index.fields.title.analyzer", "standard");
+ ESIndexOptions options = new ESIndexOptions(FIELDS, Options.fromMap(m));
+
+ FieldIndexConfig emb = options.getConfig("embedding");
+ assertThat(emb.indexType()).isEqualTo(FieldIndexConfig.IndexType.VECTOR);
+ assertThat(emb.algorithm()).isEqualTo(VectorAlgorithm.DISKBBQ);
+ assertThat(emb.dimension()).isEqualTo(128);
+ assertThat(options.getConfig("title").indexType())
+ .isEqualTo(FieldIndexConfig.IndexType.FULLTEXT);
+ }
+
+ @Test
+ void nativeVectorAlgorithmIsRejectedUpFront() {
+ Map<String, String> m = new HashMap<>();
+ m.put("global-index.es-index.fields.embedding.algorithm", "native");
+
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(m)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Vector algorithm 'native' is not supported")
+ .hasMessageContaining("hnsw")
+ .hasMessageContaining("diskbbq");
+ }
+
+ @Test
+ void hnswConstructionParametersAreRead() {
+ Map<String, String> m = new HashMap<>();
+ m.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ m.put("global-index.es-index.fields.embedding.dimension", "128");
+ m.put("global-index.es-index.fields.embedding.m", "30");
+ m.put("global-index.es-index.fields.embedding.ef_construction", "360");
+
+ FieldIndexConfig emb =
+ new ESIndexOptions(FIELDS, Options.fromMap(m)).getConfig("embedding");
+
+ assertThat(emb.algorithm()).isEqualTo(VectorAlgorithm.HNSW);
+ assertThat(emb.getIntParam("m", 16)).isEqualTo(30);
+ assertThat(emb.getIntParam("ef_construction", 100)).isEqualTo(360);
+ }
+
+ @Test
+ void hnswConstructionParametersReachLuceneCodec() {
+ Map<String, String> m = new HashMap<>();
+ m.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ m.put("global-index.es-index.fields.embedding.dimension", "128");
+ m.put("global-index.es-index.fields.embedding.m", "30");
+ m.put("global-index.es-index.fields.embedding.ef_construction", "360");
+
+ ESIndexOptions options = new ESIndexOptions(FIELDS, Options.fromMap(m));
+ PaimonLucene9Codec codec = new PaimonLucene9Codec(options.getFieldConfigs());
+ PerFieldKnnVectorsFormat perField = (PerFieldKnnVectorsFormat) codec.knnVectorsFormat();
+ KnnVectorsFormat format = perField.getKnnVectorsFormatForField("embedding");
+
+ assertThat(format).isInstanceOf(PaimonHnswVectorsFormat.class);
+ assertThat(format.toString()).contains("maxConn=30").contains("beamWidth=360");
+ }
+
+ @Test
+ void invalidHnswConstructionParametersAreRejected() {
+ Map<String, String> nonPositive = new HashMap<>();
+ nonPositive.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ nonPositive.put("global-index.es-index.fields.embedding.m", "0");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(nonPositive)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("between 1 and 512");
+
+ Map<String, String> excessiveM = new HashMap<>();
+ excessiveM.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ excessiveM.put("global-index.es-index.fields.embedding.m", "513");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(excessiveM)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("between 1 and 512");
+
+ Map<String, String> excessiveEfConstruction = new HashMap<>();
+ excessiveEfConstruction.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ excessiveEfConstruction.put(
+ "global-index.es-index.fields.embedding.ef_construction", "3201");
+ assertThatThrownBy(
+ () -> new ESIndexOptions(FIELDS, Options.fromMap(excessiveEfConstruction)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("between 1 and 3200");
+
+ Map<String, String> nonInteger = new HashMap<>();
+ nonInteger.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ nonInteger.put("global-index.es-index.fields.embedding.ef_construction", "wide");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(nonInteger)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("between 1 and 3200");
+
+ Map<String, String> wrongAlgorithm = new HashMap<>();
+ wrongAlgorithm.put("global-index.es-index.fields.embedding.algorithm", "diskbbq");
+ wrongAlgorithm.put("global-index.es-index.fields.embedding.ef_construction", "360");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(wrongAlgorithm)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("require algorithm=hnsw");
+ }
+
+ @Test
+ void tableLevelAlgorithmParametersOnlyApplyToMatchingAlgorithms() {
+ Map<String, String> diskOptions = new HashMap<>();
+ diskOptions.put("global-index.es-index.m", "30");
+ diskOptions.put("global-index.es-index.ef_construction", "360");
+ diskOptions.put("global-index.es-index.fields.embedding.algorithm", "diskbbq");
+ diskOptions.put("global-index.es-index.centroids_per_parent_cluster", "24");
+ FieldIndexConfig diskConfig =
+ new ESIndexOptions(FIELDS, Options.fromMap(diskOptions)).getConfig("embedding");
+ assertThat(diskConfig.algorithmParams()).doesNotContainKeys("m", "ef_construction");
+ assertThat(diskConfig.algorithmParams())
+ .containsEntry("centroids_per_parent_cluster", "24");
+
+ Map<String, String> hnswOptions = new HashMap<>();
+ hnswOptions.put("global-index.es-index.vectors_per_cluster", "384");
+ hnswOptions.put("global-index.es-index.centroids_per_parent_cluster", "24");
+ hnswOptions.put("global-index.es-index.fields.embedding.algorithm", "hnsw");
+ FieldIndexConfig hnswConfig =
+ new ESIndexOptions(FIELDS, Options.fromMap(hnswOptions)).getConfig("embedding");
+ assertThat(hnswConfig.algorithmParams())
+ .doesNotContainKeys("vectors_per_cluster", "centroids_per_parent_cluster");
+ }
+
+ @Test
+ void indexTypeLevelKeyIsDefaultAndFieldLevelOverrides() {
+ Map<String, String> m = new HashMap<>();
+ // index-type-level default for all vector fields ...
+ m.put("global-index.es-index.metric", "l2");
+ m.put("global-index.es-index.fields.embedding.algorithm", "diskbbq");
+ m.put("global-index.es-index.fields.embedding.dimension", "128");
+ // ... overridden per-field
+ m.put("global-index.es-index.fields.embedding.metric", "cosine");
+ ESIndexOptions options = new ESIndexOptions(FIELDS, Options.fromMap(m));
+
+ FieldIndexConfig emb = options.getConfig("embedding");
+ assertThat(emb.metric()).isEqualTo("cosine"); // field-level wins over es-index.metric=l2
+ }
+
+ @Test
+ void globalIndexPrefixedKeysAreRead() {
+ // Keys use the full global-index.es-index. prefix (the persisted table-property / ES-mount
+ // convention). The parser reads the explicitly configured analyzer.
+ Map<String, String> m = new HashMap<>();
+ m.put("global-index.es-index.fields.title.analyzer", "standard");
+ ESIndexOptions options = new ESIndexOptions(FIELDS, Options.fromMap(m));
+ assertThat(options.getConfig("title").indexType())
+ .isEqualTo(FieldIndexConfig.IndexType.FULLTEXT);
+ }
+
+ @Test
+ void textFieldsGetComplementaryMultiField() {
+ ESIndexOptions fulltext = new ESIndexOptions(FIELDS, Options.fromMap(new HashMap<>()));
+ assertThat(fulltext.keywordSubField("title")).isEqualTo("title.keyword");
+ assertThat(fulltext.fullTextSearchField("title")).isEqualTo("title");
+
+ Map<String, String> keywordOptions = new HashMap<>();
+ keywordOptions.put("global-index.es-index.fields.title.type", "keyword");
+ ESIndexOptions keyword = new ESIndexOptions(FIELDS, Options.fromMap(keywordOptions));
+ assertThat(keyword.fullTextSubField("title")).isEqualTo("title.fulltext");
+ assertThat(keyword.fullTextSearchField("title")).isEqualTo("title.fulltext");
+ }
+
+ @Test
+ void unsupportedScalarTypeIsRejectedUpFront() {
+ // A BOOLEAN companion column is not a supported es-index scalar type. It must be rejected
+ // when options are parsed, not silently mapped to KEYWORD and then blow up at build time in
+ // ESIndexGlobalIndexWriter.extractScalar (getString -> ClassCastException).
+ List<DataField> fields = Arrays.asList(new DataField(0, "flag", DataTypes.BOOLEAN()));
+ assertThatThrownBy(() -> new ESIndexOptions(fields, Options.fromMap(new HashMap<>())))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unsupported scalar type");
+ }
+
+ @Test
+ void textFieldDefaultsToFullTextWithKeywordSubField() {
+ // A String column with no analyzer/type configured defaults to FULLTEXT (with a keyword
+ // sub-field), so full-text search always works on text columns and exact filters use the
+ // sub-field. This closes the coverage gap where a text column carried only as an extra
+ // field would otherwise default to KEYWORD yet still be counted as full-text coverage.
+ ESIndexOptions options = new ESIndexOptions(FIELDS, Options.fromMap(new HashMap<>()));
+ assertThat(options.getConfig("title").indexType())
+ .isEqualTo(FieldIndexConfig.IndexType.FULLTEXT);
+ assertThat(options.keywordSubField("title")).isEqualTo("title.keyword");
+ }
+
+ @Test
+ void textFieldConfiguredAsKeywordStillGetsFullTextSubField() {
+ Map<String, String> m = new HashMap<>();
+ m.put("global-index.es-index.fields.title.type", "keyword");
+ ESIndexOptions options = new ESIndexOptions(FIELDS, Options.fromMap(m));
+ assertThat(options.getConfig("title").indexType())
+ .isEqualTo(FieldIndexConfig.IndexType.KEYWORD);
+ assertThat(options.keywordSubField("title")).isNull();
+ assertThat(options.fullTextSubField("title")).isEqualTo("title.fulltext");
+ }
+
+ @Test
+ void temporalFieldsUseLongScalarForFiltering() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "d", DataTypes.DATE()),
+ new DataField(1, "ts", DataTypes.TIMESTAMP(3)));
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(new HashMap<>()));
+
+ assertThat(options.getConfig("d").indexType()).isEqualTo(FieldIndexConfig.IndexType.SCALAR);
+ assertThat(options.getConfig("d").scalarType()).isEqualTo(ScalarFieldType.LONG);
+ assertThat(options.getConfig("ts").indexType())
+ .isEqualTo(FieldIndexConfig.IndexType.SCALAR);
+ assertThat(options.getConfig("ts").scalarType()).isEqualTo(ScalarFieldType.LONG);
+
+ Map<String, String> explicitDate = new HashMap<>();
+ explicitDate.put("global-index.es-index.fields.d.type", "date");
+ ESIndexOptions explicitOptions = new ESIndexOptions(fields, Options.fromMap(explicitDate));
+ assertThat(explicitOptions.getConfig("d").indexType())
+ .isEqualTo(FieldIndexConfig.IndexType.SCALAR);
+ assertThat(explicitOptions.getConfig("d").scalarType()).isEqualTo(ScalarFieldType.LONG);
+ }
+
+ @Test
+ void arrayVectorRequiresDimensionAndGetsPresenceField() {
+ List<DataField> fields =
+ Arrays.asList(new DataField(0, "embedding", DataTypes.ARRAY(DataTypes.FLOAT())));
+
+ assertThatThrownBy(() -> new ESIndexOptions(fields, new Options()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("requires a positive dimension");
+
+ Map<String, String> configured = new HashMap<>();
+ configured.put("global-index.es-index.fields.embedding.dimension", "4");
+ ESIndexOptions options = new ESIndexOptions(fields, Options.fromMap(configured));
+ String presenceField = options.arrayPresenceField("embedding");
+ assertThat(presenceField).isEqualTo("embedding.__paimon_array_present");
+ assertThat(options.getConfig(presenceField).scalarType()).isEqualTo(ScalarFieldType.INT);
+ }
+
+ @Test
+ void invalidVectorDimensionMetricAndDiskBBQParametersAreRejected() {
+ Map<String, String> wrongDimension = new HashMap<>();
+ wrongDimension.put("global-index.es-index.fields.embedding.dimension", "64");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(wrongDimension)))
+ .hasMessageContaining("does not match the VECTOR type dimension 128");
+
+ List<DataField> oversizedVector =
+ Arrays.asList(
+ new DataField(0, "embedding", DataTypes.VECTOR(4097, DataTypes.FLOAT())));
+ assertThatThrownBy(() -> new ESIndexOptions(oversizedVector, new Options()))
+ .hasMessageContaining("maximum supported dimension 4096");
+
+ Map<String, String> badMetric = new HashMap<>();
+ badMetric.put("global-index.es-index.fields.embedding.metric", "angular-ish");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(badMetric)))
+ .hasMessageContaining("Unknown vector metric 'angular-ish'");
+
+ Map<String, String> badVpc = new HashMap<>();
+ badVpc.put("global-index.es-index.fields.embedding.algorithm", "diskbbq");
+ badVpc.put("global-index.es-index.fields.embedding.vectors_per_cluster", "many");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(badVpc)))
+ .hasMessageContaining("vectors_per_cluster")
+ .hasMessageContaining("between 64 and 65536");
+
+ Map<String, String> badCpc = new HashMap<>();
+ badCpc.put("global-index.es-index.fields.embedding.algorithm", "diskbbq");
+ badCpc.put("global-index.es-index.fields.embedding.centroids_per_parent_cluster", "1");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(badCpc)))
+ .hasMessageContaining("centroids_per_parent_cluster")
+ .hasMessageContaining("between 2 and 384");
+
+ Map<String, String> hnswVpc = new HashMap<>();
+ hnswVpc.put("global-index.es-index.fields.embedding.vectors_per_cluster", "64");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(hnswVpc)))
+ .hasMessageContaining("requires algorithm=diskbbq");
+
+ Map<String, String> hnswCpc = new HashMap<>();
+ hnswCpc.put("global-index.es-index.fields.embedding.centroids_per_parent_cluster", "16");
+ assertThatThrownBy(() -> new ESIndexOptions(FIELDS, Options.fromMap(hnswCpc)))
+ .hasMessageContaining("requires algorithm=diskbbq");
+ }
+
+ @Test
+ void unsupportedIkAnalyzerIsRejectedBeforeIndexBuild() {
+ Map<String, String> options = new HashMap<>();
+ options.put("global-index.es-index.fields.title.analyzer", "ik_smart");
+
+ assertThatThrownBy(
+ () ->
+ new ESIndexOptions(
+ Arrays.asList(
+ new DataField(0, "title", DataTypes.STRING())),
+ Options.fromMap(options)))
+ .hasMessageContaining("not supported by paimon-eslib")
+ .hasMessageContaining("standard, whitespace, simple, or keyword");
+ }
+
+ @Test
+ void unknownAndIncompatibleExplicitTypesAreRejected() {
+ List<DataField> number = Arrays.asList(new DataField(0, "id", DataTypes.INT()));
+
+ Map<String, String> unknown = new HashMap<>();
+ unknown.put("global-index.es-index.fields.id.type", "mystery");
+ assertThatThrownBy(() -> new ESIndexOptions(number, Options.fromMap(unknown)))
+ .hasMessageContaining("Unknown es-index type 'mystery'");
+
+ Map<String, String> fulltextNumber = new HashMap<>();
+ fulltextNumber.put("global-index.es-index.fields.id.type", "fulltext");
+ assertThatThrownBy(() -> new ESIndexOptions(number, Options.fromMap(fulltextNumber)))
+ .hasMessageContaining("incompatible")
+ .hasMessageContaining("id");
+
+ Map<String, String> geo = new HashMap<>();
+ geo.put("global-index.es-index.fields.id.type", "geo_point");
+ assertThatThrownBy(() -> new ESIndexOptions(number, Options.fromMap(geo)))
+ .hasMessageContaining("geo_point")
+ .hasMessageContaining("not supported");
+
+ List<DataField> integerVector =
+ Arrays.asList(new DataField(0, "embedding", DataTypes.VECTOR(4, DataTypes.INT())));
+ assertThatThrownBy(() -> new ESIndexOptions(integerVector, new Options()))
+ .hasMessageContaining("Unsupported scalar type")
+ .hasMessageContaining("VECTOR");
+ }
+
+ @Test
+ void generatedFieldNamesCannotCollideWithLogicalColumns() {
+ List<DataField> keywordCollision =
+ Arrays.asList(
+ new DataField(0, "title", DataTypes.STRING()),
+ new DataField(1, "title.keyword", DataTypes.STRING()));
+ assertThatThrownBy(() -> new ESIndexOptions(keywordCollision, new Options()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("reserved es-index generated field")
+ .hasMessageContaining("title.keyword");
+
+ List<DataField> presenceCollision =
+ Arrays.asList(
+ new DataField(0, "labels", DataTypes.ARRAY(DataTypes.INT())),
+ new DataField(1, "labels.__paimon_array_present", DataTypes.INT()));
+ assertThatThrownBy(() -> new ESIndexOptions(presenceCollision, new Options()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("reserved es-index generated field")
+ .hasMessageContaining("labels.__paimon_array_present");
+ }
+
+ @Test
+ void fieldConfigViewIsImmutable() {
+ ESIndexOptions options =
+ new ESIndexOptions(
+ Arrays.asList(new DataField(0, "id", DataTypes.INT())), new Options());
+
+ assertThatThrownBy(() -> options.getFieldConfigs().clear())
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThat(options.getConfig("id")).isNotNull();
+ }
+}
diff --git a/pom.xml b/pom.xml
index 26d9a897b0..03b4c4ee2f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -520,6 +520,21 @@ under the License.
</activation>
</profile>
+ <profile>
+ <!-- paimon-eslib depends on Java-11 eslib-api/eslib-core (Lucene 9.12), so it must be
+ built with JDK 11+. The [11,) range activates on JDK 11/17/21; a bare
+ <jdk>11</jdk> is a prefix match that only fires on JDK 11, silently skipping the
+ module (no compile, no test) on 17/21. The default JDK-8 build still skips it, so
+ the RequireJavaVersion enforcer rule is not tripped. -->
+ <id>paimon-eslib</id>
+ <modules>
+ <module>paimon-eslib</module>
+ </modules>
+ <activation>
+ <jdk>[11,)</jdk>
+ </activation>
+ </profile>
+
<profile>
<!-- JDK 9+ supports the ignore-source-errors flag for javadoc -->
<id>javadoc-jdk9+</id>