map = new LinkedHashMap<>();
+ for (int i = 0; i < fields.size(); i++) {
+ Object value = row.getField(i, fields.get(i).getType());
+ Object normalized = value == null ? null : normalize(value);
+ map.put(fields.get(i).getName(), normalized);
+ }
+ return GSON.toJson(map);
+ }
+
+ private static Object normalize(Object value) {
+ // BinaryString and other non-primitive value holders are serialized via their toString so
+ // that Gson escapes them as plain json strings.
+ if (value instanceof Number || value instanceof Boolean || value instanceof String) {
+ return value;
+ }
+ return value.toString();
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector
new file mode 100644
index 000000000..855b0ca01
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+org.apache.geaflow.dsl.connector.doris.DorisTableConnector
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java
new file mode 100644
index 000000000..e03fdce6a
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.geaflow.dsl.connector.doris;
+
+import org.apache.geaflow.common.config.Configuration;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class DorisConfigKeysTest {
+
+ @Test
+ public void testSinkDefaults() {
+ Configuration conf = new Configuration();
+ Assert.assertEquals(conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS), 10000L);
+ Assert.assertEquals(conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_BYTES),
+ 10485760L);
+ Assert.assertEquals(conf.getInteger(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_RETRIES), 3);
+ Assert.assertEquals(conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_FORMAT), "json");
+ Assert.assertEquals(conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME), "root");
+ }
+
+ @Test
+ public void testSourcePartitionDefaults() {
+ Configuration conf = new Configuration();
+ Assert.assertEquals(
+ conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM), 1L);
+ Assert.assertEquals(
+ conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN), "id");
+ }
+
+ @Test
+ public void testOverrideValue() {
+ Configuration conf = new Configuration();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS.getKey(), "500");
+ Assert.assertEquals(conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS), 500L);
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java
new file mode 100644
index 000000000..bb1edce84
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java
@@ -0,0 +1,342 @@
+/*
+ * 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.geaflow.dsl.connector.doris;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.type.Types;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.data.impl.ObjectRow;
+import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.dsl.common.types.TableField;
+import org.apache.geaflow.dsl.common.types.TableSchema;
+import org.apache.geaflow.dsl.connector.api.FetchData;
+import org.apache.geaflow.dsl.connector.api.Partition;
+import org.apache.geaflow.dsl.connector.api.window.AllFetchWindow;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testng.Assert;
+import org.testng.SkipException;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Integration test for the Doris connector. It boots a real Doris (all-in-one) container using
+ * Testcontainers and exercises the Stream Load sink and the partitioned source end to end. The
+ * container requires host networking (so the FE-to-BE Stream Load redirect is reachable), which is
+ * only available on Linux; the test skips itself when Docker or host networking is not available.
+ *
+ * An external Doris can be targeted instead by setting the {@code doris.it.fenodes} and
+ * {@code doris.it.jdbcUrl} system properties.
+ */
+public class DorisConnectorIntegrationTest {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DorisConnectorIntegrationTest.class);
+
+ private static final String DORIS_IMAGE = "apache/doris:doris-all-in-one-2.1.0";
+ private static final String DATABASE = "geaflow_it";
+ private static final String TABLE = "person";
+ private static final String USERNAME = "root";
+ private static final String PASSWORD = "";
+
+ private static GenericContainer dorisContainer;
+ private static String feNodes;
+ private static String jdbcUrl;
+
+ @BeforeClass
+ public void setUp() throws Exception {
+ String externalFe = System.getProperty("doris.it.fenodes");
+ String externalJdbc = System.getProperty("doris.it.jdbcUrl");
+ if (externalFe != null && externalJdbc != null) {
+ feNodes = externalFe;
+ jdbcUrl = externalJdbc;
+ } else if (Boolean.getBoolean("doris.it.enabled")) {
+ // The container is heavy (multi-GB image) and needs host networking so the FE-to-BE
+ // Stream Load redirect is reachable, so it is opt-in and never runs in a normal CI.
+ if (!DockerClientFactory.instance().isDockerAvailable()) {
+ throw new SkipException("Docker is not available, skip Doris integration test.");
+ }
+ if (!System.getProperty("os.name", "").toLowerCase().contains("linux")) {
+ throw new SkipException("Doris Stream Load redirect needs host networking, only "
+ + "supported on Linux, skip integration test.");
+ }
+ dorisContainer = new GenericContainer<>(DORIS_IMAGE)
+ .withNetworkMode("host")
+ .waitingFor(Wait.forLogMessage(".*get heartbeat response.*", 1)
+ .withStartupTimeout(java.time.Duration.ofMinutes(5)));
+ dorisContainer.start();
+ feNodes = "127.0.0.1:8030";
+ jdbcUrl = "jdbc:mysql://127.0.0.1:9030/";
+ } else {
+ throw new SkipException("Doris integration test is disabled by default. Enable it with "
+ + "-Ddoris.it.enabled=true (requires Docker on a Linux host), or point it at an "
+ + "external Doris with -Ddoris.it.fenodes and -Ddoris.it.jdbcUrl.");
+ }
+ waitForBackendAlive();
+ prepareSchema();
+ }
+
+ @AfterClass
+ public void tearDown() {
+ if (dorisContainer != null) {
+ dorisContainer.stop();
+ }
+ }
+
+ private Connection newConnection(String url) throws Exception {
+ Class.forName("com.mysql.cj.jdbc.Driver");
+ return DriverManager.getConnection(url, USERNAME, PASSWORD);
+ }
+
+ private void waitForBackendAlive() throws Exception {
+ long deadline = System.currentTimeMillis() + java.time.Duration.ofMinutes(3).toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ try (Connection connection = newConnection(jdbcUrl);
+ Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery("SHOW BACKENDS")) {
+ while (rs.next()) {
+ if ("true".equalsIgnoreCase(rs.getString("Alive"))) {
+ return;
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.info("waiting for doris backend alive: {}", e.getMessage());
+ }
+ Thread.sleep(5000);
+ }
+ throw new IllegalStateException("Doris backend did not become alive in time.");
+ }
+
+ private void prepareSchema() throws Exception {
+ try (Connection connection = newConnection(jdbcUrl);
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE IF NOT EXISTS " + DATABASE);
+ statement.execute("DROP TABLE IF EXISTS " + DATABASE + "." + TABLE);
+ statement.execute("CREATE TABLE " + DATABASE + "." + TABLE + " ("
+ + "id BIGINT, name VARCHAR(64), age INT) "
+ + "UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 "
+ + "PROPERTIES(\"replication_num\" = \"1\")");
+ }
+ }
+
+ private StructType schema() {
+ return new StructType(
+ new TableField("id", Types.LONG, false),
+ new TableField("name", Types.BINARY_STRING, true),
+ new TableField("age", Types.INTEGER, true));
+ }
+
+ private Configuration sinkConfig() {
+ Configuration conf = new Configuration();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, feNodes);
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, DATABASE);
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, TABLE);
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME, USERNAME);
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_PASSWORD, PASSWORD);
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS, "1000");
+ return conf;
+ }
+
+ private long countRows() throws Exception {
+ try (Connection connection = newConnection(jdbcUrl);
+ Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery(
+ "SELECT COUNT(*) FROM " + DATABASE + "." + TABLE)) {
+ rs.next();
+ return rs.getLong(1);
+ }
+ }
+
+ private void truncate() throws Exception {
+ try (Connection connection = newConnection(jdbcUrl);
+ Statement statement = connection.createStatement()) {
+ statement.execute("TRUNCATE TABLE " + DATABASE + "." + TABLE);
+ }
+ }
+
+ @Test
+ public void testSinkAndSource() throws Exception {
+ truncate();
+ int rowCount = 2000;
+ DorisTableSink sink = new DorisTableSink();
+ sink.init(sinkConfig(), schema());
+ sink.open(null);
+ for (int i = 0; i < rowCount; i++) {
+ sink.write(ObjectRow.create((long) i, "name_" + i, i % 100));
+ }
+ sink.finish();
+ sink.close();
+ Assert.assertEquals(countRows(), rowCount);
+
+ Configuration sourceConf = new Configuration();
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_JDBC_URL, jdbcUrl + DATABASE);
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, DATABASE);
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, TABLE);
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME, USERNAME);
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_PASSWORD, PASSWORD);
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM, "4");
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN, "id");
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND, "0");
+ sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND,
+ String.valueOf(rowCount));
+
+ DorisTableSource source = new DorisTableSource();
+ source.init(sourceConf, new TableSchema(schema()));
+ source.open(null);
+ List partitions = source.listPartitions();
+ Assert.assertTrue(partitions.size() > 1, "source should produce parallel partitions.");
+ List allRows = new ArrayList<>();
+ for (Partition partition : partitions) {
+ FetchData fetchData =
+ source.fetch(partition, Optional.empty(), new AllFetchWindow(0));
+ Iterator iterator = fetchData.getDataIterator();
+ while (iterator.hasNext()) {
+ allRows.add(iterator.next());
+ }
+ }
+ source.close();
+ Assert.assertEquals(allRows.size(), rowCount);
+ }
+
+ @Test
+ public void testStreamLoadThroughputBenchmark() throws Exception {
+ int rowCount = 20000;
+
+ truncate();
+ DorisTableSink sink = new DorisTableSink();
+ sink.init(sinkConfig(), schema());
+ sink.open(null);
+ long streamLoadStart = System.currentTimeMillis();
+ for (int i = 0; i < rowCount; i++) {
+ sink.write(ObjectRow.create((long) i, "name_" + i, i % 100));
+ }
+ sink.finish();
+ sink.close();
+ long streamLoadCost = System.currentTimeMillis() - streamLoadStart;
+ Assert.assertEquals(countRows(), rowCount);
+
+ truncate();
+ long jdbcStart = System.currentTimeMillis();
+ try (Connection connection = newConnection(jdbcUrl);
+ Statement statement = connection.createStatement()) {
+ for (int i = 0; i < rowCount; i++) {
+ statement.execute(String.format(
+ "INSERT INTO %s.%s (id, name, age) VALUES (%d, '%s', %d)",
+ DATABASE, TABLE, i, "name_" + i, i % 100));
+ }
+ }
+ long jdbcCost = System.currentTimeMillis() - jdbcStart;
+
+ LOGGER.info("Doris write benchmark for {} rows: streamLoad={}ms ({} rows/s), "
+ + "row-by-row JDBC={}ms ({} rows/s)", rowCount, streamLoadCost,
+ rowCount * 1000L / Math.max(1, streamLoadCost), jdbcCost,
+ rowCount * 1000L / Math.max(1, jdbcCost));
+ Assert.assertTrue(streamLoadCost < jdbcCost,
+ "Stream Load should be faster than row-by-row JDBC insert.");
+ }
+
+ @Test(expectedExceptions = GeaFlowDSLException.class)
+ public void testTableNotFound() throws Exception {
+ Configuration conf = sinkConfig();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "no_such_table");
+ DorisTableSink sink = new DorisTableSink();
+ sink.init(conf, schema());
+ sink.open(null);
+ sink.write(ObjectRow.create(1L, "a", 1));
+ sink.finish();
+ }
+
+ @Test
+ public void testFeFailover() throws Exception {
+ truncate();
+ // Put an unreachable FE first, the sink must fail over to the real FE and still succeed.
+ Configuration conf = sinkConfig();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, "127.0.0.1:1," + feNodes);
+ DorisTableSink sink = new DorisTableSink();
+ sink.init(conf, schema());
+ sink.open(null);
+ for (int i = 0; i < 100; i++) {
+ sink.write(ObjectRow.create((long) i, "name_" + i, i));
+ }
+ sink.finish();
+ sink.close();
+ Assert.assertEquals(countRows(), 100);
+ }
+
+ @Test
+ public void testDataTypeFidelity() throws Exception {
+ String typesTable = "types_table";
+ try (Connection connection = newConnection(jdbcUrl);
+ Statement statement = connection.createStatement()) {
+ statement.execute("DROP TABLE IF EXISTS " + DATABASE + "." + typesTable);
+ statement.execute("CREATE TABLE " + DATABASE + "." + typesTable + " ("
+ + "id BIGINT, c_int INT, c_double DOUBLE, c_varchar VARCHAR(256)) "
+ + "UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 "
+ + "PROPERTIES(\"replication_num\" = \"1\")");
+ }
+ StructType typesSchema = new StructType(
+ new TableField("id", Types.LONG, false),
+ new TableField("c_int", Types.INTEGER, true),
+ new TableField("c_double", Types.DOUBLE, true),
+ new TableField("c_varchar", Types.BINARY_STRING, true));
+
+ Configuration conf = sinkConfig();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, typesTable);
+ String tricky = "a\nb\tc\"d\\e\u4e2d\u6587\ud83d\ude00";
+ DorisTableSink sink = new DorisTableSink();
+ sink.init(conf, typesSchema);
+ sink.open(null);
+ sink.write(ObjectRow.create(1L, 10, 1.5, tricky));
+ sink.write(ObjectRow.create(2L, null, null, null));
+ sink.finish();
+ sink.close();
+
+ try (Connection connection = newConnection(jdbcUrl);
+ Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery("SELECT id, c_int, c_double, c_varchar FROM "
+ + DATABASE + "." + typesTable + " ORDER BY id")) {
+ Assert.assertTrue(rs.next());
+ Assert.assertEquals(rs.getLong("id"), 1L);
+ Assert.assertEquals(rs.getInt("c_int"), 10);
+ Assert.assertEquals(rs.getDouble("c_double"), 1.5);
+ Assert.assertEquals(rs.getString("c_varchar"), tricky);
+ Assert.assertTrue(rs.next());
+ Assert.assertEquals(rs.getLong("id"), 2L);
+ rs.getObject("c_int");
+ Assert.assertTrue(rs.wasNull(), "c_int should round-trip as NULL");
+ rs.getString("c_varchar");
+ Assert.assertTrue(rs.wasNull(), "c_varchar should round-trip as NULL");
+ Assert.assertFalse(rs.next());
+ }
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java
new file mode 100644
index 000000000..f5b0897bf
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.geaflow.dsl.connector.doris;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class DorisStreamLoadTest {
+
+ private DorisStreamLoad newClient(java.util.List feNodes) {
+ return new DorisStreamLoad(feNodes, "test_db", "test_table", "root", "",
+ DorisConstants.FORMAT_CSV, "\t", "\n", Arrays.asList("id", "name"), 1000, 1000, 3);
+ }
+
+ @Test
+ public void testLoadUrlWithHostPort() throws IOException {
+ try (DorisStreamLoad client = newClient(Collections.singletonList("127.0.0.1:8030"))) {
+ Assert.assertEquals(client.getLoadUrl(),
+ "http://127.0.0.1:8030/api/test_db/test_table/_stream_load");
+ }
+ }
+
+ @Test
+ public void testLoadUrlWithScheme() throws IOException {
+ try (DorisStreamLoad client = newClient(Collections.singletonList("http://doris-fe:8030"))) {
+ Assert.assertEquals(client.getLoadUrl(),
+ "http://doris-fe:8030/api/test_db/test_table/_stream_load");
+ }
+ }
+
+ @Test
+ public void testMultipleFeNodesForFailover() throws IOException {
+ try (DorisStreamLoad client =
+ newClient(Arrays.asList("fe1:8030", "fe2:8030", "fe3:8030"))) {
+ Assert.assertEquals(client.getLoadUrls().size(), 3);
+ Assert.assertEquals(client.getLoadUrls().get(1),
+ "http://fe2:8030/api/test_db/test_table/_stream_load");
+ }
+ }
+
+ @Test(expectedExceptions = GeaFlowDSLException.class)
+ public void testEmptyFeNodesThrows() throws IOException {
+ try (DorisStreamLoad client = newClient(Collections.emptyList())) {
+ Assert.fail("should have thrown for empty fenodes");
+ }
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSinkTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSinkTest.java
new file mode 100644
index 000000000..7ae21a5be
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSinkTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.geaflow.dsl.connector.doris;
+
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.type.Types;
+import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.dsl.common.types.TableField;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class DorisTableSinkTest {
+
+ private DorisTableSink sink;
+ private StructType schema;
+
+ @BeforeMethod
+ public void setUp() {
+ sink = new DorisTableSink();
+ schema = new StructType(
+ new TableField("id", Types.LONG, false),
+ new TableField("name", Types.BINARY_STRING, true));
+ }
+
+ private Configuration validConfig() {
+ Configuration conf = new Configuration();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, "127.0.0.1:8030");
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, "test_db");
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "test_table");
+ return conf;
+ }
+
+ @Test
+ public void testInit() {
+ sink.init(validConfig(), schema);
+ Assert.assertNotNull(sink);
+ }
+
+ @Test(expectedExceptions = GeaFlowDSLException.class)
+ public void testInitWithoutFeNodes() {
+ Configuration conf = new Configuration();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, "test_db");
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "test_table");
+ sink.init(conf, schema);
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSourceTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSourceTest.java
new file mode 100644
index 000000000..eff52a8f4
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSourceTest.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.geaflow.dsl.connector.doris;
+
+import java.util.List;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.type.Types;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.dsl.common.types.TableField;
+import org.apache.geaflow.dsl.common.types.TableSchema;
+import org.apache.geaflow.dsl.connector.api.Partition;
+import org.apache.geaflow.dsl.connector.doris.DorisTableSource.DorisPartition;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class DorisTableSourceTest {
+
+ private TableSchema schema() {
+ return new TableSchema(new StructType(
+ new TableField("id", Types.LONG, false),
+ new TableField("name", Types.BINARY_STRING, true)));
+ }
+
+ private DorisTableSource newSource(Configuration conf) {
+ Configuration base = conf;
+ base.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_JDBC_URL, "jdbc:mysql://127.0.0.1:9030/db");
+ base.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, "db");
+ base.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "t");
+ DorisTableSource source = new DorisTableSource();
+ source.init(base, schema());
+ return source;
+ }
+
+ @Test
+ public void testSinglePartition() {
+ List partitions = newSource(new Configuration()).listPartitions();
+ Assert.assertEquals(partitions.size(), 1);
+ Assert.assertEquals(((DorisPartition) partitions.get(0)).getWhereClause(), "");
+ }
+
+ @Test
+ public void testRangePartitions() {
+ Configuration conf = new Configuration();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM, "4");
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN, "id");
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND, "0");
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND, "100");
+ List partitions = newSource(conf).listPartitions();
+ Assert.assertEquals(partitions.size(), 4);
+ // The first partition also covers rows whose partition column is entirely NULL.
+ Assert.assertTrue(((DorisPartition) partitions.get(0)).getWhereClause()
+ .contains("id IS NULL"));
+ }
+
+ @Test
+ public void testCustomPartitions() {
+ Configuration conf = new Configuration();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_MODE, "custom");
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_CLAUSES,
+ "dt='2024-01-01';dt='2024-01-02';dt='2024-01-03'");
+ List partitions = newSource(conf).listPartitions();
+ Assert.assertEquals(partitions.size(), 3);
+ Assert.assertEquals(((DorisPartition) partitions.get(1)).getWhereClause(),
+ "WHERE dt='2024-01-02'");
+ }
+
+ @Test
+ public void testCustomPartitionsEmptyFallsBackToSingle() {
+ Configuration conf = new Configuration();
+ conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_MODE, "custom");
+ List partitions = newSource(conf).listPartitions();
+ Assert.assertEquals(partitions.size(), 1);
+ Assert.assertEquals(((DorisPartition) partitions.get(0)).getWhereClause(), "");
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java
new file mode 100644
index 000000000..e004a1af1
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.geaflow.dsl.connector.doris;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.apache.geaflow.common.type.Types;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.data.impl.ObjectRow;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.dsl.common.types.TableField;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class DorisUtilsTest {
+
+ private static final Gson GSON = new Gson();
+
+ private StructType buildSchema() {
+ return new StructType(
+ new TableField("id", Types.LONG, false),
+ new TableField("name", Types.BINARY_STRING, true),
+ new TableField("score", Types.DOUBLE, true));
+ }
+
+ @Test
+ public void testRowToCsv() {
+ StructType schema = buildSchema();
+ Row row = ObjectRow.create(1L, "alice", 9.5);
+ String csv = DorisUtils.rowToCsv(row, schema, "\t");
+ Assert.assertEquals(csv, "1\talice\t9.5");
+ }
+
+ @Test
+ public void testRowToCsvWithNull() {
+ StructType schema = buildSchema();
+ Row row = ObjectRow.create(2L, null, null);
+ String csv = DorisUtils.rowToCsv(row, schema, "\t");
+ Assert.assertEquals(csv, "2\t\\N\t\\N");
+ }
+
+ @Test
+ public void testRowToCsvWithCustomSeparator() {
+ StructType schema = buildSchema();
+ Row row = ObjectRow.create(3L, "bob", 1.0);
+ String csv = DorisUtils.rowToCsv(row, schema, ",");
+ Assert.assertEquals(csv, "3,bob,1.0");
+ }
+
+ @Test
+ public void testRowToJson() {
+ StructType schema = buildSchema();
+ Row row = ObjectRow.create(4L, "carol", 8.0);
+ String json = DorisUtils.rowToJson(row, schema);
+ Assert.assertTrue(json.contains("\"id\":4"));
+ Assert.assertTrue(json.contains("\"name\":\"carol\""));
+ Assert.assertTrue(json.contains("\"score\":8.0"));
+ }
+
+ @Test
+ public void testRowToJsonEscapesSpecialChars() {
+ StructType schema = buildSchema();
+ // A value with a newline, a tab, a double quote, a backslash and unicode/emoji.
+ String tricky = "line1\nline2\tcol\"q\\b\u4e2d\u6587\ud83d\ude00";
+ Row row = ObjectRow.create(1L, tricky, 1.0);
+ String json = DorisUtils.rowToJson(row, schema);
+ // Round-trip: parsing the produced json back must yield the exact original value.
+ JsonObject parsed = GSON.fromJson(json, JsonObject.class);
+ Assert.assertEquals(parsed.get("name").getAsString(), tricky);
+ }
+
+ @Test
+ public void testRowToJsonKeepsNullAndEmptyString() {
+ StructType schema = buildSchema();
+ Row rowWithNull = ObjectRow.create(2L, null, 2.0);
+ JsonObject parsedNull = GSON.fromJson(DorisUtils.rowToJson(rowWithNull, schema),
+ JsonObject.class);
+ Assert.assertTrue(parsedNull.has("name"), "null column key must be kept");
+ Assert.assertTrue(parsedNull.get("name").isJsonNull(), "null column must be json null");
+
+ Row rowWithEmpty = ObjectRow.create(3L, "", 3.0);
+ JsonObject parsedEmpty = GSON.fromJson(DorisUtils.rowToJson(rowWithEmpty, schema),
+ JsonObject.class);
+ Assert.assertEquals(parsedEmpty.get("name").getAsString(), "");
+ }
+
+ @Test
+ public void testRowToCsvKeepsEmptyString() {
+ StructType schema = buildSchema();
+ Row row = ObjectRow.create(7L, "", 1.0);
+ String csv = DorisUtils.rowToCsv(row, schema, "\t");
+ // Empty string stays as an empty field (not the null placeholder).
+ Assert.assertEquals(csv, "7\t\t1.0");
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml
index e1f262b9d..4fe03c6de 100644
--- a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml
@@ -49,6 +49,7 @@
geaflow-dsl-connector-paimon
geaflow-dsl-connector-neo4j
geaflow-dsl-connector-elasticsearch
+ geaflow-dsl-connector-doris
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml
index e9863eb2e..87c9f040a 100644
--- a/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml
+++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml
@@ -81,6 +81,11 @@
geaflow-dsl-connector-jdbc
+
+ org.apache.geaflow
+ geaflow-dsl-connector-doris
+
+
org.apache.geaflow
geaflow-dsl-connector-odps
diff --git a/geaflow/geaflow-dsl/pom.xml b/geaflow/geaflow-dsl/pom.xml
index f79ea4e0d..a7a37063f 100644
--- a/geaflow/geaflow-dsl/pom.xml
+++ b/geaflow/geaflow-dsl/pom.xml
@@ -188,6 +188,12 @@
${project.version}
+
+ org.apache.geaflow
+ geaflow-dsl-connector-doris
+ ${project.version}
+
+