This is an automated email from the ASF dual-hosted git repository.
corgy-w 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 72f32edde0 [Fix][Connector-V2][File] Respect custom filename for
binary sink (#10817)
72f32edde0 is described below
commit 72f32edde0a12c9dcffcca317b2f3ce728483eac
Author: Jast <[email protected]>
AuthorDate: Fri May 8 10:36:38 2026 +0800
[Fix][Connector-V2][File] Respect custom filename for binary sink (#10817)
---
docs/en/connectors/sink/LocalFile.md | 2 +
docs/zh/connectors/sink/LocalFile.md | 4 +-
.../seatunnel/file/config/BaseFileSinkConfig.java | 2 +
.../seatunnel/file/sink/BaseFileSinkWriter.java | 23 +++
.../file/sink/writer/BinaryWriteStrategy.java | 31 ++++
.../file/local/LocalFileBinarySinkTest.java | 193 +++++++++++++++++++++
6 files changed, 254 insertions(+), 1 deletion(-)
diff --git a/docs/en/connectors/sink/LocalFile.md
b/docs/en/connectors/sink/LocalFile.md
index a22fd057e1..115159756b 100644
--- a/docs/en/connectors/sink/LocalFile.md
+++ b/docs/en/connectors/sink/LocalFile.md
@@ -94,6 +94,8 @@ Only used when `custom_filename` is `true`
`file_name_expression` describes the file expression which will be created
into the `path`. We can add the variable `${now}` or `${uuid}` in the
`file_name_expression`, like `test_${uuid}_${now}`,
`${now}` represents the current time, and its format can be defined by
specifying the option `filename_time_format`.
+For `file_format_type = "binary"`, `custom_filename = true` is intended for
renaming a single source file in each sink task. When multiple source files are
written by the same sink task, keep `custom_filename = false` to preserve
source relative paths. When parallel sink subtasks are used, include
`${transactionId}` or `${uuid}` in `file_name_expression` to avoid multiple
subtasks writing the same final file.
+
Please note that, If `is_enable_transaction` is `true`, we will auto add
`${transactionId}_` in the head of the file.
### filename_time_format [string]
diff --git a/docs/zh/connectors/sink/LocalFile.md
b/docs/zh/connectors/sink/LocalFile.md
index 648999e659..0e1549a751 100644
--- a/docs/zh/connectors/sink/LocalFile.md
+++ b/docs/zh/connectors/sink/LocalFile.md
@@ -91,6 +91,8 @@ import ChangeLog from '../changelog/connector-file-local.md';
`file_name_expression` 描述将创建到 `path` 中的文件表达式。我们可以在 `file_name_expression`
中添加变量 `${now}` 或 `${uuid}`，例如 `test_${uuid}_${now}`，`${now}` 表示当前时间，其格式可以通过指定
`filename_time_format` 选项来定义。
+对于 `file_format_type = "binary"`，`custom_filename = true` 适用于在每个 sink task
中重命名单个源文件。当同一个 sink task 会写入多个源文件时，请保持 `custom_filename =
false`，以保留源文件相对路径。当使用并行 sink subtask 时，请在 `file_name_expression` 中包含
`${transactionId}` 或 `${uuid}`，避免多个 subtask 写入同一个最终文件。
+
请注意，如果 `is_enable_transaction` 为 `true`，我们将自动在文件名的头部添加 `${transactionId}_`。
### filename_time_format [string]
@@ -340,4 +342,4 @@ LocalFile {
## 变更日志
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSinkConfig.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSinkConfig.java
index b5f12771db..488cf83a89 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSinkConfig.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseFileSinkConfig.java
@@ -40,6 +40,7 @@ public class BaseFileSinkConfig implements DelimiterConfig,
Serializable {
protected int batchSize;
protected String path;
protected String fileNameExpression;
+ protected boolean customFilename;
protected boolean singleFileMode;
protected boolean createEmptyFileWhenNoData;
protected FileFormat fileFormat;
@@ -60,6 +61,7 @@ public class BaseFileSinkConfig implements DelimiterConfig,
Serializable {
this.path = "";
}
this.fileNameExpression =
pluginConfig.get(FileBaseSinkOptions.FILE_NAME_EXPRESSION);
+ this.customFilename =
pluginConfig.get(FileBaseSinkOptions.CUSTOM_FILENAME);
this.singleFileMode =
pluginConfig.get(FileBaseSinkOptions.SINGLE_FILE_MODE);
this.createEmptyFileWhenNoData =
pluginConfig.get(FileBaseSinkOptions.CREATE_EMPTY_FILE_WHEN_NO_DATA);
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSinkWriter.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSinkWriter.java
index c880026cc3..6849205009 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSinkWriter.java
@@ -22,9 +22,11 @@ import
org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
import org.apache.seatunnel.api.sink.SinkWriter;
import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.Constants;
import org.apache.seatunnel.common.exception.CommonError;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy;
@@ -114,6 +116,21 @@ public class BaseFileSinkWriter
}
private void preCheckConfig(SinkWriter.Context context) {
+ if (writeStrategy.getFileSinkConfig().getFileFormat() ==
FileFormat.BINARY
+ && writeStrategy.getFileSinkConfig().isCustomFilename()
+ && context.getNumberOfParallelSubtasks() > 1
+ && !hasParallelUniqueFilenameExpression(
+
writeStrategy.getFileSinkConfig().getFileNameExpression())) {
+ throw new IllegalArgumentException(
+ "Binary custom filename requires a unique filename
expression when parallel "
+ + "subtasks are used. Please include "
+ + DEFAULT_FILE_NAME_EXPRESSION
+ + " or ${"
+ + Constants.UUID
+ + "} in "
+ + FILE_NAME_EXPRESSION.key()
+ + ".");
+ }
if (writeStrategy.getFileSinkConfig().isSingleFileMode()
&& context.getNumberOfParallelSubtasks() > 1) {
if
(StringUtils.isNotEmpty(writeStrategy.getFileSinkConfig().getFileNameExpression())
@@ -131,6 +148,12 @@ public class BaseFileSinkWriter
}
}
+ private boolean hasParallelUniqueFilenameExpression(String
fileNameExpression) {
+ return StringUtils.isBlank(fileNameExpression)
+ || fileNameExpression.contains(DEFAULT_FILE_NAME_EXPRESSION)
+ || fileNameExpression.contains("${" + Constants.UUID + "}");
+ }
+
private List<String> findTransactionList(
String jobId, String uuidPrefix, HadoopFileSystemProxy
hadoopFileSystemProxy)
throws IOException {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
index 1fe307e33d..d1904f3c94 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
@@ -21,6 +21,7 @@ import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.common.exception.CommonError;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSinkOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import
org.apache.seatunnel.connectors.seatunnel.file.sink.config.FileSinkConfig;
@@ -85,6 +86,9 @@ public class BinaryWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStrea
}
public String getOrCreateFilePathBeingWritten(String relativePath) {
+ if (fileSinkConfig.isCustomFilename()) {
+ return getOrCreateCustomFilePathBeingWritten(relativePath);
+ }
String beingWrittenFilePath = beingWrittenFile.get(relativePath);
if (beingWrittenFilePath != null) {
return beingWrittenFilePath;
@@ -96,6 +100,33 @@ public class BinaryWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStrea
}
}
+ private String getOrCreateCustomFilePathBeingWritten(String relativePath) {
+ String beingWrittenFilePath = beingWrittenFile.get(relativePath);
+ if (beingWrittenFilePath != null) {
+ return beingWrittenFilePath;
+ } else {
+ if (!beingWrittenFile.isEmpty()) {
+ String existingRelativePath =
beingWrittenFile.keySet().iterator().next();
+ throw new FileConnectorException(
+ CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+ "Binary sink custom_filename only supports a single
source file in each "
+ + "sink task. Existing source file: "
+ + existingRelativePath
+ + ", new source file: "
+ + relativePath);
+ }
+ String[] pathSegments =
+ new String[] {
+ transactionDirectory,
+ FileBaseSinkOptions.NON_PARTITION,
+ generateFileName(transactionId)
+ };
+ String newBeingWrittenFilePath = String.join(File.separator,
pathSegments);
+ beingWrittenFile.put(relativePath, newBeingWrittenFilePath);
+ return newBeingWrittenFilePath;
+ }
+ }
+
@Override
public FSDataOutputStream getOrCreateOutputStream(@NonNull String
filePath) {
FSDataOutputStream fsDataOutputStream =
beingWrittenOutputStream.get(filePath);
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileBinarySinkTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileBinarySinkTest.java
new file mode 100644
index 0000000000..69662e49c2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileBinarySinkTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.file.local;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.common.utils.FileUtils;
+import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.file.local.sink.LocalFileSinkFactory;
+import org.apache.seatunnel.connectors.seatunnel.sink.SinkFlowTestUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@DisabledOnOs(
+ value = OS.WINDOWS,
+ disabledReason =
+ "Hadoop has windows problem, please refer
https://cwiki.apache.org/confluence/display/HADOOP2/WindowsProblems";)
+class LocalFileBinarySinkTest {
+
+ private final CatalogTable binaryCatalogTable =
+ CatalogTable.of(
+ TableIdentifier.of("catalog", "database", "table"),
+ TableSchema.builder()
+ .columns(
+ Arrays.asList(
+ PhysicalColumn.of(
+ "data",
+
PrimitiveByteArrayType.INSTANCE,
+ (Long) null,
+ true,
+ null,
+ ""),
+ PhysicalColumn.of(
+ "relativePath",
+ BasicType.STRING_TYPE,
+ 1L,
+ true,
+ null,
+ ""),
+ PhysicalColumn.of(
+ "partIndex",
+ BasicType.LONG_TYPE,
+ 1L,
+ true,
+ null,
+ "")))
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "comment");
+
+ @Test
+ void testBinarySinkUsesCustomFilename() throws Exception {
+ String testPath =
"/tmp/seatunnel/LocalFileBinarySinkTest/customFilename";
+ FileUtils.deleteFile(testPath);
+
+ Map<String, Object> options = createBinarySinkOptions(testPath);
+ options.put("single_file_mode", true);
+ options.put("custom_filename", true);
+ options.put("file_name_expression", "expected_binary_file");
+ options.put("filename_extension", ".raw");
+
+ byte[] data = "binary-content".getBytes(StandardCharsets.UTF_8);
+ SinkFlowTestUtils.runBatchWithCheckpointDisabled(
+ binaryCatalogTable,
+ ReadonlyConfig.fromMap(options),
+ new LocalFileSinkFactory(),
+ Collections.singletonList(new SeaTunnelRow(new Object[] {data,
"source.raw", 0L})));
+
+ Assertions.assertArrayEquals(
+ data, Files.readAllBytes(Paths.get(testPath,
"expected_binary_file.raw")));
+ Assertions.assertFalse(FileUtils.isFileExist(Paths.get(testPath,
"source.raw").toString()));
+ }
+
+ @Test
+ void testBinarySinkKeepsSourceRelativePathByDefault() throws Exception {
+ String testPath =
"/tmp/seatunnel/LocalFileBinarySinkTest/defaultRelativePath";
+ FileUtils.deleteFile(testPath);
+
+ byte[] data = "binary-content".getBytes(StandardCharsets.UTF_8);
+ SinkFlowTestUtils.runBatchWithCheckpointDisabled(
+ binaryCatalogTable,
+ ReadonlyConfig.fromMap(createBinarySinkOptions(testPath)),
+ new LocalFileSinkFactory(),
+ Collections.singletonList(
+ new SeaTunnelRow(new Object[] {data,
"nested/source.raw", 0L})));
+
+ Assertions.assertArrayEquals(
+ data, Files.readAllBytes(Paths.get(testPath, "nested",
"source.raw")));
+ }
+
+ @Test
+ void testBinarySinkRejectsMultipleSourceFilesWhenUsingCustomFilename()
throws Exception {
+ String testPath =
"/tmp/seatunnel/LocalFileBinarySinkTest/customFilenameMultipleFiles";
+ FileUtils.deleteFile(testPath);
+
+ Map<String, Object> options = createBinarySinkOptions(testPath);
+ options.put("single_file_mode", true);
+ options.put("custom_filename", true);
+ options.put("file_name_expression", "expected_binary_file");
+ options.put("filename_extension", ".raw");
+
+ byte[] data = "binary-content".getBytes(StandardCharsets.UTF_8);
+ SeaTunnelRuntimeException exception =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () ->
+
SinkFlowTestUtils.runBatchWithCheckpointDisabled(
+ binaryCatalogTable,
+ ReadonlyConfig.fromMap(options),
+ new LocalFileSinkFactory(),
+ Arrays.asList(
+ new SeaTunnelRow(
+ new Object[] {data,
"source-1.raw", 0L}),
+ new SeaTunnelRow(
+ new Object[] {data,
"source-2.raw", 0L}))));
+
+ Assertions.assertInstanceOf(FileConnectorException.class,
exception.getCause());
+ Assertions.assertTrue(
+ exception.getCause().getMessage().contains("only supports a
single source file"));
+ }
+
+ @Test
+ void testBinarySinkRejectsUnsafeCustomFilenameWithParallelSubtasks()
throws Exception {
+ String testPath =
"/tmp/seatunnel/LocalFileBinarySinkTest/parallelCustomFilename";
+ FileUtils.deleteFile(testPath);
+
+ Map<String, Object> options = createBinarySinkOptions(testPath);
+ options.put("custom_filename", true);
+ options.put("file_name_expression", "expected_binary_file");
+ options.put("filename_extension", ".raw");
+
+ byte[] data = "binary-content".getBytes(StandardCharsets.UTF_8);
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+
SinkFlowTestUtils.runParallelSubtasksBatchWithCheckpointDisabled(
+ binaryCatalogTable,
+ ReadonlyConfig.fromMap(options),
+ new LocalFileSinkFactory(),
+ Collections.singletonList(
+ new SeaTunnelRow(
+ new Object[] {data,
"source.raw", 0L})),
+ 2));
+
+ Assertions.assertTrue(
+ exception
+ .getMessage()
+ .contains("Binary custom filename requires a unique
filename expression"));
+ }
+
+ private Map<String, Object> createBinarySinkOptions(String path) {
+ Map<String, Object> options = new HashMap<>();
+ options.put("path", path);
+ options.put("file_format_type", "binary");
+ options.put("is_enable_transaction", false);
+ return options;
+ }
+}
