This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 909a217c09 [Fix][Connector-V2] Keep the column of a null field in the
Druid sink (#13046)
909a217c09 is described below
commit 909a217c09e628631afe6edf29a01ec85de1cbc9
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Fri Oct 30 02:39:52 2026 +0000
[Fix][Connector-V2] Keep the column of a null field in the Druid sink
(#13046)
---
docs/en/connectors/sink/Druid.md | 2 +
docs/zh/connectors/sink/Druid.md | 2 +
.../connectors/druid/sink/DruidWriter.java | 7 +-
.../connectors/druid/sink/DruidWriterTest.java | 166 +++++++++++++++++++++
.../seatunnel/e2e/connector/druid/DruidIT.java | 48 ++++++
.../resources/fakesource_to_druid_with_null.conf | 61 ++++++++
6 files changed, 283 insertions(+), 3 deletions(-)
diff --git a/docs/en/connectors/sink/Druid.md b/docs/en/connectors/sink/Druid.md
index 8fe4d19499..978390038d 100644
--- a/docs/en/connectors/sink/Druid.md
+++ b/docs/en/connectors/sink/Druid.md
@@ -79,6 +79,8 @@ Only the data types listed in [Data Type
Mapping](#data-type-mapping) are suppor
Because the connector submits inline CSV data to Druid, string values should
not contain raw commas or line breaks unless they are normalized before the
Druid sink.
+A null field is written as an empty CSV value, so it keeps its column. Druid
stores it according to `druid.generic.useDefaultValueForNull`: with the default
`true`, as `''` for string columns and `0` for numeric columns; with `false`,
as a SQL null.
+
## Example
### Write One Table
diff --git a/docs/zh/connectors/sink/Druid.md b/docs/zh/connectors/sink/Druid.md
index 8e82ba40eb..806c4cf10c 100644
--- a/docs/zh/connectors/sink/Druid.md
+++ b/docs/zh/connectors/sink/Druid.md
@@ -79,6 +79,8 @@ Druid 写入需要主时间列。连接器会自动追加一个名为 `timestamp
由于连接器会把数据以内联 CSV 的形式提交给 Druid，字符串字段中不建议直接包含英文逗号或换行符；如有这类内容，建议在进入 Druid Sink
前先做清洗或替换。
+值为 null 的字段会写成空的 CSV 值，因此仍占据原来的列。Druid 按
`druid.generic.useDefaultValueForNull` 存储该值：默认值为 `true` 时，字符串列存为 `''`，数值列存为
`0`；为 `false` 时存为 SQL null。
+
## 示例
### 写入单表
diff --git
a/seatunnel-connectors-v2/connector-druid/src/main/java/org/apache/seatunnel/connectors/druid/sink/DruidWriter.java
b/seatunnel-connectors-v2/connector-druid/src/main/java/org/apache/seatunnel/connectors/druid/sink/DruidWriter.java
index af8e3a330f..bb2112111a 100644
---
a/seatunnel-connectors-v2/connector-druid/src/main/java/org/apache/seatunnel/connectors/druid/sink/DruidWriter.java
+++
b/seatunnel-connectors-v2/connector-druid/src/main/java/org/apache/seatunnel/connectors/druid/sink/DruidWriter.java
@@ -112,9 +112,10 @@ public class DruidWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
final StringJoiner joiner = new StringJoiner(DEFAULT_FIELD_DELIMITER,
"", "");
for (int i = 0; i < element.getArity(); i++) {
final Object v = element.getField(i);
- if (v != null) {
- joiner.add(v.toString());
- }
+ // A null is written as an empty field so that it keeps its
column. Leaving it out
+ // shifts every later value one column left, the timestamp column
comes out missing,
+ // and Druid drops the row as unparseable without failing the task.
+ joiner.add(v == null ? "" : v.toString());
}
// timestamp column is a required field to add in Druid.
// See
https://druid.apache.org/docs/24.0.0/ingestion/data-model.html#primary-timestamp
diff --git
a/seatunnel-connectors-v2/connector-druid/src/test/java/org/apache/seatunnel/connectors/druid/sink/DruidWriterTest.java
b/seatunnel-connectors-v2/connector-druid/src/test/java/org/apache/seatunnel/connectors/druid/sink/DruidWriterTest.java
new file mode 100644
index 0000000000..b1885cc045
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-druid/src/test/java/org/apache/seatunnel/connectors/druid/sink/DruidWriterTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.druid.sink;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.data.input.ColumnsFilter;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.impl.CsvInputFormat;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.InlineInputSource;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpServer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class DruidWriterTest {
+
+ private static final SeaTunnelRowType ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"id", "name", "amount"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.INT_TYPE, BasicType.STRING_TYPE,
BasicType.LONG_TYPE
+ });
+
+ private static final List<SeaTunnelRow> ROWS =
+ Arrays.asList(
+ new SeaTunnelRow(new Object[] {1, "a", 10L}),
+ new SeaTunnelRow(new Object[] {2, null, 20L}),
+ new SeaTunnelRow(new Object[] {3, "c", null}));
+
+ @Test
+ public void aNullFieldKeepsItsColumn() throws IOException {
+ JsonNode ioConfig = submitIoConfig(ROWS);
+ String[] lines =
ioConfig.get("inputSource").get("data").asText().split("\n");
+
+ Assertions.assertEquals(3, lines.length);
+ // The last field is the processing time the writer appends, the same
on every line.
+ String timestamp = lines[0].substring(lines[0].lastIndexOf(',') + 1);
+ Assertions.assertEquals("1,a,10," + timestamp, lines[0]);
+ Assertions.assertEquals("2,,20," + timestamp, lines[1]);
+ Assertions.assertEquals("3,c,," + timestamp, lines[2]);
+ }
+
+ @Test
+ public void druidReadsEveryRowThatHasANullField(@TempDir Path
temporaryDirectory)
+ throws IOException {
+ JsonNode ioConfig = submitIoConfig(ROWS);
+ List<String> columns = new ArrayList<>();
+ ioConfig.get("inputFormat").get("columns").forEach(column ->
columns.add(column.asText()));
+ Assertions.assertEquals(Arrays.asList("id", "name", "amount",
"timestamp"), columns);
+
+ // Read the submitted data with Druid's own CSV reader and the columns
the task declares.
+ // A row that comes out a field short is rejected there as having no
timestamp.
+ NullHandling.initializeForTests();
+ InputRowSchema schema =
+ new InputRowSchema(
+ new TimestampSpec("timestamp", "auto", null),
+ new DimensionsSpec(
+ Arrays.asList(
+ new StringDimensionSchema("id"),
+ new StringDimensionSchema("name"),
+ new StringDimensionSchema("amount"))),
+ ColumnsFilter.all());
+ List<InputRow> rows = new ArrayList<>();
+ try (CloseableIterator<InputRow> iterator =
+ new
InlineInputSource(ioConfig.get("inputSource").get("data").asText())
+ .reader(
+ schema,
+ new CsvInputFormat(columns, "\n", null, false,
0),
+ temporaryDirectory.toFile())
+ .read()) {
+ iterator.forEachRemaining(rows::add);
+ }
+
+ Assertions.assertEquals(3, rows.size());
+ Assertions.assertNull(rows.get(1).getRaw("name"));
+ Assertions.assertEquals("20", rows.get(1).getRaw("amount"));
+ Assertions.assertEquals("c", rows.get(2).getRaw("name"));
+ Assertions.assertNull(rows.get(2).getRaw("amount"));
+ }
+
+ /**
+ * Writes the rows through a {@link DruidWriter} pointed at a local
stand-in for the Druid
+ * Overlord, and returns the {@code ioConfig} of the indexing task it
submits on close.
+ */
+ private static JsonNode submitIoConfig(List<SeaTunnelRow> rows) throws
IOException {
+ List<String> requestBodies = new ArrayList<>();
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext(
+ "/druid/indexer/v1/task",
+ exchange -> {
+ try (InputStream in = exchange.getRequestBody()) {
+ ByteArrayOutputStream body = new
ByteArrayOutputStream();
+ byte[] buffer = new byte[8192];
+ int read;
+ while ((read = in.read(buffer)) > 0) {
+ body.write(buffer, 0, read);
+ }
+ requestBodies.add(new String(body.toByteArray(),
StandardCharsets.UTF_8));
+ }
+ byte[] response =
"{\"task\":\"test\"}".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(response);
+ }
+ });
+ server.start();
+ try {
+ // A batch size above the row count, so the only submission
happens on close.
+ DruidWriter writer =
+ new DruidWriter(
+ ROW_TYPE,
+ "127.0.0.1:" + server.getAddress().getPort(),
+ "test_datasource",
+ rows.size() + 1);
+ for (SeaTunnelRow row : rows) {
+ writer.write(row);
+ }
+ writer.close();
+ } finally {
+ server.stop(0);
+ }
+
+ Assertions.assertEquals(1, requestBodies.size());
+ return new
ObjectMapper().readTree(requestBodies.get(0)).get("spec").get("ioConfig");
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-druid-e2e/src/test/java/org/apache/seatunnel/e2e/connector/druid/DruidIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-druid-e2e/src/test/java/org/apache/seatunnel/e2e/connector/druid/DruidIT.java
index 87390f0f46..e9ff70efb1 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-druid-e2e/src/test/java/org/apache/seatunnel/e2e/connector/druid/DruidIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-druid-e2e/src/test/java/org/apache/seatunnel/e2e/connector/druid/DruidIT.java
@@ -18,6 +18,10 @@
package org.apache.seatunnel.e2e.connector.druid;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ArrayNode;
+
+import org.apache.seatunnel.common.utils.JsonUtils;
import org.apache.seatunnel.e2e.common.TestResource;
import org.apache.seatunnel.e2e.common.TestSuiteBase;
import org.apache.seatunnel.e2e.common.container.EngineType;
@@ -46,7 +50,11 @@ import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.time.Duration;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.given;
@@ -60,6 +68,8 @@ public class DruidIT extends TestSuiteBase implements
TestResource {
private static final String DATASOURCE = "testDataSource";
private static final String MULTI_DATASOURCE_1 = "druid_sink_1";
private static final String MULTI_DATASOURCE_2 = "druid_sink_2";
+ private static final String NULL_FIELD_DATASOURCE_PREFIX =
"druid_null_field_";
+ private static final String NULL_FIELD_DATASOURCE_VARIABLE =
"nullFieldDatasource";
private static final String SQL_QUERY_TEMPLATE = "SELECT * FROM ";
private static final String DRUID_SERVICE_NAME = "router";
private static final int DRUID_SERVICE_PORT = 8888;
@@ -161,6 +171,44 @@ public class DruidIT extends TestSuiteBase implements
TestResource {
});
}
+ @TestTemplate
+ public void testDruidSinkWithNullFields(TestContainer container) throws
Exception {
+ // The Druid cluster is shared by every engine this template runs on,
so each run writes
+ // to its own datasource. Otherwise a run could pass on rows an
earlier run indexed.
+ String datasource =
+ NULL_FIELD_DATASOURCE_PREFIX
+ +
container.identifier().name().toLowerCase(Locale.ROOT);
+ Container.ExecResult execResult =
+ container.executeJob(
+ "/fakesource_to_druid_with_null.conf",
+ Arrays.asList(
+ COORDINATOR_URL_VARIABLE + "=" +
coordinatorURL,
+ NULL_FIELD_DATASOURCE_VARIABLE + "=" +
datasource));
+ Assertions.assertEquals(0, execResult.getExitCode());
+ given().ignoreExceptions()
+ .await()
+ .atLeast(100L, TimeUnit.MILLISECONDS)
+ .atMost(400L, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ ArrayNode rows =
JsonUtils.parseArray(getSelectResponse(datasource));
+ // Rows 2 and 3 each have a null. Before the fix
both were dropped,
+ // because the null was left out of the CSV line.
How Druid renders
+ // the null itself depends on its null handling,
so it is not
+ // asserted.
+ Assertions.assertEquals(3, rows.size(),
rows.toString());
+ Map<Long, JsonNode> rowsById = new HashMap<>();
+ for (JsonNode row : rows) {
+ rowsById.put(row.get("id").asLong(), row);
+ }
+ Assertions.assertEquals("a",
rowsById.get(1L).get("name").asText());
+ Assertions.assertEquals(10L,
rowsById.get(1L).get("amount").asLong());
+ // The value after the null name stayed in its own
column.
+ Assertions.assertEquals(20L,
rowsById.get(2L).get("amount").asLong());
+ Assertions.assertEquals("c",
rowsById.get(3L).get("name").asText());
+ });
+ }
+
private String getSelectResponse(String datasource) throws IOException {
try (CloseableHttpClient client = HttpClients.createDefault()) {
HttpPost request = new HttpPost("http://"; + coordinatorURL +
"/druid/v2/sql");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-druid-e2e/src/test/resources/fakesource_to_druid_with_null.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-druid-e2e/src/test/resources/fakesource_to_druid_with_null.conf
new file mode 100644
index 0000000000..d542bc4fe1
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-druid-e2e/src/test/resources/fakesource_to_druid_with_null.conf
@@ -0,0 +1,61 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ plugin_output = "fake"
+ schema = {
+ fields {
+ id = int
+ name = string
+ amount = bigint
+ }
+ }
+ # The second row has a null string and the third a null number. Each must
still be ingested,
+ # and every value after the null must stay in its own column.
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, "a", 10]
+ },
+ {
+ kind = INSERT
+ fields = [2, null, 20]
+ },
+ {
+ kind = INSERT
+ fields = [3, "c", null]
+ }
+ ]
+ }
+}
+
+transform {
+}
+
+sink {
+ Druid {
+ coordinatorUrl = "${druidCoordinatorUrl}"
+ # Set per engine run by DruidIT, so each run reads back only the rows it
wrote.
+ datasource = "${nullFieldDatasource}"
+ }
+}
