Merge pull request #51 from zilliztech/opensearch

add vector support for opensearch
zilliztech · Feb 6, 2025 · 677c1f3 · 677c1f3
2 parents f8c5729 + 44083a1
commit 677c1f3
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 2 deletions.
diff --git a/...ache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchTypeConverter.java b/...ache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchTypeConverter.java
@@ -55,6 +55,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.IP_RANGE;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.JOIN;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.KEYWORD;
+import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.KNN_VECTOR;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.LONG;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.LONG_RANGE;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.MATCH_ONLY_TEXT;
@@ -107,8 +108,9 @@ public Column convert(BasicTypeDefine<EsType> typeDefine) {
                 break;
             case DENSE_VECTOR:
             case VECTOR:
-                String elementType =
-                        typeDefine.getNativeType().getOptions().get("element_type").toString();
+            case KNN_VECTOR:
+                String elementType = typeDefine.getNativeType().getOptions().containsKey("element_type") ?
+                        String.valueOf(typeDefine.getNativeType().getOptions().get("element_type")) : "";
                 if (elementType.equals("byte")) {
                     builder.dataType(ArrayType.BYTE_ARRAY_TYPE);
                 } else {

diff --git a/...ain/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java b/...ain/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
@@ -20,6 +20,7 @@
 import org.apache.http.Header;
 import org.apache.http.HttpRequestInterceptor;
 import org.apache.http.message.BasicHeader;
+import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.KNN_VECTOR;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsType.VECTOR;
 import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper;
@@ -680,6 +681,19 @@ private static Map<String, BasicTypeDefine<EsType>> getFieldTypeMappingFromPrope
                                     // es dense_vector dim
                                     typeDefine.scale(fieldProperty.get("dimension").asInt());
 
+                                } else if (type.equalsIgnoreCase(KNN_VECTOR)) {
+                                //adapt for huawei cloud elastic search
+                                String elementType =
+                                        fieldProperty.get("data_type") == null
+                                                ? "float"
+                                                : fieldProperty.get("data_type").asText();
+                                Map<String, Object> options = new HashMap<>();
+                                options.put("element_type", elementType);
+                                typeDefine.nativeType(new EsType(type, options));
+
+                                // es dense_vector dim
+                                typeDefine.scale(fieldProperty.get("dimension").asInt());
+
                                 } else if (type.equalsIgnoreCase(DATE)
                                         || type.equalsIgnoreCase(DATE_NANOS)) {
                                     String format =

diff --git a/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsType.java b/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsType.java
@@ -36,6 +36,7 @@ public class EsType {
     public static final String DATE_NANOS = "date_nanos";
     public static final String DENSE_VECTOR = "dense_vector";
     public static final String VECTOR = "vector";
+    public static final String KNN_VECTOR = "knn_vector";
     public static final String DOUBLE = "double";
     public static final String FLATTENED = "flattened";
     public static final String FLOAT = "float";