Alibaba Cloud Elasticsearch accelerates Paimon multi-modal data lake search

How it works

The raw data in the Paimon table remains in Object Storage Service (OSS) or other lakehouse storage. After Elasticsearch reads the Paimon table snapshot and global index metadata, it mounts the data lake table index as a queryable Elasticsearch index. You can use Elasticsearch Query DSL, KNN queries, filter conditions, and field return capabilities to access data in the data lake without synchronizing all the raw data to an online Elasticsearch index.

The implementation works as follows:

• Build the vector columns in the Paimon table into vector indexes queryable by Elasticsearch.

• Write scalar columns such as business IDs, tags, URI paths, timestamps, and categories into the global index for filtering and retrieval.

• Mount the Paimon table as an Elasticsearch index by using /_paimon/mount.

• Use Elasticsearch KNN queries for vector recall combined with filter conditions such as term, terms, and range.

• After hitting TopK results, read fields from the Paimon table on demand to reduce redundant storage of raw data in the online index.

• Use Spark SQL vector_search on the Paimon side to verify recall results for the same query vector.

A typical pipeline is as follows:

Business data / images / text / Embedding
  -> Write to Paimon table
  -> Build Paimon global index
  -> Elasticsearch mounts the Paimon table
  -> Query using ES Query DSL / KNN
  -> Return scalar columns or source fields on demand after hits

Scenarios

This feature is applicable to the following scenarios:

• Multi-modal data such as images, text, and video frames are stored in Paimon/OSS, and you want to provide online similarity search services through Elasticsearch.

• Your business requires both vector search and structured filtering, such as "similar images + category filtering" and "similar documents + tag filtering."

• You do not want to synchronize all fields from the data lake table to Elasticsearch and only want to keep the search index and necessary return fields.

• You need to use Elasticsearch query DSL, sorting, filtering, aggregation, Kibana, or RAG application ecosystem as the search entry for data in the data lake.

Prerequisites

The following table describes the overall requirements for using this feature. Step 1 and Step 2 in this tutorial demonstrate how to create a Paimon table from scratch and build an es-index global index by using the example tablepaimon_vector_demo. If you already have a Paimon table and an es-index global index that meet the requirements, you can start from Step 3.

Limits

• If you need to backfill fields through _source or Paimon source, you must correctly enable row-tracking.enabled=true and data-evolution.enabled=true in the table creation and write pipeline. Otherwise, data files may lack firstRowId, which prevents the source stage from establishing the mapping from rowId to data files.

• Accompanying scalar columns that need to participate in filtering or be written to the global index must be declared together with the vector column in index_column in Spark SQL. Only placing a field in return_fields or _source does not guarantee that it can be used for filter queries such as term and range.

• If the vector field uses dot_product as the similarity metric, both the vectors written to the table and the query vectors typically need to be L2 normalized first to ensure that scores and rankings are comparable.

• The dimension limits, recall parameters, and performance characteristics of vector index algorithms such as HNSW and DiskBBQ are subject to the current Elasticsearch instance version.

• If large-scale HNSW indexes use the heap mount mode, you need to evaluate JVM heap, shard count, and vector graph size in advance. For feature verification, use the mmap or hybrid mode first.

• The query vector of Spark SQL vector_search must be written as Float/Double array literals parseable by Spark, for example array(1.234E-2D, -3.456E-2D), to avoid being parsed as DECIMAL or expressions.

Step 1: Prepare the Paimon table

The following example uses an OSS filesystem catalog. For DLF catalog scenarios, replace the catalog and table paths with your actual configurations.

# Use the Paimon Spark extension
spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions

# Configure the OSS filesystem catalog
spark.sql.catalog.oss=org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.oss.metastore=filesystem
spark.sql.catalog.oss.warehouse=oss://<bucket>/<warehouse>
spark.sql.catalog.oss.fs.oss.endpoint=oss-<region>-internal.aliyuncs.com
spark.sql.catalog.oss.fs.oss.accessKeyId=${OSS_AK_ID}
spark.sql.catalog.oss.fs.oss.accessKeySecret=${OSS_AK_SECRET}

# Hadoop OSS read/write configuration
spark.hadoop.fs.oss.endpoint=oss-<region>-internal.aliyuncs.com
spark.hadoop.fs.oss.accessKeyId=${OSS_AK_ID}
spark.hadoop.fs.oss.accessKeySecret=${OSS_AK_SECRET}

# Prevent Paimon v1 function from preempting external function resolution
spark.paimon.v1Function.enabled=false

