[FLINK-38721] Support vector search for es connector. by wenjin272 · Pull Request #137 · apache/flink-connector-elasticsearch

wenjin272 · 2025-12-24T09:01:01Z

No description provided.

boring-cyborg · 2025-12-24T09:01:04Z

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

fsk119 · 2026-01-05T12:45:39Z

+        this.networkClientConfig = networkClientConfig;
+        this.hosts = hosts;
+        this.callBridge = callBridge;
+        this.cosineSimilarity =


Does es7 only support cosine as the metric?

fsk119 · 2026-01-05T12:53:13Z

-    private final ElasticsearchConfiguration config;
+    protected final DecodingFormat<DeserializationSchema<RowData>> format;
+    protected final ElasticsearchConfiguration config;
    private final int lookupMaxRetryTimes;


can we use same retry times for lookup and vector search?

xishuaidelin

Hi @wenjin272 , thanks for your contribution. I've left some comments below.

xishuaidelin · 2026-03-09T06:17:04Z

+            ElasticsearchApiCallBridge<RestHighLevelClient> callBridge) {
+
+        checkNotNull(deserializationSchema, "No DeserializationSchema supplied.");
+        checkNotNull(maxRetryTimes, "No maxRetryTimes supplied.");


No need to check type 'int'.

xishuaidelin · 2026-03-09T06:18:20Z

+            NetworkConfig networkConfig) {
+
+        checkNotNull(deserializationSchema, "No DeserializationSchema supplied.");
+        checkNotNull(maxRetryTimes, "No maxRetryTimes supplied.");


No need to check type int.

xishuaidelin · 2026-03-09T06:33:42Z

+                            if (hit.source() != null) {
+                                return new SearchResult(hit.source().toJson().toString(), hit.score());
+                            } else {
+                                return new SearchResult(null, hit.score());


Maybe we should filter the data with source=null to avoid NPE in method parseSearchResult.

searchHits.stream() .filter(hit -> hit.source() != null) .map(hit -> new SearchResult(hit.source().toJson().toString(), hit.score())) .toArray(SearchResult[]::new);

xishuaidelin · 2026-03-09T06:39:46Z

+    private SearchSourceBuilder searchSourceBuilder;
+
+    private final ElasticsearchApiCallBridge<RestHighLevelClient> callBridge;
+    private final NetworkClientConfig networkClientConfig;


This function in ES8 uses NetworkConfig here. Are there any differences that require us to use a different config?

ES7 and ES8 use different client stacks, so the two configs live on opposite sides of the ApiCallBridge boundary:

ES7 ES8

Target client RestHighLevelClient co.elastic.clients.elasticsearch.ElasticsearchClient

Client creation path ElasticsearchApiCallBridge.createClient(NetworkClientConfig, hosts) NetworkConfig.createEsSyncClient() / createEsClient()

Config role Pure data holder Data holder + client factory

Already used by ES7 lookup function ES8 sink and lookup

The two configs carry roughly the same fields, but:

NetworkClientConfig is consumed by ApiCallBridge, which only ES7 implements.

NetworkConfig self-builds the ES8 Java API Client and is what the existing ES8 sink/lookup already depend on.

So this PR follows each module's existing convention rather than introducing a new split.

xishuaidelin · 2026-03-09T06:47:53Z

+        return VectorSearchFunctionProvider.of(vectorSearchFunction);
+    }
+
+    private String getSearchColumn(VectorSearchContext vectorSearchContext) {


This function appears to be identical to the one in ES8. Would it make sense to extract a base class to avoid duplication?

This still seems unresolved.

xishuaidelin · 2026-03-09T06:55:15Z

+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** The {@link VectorSearchFunction} implementation for Elasticsearch. */
+public class ElasticsearchRowDataVectorSearchFunction extends VectorSearchFunction {


Could we introduce an AbstractElasticsearchVectorSearchFunction base class for both the ES7 and ES8 implementations? This would allow us to extract and reuse the shared logic.

…class. Address PR apache#137 review feedback from xishuaidelin: - Introduce AbstractElasticsearchVectorSearchFunction in the base module so ES7 / ES8 share the retry loop, result decoding and SearchResult type; each subclass now only supplies client initialization and the version-specific search call. - Filter hits whose source is null to avoid NPE when deserializing. - Drop redundant checkNotNull on the primitive maxRetryTimes parameter. Also apply spotless formatting drift picked up on adjacent files. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

wenjin272 · 2026-05-15T09:16:19Z

hi, @reswqa. Sorry to bother you. I noticed that you have made significant contributions to the flink-connector-elasticsearch project and have recently been preparing for a release, so I’d like to ask you a question.

FLINK-38721 introduces VectorSearchFunction usage which is only available since Flink 2.2 (FLIP-548). Currently, we aim to support vector search in the flink-connector-elasticsearch module. Although the Flink version bumped on the main branch is 2.2.1, I noticed that CI runs against both Flink 2.1 and 2.2 on the main branch, which causes compilation errors when building against Flink 2.1. I’d like to ask how the community typically resolves this issue.

reswqa · 2026-05-15T09:53:35Z

Hi @wenjin272, Thanks for informing me.

As one ES connector version can be compatible with multiple flink versions. Currently, ES 3.1 supports Flink 1.18, 1.19 and 1.20. And ES 4.0 is the first version that supports flink 2.0.

I originally planned to release 4.1 which would support both 2.1, 2.2 and 2.3 simultaneously. But considering that vector search is a relatively important feature, I want to include it in this round of release.

I think it could be like this: synchronize other important features from main branch to the v4.0 branch, and then release version 4.0.1 (supporting Flink 2.0 and 2.1) and version 4.1 or 5.0 (supporting Flink >= 2.2) at the same time. In this case, the testing version of flink for the main branch are only 2.2 and 2.3.

What do you think?

wenjin272 · 2026-05-16T03:59:34Z

I think it could be like this: synchronize other important features from main branch to the v4.0 branch, and then release version 4.0.1 (supporting Flink 2.0 and 2.1) and version 4.1 (supporting Flink >= 2.2) at the same time. In this case, the testing version of flink for the main branch are only 2.2 and 2.3.

Ty for your kind reply @reswqa. It makes sense to me. I will do my best to push this PR forward in an effort to get it merged next week.

xishuaidelin · 2026-05-19T02:13:18Z

+package org.apache.flink.connector.elasticsearch.table.search;
+
+/** Metric for vector search. */
+public enum SearchMetric {


Missing Apache license header. Several other files have the same issue.

xishuaidelin · 2026-05-19T02:23:37Z

+
+    @Override
+    protected void doOpen(FunctionContext context) {
+        this.client = callBridge.createClient(networkClientConfig, hosts);


The client is created but never closed.

xishuaidelin · 2026-05-19T02:27:23Z

+ * A {@link DynamicTableSource} that describes how to create a {@link Elasticsearch7DynamicSource}
+ * from a logical description.
+ */
+public class Elasticsearch7DynamicSource extends ElasticsearchDynamicSource


This class does not override copy(). If copied, new source would lose the vectorSearch capability.

xishuaidelin · 2026-05-19T02:39:57Z

+        return VectorSearchFunctionProvider.of(vectorSearchFunction);
+    }
+
+    private String getSearchColumn(VectorSearchContext vectorSearchContext) {


This still seems unresolved.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Deduplicate identical getSearchColumn logic from Elasticsearch7DynamicSource and Elasticsearch8DynamicSource into a shared static helper in the base module. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Without overriding copy(), a copied source would fall back to the base ElasticsearchDynamicSource and lose its VectorSearchTableSource capability. Promote lookupCache / docType / summaryString in the base class to protected so the subclass can reconstruct an Elasticsearch7DynamicSource. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

xishuaidelin · 2026-05-19T08:26:29Z

LGTM！

boring-cyborg · 2026-05-19T11:46:55Z

Awesome work, congrats on your first merged pull request!

boring-cyborg Bot added the component=Connectors/Elasticsearch label Dec 24, 2025

fsk119 reviewed Jan 5, 2026

View reviewed changes

wenjin272 force-pushed the support-vector-search branch from d7abd81 to 17b9210 Compare February 2, 2026 07:01

xishuaidelin reviewed Mar 9, 2026

View reviewed changes

wenjin272 force-pushed the support-vector-search branch from 17b75f1 to ae37638 Compare April 21, 2026 06:30

wenjin272 force-pushed the support-vector-search branch from ae37638 to 39368cb Compare April 21, 2026 07:19

boring-cyborg Bot added the component=BuildSystem label Apr 21, 2026

wenjin272 force-pushed the support-vector-search branch 2 times, most recently from 722bc8c to 736b411 Compare April 21, 2026 09:26

wenjin272 requested a review from xishuaidelin April 22, 2026 10:45

wenjin272 and others added 2 commits May 15, 2026 16:26

[FLINK-38721] Support vector search for es7 connector.

61446a9

[FLINK-38721] Support vector search for es8 connector.

5a3737b

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

wenjin272 force-pushed the support-vector-search branch from 736b411 to 5a3737b Compare May 15, 2026 08:27

[hotfix] Update CI run flink versions for vector search

638a93a

xishuaidelin reviewed May 19, 2026

View reviewed changes

reswqa mentioned this pull request May 19, 2026

[FLINK-39706][ci] Remove flink 2.1 testing from es connector main branch #149

Open

wenjin272 and others added 4 commits May 19, 2026 12:58

Add Apache license header to vector search files

0d57ed1

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Extract shared search column resolution to VectorSearchUtils

cb59a13

Deduplicate identical getSearchColumn logic from Elasticsearch7DynamicSource and Elasticsearch8DynamicSource into a shared static helper in the base module. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Close Elasticsearch client on vector search function close

1c0e4a2

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

wenjin272 merged commit 08b7d66 into apache:main May 19, 2026
4 checks passed

	ES7	ES8
Target client	`RestHighLevelClient`	`co.elastic.clients.elasticsearch.ElasticsearchClient`
Client creation path	`ElasticsearchApiCallBridge.createClient(NetworkClientConfig, hosts)`	`NetworkConfig.createEsSyncClient()` / `createEsClient()`
Config role	Pure data holder	Data holder + client factory
Already used by	ES7 lookup function	ES8 sink and lookup

Conversation

wenjin272 commented Dec 24, 2025

Uh oh!

boring-cyborg Bot commented Dec 24, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xishuaidelin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wenjin272 commented May 15, 2026

Uh oh!

reswqa commented May 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

wenjin272 commented May 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xishuaidelin commented May 19, 2026

Uh oh!

Uh oh!

boring-cyborg Bot commented May 19, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

reswqa commented May 15, 2026 •

edited

Loading

wenjin272 commented May 16, 2026 •

edited

Loading