[Feature][Connector-V2] Add tencent vector db source connector

.github/workflows/labeler/label-scope-conf.yml

@@ -264,6 +264,12 @@ qdrant:
           - any-glob-to-any-file: seatunnel-connectors-v2/connector-qdrant/**
           - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(qdrant)/**'
 
+tencentvectordb:
+  - all:
+      - changed-files:
+          - any-glob-to-any-file: seatunnel-connectors-v2/connector-tencent-vectordb/**
+          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(tencent-vectordb)/**'
+
 typesense:
   - all:
       - changed-files:

config/plugin_config

@@ -92,3 +92,4 @@ connector-sls
 connector-qdrant
 connector-typesense
 connector-cdc-opengauss
+connector-tencent-vectordb

docs/en/connector-v2/source/TencentVectorDB.md

@@ -0,0 +1,46 @@
+# Tencent VectorDB
+
+> Tencent VectorDB source connector
+
+## Description
+
+This Tencent VectorDB source connector reads data from Tencent VectorDB
+## Key Features
+
+- [x] [batch](../../concept/connector-v2-features.md)
+- [x] [exactly-once](../../concept/connector-v2-features.md)
+- [ ] [column projection](../../concept/connector-v2-features.md)
+
+## Data Type Mapping
+
+| Pinecone Data Type  | SeaTunnel Data Type |
+|---------------------|---------------------|
+| FLOAT_VECTOR        | FLOAT_VECTOR        |
+| SPARSE_FLOAT_VECTOR | SPARSE_FLOAT_VECTOR |
+
+## Source Options
+
+| Name       | Type    | Required | Default | Description                 |
+|------------|---------|----------|---------|-----------------------------|
+| url        | String  | Yes      | -       | endpoint                    |
+| user_name  | String  | Yes      | -       | user name                   |
+| api_key    | String  | Yes      | -       | api key for authentication. |
+| database   | String  | Yes      | -       | database                    |
+| collection | String  | Yes      | -       | collection name             |
+| batch_size | Integer | No       | 100     | Batch size to read          |
+
+## Task Example
+
+```bash
+source {
+  TencentVectorDB {
+    url = "*****"
+    user_name = "root"
+    api_key = "*****"
+    database = "default"
+    collection = "movie_reviews"
+  }
+}
+```
+
+## Changelog

plugin-mapping.properties

@@ -139,6 +139,7 @@ seatunnel.sink.Sls = connector-sls
 seatunnel.source.Typesense = connector-typesense
 seatunnel.sink.Typesense = connector-typesense
 seatunnel.source.Opengauss-CDC = connector-cdc-opengauss
+seatunnel.source.TencentVectorDB = connector-tencent-vectordb
 
 seatunnel.transform.Sql = seatunnel-transforms-v2
 seatunnel.transform.FieldMapper = seatunnel-transforms-v2

seatunnel-connectors-v2/connector-tencent-vectordb/pom.xml

@@ -0,0 +1,60 @@
+<?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>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-tencent-vectordb</artifactId>
+    <name>SeaTunnel : Connectors V2 : TencentVectorDB</name>
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>com.google.code.gson</groupId>
+                <artifactId>gson</artifactId>
+                <version>2.10.1</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+    <dependencies>
+        <dependency>
+            <groupId>com.tencent.tcvectordb</groupId>
+            <artifactId>vectordatabase-sdk-java</artifactId>
+            <version>2.0.4</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.code.gson</groupId>
+                    <artifactId>gson</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.10.1</version>
+        </dependency>
+    </dependencies>
+
+</project>

seatunnel-connectors-v2/connector-tencent-vectordb/src/main/java/org/apache/seatunnel/connectors/tencent/vectordb/config/TencentVectorDBSourceConfig.java

@@ -0,0 +1,57 @@
+/*
+ * 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.seatunnel.connectors.tencent.vectordb.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+public class TencentVectorDBSourceConfig {
+    public static final String CONNECTOR_IDENTITY = "TencentVectorDB";
+
+    public static final Option<String> URL =
+            Options.key("url").stringType().noDefaultValue().withDescription("url");
+
+    public static final Option<String> USER_NAME =
+            Options.key("user_name")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("user name for authentication");
+
+    public static final Option<String> API_KEY =
+            Options.key("api_key")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("token for authentication");
+
+    public static final Option<String> DATABASE =
+            Options.key("database")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Tencent Vector DB database name");
+
+    public static final Option<String> COLLECTION =
+            Options.key("collection")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Tencent Vector DB collection name");
+    public static final Option<Integer> BATCH_SIZE =
+            Options.key("batch_size")
+                    .intType()
+                    .defaultValue(100)
+                    .withDescription("Tencent Vector DB reader batch size");
+}

seatunnel-connectors-v2/connector-tencent-vectordb/src/main/java/org/apache/seatunnel/connectors/tencent/vectordb/exception/TencentVectorDBConnectorErrorCode.java

@@ -0,0 +1,36 @@
+/*
+ * 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.seatunnel.connectors.tencent.vectordb.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+
+import lombok.Getter;
+
+@Getter
+public enum TencentVectorDBConnectorErrorCode implements SeaTunnelErrorCode {
+    SOURCE_TABLE_SCHEMA_IS_NULL("TC-VECTORDB-01", "Source table schema is null"),
+    READ_DATA_FAIL("TC-VECTORDB-02", "Read data fail");
+
+    private final String code;
+    private final String description;
+
+    TencentVectorDBConnectorErrorCode(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+}

seatunnel-connectors-v2/connector-tencent-vectordb/src/main/java/org/apache/seatunnel/connectors/tencent/vectordb/exception/TencentVectorDBConnectorException.java

@@ -0,0 +1,32 @@
+/*
+ * 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.seatunnel.connectors.tencent.vectordb.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class TencentVectorDBConnectorException extends SeaTunnelRuntimeException {
+    public TencentVectorDBConnectorException(SeaTunnelErrorCode seaTunnelErrorCode) {
+        super(seaTunnelErrorCode, seaTunnelErrorCode.getErrorMessage());
+    }
+
+    public TencentVectorDBConnectorException(
+            SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
+        super(seaTunnelErrorCode, seaTunnelErrorCode.getErrorMessage(), cause);
+    }
+}

seatunnel-connectors-v2/connector-tencent-vectordb/src/main/java/org/apache/seatunnel/connectors/tencent/vectordb/source/TencentVectorDBSource.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.seatunnel.connectors.tencent.vectordb.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.source.Boundedness;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.api.source.SupportColumnProjection;
+import org.apache.seatunnel.api.source.SupportParallelism;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.tencent.vectordb.config.TencentVectorDBSourceConfig;
+import org.apache.seatunnel.connectors.tencent.vectordb.utils.ConnectorUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class TencentVectorDBSource
+        implements SeaTunnelSource<
+                        SeaTunnelRow, TencentVectorDBSourceSplit, TencentVectorDBSourceState>,
+                SupportParallelism,
+                SupportColumnProjection {
+    private final ReadonlyConfig config;
+    private final Map<TablePath, CatalogTable> sourceTables;
+
+    public TencentVectorDBSource(ReadonlyConfig config) {
+        this.config = config;
+        ConnectorUtils connectorUtils = new ConnectorUtils(config);
+        this.sourceTables = connectorUtils.getSourceTables();
+    }
+
+    /**
+     * Get the boundedness of this source.
+     *
+     * @return the boundedness of this source.
+     */
+    @Override
+    public Boundedness getBoundedness() {
+        return Boundedness.BOUNDED;
+    }
+
+    /**
+     * Create source reader, used to produce data.
+     *
+     * @param readerContext reader context.
+     * @return source reader.
+     * @throws Exception when create reader failed.
+     */
+    @Override
+    public SourceReader<SeaTunnelRow, TencentVectorDBSourceSplit> createReader(
+            SourceReader.Context readerContext) throws Exception {
+        return new TencentVectorDBSourceReader(readerContext, config, sourceTables);
+    }
+
+    @Override
+    public List<CatalogTable> getProducedCatalogTables() {
+        return new ArrayList<>(sourceTables.values());
+    }
+
+    /**
+     * Create source split enumerator, used to generate splits. This method will be called only once
+     * when start a source.
+     *
+     * @param enumeratorContext enumerator context.
+     * @return source split enumerator.
+     * @throws Exception when create enumerator failed.
+     */
+    @Override
+    public SourceSplitEnumerator<TencentVectorDBSourceSplit, TencentVectorDBSourceState>
+            createEnumerator(
+                    SourceSplitEnumerator.Context<TencentVectorDBSourceSplit> enumeratorContext)
+                    throws Exception {
+        return new TencentVectorDBSourceSplitEnumertor(
+                enumeratorContext, config, sourceTables, null);
+    }
+
+    /**
+     * Create source split enumerator, used to generate splits. This method will be called when
+     * restore from checkpoint.
+     *
+     * @param enumeratorContext enumerator context.
+     * @param checkpointState checkpoint state.
+     * @return source split enumerator.
+     * @throws Exception when create enumerator failed.
+     */
+    @Override
+    public SourceSplitEnumerator<TencentVectorDBSourceSplit, TencentVectorDBSourceState>
+            restoreEnumerator(
+                    SourceSplitEnumerator.Context<TencentVectorDBSourceSplit> enumeratorContext,
+                    TencentVectorDBSourceState checkpointState)
+                    throws Exception {
+        return new TencentVectorDBSourceSplitEnumertor(
+                enumeratorContext, config, sourceTables, checkpointState);
+    }
+
+    /**
+     * Returns a unique identifier among same factory interfaces.
+     *
+     * <p>For consistency, an identifier should be declared as one lower case word (e.g. {@code
+     * kafka}). If multiple factories exist for different versions, a version should be appended
+     * using "-" (e.g. {@code elasticsearch-7}).
+     */
+    @Override
+    public String getPluginName() {
+        return TencentVectorDBSourceConfig.CONNECTOR_IDENTITY;
+    }
+}