CREATE TABLE oss.default.paimon_vector_demo (
  id          BIGINT,
  label       BIGINT,
  content_uri STRING,
  emb         ARRAY<DOUBLE>,
  emb_norm    ARRAY<FLOAT>
) TBLPROPERTIES (
  'write-mode' = 'append-only',
  'bucket' = '-1',
  'file.format' = 'parquet',
  'row-tracking.enabled' = 'true',
  'data-evolution.enabled' = 'true',

  'global-index.es-index.fields.emb_norm.algorithm' = 'hnsw',
  'global-index.es-index.fields.emb_norm.dimension' = '768',
  'global-index.es-index.fields.emb_norm.metric' = 'dot_product'
);

INSERT INTO oss.default.paimon_vector_demo
SELECT
  id,
  label,
  content_uri,
  emb,
  vector_l2_normalize(emb) AS emb_norm
FROM source_table;

INSERT INTO oss.default.paimon_vector_demo
SELECT
  id,
  label,
  content_uri,
  emb,
  transform(
    emb,
    x -> CAST(x / sqrt(aggregate(emb, CAST(0 AS DOUBLE), (acc, v) -> acc + v * v)) AS FLOAT)
  ) AS emb_norm
FROM source_table;

SELECT count(*) AS total_count
FROM oss.default.paimon_vector_demo;

SELECT
  id,
  size(emb_norm) AS dim,
  aggregate(emb_norm, CAST(0 AS FLOAT), (acc, v) -> acc + v * v) AS sumsq
FROM oss.default.paimon_vector_demo
LIMIT 10;

Step 2: Build the Paimon global index

Call the Paimon create_global_index stored procedure in EMR Serverless Spark SQL to build the es-index (in the example,oss.sys.create_global_index, where oss is the catalog name configured in Step 1).

Currently, the index type of products automatically built by managed DLF is inconsistent with the es-index required by the Elasticsearch paimon-store, and cannot directly replace this step. The DLF catalog can still be used to manage Paimon table metadata.

Configure the Spark job and run the build command:

# Older versions of EMR Serverless Spark: exclude and replace the built-in Paimon
spark.emr.serverless.excludedModules             paimon
spark.emr.serverless.user.defined.jars           oss://<bucket>/jars/<paimon-es-index-bundle>.jar
spark.sql.extensions                             org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions

-- Use Spark SQL to build the es-index
CALL oss.sys.create_global_index(
  table        => 'default.paimon_vector_demo',
  index_column => 'emb_norm,id,label,content_uri',
  index_type   => 'es-index',
  options      => 'global-index.row-count-per-shard=100000,global-index.es-index.fields.emb_norm.m=30,global-index.es-index.fields.emb_norm.ef_construction=360'
);

The Spark configuration and SQL parameter descriptions are as follows.

SELECT
  index_type,
  index_field_name,
  row_count,
  file_name
FROM oss.default.`paimon_vector_demo$table_indexes`
WHERE index_type = 'es-index';

Step 3: Mount the Paimon table as an Elasticsearch index

Call the /_paimon/mount API to mount the Paimon table as an Elasticsearch index.

POST /_paimon/mount
{
  "auth_type": "direct",
  "table_path": "oss://<bucket>/<warehouse>/default.db/paimon_vector_demo",
  "oss_endpoint": "oss-<region>-internal.aliyuncs.com",
  "oss_bucket": "<bucket>",
  "oss_access_key_id": "${OSS_AK_ID}",
  "oss_access_key_secret": "${OSS_AK_SECRET}",
  "index_name": "paimon_vector_demo_index",
  "vector_field_name": "emb_norm",
  "storage_mode": "mmap",
  "source_enabled": true,
  "return_fields": [
    "id",
    "label",
    "content_uri"
  ],
  "timeout": "600"
}

If the call is successful, a response similar to the following is returned:

