This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13046-f08a248b12c71f39d9b13eab1691bd0a7ea613b5 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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}" + } +}
