`），并按属性名排序。适合「整 label 全字段 dump」而又不想逐字段手写声明的场景。
+
+```hocon
+source {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ label = "person"
+ label_type = "VERTEX"
+ # 不写 schema：读取 "person" 的全部属性
+ }
+}
+```
+
+注意：
+
+- 该 label 必须已存在于服务端，否则作业在构建阶段失败。
+- 无任何属性键的 label 只会产生保留列（`~id`、`~label` 等）。
+- 当只想读取部分属性、固定列顺序或指定类型时，请显式声明 `schema.fields`。
+
+## 读取全部 label
+
+省略 `label` 即可在单个作业中读取 `label_type`（默认 `VERTEX`）下的**全部** label——适合整图迁移 / 备份，而无需为每个 label 各配一个 source。作业构建阶段 connector 会从服务端 schema 列出该类型的所有 label，为每个 label 产出一张输出表，各自按 [Schema 自动发现](#schema-自动发现)推断列。每行都会带上其 label 对应的 table id，因此下游多表 sink 可据此将行路由到对应表。
+
+```hocon
+source {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ label_type = "VERTEX"
+ # 不写 label：读取全部顶点 label，每个 label 一张表
+ }
+}
+```
+
+注意：
+
+- 一个作业读取顶点**或**边，不能混读：设置 `label_type = "EDGE"` 以读取全部边 label。
+- 不允许配置 `schema`（单一 schema 无法描述多个 label），列始终按 label 自动发现。
+- 不允许配置 `filter`（属性等值过滤要求该属性存在于每个 label）。
+- 每个 label 对应一个 `LABEL_LIST` split，分配给各 Reader（并行度上限为 label 数量）。此模式不使用单个 label 内部的 shard 级并行。
+- 若图中不存在该类型的任何 label，作业在构建阶段失败。
+
+## 并行读取
+
+对于大图，设置 `parallelism > 1` 可并行读取一个 label。Enumerator 请求 HugeGraph 将该 label 的 keyspace 切分为大小约为 `split_size` 字节的多个 shard，并以 round-robin 方式分配给各 Reader，使吞吐随并行度提升，而不再受单一分页游标限制。
+
+```hocon
+source {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ label = "person"
+ label_type = "VERTEX"
+ parallelism = 8
+ split_size = 1048576
+ schema = {
+ fields = {
+ name = "string"
+ age = "int"
+ }
+ }
+ }
+}
+```
+
+注意：
+
+- Shard 扫描需要支持 scan 的后端（RocksDB / HBase / Cassandra）；`memory` 后端不支持 shard 切分，请在其上使用 `parallelism = 1`。
+- Shard 扫描会返回 key-range 内所有 label 的元素，connector 仅保留配置的 `label`。当目标 label 只占全图很小比例时，单并行度的 `filter` 读取可能搬运更少数据（尽管无法并行）。
+- `filter` 不能与 `parallelism > 1` 同时使用；要用服务端过滤请保持 `parallelism = 1`，要并行请去掉 filter。
+- 调优 `split_size`：值越小 shard 越多越小（负载更均衡、请求更多）；值越大 shard 越少越大。最小值为 `1048576`（1 MiB），更小的值会被拒绝，以避免把 keyspace 切分成过多的 shard。
+
+## Changelog
+
+
diff --git a/plugin-mapping.properties b/plugin-mapping.properties
index af93a9c8f781..d4b0b1e138a3 100644
--- a/plugin-mapping.properties
+++ b/plugin-mapping.properties
@@ -159,6 +159,7 @@ seatunnel.source.GraphQL = connector-graphql
seatunnel.sink.GraphQL = connector-graphql
seatunnel.sink.Aerospike = connector-aerospike
seatunnel.sink.SensorsData = connector-sensorsdata
+seatunnel.source.HugeGraph = connector-hugegraph
seatunnel.sink.HugeGraph = connector-hugegraph
seatunnel.source.Fluss = connector-fluss
seatunnel.sink.Fluss = connector-fluss
diff --git a/pom.xml b/pom.xml
index 058d75d41ada..8a49c5f06a33 100644
--- a/pom.xml
+++ b/pom.xml
@@ -148,7 +148,7 @@
2.12.15
9.4.56.v20240826
4.0.4
- 1.5.0
+ 1.7.0
false
true
diff --git a/seatunnel-connectors-v2/connector-hugegraph/pom.xml b/seatunnel-connectors-v2/connector-hugegraph/pom.xml
index d8cf4077fd05..2faf3ae6c9ab 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/pom.xml
+++ b/seatunnel-connectors-v2/connector-hugegraph/pom.xml
@@ -29,10 +29,6 @@
connector-hugegraph
SeaTunnel : Connectors V2 : HugeGraph
-
- 1.5.0
-
-
org.apache.seatunnel
@@ -44,12 +40,53 @@
org.apache.hugegraph
hugegraph-client
${hugegraph.client.version}
+
+
+
+ junit
+ junit
+
+
+ org.hamcrest
+ hamcrest-core
+
+
+
+ org.apache.hugegraph
+ hg-pd-client
+
+
+ org.apache.hugegraph
+ hg-pd-common
+
+
+ org.apache.hugegraph
+ hg-pd-grpc
+
+
+ io.grpc
+ *
+
+
org.apache.hugegraph
hugegraph-common
${hugegraph.client.version}
+
+
+ junit
+ junit
+
+
+ org.hamcrest
+ hamcrest-core
+
+
@@ -67,6 +104,68 @@
+
+ org.apache.maven.plugins
+ maven-shade-plugin
+
+
+
+ shade
+
+ package
+
+
+
+
+ com.google.common
+ ${seatunnel.shade.package}.hugegraph.com.google.common
+
+
+ com.google.thirdparty
+ ${seatunnel.shade.package}.hugegraph.com.google.thirdparty
+
+
+ com.fasterxml.jackson
+ ${seatunnel.shade.package}.hugegraph.com.fasterxml.jackson
+
+
+
+ okhttp3
+ ${seatunnel.shade.package}.hugegraph.okhttp3
+
+
+ okio
+ ${seatunnel.shade.package}.hugegraph.okio
+
+
+
+ javassist
+ ${seatunnel.shade.package}.hugegraph.javassist
+
+
+ org.joda.time
+ ${seatunnel.shade.package}.hugegraph.org.joda.time
+
+
+
+
+
+
+
+ *:*
+
+ META-INF/maven/**
+
+
+
+
+
+
+
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/BatchBuffer.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/BatchBuffer.java
index c3851129390a..bd6575488e3c 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/BatchBuffer.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/BatchBuffer.java
@@ -18,30 +18,50 @@
package org.apache.seatunnel.connectors.seatunnel.hugegraph.buffer;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.client.HugeGraphClient;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.MappingConfig.LabelType;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorErrorCode;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorException;
import org.apache.hugegraph.structure.GraphElement;
import org.apache.hugegraph.structure.graph.Edge;
+import org.apache.hugegraph.structure.graph.UpdateStrategy;
import org.apache.hugegraph.structure.graph.Vertex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+/**
+ * Dual-bucket batch buffer that independently accumulates and flushes vertices and edges. Each
+ * bucket triggers flush when reaching batch_size; both buckets are flushed on timer, prepareCommit,
+ * or close.
+ *
+ * Vertex-before-edge ordering is enforced only when {@code check_vertex} is true — the server
+ * then rejects edges whose endpoint vertices do not yet exist, so a filling edge bucket first
+ * flushes any pending vertices, and {@link #flush()} writes vertices before edges. When {@code
+ * check_vertex} is false (the default) the server already accepts orphan edges, so the buckets
+ * flush independently for higher throughput (no forced, undersized vertex flushes).
+ */
public class BatchBuffer implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(BatchBuffer.class);
- private final List buffer = new ArrayList<>();
+ private final List vertexBuffer = new ArrayList<>();
+ private final List edgeBuffer = new ArrayList<>();
private final int batchSize;
private final ScheduledExecutorService scheduler;
private final ScheduledFuture scheduledFuture;
@@ -49,11 +69,61 @@ public class BatchBuffer implements AutoCloseable {
private volatile boolean closed = false;
private volatile Exception flushException;
private final HugeGraphClient client;
+ private final boolean batchFailureFallback;
+ private final boolean checkVertex;
+ // Fail the task once this many records have been skipped by the single-record fallback;
+ // negative means unlimited. Guards against the previously unbounded silent skipping.
+ private final int maxInsertErrors;
+ // Optional directory for skipped-record failure samples; null = do not persist.
+ private final String failureDataPath;
+ private final int subtaskIndex;
+ // Cumulative count of records skipped by the fallback across this writer's lifetime. Only
+ // mutated inside synchronized flush paths, so a plain long is sufficient.
+ private long insertFailureCount;
+ // Lazily opened on the first persisted sample; disabled after an I/O error so a broken
+ // failure-log path never turns into a second failure that masks the real one.
+ private BufferedWriter failureWriter;
+ private boolean failureWriterDisabled;
+ /**
+ * Backward-compatible constructor that retains the original 3-argument signature. Defaults
+ * {@code batchFailureFallback} and {@code checkVertex} to {@code false}, matching the pre-2.x
+ * behaviour where neither feature existed.
+ *
+ * @deprecated Use {@link #BatchBuffer(HugeGraphClient, int, long, boolean, boolean)} instead so
+ * callers explicitly opt into failure-fallback and vertex-checking semantics.
+ */
+ @Deprecated
public BatchBuffer(HugeGraphClient client, int batchSize, long batchIntervalMs) {
+ this(client, batchSize, batchIntervalMs, false, false);
+ }
+
+ public BatchBuffer(
+ HugeGraphClient client,
+ int batchSize,
+ long batchIntervalMs,
+ boolean batchFailureFallback,
+ boolean checkVertex) {
+ this(client, batchSize, batchIntervalMs, batchFailureFallback, checkVertex, -1, null, 0);
+ }
+ public BatchBuffer(
+ HugeGraphClient client,
+ int batchSize,
+ long batchIntervalMs,
+ boolean batchFailureFallback,
+ boolean checkVertex,
+ int maxInsertErrors,
+ String failureDataPath,
+ int subtaskIndex) {
this.batchSize = batchSize;
this.client = client;
+ this.batchFailureFallback = batchFailureFallback;
+ this.checkVertex = checkVertex;
+ this.maxInsertErrors = maxInsertErrors;
+ this.failureDataPath = failureDataPath;
+ this.subtaskIndex = subtaskIndex;
+ this.insertFailureCount = 0;
if (batchIntervalMs > 0) {
this.scheduler =
@@ -81,7 +151,7 @@ public BatchBuffer(HugeGraphClient client, int batchSize, long batchIntervalMs)
}
}
- public synchronized void add(GraphElement element) throws IOException {
+ public synchronized void add(GraphElementEnvelope envelope) throws IOException {
checkFlushException();
if (closed) {
throw new HugeGraphConnectorException(
@@ -90,9 +160,25 @@ public synchronized void add(GraphElement element) throws IOException {
}
try {
- buffer.add(element);
- if (buffer.size() >= batchSize) {
- doFlush();
+ if (envelope.getElementType() == LabelType.VERTEX) {
+ vertexBuffer.add(envelope);
+ if (vertexBuffer.size() >= batchSize) {
+ doFlushVertices();
+ }
+ } else {
+ edgeBuffer.add(envelope);
+ if (edgeBuffer.size() >= batchSize) {
+ // Topology safety only matters when the server validates endpoints: with
+ // check_vertex=true, flush pending vertices before the edges so no edge is sent
+ // before its endpoints exist. With check_vertex=false the server already
+ // accepts
+ // orphan edges, so skip the forced (undersized) vertex flush and let the vertex
+ // bucket accumulate to a full batch — fewer, fuller vertex requests.
+ if (checkVertex && !vertexBuffer.isEmpty()) {
+ doFlushVertices();
+ }
+ doFlushEdges();
+ }
}
} catch (Exception e) {
throw new HugeGraphConnectorException(
@@ -100,40 +186,267 @@ public synchronized void add(GraphElement element) throws IOException {
}
}
+ /**
+ * Backward-compatible overload that wraps a plain {@link GraphElement} in a minimal envelope.
+ *
+ * @deprecated Use {@link #add(GraphElementEnvelope)} instead so the buffer receives complete
+ * mapping context (label name, element type) for failure diagnostics.
+ */
+ @Deprecated
+ public synchronized void add(GraphElement element) throws IOException {
+ LabelType type = element instanceof Vertex ? LabelType.VERTEX : LabelType.EDGE;
+ add(new GraphElementEnvelope(null, type, element));
+ }
+
public synchronized void flush() throws IOException {
checkFlushException();
- if (closed && buffer.isEmpty()) {
+ if (closed && vertexBuffer.isEmpty() && edgeBuffer.isEmpty()) {
return;
}
- doFlush();
+ doFlushVertices();
+ doFlushEdges();
}
- private void doFlush() {
- if (buffer.isEmpty()) {
+ private void doFlushVertices() {
+ if (vertexBuffer.isEmpty()) {
return;
}
+ List batch = new ArrayList<>(vertexBuffer);
+ vertexBuffer.clear();
+ // Route by each element's own mapping strategy: a group with no strategy is a plain insert,
+ // a group with a strategy is an upsert. HugeGraph applies one strategy map per batch call,
+ // so elements with different strategies must go in separate calls.
+ for (Map.Entry, List> group :
+ groupByStrategy(batch).entrySet()) {
+ flushVertexGroup(group.getValue(), group.getKey());
+ }
+ }
+
+ private void flushVertexGroup(
+ List batch, Map updateStrategies) {
try {
- GraphElement firstElement = buffer.get(0);
- if (firstElement instanceof Vertex) {
- List vertices =
- buffer.stream()
- .map(element -> (Vertex) element)
- .collect(Collectors.toList());
+ List vertices =
+ batch.stream()
+ .map(env -> (Vertex) env.getElement())
+ .collect(Collectors.toList());
+ if (updateStrategies.isEmpty()) {
client.batchWriteVertices(vertices);
} else {
- List edges =
- buffer.stream().map(element -> (Edge) element).collect(Collectors.toList());
- client.batchWriteEdges(edges);
+ client.batchUpdateVertices(vertices, updateStrategies);
+ }
+ } catch (Exception e) {
+ if (!batchFailureFallback) {
+ logBatchFailure(batch, e);
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Failed to write vertex batch",
+ e);
}
+ fallbackInsertSingly(batch, e);
+ }
+ }
+
+ private void doFlushEdges() {
+ if (edgeBuffer.isEmpty()) {
+ return;
+ }
+ List batch = new ArrayList<>(edgeBuffer);
+ edgeBuffer.clear();
+ for (Map.Entry, List> group :
+ groupByStrategy(batch).entrySet()) {
+ flushEdgeGroup(group.getValue(), group.getKey());
+ }
+ }
- buffer.clear();
+ private void flushEdgeGroup(
+ List batch, Map updateStrategies) {
+ try {
+ List edges =
+ batch.stream().map(env -> (Edge) env.getElement()).collect(Collectors.toList());
+ if (updateStrategies.isEmpty()) {
+ client.batchWriteEdges(edges, checkVertex);
+ } else {
+ client.batchUpdateEdges(edges, updateStrategies, checkVertex);
+ }
} catch (Exception e) {
- LOG.error("Failed to write batch data to HugeGraph", e);
+ if (!batchFailureFallback) {
+ logBatchFailure(batch, e);
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Failed to write edge batch",
+ e);
+ }
+ fallbackInsertSingly(batch, e);
+ }
+ }
+
+ /**
+ * Groups a batch by its elements' update-strategy map, preserving first-seen order so a flush
+ * stays deterministic. Elements sharing the same strategy map flush together in one server
+ * call.
+ */
+ private static Map, List> groupByStrategy(
+ List batch) {
+ Map, List> groups =
+ new java.util.LinkedHashMap<>();
+ for (GraphElementEnvelope envelope : batch) {
+ groups.computeIfAbsent(envelope.getUpdateStrategies(), key -> new ArrayList<>())
+ .add(envelope);
+ }
+ return groups;
+ }
+
+ /**
+ * A batch insert failed; retry each element on its own so a single poison record no longer
+ * fails the whole batch. Failed records are logged and skipped; the rest succeed. If
+ * every record fails, the failure is systemic (bad connection / schema), not a poison
+ * record, so it is rethrown instead of silently dropping the whole batch.
+ */
+ private void fallbackInsertSingly(List batch, Exception batchFailure) {
+ LOG.warn(
+ "Batch write failed ({} element(s)); falling back to single-record insert. cause={}",
+ batch.size(),
+ batchFailure.getMessage());
+ int failed = 0;
+ Exception lastFailure = null;
+ for (GraphElementEnvelope envelope : batch) {
+ Map updateStrategies = envelope.getUpdateStrategies();
+ try {
+ if (envelope.getElementType() == LabelType.VERTEX) {
+ if (updateStrategies.isEmpty()) {
+ client.writeVertex((Vertex) envelope.getElement());
+ } else {
+ client.updateVertex((Vertex) envelope.getElement(), updateStrategies);
+ }
+ } else {
+ if (updateStrategies.isEmpty()) {
+ client.writeEdge((Edge) envelope.getElement(), checkVertex);
+ } else {
+ client.updateEdge(
+ (Edge) envelope.getElement(), updateStrategies, checkVertex);
+ }
+ }
+ } catch (Exception single) {
+ failed++;
+ lastFailure = single;
+ insertFailureCount++;
+ LOG.error(
+ "Single-record write failure — {}",
+ formatFailureDiagnostic(envelope, single));
+ writeFailureSample(envelope, single);
+ // Bound the previously unlimited silent skipping: once the cumulative number of
+ // skipped records reaches max_insert_errors, stop and fail the task instead of
+ // continuing to drop data. Negative means unlimited.
+ if (maxInsertErrors >= 0 && insertFailureCount >= maxInsertErrors) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ String.format(
+ "Aborting: cumulative single-insert failures (%d) reached "
+ + "max_insert_errors (%d). Last error: %s",
+ insertFailureCount, maxInsertErrors, single.getMessage()),
+ single);
+ }
+ }
+ }
+ if (failed == batch.size()) {
throw new HugeGraphConnectorException(
- HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED, e.getMessage(), e);
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ String.format(
+ "All %d record(s) in the batch failed single-insert fallback",
+ batch.size()),
+ lastFailure);
+ }
+ if (failed > 0) {
+ LOG.warn(
+ "Single-record fallback completed: {} succeeded, {} failed and were skipped "
+ + "({} skipped in total so far)",
+ batch.size() - failed,
+ failed,
+ insertFailureCount);
+ }
+ }
+
+ /**
+ * Appends one line describing a skipped record — the mapped element's id/label/properties plus
+ * the server error — to the per-subtask failure file when {@code failure_data_path} is set.
+ * Best-effort: a write/open error disables further persistence rather than failing the task, so
+ * a broken debug path can never mask the real insert failure.
+ */
+ private void writeFailureSample(GraphElementEnvelope envelope, Exception failure) {
+ if (failureDataPath == null || failureDataPath.isEmpty() || failureWriterDisabled) {
+ return;
+ }
+ try {
+ if (failureWriter == null) {
+ File dir = new File(failureDataPath);
+ if (!dir.exists() && !dir.mkdirs() && !dir.exists()) {
+ throw new IOException("Failed to create failure data directory: " + dir);
+ }
+ File file =
+ new File(dir, "hugegraph-sink-failures-subtask-" + subtaskIndex + ".log");
+ failureWriter =
+ new BufferedWriter(
+ new OutputStreamWriter(
+ new FileOutputStream(file, true), StandardCharsets.UTF_8));
+ LOG.info("Persisting skipped-record failure samples to {}", file.getAbsolutePath());
+ }
+ failureWriter.write(formatFailureSample(envelope, failure));
+ failureWriter.newLine();
+ // Flush per record: failures are rare and losing samples on an abrupt crash defeats
+ // their purpose.
+ failureWriter.flush();
+ } catch (IOException e) {
+ failureWriterDisabled = true;
+ LOG.warn(
+ "Failed to persist failure sample to '{}'; disabling failure-data persistence. cause={}",
+ failureDataPath,
+ e.getMessage());
}
}
+ /**
+ * One-line, tab-delimited failure sample. Newlines are stripped to keep one record per line.
+ */
+ static String formatFailureSample(GraphElementEnvelope envelope, Exception failure) {
+ GraphElement element = envelope.getElement();
+ String line =
+ String.format(
+ "mapping=%s\ttype=%s\tid=%s\tlabel=%s\tproperties=%s\terror=%s",
+ envelope.getMappingLabel(),
+ envelope.getElementType(),
+ element == null ? null : element.id(),
+ element == null ? null : element.label(),
+ element == null ? null : element.properties(),
+ failure.getMessage());
+ return line.replace('\n', ' ').replace('\r', ' ');
+ }
+
+ private void logBatchFailure(List batch, Exception e) {
+ LOG.error(
+ "Batch write failure — {} element(s), failureType={}, serverError={}",
+ batch.size(),
+ e.getClass().getName(),
+ e.getMessage());
+ for (GraphElementEnvelope envelope : batch) {
+ LOG.error("Graph element write failure — {}", formatFailureDiagnostic(envelope, e));
+ }
+ }
+
+ static String formatFailureDiagnostic(GraphElementEnvelope envelope, Exception failure) {
+ // Log only the mapped graph element's id/label — bounded and non-sensitive. The raw source
+ // row is intentionally not retained (see GraphElementEnvelope) to avoid unbounded memory
+ // and
+ // leaking excluded field content into logs.
+ return String.format(
+ "mapping=%s, elementType=%s, elementId=%s, elementLabel=%s, failureType=%s, serverError=%s",
+ envelope.getMappingLabel(),
+ envelope.getElementType(),
+ envelope.getElement() == null ? null : envelope.getElement().id(),
+ envelope.getElement() == null ? null : envelope.getElement().label(),
+ failure.getClass().getName(),
+ failure.getMessage());
+ }
+
@Override
public void close() throws IOException {
synchronized (this) {
@@ -158,11 +471,27 @@ public void close() throws IOException {
}
}
LOG.info("Closing BatchBuffer, performing final flush...");
- flush();
- checkFlushException();
+ try {
+ flush();
+ checkFlushException();
+ } finally {
+ closeFailureWriter();
+ }
LOG.info("BatchBuffer closed.");
}
+ private void closeFailureWriter() {
+ if (failureWriter != null) {
+ try {
+ failureWriter.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close failure-data writer. cause={}", e.getMessage());
+ } finally {
+ failureWriter = null;
+ }
+ }
+ }
+
private void checkFlushException() {
if (flushException != null) {
throw new HugeGraphConnectorException(
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/GraphElementEnvelope.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/GraphElementEnvelope.java
new file mode 100644
index 000000000000..3b1bc2ad0e0b
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/GraphElementEnvelope.java
@@ -0,0 +1,79 @@
+/*
+ * 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.seatunnel.hugegraph.buffer;
+
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.MappingConfig.LabelType;
+
+import org.apache.hugegraph.structure.GraphElement;
+import org.apache.hugegraph.structure.graph.UpdateStrategy;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Wraps a graph element with non-sensitive mapping context for failure diagnostics.
+ *
+ * Deliberately does NOT retain the source {@link
+ * org.apache.seatunnel.api.table.type.SeaTunnelRow}: only the mapped {@link GraphElement} is ever
+ * sent, and envelopes stay alive until the batch is flushed (by size/timer/checkpoint/close).
+ * Keeping the raw row would pin fields that were excluded by {@code mapping.properties} (e.g. large
+ * BYTES payloads) in memory for the whole batch and leak their content into failure logs.
+ */
+public class GraphElementEnvelope {
+
+ private final String mappingLabel;
+ private final LabelType elementType;
+ private final GraphElement element;
+ // Per-mapping update strategies (property name -> strategy). Empty means plain insert. Carried
+ // on the envelope so the buffer can route each element by its own mapping's strategy instead of
+ // one merged global map — a strategy on one mapping no longer forces upsert on every mapping,
+ // and two mappings may assign different strategies to the same property name.
+ private final Map updateStrategies;
+
+ public GraphElementEnvelope(String mappingLabel, LabelType elementType, GraphElement element) {
+ this(mappingLabel, elementType, element, Collections.emptyMap());
+ }
+
+ public GraphElementEnvelope(
+ String mappingLabel,
+ LabelType elementType,
+ GraphElement element,
+ Map updateStrategies) {
+ this.mappingLabel = mappingLabel;
+ this.elementType = elementType;
+ this.element = element;
+ this.updateStrategies =
+ updateStrategies == null ? Collections.emptyMap() : updateStrategies;
+ }
+
+ public String getMappingLabel() {
+ return mappingLabel;
+ }
+
+ public LabelType getElementType() {
+ return elementType;
+ }
+
+ public GraphElement getElement() {
+ return element;
+ }
+
+ public Map getUpdateStrategies() {
+ return updateStrategies;
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/HugeGraphClient.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/HugeGraphClient.java
index ed4491eaef3e..d173e9bca0c9 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/HugeGraphClient.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/HugeGraphClient.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -17,18 +17,31 @@
package org.apache.seatunnel.connectors.seatunnel.hugegraph.client;
-import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.HugeGraphSinkConfig;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.HugeGraphConnectionConfig;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.LabelOptions;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorErrorCode;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorException;
+import org.apache.hugegraph.api.graph.EdgeAPI;
+import org.apache.hugegraph.api.graph.VertexAPI;
+import org.apache.hugegraph.client.RestClient;
import org.apache.hugegraph.driver.GraphManager;
import org.apache.hugegraph.driver.HugeClient;
import org.apache.hugegraph.driver.SchemaManager;
import org.apache.hugegraph.exception.ServerException;
import org.apache.hugegraph.rest.ClientException;
+import org.apache.hugegraph.rest.RestClientConfig;
+import org.apache.hugegraph.structure.constant.DataType;
+import org.apache.hugegraph.structure.constant.Frequency;
import org.apache.hugegraph.structure.constant.IdStrategy;
+import org.apache.hugegraph.structure.graph.BatchEdgeRequest;
+import org.apache.hugegraph.structure.graph.BatchVertexRequest;
import org.apache.hugegraph.structure.graph.Edge;
+import org.apache.hugegraph.structure.graph.Edges;
+import org.apache.hugegraph.structure.graph.Shard;
+import org.apache.hugegraph.structure.graph.UpdateStrategy;
import org.apache.hugegraph.structure.graph.Vertex;
+import org.apache.hugegraph.structure.graph.Vertices;
import org.apache.hugegraph.structure.schema.EdgeLabel;
import org.apache.hugegraph.structure.schema.PropertyKey;
import org.apache.hugegraph.structure.schema.VertexLabel;
@@ -36,39 +49,61 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
-public final class HugeGraphClient {
+public final class HugeGraphClient implements HugeGraphOperations {
- // TODO: Add handling for schema fetch failures.
private static final Logger LOG = LoggerFactory.getLogger(HugeGraphClient.class);
+ /** HugeGraph server per-request batch cap (server option batch.max_vertices_per_batch). */
+ private static final int MAX_RECORDS_PER_BATCH_REQUEST = 500;
+
private HugeClient client;
+ private RestClient restClient;
+ private VertexAPI vertexAPI;
+ private EdgeAPI edgeAPI;
private SchemaManager schema;
- private final HugeGraphSinkConfig config;
+ private final HugeGraphConnectionConfig config;
private final int maxRetries;
private final long retryBackoffMs;
+ private final long retryBackoffMaxMs;
- public HugeGraphClient(HugeGraphSinkConfig config) {
+ public HugeGraphClient(HugeGraphConnectionConfig config) {
this.client = null;
+ this.restClient = null;
+ this.vertexAPI = null;
+ this.edgeAPI = null;
this.schema = null;
this.config = config;
- this.maxRetries = config.getMaxRetries() > 0 ? config.getMaxRetries() : 3;
- this.retryBackoffMs = config.getRetryBackoffMs() > 0 ? config.getRetryBackoffMs() : 5000L;
+ this.maxRetries = Math.max(0, config.getMaxRetries());
+ this.retryBackoffMs = Math.max(0, config.getRetryBackoffMs());
+ this.retryBackoffMaxMs = Math.max(0, config.getRetryBackoffMaxMs());
}
- private HugeClient createClient(HugeGraphSinkConfig config) {
+ /** Default graph space per HugeGraphOptions.GRAPH_SPACE.defaultValue(). */
+ private static final String DEFAULT_GRAPH_SPACE = "DEFAULT";
+
+ private HugeClient createClient(HugeGraphConnectionConfig config) {
try {
- String url = String.format("http://%s:%d", config.getHost(), config.getPort());
- LOG.debug("Creating new HugeClient for url: {}, graph: {}", url, config.getGraphName());
+ String url = buildServerUrl(config);
+ String graphSpace =
+ config.getGraphSpace() != null ? config.getGraphSpace() : DEFAULT_GRAPH_SPACE;
+ LOG.debug(
+ "Creating new HugeClient for url: {}, graphSpace: {}, graph: {}",
+ url,
+ graphSpace,
+ config.getGraphName());
HugeClient client =
- HugeClient.builder(url, config.getGraphName())
+ HugeClient.builder(url, graphSpace, config.getGraphName())
.configUser(config.getUsername(), config.getPassword())
.configIdleTime(60)
.build();
- client.graph().listVertices();
LOG.info("Successfully created and validated HugeClient instance.");
return client;
} catch (Exception e) {
@@ -83,14 +118,23 @@ private interface GraphOperation {
void execute(GraphManager graph) throws ServerException, ClientException;
}
+ @FunctionalInterface
+ private interface ReadOperation {
+ T execute() throws ServerException, ClientException;
+ }
+
private void ensureClientInitialized() throws HugeGraphConnectorException {
if (this.client == null) {
LOG.info("Client not initialized. Attempting to connect...");
try {
this.client = createClient(this.config);
this.schema = this.client.schema();
+ createPageApis(this.config);
LOG.info("HugeClient initialized successfully.");
} catch (Exception e) {
+ // Avoid leaking a partially-opened client (e.g. createPageApis failed after the
+ // HugeClient was created) — release everything before surfacing the failure.
+ reconnect();
throw new HugeGraphConnectorException(
HugeGraphConnectorErrorCode.BUILD_CLIENT_FAILED,
"Failed to establish initial connection",
@@ -109,52 +153,263 @@ private void reconnect() {
}
}
this.client = null;
+ if (this.restClient != null) {
+ try {
+ this.restClient.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing potentially broken REST client: {}", e.getMessage());
+ }
+ }
+ this.restClient = null;
+ this.vertexAPI = null;
+ this.edgeAPI = null;
this.schema = null;
}
- private void executeGraphOperation(GraphOperation operation) {
- for (int attempt = 1; attempt <= this.maxRetries; attempt++) {
+ private void createPageApis(HugeGraphConnectionConfig config) {
+ String url = buildServerUrl(config);
+ String graphSpace =
+ config.getGraphSpace() != null ? config.getGraphSpace() : DEFAULT_GRAPH_SPACE;
+ RestClientConfig restClientConfig =
+ RestClientConfig.builder()
+ .user(config.getUsername() == null ? "" : config.getUsername())
+ .password(config.getPassword() == null ? "" : config.getPassword())
+ .build();
+ this.restClient = new RestClient(url, restClientConfig);
+ this.vertexAPI = new VertexAPI(this.restClient, graphSpace, config.getGraphName());
+ this.edgeAPI = new EdgeAPI(this.restClient, graphSpace, config.getGraphName());
+ }
+
+ static String buildServerUrl(HugeGraphConnectionConfig config) {
+ String protocol =
+ config.getProtocol() == null || config.getProtocol().isEmpty()
+ ? "http"
+ : config.getProtocol().toLowerCase(java.util.Locale.ROOT);
+ return String.format("%s://%s:%d", protocol, config.getHost(), config.getPort());
+ }
+
+ /**
+ * Executes a write operation that is safe to retry: UPSERT (updateVertices/updateEdges with
+ * createIfNotExist=true) and DELETE (removeVertex/removeEdge). Idempotent operations are
+ * retried on retryable errors because a second attempt cannot create duplicates.
+ */
+ private void executeIdempotentWrite(GraphOperation operation) {
+ executeGraphOperation(operation, true);
+ }
+
+ /**
+ * Executes a write operation that is NOT safe to retry: plain INSERT (addVertex/addVertices/
+ * addEdge/addEdges). A retry after a server-committed-but-client-timed-out response would
+ * create a duplicate element. Non-idempotent writes fail fast — the caller's single-record
+ * fallback handles them individually instead.
+ */
+ private void executeNonIdempotentWrite(GraphOperation operation) {
+ try {
+ ensureClientInitialized();
+ operation.execute(this.client.graph());
+ } catch (ServerException | ClientException e) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Non-idempotent write failed (not retried to avoid duplicates): "
+ + e.getMessage(),
+ e);
+ }
+ }
+
+ /**
+ * Executes a graph write with optional retry. When {@code idempotent} is true, retryable server
+ * errors (status ≥ 500, 408, 425, 429) are retried up to {@code maxRetries} times with
+ * exponential backoff. When false, the operation is attempted once — if it fails the exception
+ * propagates immediately so the caller can route through the single-record fallback or skip the
+ * record.
+ */
+ private void executeGraphOperation(GraphOperation operation, boolean idempotent) {
+ int totalAttempts = idempotent ? this.maxRetries + 1 : 1;
+ for (int attempt = 1; attempt <= totalAttempts; attempt++) {
try {
ensureClientInitialized();
operation.execute(this.client.graph());
return;
} catch (ServerException | ClientException e) {
+ if (!isRetryable(e) || !idempotent) {
+ LOG.error(
+ "Server rejected the request ({}): {}",
+ idempotent ? "non-retryable" : "non-idempotent, not retrying",
+ e.getMessage());
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Server rejected the request"
+ + (idempotent ? " (non-retryable)" : " (non-idempotent)")
+ + ": "
+ + e.getMessage(),
+ e);
+ }
LOG.warn(
"Graph operation failed on attempt {}/{}. Error: {}",
attempt,
- this.maxRetries,
+ totalAttempts,
e.getMessage());
reconnect();
- if (attempt == this.maxRetries) {
+ if (attempt == totalAttempts) {
LOG.error("Max retries ({}) reached. Failing task.", this.maxRetries);
throw new HugeGraphConnectorException(
HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
"Failed to execute graph operation after "
- + this.maxRetries
- + " attempts",
+ + totalAttempts
+ + " attempt(s). Last error: "
+ + e.getMessage(),
e);
}
- try {
- LOG.info("Will retry in {} ms...", retryBackoffMs);
- Thread.sleep(retryBackoffMs);
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
+ sleepBeforeRetry(attempt);
+ } catch (HugeGraphConnectorException e) {
+ if (!HugeGraphConnectorErrorCode.BUILD_CLIENT_FAILED
+ .getCode()
+ .equals(e.getSeaTunnelErrorCode().getCode())) {
+ throw e;
+ }
+ if (!idempotent) {
+ throw e;
+ }
+ reconnect();
+ if (attempt == totalAttempts) {
throw new HugeGraphConnectorException(
- HugeGraphConnectorErrorCode.OPERATION_RETRY_INTERRUPTED,
- "Graph operation retry was interrupted",
- ie);
+ HugeGraphConnectorErrorCode.BUILD_CLIENT_FAILED,
+ "Failed to establish HugeGraph connection after "
+ + totalAttempts
+ + " attempt(s)",
+ e);
}
-
+ LOG.warn(
+ "HugeGraph connection failed on attempt {}/{}. Error: {}",
+ attempt,
+ totalAttempts,
+ e.getMessage());
+ sleepBeforeRetry(attempt);
} catch (Exception e) {
LOG.error("Non-retryable error executing graph operation: {}", e.getMessage(), e);
throw new HugeGraphConnectorException(
HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
- "Non-retryable error executing graph operation",
+ "Non-retryable error executing graph operation: " + e.getMessage(),
+ e);
+ }
+ }
+ }
+
+ /**
+ * Deterministic 4xx responses (bad request, semantic rejection such as exceeding the server
+ * batch size cap) cannot succeed on retry. Only connection-level failures and 5xx server errors
+ * are worth retrying.
+ */
+ static boolean isRetryable(Exception e) {
+ if (e instanceof ServerException) {
+ int status = ((ServerException) e).status();
+ return status == 408 || status == 425 || status == 429 || status >= 500;
+ }
+ return true;
+ }
+
+ private T executeReadOperation(ReadOperation operation) {
+ int totalAttempts = this.maxRetries + 1;
+ for (int attempt = 1; attempt <= totalAttempts; attempt++) {
+ try {
+ ensureClientInitialized();
+ return operation.execute();
+ } catch (ServerException | ClientException e) {
+ if (!isRetryable(e)) {
+ LOG.error("Server rejected the request (non-retryable): {}", e.getMessage());
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Server rejected the request (non-retryable): " + e.getMessage(),
+ e);
+ }
+ LOG.warn(
+ "Graph read operation failed on attempt {}/{}. Error: {}",
+ attempt,
+ totalAttempts,
+ e.getMessage());
+ reconnect();
+
+ if (attempt == totalAttempts) {
+ LOG.error("Max retries ({}) reached. Failing task.", this.maxRetries);
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Failed to execute graph read operation after "
+ + totalAttempts
+ + " attempt(s). Last error: "
+ + e.getMessage(),
+ e);
+ }
+
+ sleepBeforeRetry(attempt);
+ } catch (HugeGraphConnectorException e) {
+ if (!HugeGraphConnectorErrorCode.BUILD_CLIENT_FAILED
+ .getCode()
+ .equals(e.getSeaTunnelErrorCode().getCode())) {
+ throw e;
+ }
+ reconnect();
+ if (attempt == totalAttempts) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.BUILD_CLIENT_FAILED,
+ "Failed to establish HugeGraph connection after "
+ + totalAttempts
+ + " attempt(s)",
+ e);
+ }
+ LOG.warn(
+ "HugeGraph connection failed on attempt {}/{}. Error: {}",
+ attempt,
+ totalAttempts,
+ e.getMessage());
+ sleepBeforeRetry(attempt);
+ } catch (Exception e) {
+ LOG.error(
+ "Non-retryable error executing graph read operation: {}",
+ e.getMessage(),
+ e);
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Non-retryable error executing graph read operation",
e);
}
}
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.GRAPH_OPERATION_FAILED,
+ "Failed to execute graph read operation");
+ }
+
+ private void sleepBeforeRetry(int attempt) {
+ long delay = computeBackoffMs(retryBackoffMs, retryBackoffMaxMs, attempt);
+ try {
+ LOG.info("Will retry in {} ms (attempt {})...", delay, attempt);
+ Thread.sleep(delay);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.OPERATION_RETRY_INTERRUPTED,
+ "Graph operation retry was interrupted",
+ ie);
+ }
+ }
+
+ /**
+ * Exponential backoff: {@code baseMs * 2^(attempt-1)}, capped at {@code maxMs} (a non-positive
+ * {@code maxMs} means no cap). The shift is bounded so a large {@code maxRetries} cannot
+ * overflow. {@code attempt} is 1-based (the first retry uses the base delay).
+ */
+ static long computeBackoffMs(long baseMs, long maxMs, int attempt) {
+ if (baseMs <= 0) {
+ return 0;
+ }
+ int shift = Math.min(Math.max(attempt - 1, 0), 30);
+ long scaled = baseMs << shift;
+ if (scaled < 0) {
+ // Overflow guard (defensive; the shift cap already prevents this for int-range bases).
+ return maxMs > 0 ? maxMs : Long.MAX_VALUE;
+ }
+ return (maxMs > 0) ? Math.min(scaled, maxMs) : scaled;
}
private SchemaManager getSchema() {
@@ -162,51 +417,483 @@ private SchemaManager getSchema() {
return this.schema;
}
+ // --- Schema read operations ---
+
public PropertyKey getPropertyKey(String propertyName) {
- return getSchema().getPropertyKey(propertyName);
+ return executeReadOperation(() -> getSchema().getPropertyKey(propertyName));
}
public VertexLabel getVertexLabel(String label) {
- return getSchema().getVertexLabel(label);
+ VertexLabel vertexLabel = getVertexLabelOrNull(label);
+ if (vertexLabel == null) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.INVALID_GRAPH_SCHEMA,
+ "Vertex label '"
+ + label
+ + "' does not exist in HugeGraph. "
+ + "Please create it first or check your configuration.");
+ }
+ return vertexLabel;
}
public EdgeLabel getEdgeLabel(String label) {
- return getSchema().getEdgeLabel(label);
+ EdgeLabel edgeLabel = getEdgeLabelOrNull(label);
+ if (edgeLabel == null) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.INVALID_GRAPH_SCHEMA,
+ "Edge label '"
+ + label
+ + "' does not exist in HugeGraph. "
+ + "Please create it first or check your configuration.");
+ }
+ return edgeLabel;
}
public String getVertexLabelId(String label) {
- VertexLabel vertexLabel = getSchema().getVertexLabel(label);
+ VertexLabel vertexLabel = getVertexLabel(label);
return String.valueOf(vertexLabel.id());
}
public String getEdgeLabelId(String label) {
- EdgeLabel edgeLabel = getSchema().getEdgeLabel(label);
+ EdgeLabel edgeLabel = getEdgeLabel(label);
return String.valueOf(edgeLabel.id());
}
public IdStrategy getIdStrategy(String label) {
- VertexLabel vertexLabel = getSchema().getVertexLabel(label);
+ VertexLabel vertexLabel = getVertexLabel(label);
return vertexLabel.idStrategy();
}
+ // --- Schema creation operations (idempotent, ifNotExist) ---
+
+ public PropertyKey createPropertyKeyIfNotExist(
+ String name,
+ DataType dataType,
+ org.apache.hugegraph.structure.constant.Cardinality cardinality) {
+ return executeReadOperation(
+ () ->
+ getSchema()
+ .propertyKey(name)
+ .dataType(dataType)
+ .cardinality(cardinality)
+ .ifNotExist()
+ .create());
+ }
+
+ public VertexLabel createVertexLabelIfNotExist(
+ String label,
+ IdStrategy idStrategy,
+ List primaryKeys,
+ List propertyNames,
+ List nullableKeys,
+ LabelOptions options) {
+ return executeReadOperation(
+ () -> {
+ VertexLabel.Builder builder =
+ getSchema().vertexLabel(label).idStrategy(idStrategy);
+ if (idStrategy == IdStrategy.PRIMARY_KEY && primaryKeys != null) {
+ builder.primaryKeys(primaryKeys.toArray(new String[0]));
+ }
+ if (propertyNames != null && !propertyNames.isEmpty()) {
+ builder.properties(propertyNames.toArray(new String[0]));
+ }
+ if (nullableKeys != null && !nullableKeys.isEmpty()) {
+ builder.nullableKeys(nullableKeys.toArray(new String[0]));
+ }
+ if (options != null) {
+ if (options.getTtl() != null && options.getTtl() > 0) {
+ builder.ttl(options.getTtl());
+ if (options.getTtlStartTime() != null
+ && !options.getTtlStartTime().isEmpty()) {
+ builder.ttlStartTime(options.getTtlStartTime());
+ }
+ }
+ if (options.getEnableLabelIndex() != null) {
+ builder.enableLabelIndex(options.getEnableLabelIndex());
+ }
+ if (options.getUserdata() != null) {
+ for (Map.Entry entry :
+ options.getUserdata().entrySet()) {
+ builder.userdata(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+ return builder.ifNotExist().create();
+ });
+ }
+
+ public EdgeLabel createEdgeLabelIfNotExist(
+ String label,
+ String sourceLabel,
+ String targetLabel,
+ Frequency frequency,
+ List sortKeys,
+ List propertyNames,
+ List nullableKeys,
+ LabelOptions options) {
+ return executeReadOperation(
+ () -> {
+ EdgeLabel.Builder builder =
+ getSchema()
+ .edgeLabel(label)
+ .sourceLabel(sourceLabel)
+ .targetLabel(targetLabel);
+ if (frequency != null) {
+ builder.frequency(frequency);
+ }
+ if (sortKeys != null && !sortKeys.isEmpty()) {
+ builder.sortKeys(sortKeys.toArray(new String[0]));
+ }
+ if (propertyNames != null && !propertyNames.isEmpty()) {
+ builder.properties(propertyNames.toArray(new String[0]));
+ }
+ if (nullableKeys != null && !nullableKeys.isEmpty()) {
+ builder.nullableKeys(nullableKeys.toArray(new String[0]));
+ }
+ if (options != null) {
+ if (options.getTtl() != null && options.getTtl() > 0) {
+ builder.ttl(options.getTtl());
+ if (options.getTtlStartTime() != null
+ && !options.getTtlStartTime().isEmpty()) {
+ builder.ttlStartTime(options.getTtlStartTime());
+ }
+ }
+ if (options.getEnableLabelIndex() != null) {
+ builder.enableLabelIndex(options.getEnableLabelIndex());
+ }
+ if (options.getUserdata() != null) {
+ for (Map.Entry entry :
+ options.getUserdata().entrySet()) {
+ builder.userdata(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+ return builder.ifNotExist().create();
+ });
+ }
+
+ /** Check if a property key exists. Returns null if not found. */
+ public PropertyKey getPropertyKeyOrNull(String name) {
+ return executeReadOperation(
+ () -> {
+ try {
+ return getSchema().getPropertyKey(name);
+ } catch (ServerException e) {
+ if (e.status() == 404
+ || (e.getMessage() != null
+ && e.getMessage().contains("does not exist"))) {
+ return null;
+ }
+ throw e;
+ }
+ });
+ }
+
+ /** Check if a vertex label exists. Returns null if not found. */
+ public VertexLabel getVertexLabelOrNull(String label) {
+ return executeReadOperation(
+ () -> {
+ try {
+ return getSchema().getVertexLabel(label);
+ } catch (ServerException e) {
+ if (e.status() == 404
+ || (e.getMessage() != null
+ && e.getMessage().contains("does not exist"))) {
+ return null;
+ }
+ throw e;
+ }
+ });
+ }
+
+ /** Check if an edge label exists. Returns null if not found. */
+ public EdgeLabel getEdgeLabelOrNull(String label) {
+ return executeReadOperation(
+ () -> {
+ try {
+ return getSchema().getEdgeLabel(label);
+ } catch (ServerException e) {
+ if (e.status() == 404
+ || (e.getMessage() != null
+ && e.getMessage().contains("does not exist"))) {
+ return null;
+ }
+ throw e;
+ }
+ });
+ }
+
+ @Override
+ public Set getVertexLabelPropertiesOrNull(String label) {
+ VertexLabel vertexLabel = getVertexLabelOrNull(label);
+ return vertexLabel == null ? null : vertexLabel.properties();
+ }
+
+ @Override
+ public Set getEdgeLabelPropertiesOrNull(String label) {
+ EdgeLabel edgeLabel = getEdgeLabelOrNull(label);
+ return edgeLabel == null ? null : edgeLabel.properties();
+ }
+
+ @Override
+ public List listVertexLabels() {
+ return executeReadOperation(
+ () -> {
+ List names = new ArrayList<>();
+ for (VertexLabel vertexLabel : getSchema().getVertexLabels()) {
+ names.add(vertexLabel.name());
+ }
+ return names;
+ });
+ }
+
+ @Override
+ public List listEdgeLabels() {
+ return executeReadOperation(
+ () -> {
+ List names = new ArrayList<>();
+ for (EdgeLabel edgeLabel : getSchema().getEdgeLabels()) {
+ names.add(edgeLabel.name());
+ }
+ return names;
+ });
+ }
+
+ @Override
+ public DataType getPropertyDataType(String propertyName) {
+ return getPropertyKey(propertyName).dataType();
+ }
+
+ @Override
+ public org.apache.hugegraph.structure.constant.Cardinality getPropertyCardinality(
+ String propertyName) {
+ return getPropertyKey(propertyName).cardinality();
+ }
+
+ // --- Graph write operations ---
+
+ /**
+ * Plain vertex insert — NOT idempotent. A retry after a server-committed-but-client-timed-out
+ * response would create a duplicate. Fails fast on the first error; the caller's single-record
+ * fallback handles the record individually instead.
+ */
public void writeVertex(Vertex vertex) {
- executeGraphOperation(graph -> graph.addVertex(vertex));
+ executeNonIdempotentWrite(graph -> graph.addVertex(vertex));
+ }
+
+ /** Plain edge insert — NOT idempotent. See {@link #writeVertex}. */
+ public void writeEdge(Edge edge, boolean checkVertex) {
+ // Route through addEdges so the single-insert path honors checkVertex the same way the
+ // batch path does (GraphManager.addEdge has no checkVertex overload).
+ executeNonIdempotentWrite(
+ graph -> graph.addEdges(Collections.singletonList(edge), checkVertex));
}
- public void writeEdge(Edge edge) {
- executeGraphOperation(graph -> graph.addEdge(edge));
+ /** Single-vertex property-merge upsert; idempotent — see {@link #batchUpdateVertices}. */
+ public void updateVertex(Vertex vertex, Map updateStrategies) {
+ batchUpdateVertices(Collections.singletonList(vertex), updateStrategies);
}
+ /** Single-edge property-merge upsert; idempotent — see {@link #batchUpdateEdges}. */
+ public void updateEdge(
+ Edge edge, Map updateStrategies, boolean checkVertex) {
+ batchUpdateEdges(Collections.singletonList(edge), updateStrategies, checkVertex);
+ }
+
+ /**
+ * Upserts vertices with per-property merge strategies (OVERRIDE / APPEND / SUM / UNION / ...)
+ * instead of overwriting. Existing vertices are merged; missing ones are created
+ * (createIfNotExist). Idempotent — safe to retry. Chunked like {@link #batchWriteVertices}.
+ */
+ public void batchUpdateVertices(
+ List buffer, Map updateStrategies) {
+ for (int start = 0; start < buffer.size(); start += MAX_RECORDS_PER_BATCH_REQUEST) {
+ List chunk =
+ buffer.subList(
+ start, Math.min(start + MAX_RECORDS_PER_BATCH_REQUEST, buffer.size()));
+ BatchVertexRequest request =
+ new BatchVertexRequest.Builder()
+ .vertices(chunk)
+ .updatingStrategies(updateStrategies)
+ .createIfNotExist(true)
+ .build();
+ executeIdempotentWrite(graph -> graph.updateVertices(request));
+ }
+ }
+
+ /**
+ * Upserts edges with per-property merge strategies. Idempotent. See {@link
+ * #batchUpdateVertices}.
+ */
+ public void batchUpdateEdges(
+ List buffer, Map updateStrategies, boolean checkVertex) {
+ for (int start = 0; start < buffer.size(); start += MAX_RECORDS_PER_BATCH_REQUEST) {
+ List chunk =
+ buffer.subList(
+ start, Math.min(start + MAX_RECORDS_PER_BATCH_REQUEST, buffer.size()));
+ BatchEdgeRequest request =
+ new BatchEdgeRequest.Builder()
+ .edges(chunk)
+ .updatingStrategies(updateStrategies)
+ .checkVertex(checkVertex)
+ .createIfNotExist(true)
+ .build();
+ executeIdempotentWrite(graph -> graph.updateEdges(request));
+ }
+ }
+
+ /**
+ * Writes vertices in chunks of at most {@link #MAX_RECORDS_PER_BATCH_REQUEST}. The HugeGraph
+ * server rejects batch requests above its per-request cap (default 500, see server option
+ * batch.max_vertices_per_batch), so a user-configured batch_size larger than the cap is split
+ * client-side instead of failing wholesale.
+ *
+ * NOT idempotent — a retry after a server-committed-but-client-timed-out response would
+ * create duplicates. Fails fast; the caller's single-record fallback handles each record.
+ */
+ public void batchWriteVertices(List buffer) {
+ for (int start = 0; start < buffer.size(); start += MAX_RECORDS_PER_BATCH_REQUEST) {
+ List chunk =
+ buffer.subList(
+ start, Math.min(start + MAX_RECORDS_PER_BATCH_REQUEST, buffer.size()));
+ executeNonIdempotentWrite(graph -> graph.addVertices(chunk));
+ }
+ }
+
+ /**
+ * Writes edges in server-cap-sized chunks. NOT idempotent. See {@link #batchWriteVertices} and
+ * {@link #batchWriteEdges}. When {@code checkVertex} is true the server verifies that each
+ * edge's source/target vertices exist, rejecting orphan edges instead of silently writing them
+ * or auto-creating phantom vertices.
+ */
+ public void batchWriteEdges(List buffer, boolean checkVertex) {
+ for (int start = 0; start < buffer.size(); start += MAX_RECORDS_PER_BATCH_REQUEST) {
+ List chunk =
+ buffer.subList(
+ start, Math.min(start + MAX_RECORDS_PER_BATCH_REQUEST, buffer.size()));
+ executeNonIdempotentWrite(graph -> graph.addEdges(chunk, checkVertex));
+ }
+ }
+
+ // --- Graph read operations ---
+
+ /**
+ * Lists one page of vertices. HugeGraph only enters paged mode when the {@code page} query
+ * parameter is present — a null first page must be sent as an empty string, otherwise the
+ * server returns a single non-paged batch without a next-page marker and the scan silently
+ * stops after {@code limit} records.
+ *
+ * When {@code filter} is non-empty it is passed as the server-side property-equality
+ * condition map, with {@code keepP=true} so the filtered properties are retained in the
+ * returned vertices (they may be part of the output schema).
+ */
+ @SuppressWarnings("unchecked")
+ @Override
+ public PageResult listVertices(
+ String label, Map filter, String page, int limit) {
+ String effectivePage = page == null ? "" : page;
+ boolean hasFilter = filter != null && !filter.isEmpty();
+ Map conditions = hasFilter ? filter : null;
+ return executeReadOperation(
+ () -> {
+ Vertices vertices =
+ this.vertexAPI.list(
+ label, conditions, hasFilter, 0, effectivePage, limit);
+ List records = (List) vertices.results();
+ return new PageResult<>(
+ records == null ? Collections.emptyList() : records, vertices.page());
+ });
+ }
+
+ /** Lists one page of edges. See {@link #listVertices} for the empty-first-page contract. */
+ @SuppressWarnings("unchecked")
+ @Override
+ public PageResult listEdges(
+ String label, Map filter, String page, int limit) {
+ String effectivePage = page == null ? "" : page;
+ boolean hasFilter = filter != null && !filter.isEmpty();
+ Map conditions = hasFilter ? filter : null;
+ return executeReadOperation(
+ () -> {
+ Edges edges =
+ this.edgeAPI.list(
+ null,
+ null,
+ label,
+ conditions,
+ hasFilter,
+ 0,
+ effectivePage,
+ limit);
+ List records = (List) edges.results();
+ return new PageResult<>(
+ records == null ? Collections.emptyList() : records, edges.page());
+ });
+ }
+
+ /**
+ * Splits the vertex keyspace into shards for parallel scanning. Delegates to the server's
+ * {@code traverser().vertexShards} API. Requires a scan-capable backend (RocksDB / HBase /
+ * Cassandra).
+ */
+ @Override
+ public List vertexShards(long splitSize) {
+ return executeReadOperation(() -> this.client.traverser().vertexShards(splitSize));
+ }
+
+ /** Splits the edge keyspace into shards. See {@link #vertexShards}. */
+ @Override
+ public List edgeShards(long splitSize) {
+ return executeReadOperation(() -> this.client.traverser().edgeShards(splitSize));
+ }
+
+ /**
+ * Scans one page of vertices within {@code shard}. The empty-first-page contract of {@link
+ * #listVertices} applies: a null page is sent as the empty string so the server enters paged
+ * mode. The scan returns vertices of all labels in the key range; label filtering is the
+ * caller's responsibility.
+ */
+ @Override
+ public PageResult scanVertices(Shard shard, String page, int limit) {
+ String effectivePage = page == null ? "" : page;
+ return executeReadOperation(
+ () -> {
+ Vertices vertices =
+ this.client.traverser().vertices(shard, effectivePage, limit);
+ List records = vertices.results();
+ return new PageResult<>(
+ records == null ? Collections.emptyList() : records, vertices.page());
+ });
+ }
+
+ /** Scans one page of edges within {@code shard}. See {@link #scanVertices}. */
+ @Override
+ public PageResult scanEdges(Shard shard, String page, int limit) {
+ String effectivePage = page == null ? "" : page;
+ return executeReadOperation(
+ () -> {
+ Edges edges = this.client.traverser().edges(shard, effectivePage, limit);
+ List records = edges.results();
+ return new PageResult<>(
+ records == null ? Collections.emptyList() : records, edges.page());
+ });
+ }
+
+ // --- Graph delete operations ---
+
+ /** Delete vertex by id — idempotent (removing an already-deleted vertex is a no-op). */
public void deleteVertex(Object vertexId) {
- executeGraphOperation(graph -> graph.removeVertex(vertexId));
+ executeIdempotentWrite(graph -> graph.removeVertex(vertexId));
}
+ /** Delete edge by id — idempotent. */
public void deleteEdge(String edgeId) {
- executeGraphOperation(graph -> graph.removeEdge(edgeId));
+ executeIdempotentWrite(graph -> graph.removeEdge(edgeId));
}
+ /** Delete vertex with its incident edges — idempotent. */
public void deleteVertexWithEdges(Object vertexId) {
- executeGraphOperation(
+ executeIdempotentWrite(
graph -> {
List edges = graph.getEdges(vertexId);
for (Edge edge : edges) {
@@ -216,20 +903,101 @@ public void deleteVertexWithEdges(Object vertexId) {
});
}
- public void batchWriteVertices(List buffer) {
- executeGraphOperation(graph -> graph.addVertices(buffer));
+ /**
+ * Returns the names of every edge label whose source or target endpoint is {@code vertexLabel}.
+ * These are the edge labels that would be cascade-deleted if every vertex of {@code
+ * vertexLabel} is removed — used by the DROP_DATA pre-flight safety check.
+ */
+ public List getConnectedEdgeLabels(String vertexLabel) {
+ return executeReadOperation(
+ () -> {
+ List connected = new ArrayList<>();
+ for (EdgeLabel edgeLabel : getSchema().getEdgeLabels()) {
+ if (vertexLabel.equals(edgeLabel.sourceLabel())
+ || vertexLabel.equals(edgeLabel.targetLabel())) {
+ connected.add(edgeLabel.name());
+ }
+ }
+ return connected;
+ });
}
- public void batchWriteEdges(List buffer) {
- executeGraphOperation(graph -> graph.addEdges(buffer));
+ /** Page size used when clearing a single label's data for data_save_mode=DROP_DATA. */
+ private static final int DELETE_PAGE_SIZE = 500;
+
+ /**
+ * Deletes every vertex of {@code label} (data only — the VertexLabel schema is preserved), used
+ * by data_save_mode=DROP_DATA to clear just the labels this job targets instead of wiping the
+ * whole graph with {@code clearGraph}. Removing a vertex also removes its incident edges on the
+ * server. Works by repeatedly deleting the first page until none remain, so it does not depend
+ * on a paging cursor staying valid across deletes.
+ */
+ public void deleteVerticesByLabel(String label) {
+ LOG.info("data_save_mode=DROP_DATA: deleting all vertices of label '{}'", label);
+ long deleted = 0;
+ while (true) {
+ List records = listVertices(label, null, "", DELETE_PAGE_SIZE).getRecords();
+ if (records.isEmpty()) {
+ break;
+ }
+ for (Vertex vertex : records) {
+ deleteVertex(vertex.id());
+ deleted++;
+ }
+ }
+ LOG.info("Deleted {} vertices of label '{}'", deleted, label);
}
+ /**
+ * Deletes every edge of {@code label} (data only — the EdgeLabel schema is preserved). See
+ * {@link #deleteVerticesByLabel} for the paging strategy; run before vertex deletion so
+ * edge-only mappings are handled even when their endpoints are out of this job's scope.
+ */
+ public void deleteEdgesByLabel(String label) {
+ LOG.info("data_save_mode=DROP_DATA: deleting all edges of label '{}'", label);
+ long deleted = 0;
+ while (true) {
+ List records = listEdges(label, null, "", DELETE_PAGE_SIZE).getRecords();
+ if (records.isEmpty()) {
+ break;
+ }
+ for (Edge edge : records) {
+ deleteEdge(edge.id());
+ deleted++;
+ }
+ }
+ LOG.info("Deleted {} edges of label '{}'", deleted, label);
+ }
+
+ @Override
public void close() {
+ RuntimeException closeFailure = null;
if (this.client != null) {
LOG.info("Closing HugeClient instance.");
- this.client.close();
+ try {
+ this.client.close();
+ } catch (RuntimeException e) {
+ closeFailure = e;
+ }
this.client = null;
- this.schema = null;
+ }
+ if (this.restClient != null) {
+ try {
+ this.restClient.close();
+ } catch (RuntimeException e) {
+ if (closeFailure == null) {
+ closeFailure = e;
+ } else {
+ closeFailure.addSuppressed(e);
+ }
+ }
+ this.restClient = null;
+ }
+ this.vertexAPI = null;
+ this.edgeAPI = null;
+ this.schema = null;
+ if (closeFailure != null) {
+ throw closeFailure;
}
}
}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/HugeGraphOperations.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/HugeGraphOperations.java
new file mode 100644
index 000000000000..2122f85eb77a
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/HugeGraphOperations.java
@@ -0,0 +1,79 @@
+/*
+ * 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.seatunnel.hugegraph.client;
+
+import org.apache.hugegraph.structure.constant.Cardinality;
+import org.apache.hugegraph.structure.constant.DataType;
+import org.apache.hugegraph.structure.graph.Edge;
+import org.apache.hugegraph.structure.graph.Shard;
+import org.apache.hugegraph.structure.graph.Vertex;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public interface HugeGraphOperations {
+
+ Set getVertexLabelPropertiesOrNull(String label);
+
+ Set getEdgeLabelPropertiesOrNull(String label);
+
+ /** Lists the names of all vertex labels defined in the graph schema. */
+ List listVertexLabels();
+
+ /** Lists the names of all edge labels defined in the graph schema. */
+ List listEdgeLabels();
+
+ DataType getPropertyDataType(String propertyName);
+
+ Cardinality getPropertyCardinality(String propertyName);
+
+ /**
+ * Lists one page of vertices of {@code label}. When {@code filter} is non-empty its entries are
+ * applied server-side as property-equality conditions; null/empty means no filtering.
+ */
+ PageResult listVertices(
+ String label, Map filter, String page, int limit);
+
+ /**
+ * Lists one page of edges of {@code label}. See {@link #listVertices} for the filter contract.
+ */
+ PageResult listEdges(String label, Map filter, String page, int limit);
+
+ /**
+ * Splits the vertex keyspace into shards of approximately {@code splitSize} bytes each, for
+ * parallel scanning. Requires a backend that supports scan (RocksDB / HBase / Cassandra); the
+ * memory backend does not.
+ */
+ List vertexShards(long splitSize);
+
+ /** Splits the edge keyspace into shards. See {@link #vertexShards}. */
+ List edgeShards(long splitSize);
+
+ /**
+ * Scans one page of vertices within {@code shard}. Unlike {@link #listVertices}, the scan is by
+ * key range and returns vertices of ALL labels in the range, so the caller must filter by label
+ * client-side; server-side property filters are not supported here.
+ */
+ PageResult scanVertices(Shard shard, String page, int limit);
+
+ /** Scans one page of edges within {@code shard}. See {@link #scanVertices}. */
+ PageResult scanEdges(Shard shard, String page, int limit);
+
+ void close();
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/PageResult.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/PageResult.java
new file mode 100644
index 000000000000..9124b9e731e5
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/client/PageResult.java
@@ -0,0 +1,29 @@
+/*
+ * 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.seatunnel.hugegraph.client;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class PageResult {
+
+ private final List records;
+ private final String nextPage;
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphConnectionConfig.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphConnectionConfig.java
new file mode 100644
index 000000000000..0a5eea1adfb0
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphConnectionConfig.java
@@ -0,0 +1,132 @@
+/*
+ * 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.seatunnel.hugegraph.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorException;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+public class HugeGraphConnectionConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private String host;
+ private int port;
+ private String protocol;
+ private String graphName;
+ private String graphSpace;
+ private String username;
+ private String password;
+ private int maxRetries;
+ private int retryBackoffMs;
+ private int retryBackoffMaxMs;
+
+ public static HugeGraphConnectionConfig of(ReadonlyConfig config) {
+ HugeGraphConnectionConfig connectionConfig = new HugeGraphConnectionConfig();
+ connectionConfig.setHost(config.get(HugeGraphOptions.HOST));
+ connectionConfig.setPort(config.get(HugeGraphOptions.PORT));
+ connectionConfig.setProtocol(
+ config.getOptional(HugeGraphOptions.PROTOCOL)
+ .orElse(HugeGraphOptions.PROTOCOL.defaultValue()));
+ connectionConfig.setGraphName(config.get(HugeGraphOptions.GRAPH_NAME));
+ connectionConfig.setGraphSpace(
+ config.getOptional(HugeGraphOptions.GRAPH_SPACE)
+ .filter(graphSpace -> !graphSpace.isEmpty())
+ .orElse(HugeGraphOptions.GRAPH_SPACE.defaultValue()));
+ config.getOptional(HugeGraphOptions.USERNAME).ifPresent(connectionConfig::setUsername);
+ config.getOptional(HugeGraphOptions.PASSWORD).ifPresent(connectionConfig::setPassword);
+ connectionConfig.setMaxRetries(
+ config.getOptional(HugeGraphOptions.MAX_RETRIES)
+ .orElse(HugeGraphOptions.MAX_RETRIES.defaultValue()));
+ connectionConfig.setRetryBackoffMs(
+ config.getOptional(HugeGraphOptions.RETRY_BACKOFF_MS)
+ .orElse(HugeGraphOptions.RETRY_BACKOFF_MS.defaultValue()));
+ connectionConfig.setRetryBackoffMaxMs(
+ config.getOptional(HugeGraphOptions.RETRY_BACKOFF_MAX_MS)
+ .orElse(HugeGraphOptions.RETRY_BACKOFF_MAX_MS.defaultValue()));
+ validate(connectionConfig);
+ return connectionConfig;
+ }
+
+ private static void validate(HugeGraphConnectionConfig config) {
+ // Fail fast at config-load with the offending option name, so the job stops before opening
+ // a client that would otherwise surface a generic connection error much later.
+ if (isBlank(config.getHost())) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Option 'host' must not be empty");
+ }
+ if (isBlank(config.getGraphName())) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Option 'graph_name' must not be empty");
+ }
+ // graph_space is not validated for emptiness: it carries a non-empty default ("DEFAULT"),
+ // HugeGraphConnectionConfig.of() coalesces blank values to that default, and
+ // HugeGraphClient additionally falls back to "DEFAULT" for any null — so it can never be
+ // empty here.
+ if (config.getPort() < 1 || config.getPort() > 65535) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ String.format(
+ "Option 'port' must be in range [1, 65535], but got %s",
+ config.getPort()));
+ }
+ if (!"http".equalsIgnoreCase(config.getProtocol())
+ && !"https".equalsIgnoreCase(config.getProtocol())) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ String.format(
+ "Option 'protocol' must be 'http' or 'https', but got '%s'",
+ config.getProtocol()));
+ }
+ // Credentials must be paired — a lone username or lone password almost always indicates a
+ // config typo and produces a confusing 401 downstream.
+ boolean userSet = !isBlank(config.getUsername());
+ boolean passwordSet = !isBlank(config.getPassword());
+ if (userSet != passwordSet) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Options 'username' and 'password' must be set together");
+ }
+ if (config.getMaxRetries() < 0) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Option 'max_retries' must be greater than or equal to 0");
+ }
+ if (config.getRetryBackoffMs() < 0) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Option 'retry_backoff_ms' must be greater than or equal to 0");
+ }
+ if (config.getRetryBackoffMaxMs() < 0) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Option 'retry_backoff_max_ms' must be greater than or equal to 0");
+ }
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.trim().isEmpty();
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphDataSaveMode.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphDataSaveMode.java
new file mode 100644
index 000000000000..c8ef5195e095
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphDataSaveMode.java
@@ -0,0 +1,37 @@
+/*
+ * 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.seatunnel.hugegraph.config;
+
+/**
+ * Controls how pre-existing data in the target graph is handled before the Sink writes. Mirrors the
+ * SeaTunnel standard {@code DataSaveMode} naming but implemented locally within the connector.
+ */
+public enum HugeGraphDataSaveMode {
+
+ /** Keep existing data; new elements are written on top (default). */
+ APPEND_DATA,
+
+ /**
+ * Before writing, delete the existing data of only the labels this job's mappings
+ * target (edges then vertices), leaving their schema and any other labels' data intact.
+ * Deleting a vertex also removes its incident edges on the server. Scoped per label so, with a
+ * multi-table sink, dropping one table does not wipe another; and on checkpoint restart the
+ * drop is not re-run, so data written before the restart survives.
+ */
+ DROP_DATA
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphOptions.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphOptions.java
index 5dcaa6c1e089..193f35a357a7 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphOptions.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphOptions.java
@@ -33,6 +33,12 @@ public class HugeGraphOptions {
public static final Option PORT =
Options.key("port").intType().noDefaultValue().withDescription("HugeGraph server port");
+ public static final Option PROTOCOL =
+ Options.key("protocol")
+ .stringType()
+ .defaultValue("http")
+ .withDescription("HugeGraph server protocol. Supported values: http, https");
+
public static final Option GRAPH_NAME =
Options.key("graph_name")
.stringType()
@@ -42,8 +48,9 @@ public class HugeGraphOptions {
public static final Option GRAPH_SPACE =
Options.key("graph_space")
.stringType()
- .noDefaultValue()
- .withDescription("The graph space of the graph to be operated on");
+ .defaultValue("DEFAULT")
+ .withDescription(
+ "The graph space the graph belongs to. Defaults to 'DEFAULT'.");
public static final Option USERNAME =
Options.key("username")
@@ -66,6 +73,52 @@ public class HugeGraphOptions {
.defaultValue(5000)
.withDescription("The batch flash period");
+ public static final Option CHECK_VERTEX =
+ Options.key("check_vertex")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether the server verifies that an edge's source/target vertices "
+ + "exist when writing edges. When false (default), edges whose "
+ + "endpoints were never loaded are written as orphan edges (or "
+ + "trigger server-side phantom vertex auto-creation). Enable to "
+ + "reject such edges.");
+
+ public static final Option BATCH_FAILURE_FALLBACK =
+ Options.key("batch_failure_fallback")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "When true, a failed batch insert falls back to inserting records one "
+ + "by one so a single bad ('poison') record no longer fails the "
+ + "whole batch. Failed records are logged and skipped; the rest "
+ + "succeed. Default false (fail-fast): any batch failure fails "
+ + "the task immediately. Opt in explicitly when record skipping "
+ + "is acceptable.");
+
+ public static final Option MAX_INSERT_ERRORS =
+ Options.key("max_insert_errors")
+ .intType()
+ .defaultValue(0)
+ .withDescription(
+ "Maximum number of records that may be skipped by the single-record "
+ + "fallback (batch_failure_fallback=true) before the task is "
+ + "failed. Default 0: any skipped record fails the task. Set to "
+ + "-1 for unlimited (never fail on skipped records). Only applies "
+ + "when batch_failure_fallback is enabled.");
+
+ public static final Option FAILURE_DATA_PATH =
+ Options.key("failure_data_path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional local directory. When set, every record skipped by the "
+ + "single-record fallback is appended (as the mapped vertex/edge "
+ + "id, label, properties and the server error) to a per-subtask "
+ + "file under this directory for offline investigation. Note: in "
+ + "cluster mode the file is created on the worker node running "
+ + "the sink subtask, not the submitting client.");
+
public static final Option MAX_RETRIES =
Options.key("max_retries").intType().defaultValue(3).withDescription("The retry times");
@@ -73,5 +126,16 @@ public class HugeGraphOptions {
Options.key("retry_backoff_ms")
.intType()
.defaultValue(5000)
- .withDescription("The retry backoff time");
+ .withDescription(
+ "The base retry backoff time in milliseconds. Backoff grows "
+ + "exponentially per attempt (retry_backoff_ms * 2^(attempt-1)), "
+ + "capped at retry_backoff_max_ms.");
+
+ public static final Option RETRY_BACKOFF_MAX_MS =
+ Options.key("retry_backoff_max_ms")
+ .intType()
+ .defaultValue(30000)
+ .withDescription(
+ "Upper bound in milliseconds for the exponential retry backoff, so a "
+ + "high max_retries cannot produce pathologically long sleeps.");
}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSchemaSaveMode.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSchemaSaveMode.java
new file mode 100644
index 000000000000..0dbb565aa0d5
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSchemaSaveMode.java
@@ -0,0 +1,31 @@
+/*
+ * 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.seatunnel.hugegraph.config;
+
+/**
+ * Controls schema management behavior during Sink initialization. Aligns with SeaTunnel standard
+ * {@code SchemaSaveMode} naming but implemented locally within the connector.
+ */
+public enum HugeGraphSchemaSaveMode {
+
+ /** Auto-create missing PropertyKey/VertexLabel/EdgeLabel; never modify existing schema. */
+ CREATE_SCHEMA_WHEN_NOT_EXIST,
+
+ /** Do not create any schema; fail immediately if schema is missing or mismatched. */
+ ERROR_WHEN_SCHEMA_NOT_EXIST
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkConfig.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkConfig.java
index dcb0fb32c303..398423ec07ca 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkConfig.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkConfig.java
@@ -18,60 +18,241 @@
package org.apache.seatunnel.connectors.seatunnel.hugegraph.config;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import lombok.Data;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
@Data
public class HugeGraphSinkConfig implements Serializable {
- private String host;
- private int port;
- private String graphName;
- private String graphSpace;
- private String username;
- private String password;
- private SchemaConfig schemaConfig;
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HugeGraphSinkConfig.class);
+
+ // Shared connection config
+ private HugeGraphConnectionConfig connectionConfig;
+
+ // Batch config
private int batchSize;
private int batchIntervalMs;
+ private boolean batchFailureFallback;
+ private boolean checkVertex;
private int maxRetries;
private int retryBackoffMs;
+ // Max records the single-record fallback may skip before the task fails (-1 = unlimited).
+ private int maxInsertErrors;
+ // Optional directory to persist skipped-record failure samples; null = do not persist.
+ private String failureDataPath;
+
+ // New: multi-mapping config
+ private List mappings;
+ private HugeGraphSchemaSaveMode schemaSaveMode;
+ private HugeGraphDataSaveMode dataSaveMode;
+ private boolean deleteVertexWithEdges;
+ private boolean allowCascadeDeleteUnmappedEdges;
- // mapping config
+ // Legacy (deprecated, kept for backward compat parsing only)
+ private SchemaConfig schemaConfig;
private List selectedFields;
private List ignoredFields;
public static HugeGraphSinkConfig of(ReadonlyConfig config) {
HugeGraphSinkConfig sinkConfig = new HugeGraphSinkConfig();
- sinkConfig.setHost(config.get(HugeGraphOptions.HOST));
- sinkConfig.setPort(config.get(HugeGraphOptions.PORT));
- sinkConfig.setGraphName(config.get(HugeGraphOptions.GRAPH_NAME));
+ // Connection
+ sinkConfig.setConnectionConfig(HugeGraphConnectionConfig.of(config));
+
+ // Batch
sinkConfig.setBatchSize(
config.getOptional(HugeGraphOptions.BATCH_SIZE)
.orElse(HugeGraphOptions.BATCH_SIZE.defaultValue()));
sinkConfig.setBatchIntervalMs(
config.getOptional(HugeGraphOptions.BATCH_INTERVAL_MS)
.orElse(HugeGraphOptions.BATCH_INTERVAL_MS.defaultValue()));
- sinkConfig.setMaxRetries(
- config.getOptional(HugeGraphOptions.MAX_RETRIES)
- .orElse(HugeGraphOptions.MAX_RETRIES.defaultValue()));
- sinkConfig.setRetryBackoffMs(
- config.getOptional(HugeGraphOptions.RETRY_BACKOFF_MS)
- .orElse(HugeGraphOptions.RETRY_BACKOFF_MS.defaultValue()));
- sinkConfig.setSchemaConfig(config.get(HugeGraphSinkOptions.SCHEMA_CONFIG));
+ sinkConfig.setBatchFailureFallback(
+ config.getOptional(HugeGraphOptions.BATCH_FAILURE_FALLBACK)
+ .orElse(HugeGraphOptions.BATCH_FAILURE_FALLBACK.defaultValue()));
+ sinkConfig.setCheckVertex(
+ config.getOptional(HugeGraphOptions.CHECK_VERTEX)
+ .orElse(HugeGraphOptions.CHECK_VERTEX.defaultValue()));
+ sinkConfig.setMaxRetries(sinkConfig.getConnectionConfig().getMaxRetries());
+ sinkConfig.setRetryBackoffMs(sinkConfig.getConnectionConfig().getRetryBackoffMs());
+ sinkConfig.setMaxInsertErrors(
+ config.getOptional(HugeGraphOptions.MAX_INSERT_ERRORS)
+ .orElse(HugeGraphOptions.MAX_INSERT_ERRORS.defaultValue()));
+ config.getOptional(HugeGraphOptions.FAILURE_DATA_PATH)
+ .ifPresent(sinkConfig::setFailureDataPath);
+
+ // Resolve mappings with backward compatibility
+ sinkConfig.setMappings(resolveMappings(config, sinkConfig));
+ applyMappingDefaults(sinkConfig.getMappings());
+
+ // Multi-table contract: source_table is an ALL-or-NOTHING switch.
+ // All absent → single-table backward-compatible (each mapping activates in the one writer).
+ // All present → multi-table (each mapping activates only in its matching writer).
+ // Mixed → misconfiguration; fail fast with a clear diagnostic.
+ if (sinkConfig.getMappings() != null) {
+ validateSourceTableConsistency(sinkConfig.getMappings());
+ }
+ boolean legacyConfig = sinkConfig.getSchemaConfig() != null;
+ sinkConfig.setSchemaSaveMode(
+ config.getOptional(HugeGraphSinkOptions.SCHEMA_SAVE_MODE)
+ .orElse(
+ legacyConfig
+ ? HugeGraphSchemaSaveMode.ERROR_WHEN_SCHEMA_NOT_EXIST
+ : HugeGraphSinkOptions.SCHEMA_SAVE_MODE.defaultValue()));
+ sinkConfig.setDeleteVertexWithEdges(
+ config.getOptional(HugeGraphSinkOptions.DELETE_VERTEX_WITH_EDGES)
+ .orElse(
+ legacyConfig
+ ? true
+ : HugeGraphSinkOptions.DELETE_VERTEX_WITH_EDGES
+ .defaultValue()));
+ sinkConfig.setAllowCascadeDeleteUnmappedEdges(
+ config.getOptional(HugeGraphSinkOptions.ALLOW_CASCADE_DELETE_UNMAPPED_EDGES)
+ .orElse(
+ HugeGraphSinkOptions.ALLOW_CASCADE_DELETE_UNMAPPED_EDGES
+ .defaultValue()));
+ sinkConfig.setDataSaveMode(
+ config.getOptional(HugeGraphSinkOptions.DATA_SAVE_MODE)
+ .orElse(HugeGraphSinkOptions.DATA_SAVE_MODE.defaultValue()));
+
+ // Deprecated fields (parse but warn)
config.getOptional(HugeGraphSinkOptions.SELECTED_FIELDS)
- .ifPresent(sinkConfig::setSelectedFields);
+ .ifPresent(
+ fields -> {
+ LOG.warn(
+ "Option 'selected_fields' is deprecated. Use 'properties' within each mapping instead.");
+ sinkConfig.setSelectedFields(fields);
+ });
config.getOptional(HugeGraphSinkOptions.IGNORED_FIELDS)
- .ifPresent(sinkConfig::setIgnoredFields);
-
- config.getOptional(HugeGraphOptions.GRAPH_SPACE).ifPresent(sinkConfig::setGraphSpace);
- config.getOptional(HugeGraphOptions.USERNAME).ifPresent(sinkConfig::setUsername);
- config.getOptional(HugeGraphOptions.PASSWORD).ifPresent(sinkConfig::setPassword);
+ .ifPresent(
+ fields -> {
+ LOG.warn(
+ "Option 'ignored_fields' is deprecated. Use 'properties' within each mapping instead.");
+ sinkConfig.setIgnoredFields(fields);
+ });
return sinkConfig;
}
+
+ /**
+ * Converts legacy global field selection into the mapping property list. The old writer ignored
+ * {@code schema_config.properties} and wrote all fields after applying selected/ignored fields,
+ * so preserving that behavior requires the input row schema.
+ */
+ public void applyLegacyFieldSelection(SeaTunnelRowType rowType) {
+ if (schemaConfig == null || mappings == null || mappings.isEmpty()) {
+ return;
+ }
+
+ List effectiveFields;
+ if (selectedFields != null && !selectedFields.isEmpty()) {
+ effectiveFields = new ArrayList<>(selectedFields);
+ } else {
+ effectiveFields = new ArrayList<>(Arrays.asList(rowType.getFieldNames()));
+ if (ignoredFields != null && !ignoredFields.isEmpty()) {
+ Set ignored = new HashSet<>(ignoredFields);
+ effectiveFields.removeIf(ignored::contains);
+ }
+ }
+ mappings.get(0).setProperties(effectiveFields);
+ }
+
+ private static List resolveMappings(
+ ReadonlyConfig config, HugeGraphSinkConfig sinkConfig) {
+ boolean hasMappings = config.getOptional(HugeGraphSinkOptions.MAPPINGS).isPresent();
+ boolean hasSchemaConfig =
+ config.getOptional(HugeGraphSinkOptions.SCHEMA_CONFIG).isPresent();
+
+ if (hasMappings) {
+ if (hasSchemaConfig) {
+ LOG.warn(
+ "Both 'mappings' and 'schema_config' are present. "
+ + "'schema_config' will be ignored. Please migrate to 'mappings'.");
+ }
+ return config.get(HugeGraphSinkOptions.MAPPINGS);
+ }
+
+ if (hasSchemaConfig) {
+ SchemaConfig schemaConfig = config.get(HugeGraphSinkOptions.SCHEMA_CONFIG);
+ sinkConfig.setSchemaConfig(schemaConfig);
+ return Collections.singletonList(MappingConfig.fromLegacySchemaConfig(schemaConfig));
+ }
+
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Either 'mappings' or 'schema_config' must be specified. "
+ + "'mappings' is the recommended option.");
+ }
+
+ private static void applyMappingDefaults(List mappings) {
+ if (mappings == null) {
+ return;
+ }
+ for (MappingConfig m : mappings) {
+ if (m.getDateFormat() == null || m.getDateFormat().isEmpty()) {
+ m.setDateFormat("yyyy-MM-dd");
+ }
+ // Leave timeZone unset when the user did not configure one; DataTypeUtil then falls
+ // back to ZoneId.systemDefault(), matching the HugeGraph Source. Hard-coding GMT+8
+ // here previously silently shifted absolute times by up to 8 hours when the Source
+ // ran on a JVM whose default zone was not Asia/Shanghai.
+ }
+ }
+
+ /**
+ * Enforces the ALL-or-NOTHING contract on {@code source_table}.
+ *
+ * All mappings with {@code source_table} set → multi-table mode: each mapping activates only
+ * in the writer whose {@code CatalogTable.getTablePath()} matches. All mappings without {@code
+ * source_table} → single-table backward-compatible: every mapping activates in the one writer.
+ * A mix of set and unset is ambiguous — the user either forgot to add {@code source_table} to
+ * some mappings, or accidentally added it to one. Refuse with a clear diagnostic.
+ */
+ static void validateSourceTableConsistency(List mappings) {
+ boolean anySet = false;
+ boolean anyUnset = false;
+ List setLabels = new ArrayList<>();
+ List unsetLabels = new ArrayList<>();
+
+ for (MappingConfig m : mappings) {
+ if (m.getSourceTable() == null || m.getSourceTable().isEmpty()) {
+ anyUnset = true;
+ unsetLabels.add(m.getLabel());
+ } else {
+ anySet = true;
+ setLabels.add(m.getLabel());
+ }
+ }
+
+ if (anySet && anyUnset) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ String.format(
+ "Inconsistent 'source_table' configuration. %d mapping(s) set it (%s), "
+ + "but %d mapping(s) are missing it (%s). "
+ + "'source_table' is an ALL-or-NOTHING switch: either every "
+ + "mapping declares it (multi-table mode — each mapping activates "
+ + "only in the matching writer), or none do (single-table mode — "
+ + "the backward-compatible default). Check that you haven't "
+ + "forgotten 'source_table' on some mappings, or added it to one "
+ + "mapping by mistake.",
+ setLabels.size(), setLabels, unsetLabels.size(), unsetLabels));
+ }
+ }
}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkOptions.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkOptions.java
index 815ca7892894..e2c1b874ff08 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkOptions.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSinkOptions.java
@@ -24,22 +24,73 @@
public class HugeGraphSinkOptions {
+ public static final Option> MAPPINGS =
+ Options.key("mappings")
+ .listType(MappingConfig.class)
+ .noDefaultValue()
+ .withDescription(
+ "List of mapping configurations. Each mapping describes how to write "
+ + "a single vertex or edge label from the input row.");
+
+ public static final Option SCHEMA_SAVE_MODE =
+ Options.key("schema_save_mode")
+ .enumType(HugeGraphSchemaSaveMode.class)
+ .defaultValue(HugeGraphSchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST)
+ .withDescription(
+ "Schema management mode. CREATE_SCHEMA_WHEN_NOT_EXIST (default) auto-creates "
+ + "missing PropertyKey/VertexLabel/EdgeLabel. ERROR_WHEN_SCHEMA_NOT_EXIST "
+ + "fails if schema is missing.");
+
+ public static final Option DATA_SAVE_MODE =
+ Options.key("data_save_mode")
+ .enumType(HugeGraphDataSaveMode.class)
+ .defaultValue(HugeGraphDataSaveMode.APPEND_DATA)
+ .withDescription(
+ "How pre-existing data is handled before writing. APPEND_DATA (default) "
+ + "keeps existing data. DROP_DATA deletes the existing data of only "
+ + "the labels this job targets (edges then vertices) at job start, "
+ + "preserving their schema and any other labels' data; the drop is "
+ + "scoped per label and is not re-run on checkpoint restart.");
+
+ public static final Option DELETE_VERTEX_WITH_EDGES =
+ Options.key("delete_vertex_with_edges")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "When true, DELETE rows for vertices will cascade-delete associated edges. "
+ + "Default false: only the vertex itself is deleted.");
+
+ public static final Option ALLOW_CASCADE_DELETE_UNMAPPED_EDGES =
+ Options.key("allow_cascade_delete_unmapped_edges")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "When data_save_mode is DROP_DATA, deleting vertices cascades to their "
+ + "incident edges — including edge labels not listed in this job's "
+ + "mappings. Default false: the job fails fast and lists the unmapped "
+ + "edge labels. Set to true to accept the destructive cascade.");
+
+ // --- Legacy options (deprecated, kept for backward compatibility) ---
+
+ public static final Option SCHEMA_CONFIG =
+ Options.key("schema_config")
+ .objectType(SchemaConfig.class)
+ .noDefaultValue()
+ .withDescription(
+ "[Deprecated] Use 'mappings' instead. Legacy schema configuration object "
+ + "that describes the mapping to a vertex or edge.");
+
public static final Option> SELECTED_FIELDS =
Options.key("selected_fields")
.listType()
.noDefaultValue()
- .withDescription("Selected Fields");
+ .withDescription(
+ "[Deprecated] Use 'properties' within each mapping instead. Selected fields.");
public static final Option> IGNORED_FIELDS =
Options.key("ignored_fields")
.listType()
.noDefaultValue()
- .withDescription("Ignored Fields");
-
- public static final Option SCHEMA_CONFIG =
- Options.key("schema_config")
- .objectType(SchemaConfig.class)
- .noDefaultValue()
.withDescription(
- "Schema configuration object that describes the mapping to a vertex or edge.");
+ "[Deprecated] Use 'properties' within each mapping instead. Ignored fields.");
}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSourceConfig.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSourceConfig.java
new file mode 100644
index 000000000000..0d6d3bcb0206
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSourceConfig.java
@@ -0,0 +1,156 @@
+/*
+ * 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.seatunnel.hugegraph.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorException;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.time.DateTimeException;
+import java.time.ZoneId;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+@Data
+public class HugeGraphSourceConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private HugeGraphConnectionConfig connectionConfig;
+ private String label;
+ private MappingConfig.LabelType labelType;
+ // Read-all-labels mode: when true, {@code label}/{@code schema} are null and {@code labels}
+ // holds every label of {@code labelType} to read (one produced table each).
+ private boolean readAllLabels;
+ private List labels;
+ private SeaTunnelRowType schema;
+ private int pageSize;
+ private long splitSize;
+ private String timeZone;
+ // Optional server-side property equality conditions; null/empty = read all elements.
+ private Map filter;
+
+ public static HugeGraphSourceConfig of(ReadonlyConfig config, SeaTunnelRowType schema) {
+ HugeGraphSourceConfig sourceConfig = new HugeGraphSourceConfig();
+ sourceConfig.setConnectionConfig(HugeGraphConnectionConfig.of(config));
+ sourceConfig.setReadAllLabels(false);
+ sourceConfig.setLabel(config.get(HugeGraphSourceOptions.LABEL));
+ sourceConfig.setLabels(Collections.singletonList(config.get(HugeGraphSourceOptions.LABEL)));
+ sourceConfig.setLabelType(
+ config.getOptional(HugeGraphSourceOptions.LABEL_TYPE)
+ .orElse(HugeGraphSourceOptions.LABEL_TYPE.defaultValue()));
+ sourceConfig.setSchema(schema);
+ sourceConfig.setPageSize(
+ config.getOptional(HugeGraphSourceOptions.PAGE_SIZE)
+ .orElse(HugeGraphSourceOptions.PAGE_SIZE.defaultValue()));
+ sourceConfig.setSplitSize(
+ config.getOptional(HugeGraphSourceOptions.SPLIT_SIZE)
+ .orElse(HugeGraphSourceOptions.SPLIT_SIZE.defaultValue()));
+ config.getOptional(HugeGraphSourceOptions.TIME_ZONE).ifPresent(sourceConfig::setTimeZone);
+ config.getOptional(HugeGraphSourceOptions.FILTER).ifPresent(sourceConfig::setFilter);
+ validate(sourceConfig);
+ return sourceConfig;
+ }
+
+ /**
+ * Read-all-labels construction: no single {@code label} and no user {@code schema}/{@code
+ * filter}; the labels are discovered from the server and each gets its own auto-discovered row
+ * type. See {@link
+ * org.apache.seatunnel.connectors.seatunnel.hugegraph.source.HugeGraphSourceFactory}.
+ */
+ public static HugeGraphSourceConfig ofReadAll(ReadonlyConfig config, List labels) {
+ HugeGraphSourceConfig sourceConfig = new HugeGraphSourceConfig();
+ sourceConfig.setConnectionConfig(HugeGraphConnectionConfig.of(config));
+ sourceConfig.setReadAllLabels(true);
+ sourceConfig.setLabel(null);
+ sourceConfig.setLabels(labels);
+ sourceConfig.setLabelType(
+ config.getOptional(HugeGraphSourceOptions.LABEL_TYPE)
+ .orElse(HugeGraphSourceOptions.LABEL_TYPE.defaultValue()));
+ sourceConfig.setSchema(null);
+ sourceConfig.setPageSize(
+ config.getOptional(HugeGraphSourceOptions.PAGE_SIZE)
+ .orElse(HugeGraphSourceOptions.PAGE_SIZE.defaultValue()));
+ sourceConfig.setSplitSize(
+ config.getOptional(HugeGraphSourceOptions.SPLIT_SIZE)
+ .orElse(HugeGraphSourceOptions.SPLIT_SIZE.defaultValue()));
+ config.getOptional(HugeGraphSourceOptions.TIME_ZONE).ifPresent(sourceConfig::setTimeZone);
+ validate(sourceConfig);
+ return sourceConfig;
+ }
+
+ private static void validate(HugeGraphSourceConfig sourceConfig) {
+ int pageSize = sourceConfig.getPageSize();
+ if (pageSize < HugeGraphSourceOptions.MIN_PAGE_SIZE
+ || pageSize > HugeGraphSourceOptions.MAX_PAGE_SIZE) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ String.format(
+ "Option 'page_size' must be in range [%s, %s], but got %s",
+ HugeGraphSourceOptions.MIN_PAGE_SIZE,
+ HugeGraphSourceOptions.MAX_PAGE_SIZE,
+ pageSize));
+ }
+
+ if (sourceConfig.getSplitSize() < HugeGraphSourceOptions.MIN_SPLIT_SIZE) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ String.format(
+ "Option 'split_size' must be at least %s bytes (the HugeGraph minimum "
+ + "shard size); a smaller value would split the keyspace into a "
+ + "huge number of shards and risk OOM / oversized checkpoints. "
+ + "Got %s.",
+ HugeGraphSourceOptions.MIN_SPLIT_SIZE, sourceConfig.getSplitSize()));
+ }
+
+ if (sourceConfig.isReadAllLabels()) {
+ // Read-all mode discovers labels from the server; there must be at least one, and no
+ // single user schema applies (each label gets its own auto-discovered row type).
+ if (sourceConfig.getLabels() == null || sourceConfig.getLabels().isEmpty()) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Read-all-labels mode requires at least one label, but none were discovered.");
+ }
+ } else if (sourceConfig.getSchema() == null) {
+ // Single-label mode: schema must be present, but an empty fields block is valid — a
+ // property-less label (e.g. a pure relationship edge, or a vertex with no properties)
+ // is exported as just the reserved columns (~id/~label/…). Requiring a fake property
+ // would make such labels unreadable.
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ "Option 'schema' is required (use 'schema = { fields {} }' for a label with no properties)");
+ }
+ if (sourceConfig.getTimeZone() != null) {
+ try {
+ ZoneId.of(sourceConfig.getTimeZone());
+ } catch (DateTimeException e) {
+ throw new HugeGraphConnectorException(
+ HugeGraphConnectorErrorCode.ILLEGAL_CONFIG_ARGUMENT,
+ String.format(
+ "Option 'time_zone' must be a valid ZoneId, but got '%s'",
+ sourceConfig.getTimeZone()),
+ e);
+ }
+ }
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSourceOptions.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSourceOptions.java
new file mode 100644
index 000000000000..5a42f1b79a0c
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/HugeGraphSourceOptions.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.seatunnel.connectors.seatunnel.hugegraph.config;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+import java.util.Map;
+
+public class HugeGraphSourceOptions {
+
+ public static final int MIN_PAGE_SIZE = 100;
+ public static final int MAX_PAGE_SIZE = 10000;
+ // Lower bound for split_size (1 MiB), matching the HugeGraph server's own minimum shard size.
+ // A smaller value shatters the keyspace into a huge number of shards — one split per shard,
+ // each
+ // persisted into every checkpoint — risking OOM / oversized checkpoints, and the server rejects
+ // it anyway; reject it up front with a clear message.
+ public static final long MIN_SPLIT_SIZE = 1048576L;
+
+ public static final Option LABEL =
+ Options.key("label")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("HugeGraph vertex label or edge label to read");
+
+ public static final Option LABEL_TYPE =
+ Options.key("label_type")
+ .enumType(MappingConfig.LabelType.class)
+ .defaultValue(MappingConfig.LabelType.VERTEX)
+ .withDescription("HugeGraph label type. Supported values are VERTEX and EDGE");
+
+ public static final Option PAGE_SIZE =
+ Options.key("page_size")
+ .intType()
+ .defaultValue(1000)
+ .withDescription("Records per HugeGraph page, must be in range [100, 10000]");
+
+ public static final Option TIME_ZONE =
+ Options.key("time_zone")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Time zone used to convert HugeGraph DATE values that the server returns "
+ + "as an epoch/Date (the instant is rendered as a local date-time "
+ + "in this zone). It does NOT apply when the server returns a DATE "
+ + "already serialized as a wall-clock string (e.g. "
+ + "'yyyy-MM-dd HH:mm:ss.SSS') — that value is kept verbatim, since "
+ + "its original zone is not carried in the string. When omitted, "
+ + "the worker JVM default time zone is used for backward "
+ + "compatibility.");
+
+ public static final Option SPLIT_SIZE =
+ Options.key("split_size")
+ .longType()
+ .defaultValue(1048576L)
+ .withDescription(
+ "Target size in bytes of each key-range shard when parallelism > 1. "
+ + "The server splits the keyspace into shards of roughly this "
+ + "size and readers scan them in parallel; a larger value yields "
+ + "fewer, bigger shards. Ignored when parallelism = 1 (which uses "
+ + "the single label-list scan). Requires a scan-capable backend "
+ + "(RocksDB / HBase / Cassandra).");
+
+ public static final Option> FILTER =
+ Options.key("filter")
+ .type(new TypeReference>() {})
+ .noDefaultValue()
+ .withDescription(
+ "Optional property equality conditions applied server-side when "
+ + "reading the label, e.g. { country = \"US\", active = \"true\" }. "
+ + "Only elements whose properties match all entries are returned. "
+ + "Every key must be a property of the configured label. When "
+ + "omitted, all elements of the label are read.");
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/LabelOptions.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/LabelOptions.java
new file mode 100644
index 000000000000..77c7d4cb019a
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/LabelOptions.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.seatunnel.hugegraph.config;
+
+import java.util.Map;
+
+/**
+ * Optional HugeGraph label attributes applied at schema-creation time (TTL, TTL start-time
+ * property, label-index toggle, and user-defined metadata). All fields are nullable; a null field
+ * means "leave at the HugeGraph server default".
+ */
+public class LabelOptions {
+
+ private final Long ttl;
+ private final String ttlStartTime;
+ private final Boolean enableLabelIndex;
+ private final Map userdata;
+
+ public LabelOptions(
+ Long ttl, String ttlStartTime, Boolean enableLabelIndex, Map userdata) {
+ this.ttl = ttl;
+ this.ttlStartTime = ttlStartTime;
+ this.enableLabelIndex = enableLabelIndex;
+ this.userdata = userdata;
+ }
+
+ public Long getTtl() {
+ return ttl;
+ }
+
+ public String getTtlStartTime() {
+ return ttlStartTime;
+ }
+
+ public Boolean getEnableLabelIndex() {
+ return enableLabelIndex;
+ }
+
+ public Map getUserdata() {
+ return userdata;
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/ListFormat.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/ListFormat.java
new file mode 100644
index 000000000000..35aae38946c0
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/ListFormat.java
@@ -0,0 +1,51 @@
+/*
+ * 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.seatunnel.hugegraph.config;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * How a raw string cell is parsed into the elements of a SET / LIST property. Defaults preserve the
+ * connector's historical behavior: an optional surrounding {@code [ ]}, comma-separated elements,
+ * with blank elements dropped.
+ */
+@Data
+public class ListFormat implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Optional leading symbol stripped before splitting (e.g. {@code [}). Empty disables it. */
+ private String startSymbol = "[";
+
+ /** Optional trailing symbol stripped before splitting (e.g. {@code ]}). Empty disables it. */
+ private String endSymbol = "]";
+
+ /** Delimiter between elements. */
+ private String elemDelimiter = ",";
+
+ /** Element values to drop after splitting (in addition to blank elements). */
+ private List ignoredElems;
+
+ public List getIgnoredElems() {
+ return ignoredElems == null ? Collections.emptyList() : ignoredElems;
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/MappingConfig.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/MappingConfig.java
index c71c2ba91427..c25649f4854d 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/MappingConfig.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/MappingConfig.java
@@ -17,21 +17,208 @@
package org.apache.seatunnel.connectors.seatunnel.hugegraph.config;
+import org.apache.hugegraph.structure.constant.Frequency;
+import org.apache.hugegraph.structure.constant.IdStrategy;
+import org.apache.hugegraph.structure.graph.UpdateStrategy;
+
import lombok.Data;
import java.io.Serializable;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@Data
public class MappingConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ // Element type
+ private LabelType type;
+ private String label;
+
+ // Optional: binds this mapping to a specific input CatalogTable.
+ // When set, the mapping only activates in a writer whose tablePath.toString()
+ // matches (multi-table sink). When absent / empty, the mapping activates in
+ // every writer — backward compatible with single-table jobs where there is
+ // only one writer. The value should be the table path string as it appears
+ // in the source's produced CatalogTable (e.g. "hugegraph.person").
+ private String sourceTable;
+
+ // Vertex-specific
+ private IdStrategy idStrategy;
+ private List idFields;
+ // Expand a list-valued id cell into one vertex per element (INSERT/append only, CUSTOMIZE ids).
+ private boolean unfold;
+
+ // Edge-specific
+ private SourceTargetConfig sourceConfig;
+ private SourceTargetConfig targetConfig;
+ private Frequency frequency;
+ private List sortKeys;
+ // Expand a list-valued source/target id cell into multiple edges (cartesian; INSERT/append
+ // only,
+ // CUSTOMIZE endpoint ids).
+ private boolean unfoldSource;
+ private boolean unfoldTarget;
+
+ // Property config. `properties` is the selected whitelist (only these source fields become
+ // properties); when empty, all input fields are used. `ignored` is the opposite blacklist
+ // (all fields except these). The two are mutually exclusive.
+ private List properties;
+ private List ignored;
+
+ // Field mapping (source field name → target property name)
private Map fieldMapping;
- private Map valueMapping;
+ // Per-field value mapping: outer key = source field name, inner map = rawValue -> mappedValue.
+ // Scoping by field prevents one column's rule from bleeding into another (e.g. gender M->male
+ // must not also rewrite status M).
+ private Map> valueMapping;
private List nullableKeys;
+ private List notNullableKeys;
private List nullValues;
- private List sortKeys;
+
+ // Per-property update-merge strategies (OVERRIDE / APPEND / SUM / UNION / ...), keyed by target
+ // property name. When set, existing elements are merged instead of overwritten.
+ private Map updateStrategies;
// Time config
private String dateFormat;
+ private List extraDateFormats;
private String timeZone;
+
+ // How raw string cells are parsed into SET/LIST elements.
+ private ListFormat listFormat;
+
+ // Label metadata (for schema creation)
+ private Long ttl;
+ private String ttlStartTime;
+ private String enableLabelIndex;
+ private Map userdata;
+
+ public enum LabelType {
+ VERTEX,
+ EDGE
+ }
+
+ @Data
+ public static class SourceTargetConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private String label;
+ private List idFields;
+ }
+
+ public Map getFieldMapping() {
+ return fieldMapping == null ? Collections.emptyMap() : fieldMapping;
+ }
+
+ public Map> getValueMapping() {
+ return valueMapping == null ? Collections.emptyMap() : valueMapping;
+ }
+
+ public List getNullValues() {
+ return nullValues == null ? Collections.emptyList() : nullValues;
+ }
+
+ public List getNullableKeys() {
+ return nullableKeys == null ? Collections.emptyList() : nullableKeys;
+ }
+
+ public List getNotNullableKeys() {
+ return notNullableKeys == null ? Collections.emptyList() : notNullableKeys;
+ }
+
+ public List getSortKeys() {
+ return sortKeys == null ? Collections.emptyList() : sortKeys;
+ }
+
+ public List getProperties() {
+ return properties == null ? Collections.emptyList() : properties;
+ }
+
+ public List getIgnored() {
+ return ignored == null ? Collections.emptyList() : ignored;
+ }
+
+ public ListFormat getListFormat() {
+ return listFormat == null ? new ListFormat() : listFormat;
+ }
+
+ public List getExtraDateFormats() {
+ return extraDateFormats == null ? Collections.emptyList() : extraDateFormats;
+ }
+
+ public Map getUpdateStrategies() {
+ return updateStrategies == null ? Collections.emptyMap() : updateStrategies;
+ }
+
+ public String getSourceTable() {
+ return sourceTable == null ? "" : sourceTable;
+ }
+
+ /**
+ * Whether this mapping is applicable to a writer serving the given table path. A mapping
+ * without {@code sourceTable} applies to every writer (backward compatible); a mapping with
+ * {@code sourceTable} only applies when the table path matches.
+ */
+ public boolean appliesTo(String tablePath) {
+ if (sourceTable == null || sourceTable.isEmpty()) {
+ return true;
+ }
+ return sourceTable.equals(tablePath);
+ }
+
+ /** Converts a legacy SchemaConfig to the new unified MappingConfig. */
+ public static MappingConfig fromLegacySchemaConfig(SchemaConfig schema) {
+ MappingConfig config = new MappingConfig();
+
+ // Element type & label
+ if (schema.getType() != null) {
+ config.setType(LabelType.valueOf(schema.getType().name()));
+ }
+ config.setLabel(schema.getLabel());
+
+ // Vertex config
+ config.setIdStrategy(schema.getIdStrategy());
+ config.setIdFields(schema.getIdFields());
+
+ // Edge config
+ if (schema.getSourceConfig() != null) {
+ SourceTargetConfig src = new SourceTargetConfig();
+ src.setLabel(schema.getSourceConfig().getLabel());
+ src.setIdFields(schema.getSourceConfig().getIdFields());
+ config.setSourceConfig(src);
+ }
+ if (schema.getTargetConfig() != null) {
+ SourceTargetConfig tgt = new SourceTargetConfig();
+ tgt.setLabel(schema.getTargetConfig().getLabel());
+ tgt.setIdFields(schema.getTargetConfig().getIdFields());
+ config.setTargetConfig(tgt);
+ }
+ config.setFrequency(schema.getFrequency());
+
+ // Properties
+ config.setProperties(schema.getProperties());
+
+ // Label metadata
+ config.setTtl(schema.getTtl());
+ config.setTtlStartTime(schema.getTtlStartTime());
+ config.setEnableLabelIndex(schema.getEnableLabelIndex());
+ config.setUserdata(schema.getUserdata());
+
+ // Flatten the nested mapping config
+ if (schema.getMapping() != null) {
+ MappingConfig legacy = schema.getMapping();
+ config.setFieldMapping(legacy.getFieldMapping());
+ config.setValueMapping(legacy.getValueMapping());
+ config.setNullableKeys(legacy.getNullableKeys());
+ config.setNullValues(legacy.getNullValues());
+ config.setSortKeys(legacy.getSortKeys());
+ config.setDateFormat(legacy.getDateFormat());
+ config.setTimeZone(legacy.getTimeZone());
+ }
+
+ return config;
+ }
}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/ReservedColumns.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/ReservedColumns.java
new file mode 100644
index 000000000000..17b1cd4cb409
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/ReservedColumns.java
@@ -0,0 +1,69 @@
+/*
+ * 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.seatunnel.hugegraph.config;
+
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Reserved output columns emitted by the HugeGraph Source. Every reserved column name starts with
+ * {@code ~}, which HugeGraph forbids for property key names, so they never collide with user
+ * properties.
+ *
+ * These columns carry the pre-assembled HugeGraph element ids (the vertex id in {@code ~id}, the
+ * edge endpoint vertex ids in {@code ~source_id}/{@code ~target_id}). When a Sink mapping sets a
+ * single reserved column as its {@code idFields} / endpoint {@code idFields}, the mapper consumes
+ * that pre-assembled id directly instead of re-building one from primary-key columns — this is what
+ * makes a lossless HugeGraph → HugeGraph clone of edges and of CUSTOMIZE-id vertices possible.
+ */
+public final class ReservedColumns {
+
+ public static final String PREFIX = "~";
+
+ public static final String ID = "~id";
+ public static final String LABEL = "~label";
+ public static final String SOURCE_ID = "~source_id";
+ public static final String SOURCE_LABEL = "~source_label";
+ public static final String TARGET_ID = "~target_id";
+ public static final String TARGET_LABEL = "~target_label";
+
+ private ReservedColumns() {}
+
+ /** Whether {@code field} is a reserved Source column (starts with {@code ~}). */
+ public static boolean isReserved(String field) {
+ return field != null && field.startsWith(PREFIX);
+ }
+
+ /**
+ * Whether an {@code idFields} list requests raw-id passthrough: exactly one field, and that
+ * field is a reserved Source column carrying a pre-assembled id.
+ */
+ public static boolean isRawIdPassthrough(List idFields) {
+ return idFields != null && idFields.size() == 1 && isReserved(idFields.get(0));
+ }
+
+ /**
+ * Removes reserved column names from the collection in-place, so callers that build a property
+ * set from all row fields (e.g. mappers and validators) can strip the non-property passthrough
+ * columns with one call. Returns {@code fields} for fluent use.
+ */
+ public static > T stripReserved(T fields) {
+ fields.removeIf(ReservedColumns::isReserved);
+ return fields;
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/SchemaConfig.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/SchemaConfig.java
index 4dfe3cddfe28..abcfe399fcc8 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/SchemaConfig.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/config/SchemaConfig.java
@@ -26,9 +26,15 @@
import java.util.List;
import java.util.Map;
+/**
+ * Legacy schema configuration object for backward compatibility. New configurations should use
+ * {@code mappings[]} with {@link MappingConfig} instead.
+ */
@Data
public class SchemaConfig implements Serializable {
+ private static final long serialVersionUID = 1L;
+
// General config
private LabelType type;
private String label;
@@ -52,7 +58,9 @@ public class SchemaConfig implements Serializable {
private SourceTargetConfig targetConfig;
private Frequency frequency;
- // Mapping Config
+ // Mapping Config (legacy nested object). Kept as MappingConfig to preserve the public
+ // getMapping()/setMapping() accessor descriptors of the previously released connector;
+ // MappingConfig already carries all the legacy nested fields.
private MappingConfig mapping;
public enum LabelType {
@@ -62,6 +70,7 @@ public enum LabelType {
@Data
public static class SourceTargetConfig implements Serializable {
+ private static final long serialVersionUID = 1L;
private String label;
private List idFields;
}
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/exception/HugeGraphConnectorErrorCode.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/exception/HugeGraphConnectorErrorCode.java
index e5608287c382..1c5205d394d9 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/exception/HugeGraphConnectorErrorCode.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/exception/HugeGraphConnectorErrorCode.java
@@ -27,6 +27,7 @@ public enum HugeGraphConnectorErrorCode implements SeaTunnelErrorCode {
BUFFER_ADD_FAILED("HUGEGRAPH-05", "BatchBuffer is already closed."),
INVALID_GRAPH_SCHEMA("HUGEGRAPH-06", "Invalid Graph Schema"),
ILLEGAL_CONFIG_ARGUMENT("HUGEGRAPH-07", "Illegal argument"),
+ SCHEMA_CREATION_FAILED("HUGEGRAPH-08", "Schema auto-creation failed"),
;
private final String code;
diff --git a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/mapper/EdgeMapper.java b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/mapper/EdgeMapper.java
index 2e22a76fa80d..ac37738d85b7 100644
--- a/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/mapper/EdgeMapper.java
+++ b/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/mapper/EdgeMapper.java
@@ -20,12 +20,15 @@
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.client.HugeGraphClient;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.MappingConfig;
-import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.SchemaConfig;
-import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.SchemaConfig.SourceTargetConfig;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.MappingConfig.SourceTargetConfig;
+import org.apache.seatunnel.connectors.seatunnel.hugegraph.config.ReservedColumns;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorErrorCode;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.exception.HugeGraphConnectorException;
import org.apache.seatunnel.connectors.seatunnel.hugegraph.utils.DataTypeUtil;
+import org.apache.hugegraph.serializer.direct.util.SplicingIdGenerator;
+import org.apache.hugegraph.structure.GraphElement;
+import org.apache.hugegraph.structure.constant.Frequency;
import org.apache.hugegraph.structure.constant.IdStrategy;
import org.apache.hugegraph.structure.graph.Edge;
import org.apache.hugegraph.structure.schema.PropertyKey;
@@ -38,107 +41,266 @@
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.stream.Collectors;
public class EdgeMapper implements GraphDataMapper {
- private final SchemaConfig schemaConfig;
private final MappingConfig mappingConfig;
private final Map fieldsIndex;
private final HugeGraphClient client;
private final String labelId;
private final Map propertyKeyCache;
+ private final Set propertySourceFields;
+ private final Set edgeIdSourceFields;
+
+ // Cached at construction time to avoid per-row schema queries
+ private final String sourceVertexLabelId;
+ private final IdStrategy sourceIdStrategy;
+ private final String targetVertexLabelId;
+ private final IdStrategy targetIdStrategy;
+ private final boolean unfoldSource;
+ private final boolean unfoldTarget;
public EdgeMapper(
- SchemaConfig schemaConfig, Map fieldsIndex, HugeGraphClient client) {
- this.schemaConfig = schemaConfig;
- this.mappingConfig = getMappingConfig();
+ MappingConfig mappingConfig, Map fieldsIndex, HugeGraphClient client) {
+ this.mappingConfig = mappingConfig;
this.client = client;
- this.labelId = client.getEdgeLabelId(schemaConfig.getLabel());
+ this.labelId = client.getEdgeLabelId(mappingConfig.getLabel());
this.fieldsIndex = fieldsIndex;
- this.propertyKeyCache = getPropertyKeyCache();
+ this.edgeIdSourceFields = resolveEdgeIdSourceFields();
+ this.propertySourceFields = resolvePropertySourceFields();
+ this.propertyKeyCache = buildPropertyKeyCache();
+ this.unfoldSource = mappingConfig.isUnfoldSource();
+ this.unfoldTarget = mappingConfig.isUnfoldTarget();
+
+ // Cache source/target vertex metadata to avoid per-row schema queries
+ this.sourceVertexLabelId =
+ client.getVertexLabelId(mappingConfig.getSourceConfig().getLabel());
+ this.sourceIdStrategy = client.getIdStrategy(mappingConfig.getSourceConfig().getLabel());
+ this.targetVertexLabelId =
+ client.getVertexLabelId(mappingConfig.getTargetConfig().getLabel());
+ this.targetIdStrategy = client.getIdStrategy(mappingConfig.getTargetConfig().getLabel());
+ }
+
+ @Override
+ public boolean isUnfoldEnabled() {
+ return unfoldSource || unfoldTarget;
+ }
+
+ private Set resolveEdgeIdSourceFields() {
+ Set fields = new HashSet<>();
+ if (mappingConfig.getSourceConfig() != null
+ && mappingConfig.getSourceConfig().getIdFields() != null) {
+ fields.addAll(mappingConfig.getSourceConfig().getIdFields());
+ }
+ if (mappingConfig.getTargetConfig() != null
+ && mappingConfig.getTargetConfig().getIdFields() != null) {
+ fields.addAll(mappingConfig.getTargetConfig().getIdFields());
+ }
+ return fields;
}
- private MappingConfig getMappingConfig() {
- MappingConfig mapping =
- schemaConfig.getMapping() == null ? new MappingConfig() : schemaConfig.getMapping();
- if (mapping.getFieldMapping() == null) {
- mapping.setFieldMapping(Collections.emptyMap());
+ private Set resolvePropertySourceFields() {
+ Set fields = new HashSet<>();
+ if (mappingConfig.getProperties().isEmpty()) {
+ // Implicit mode ("write every row field as a property") — endpoint id fields locate
+ // vertices and would otherwise be duplicated onto the edge; drop them here.
+ fields.addAll(fieldsIndex.keySet());
+ fields.removeAll(edgeIdSourceFields);
+ fields.removeAll(reservedSourceFields(fieldsIndex.keySet()));
+ // `ignored` blacklist only applies in implicit mode.
+ fields.removeAll(mappingConfig.getIgnored());
+ } else {
+ // Explicit mode — respect the user's list verbatim. If they list an endpoint field it
+ // is genuinely meant to appear as an edge property, matching what SchemaManager
+ // creates on the server.
+ fields.addAll(mappingConfig.getProperties());
}
- if (mapping.getValueMapping() == null) {
- mapping.setValueMapping(Collections.emptyMap());
+ // Sort keys are always edge properties (server-side EdgeId requires them).
+ fields.addAll(mappingConfig.getSortKeys());
+ return fields;
+ }
+
+ /**
+ * Reserved fields emitted by the HugeGraph Source (e.g. {@code ~id}, {@code ~label}). They are
+ * not valid HugeGraph property key names — including them would fail at server-side property
+ * key creation — so drop them from an implicit round-trip.
+ */
+ private static Set reservedSourceFields(Set allFields) {
+ Set reserved = new HashSet<>();
+ for (String field : allFields) {
+ if (field != null && field.startsWith("~")) {
+ reserved.add(field);
+ }
}
- schemaConfig.setMapping(mapping);
- return mapping;
+ return reserved;
}
- private HashMap getPropertyKeyCache() {
+ private HashMap buildPropertyKeyCache() {
HashMap cache = new HashMap<>();
- Map fieldMapping = mappingConfig.getFieldMapping();
- for (String fieldName : fieldsIndex.keySet()) {
- String propertyName = fieldMapping.getOrDefault(fieldName, fieldName);
- cache.put(propertyName, client.getPropertyKey(propertyName));
+ Map