```json { "acknowledged": true, "alias": "paimon_vector_demo_index", "index": "paimon_vector_demo_index_2" }

<!-- @id="mountrespnote1" -->The mount operation creates an actual index (whose name is<!-- @id="mountrn01" --> `index_name` with a numeric suffix, for example<!-- @id="mountrn02" --> `paimon_vector_demo_index_2`), and points the alias<!-- @id="mountrn03" --> `index_name` to that index. Use the alias<!-- @id="mountrn04" --> `paimon_vector_demo_index` for subsequent queries. In Step 4,<!-- @id="mountrn05" --> `_cat/indices` displays the actual index name with the numeric suffix.

The parameter descriptions are as follows. Among these, `table_path`, `oss_endpoint`, `oss_bucket`, `oss_access_key_id`, and `oss_access_key_secret` are required parameters. The API returns a `Required fields` error if any of them is missing.

| Parameter | Required | Description |
| --- | --- | --- |
| `auth_type` | No | Authentication method. The example uses `direct`, which passes OSS access credentials in the request. For production environments, use more secure managed credentials or RAM role methods. |
| `table_path` | Yes | The Paimon table path. |
| `oss_endpoint` | Yes | The OSS endpoint. Use an internal endpoint in the same region as the Elasticsearch instance. |
| `oss_bucket` | Yes | The OSS bucket where the table is located. |
| `oss_access_key_id` | Yes | The AccessKey ID for accessing OSS. |
| `oss_access_key_secret` | Yes | The AccessKey secret for accessing OSS. |
| `index_name` | No | The name of the Elasticsearch index generated after mounting. |
| `vector_field_name` | No | The name of the field used for vector search. |
| `storage_mode` | No | The index read mode. Valid values: `mmap` (relies on OS page cache, suitable for verification and low memory usage), `hybrid` (balances memory and disk access, suitable for medium to large-scale indexes), and `heap` (best performance but consumes more JVM heap). Available values are subject to the current version. For a detailed comparison, see [How to choose between mmap, hybrid, and heap?](#how-to-choose-between-mmap-hybrid-and-heap) in the FAQ section. |
| `source_enabled` | No | Whether to enable Paimon source retrieval. |
| `return_fields` | No | The fields to be returned from the Paimon table. |
| `timeout` | No | The mount timeout period, in seconds. |

(Recommended) Use `mmap` or `hybrid` for initial verification. If you need to use the `heap` mode for performance benchmarking, evaluate the index scale and JVM heap first.

## Step 4: Verify the mount result

After the mount is complete, check the index status, document count, mapping, and settings:

```bash
GET /_cat/indices/paimon_vector_demo_index?v

GET /paimon_vector_demo_index/_count

GET /paimon_vector_demo_index/_mapping

GET /paimon_vector_demo_index/_settings?flat_settings=true

Focus on the following items:

• The index status is green or as expected.

• _count is consistent with the Paimon table row count.

• The vector field mapping has the correct dimension and similarity configurations, for example dims=768 and similarity=dot_product.

• Scalar fields id, label, and content_uri appear in the mapping.

• Settings such as Paimon table path, snapshot ID, and storage mode are as expected.

Step 5: Query by using Elasticsearch

POST /paimon_vector_demo_index/_search
{
  "size": 10,
  "_source": [
    "id",
    "label",
    "content_uri"
  ],
  "knn": {
    "field": "emb_norm",
    "query_vector": [/* 768-dimensional normalized query vector */],
    "k": 10,
    "num_candidates": 100
  }
}

POST /paimon_vector_demo_index/_search
{
  "size": 10,
  "_source": [
    "id",
    "label",
    "content_uri"
  ],
  "knn": {
    "field": "emb_norm",
    "query_vector": [/* 768-dimensional normalized query vector */],
    "k": 10,
    "num_candidates": 100,
    "filter": {
      "term": {
        "label": 1
      }
    }
  }
}

{
  "terms": {
    "label": [1, 2, 3]
  }
}

{
  "range": {
    "id": {
      "lt": 100000
    }
  }
}

POST /paimon_vector_demo_index/_search
{
  "size": 10,
  "_source": true,
  "knn": {
    "field": "emb_norm",
    "query_vector": [/* 768-dimensional normalized query vector */],
    "k": 10,
    "num_candidates": 100
  }
}

Step 6: Verify query results by using Spark SQL

You can also use the same query vector in Spark SQL to verify the Paimon global index query results.

Run the following Spark SQL query:

USE oss.default;

SELECT
  id,
  label,
  content_uri,
  __paimon_search_score AS score
FROM vector_search(
  'oss.default.paimon_vector_demo',
  'emb_norm',
  array(/* 768 double literals, e.g. 1.234567E-2D */),
  10,
  map('hnsw.num_candidates', '100')
)
ORDER BY score DESC
LIMIT 10;

Vector + scalar filtering example:

SELECT
  id,
  label,
  content_uri,
  __paimon_search_score AS score
FROM vector_search(
  'oss.default.paimon_vector_demo',
  'emb_norm',
  array(/* same normalized query vector */),
  10,
  map('hnsw.num_candidates', '100')
)
WHERE label = 1
ORDER BY score DESC
LIMIT 10;

Use EXPLAIN FORMATTED to check whether the vector index is used:

EXPLAIN FORMATTED
SELECT
  id,
  label,
  __paimon_search_score AS score
FROM vector_search(
  'oss.default.paimon_vector_demo',
  'emb_norm',
  array(/* same normalized query vector */),
  10,
  map('hnsw.num_candidates', '100')
)
WHERE label = 1
ORDER BY score DESC;

The expected plan should show vector search, global index scan, or VectorSearch related information.

FAQ