This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new f3789613c03 [branch-4.1](outfile) Add opt-in file_name_padding
(#68746) (#69186)
f3789613c03 is described below
commit f3789613c03a60930179a068dfae4dfedfed2876
Author: Gabriel <[email protected]>
AuthorDate: Fri Oct 16 23:29:00 2026 +0800
[branch-4.1](outfile) Add opt-in file_name_padding (#68746) (#69186)
Backport #68746 to `branch-4.1`.
Adds opt-in `file_name_padding` (0–10) to SELECT INTO OUTFILE and INSERT
INTO FUNCTION. A positive width zero-pads each writer's file index; the
default value 0 preserves existing file names. This helps rollover files
sort by index when the configured width covers the indexes.
The Thrift conflict was resolved by preserving this branch's existing
`writer_class` definition and adding only the optional padding field.
Validation:
- Both inlined C++ formatting expressions passed the five original
boundary scenarios in a focused harness.
- FE `OutFileTest` and `BindTVFTableSinkTest` passed.
- 22 local parameter checks passed across OUTFILE and TVF: defaults,
accepted widths, invalid values, and propagation into Thrift sink
options.
- FE Checkstyle and clang-format 16 checks passed for the affected
sources.
- The new SQL regression passed locally against a master build
containing #68746; branch-specific full BE build and end-to-end runs
remain for CI.
The follow-up cleanup removes `result_file_name.h`, inlines the
formatting expression in both writers, and replaces the helper-only test
with `test_result_file_name_padding`. The regression covers both writers
and all three formats, padding defaults and boundaries, invalid values,
custom OUTFILE suffixes, real file rotation past index 9, and data
read-back.
Master follow-up: #69189.
---------
Co-authored-by: Alvaro Tuso <[email protected]>
---
be/src/exec/operator/result_sink_operator.h | 6 ++
be/src/exec/sink/writer/vfile_result_writer.cpp | 9 +-
be/src/exec/sink/writer/vtvf_table_writer.cpp | 6 +-
be/src/exec/sink/writer/vtvf_table_writer.h | 1 +
.../org/apache/doris/analysis/OutFileClause.java | 20 ++++
.../org/apache/doris/planner/TVFTableSink.java | 21 ++++
gensrc/thrift/DataSinks.thrift | 9 ++
.../export_p0/test_result_file_name_padding.groovy | 111 +++++++++++++++++++++
8 files changed, 177 insertions(+), 6 deletions(-)
diff --git a/be/src/exec/operator/result_sink_operator.h
b/be/src/exec/operator/result_sink_operator.h
index 173a5b52e82..e6f249ad8d3 100644
--- a/be/src/exec/operator/result_sink_operator.h
+++ b/be/src/exec/operator/result_sink_operator.h
@@ -66,6 +66,9 @@ struct ResultFileOptions {
//Bring BOM when exporting to CSV format
bool with_bom = false;
int64_t orc_writer_version = 0;
+ // Zero-pad the per-file index in the output file name to this many digits.
+ // 0 means no padding, preserving the historical name format.
+ int file_name_padding = 0;
ResultFileOptions(const TResultFileSinkOptions& t_opt) {
file_path = t_opt.file_path;
@@ -79,6 +82,9 @@ struct ResultFileOptions {
t_opt.__isset.delete_existing_files ?
t_opt.delete_existing_files : false;
file_suffix = t_opt.file_suffix;
with_bom = t_opt.with_bom;
+ if (t_opt.__isset.file_name_padding) {
+ file_name_padding = t_opt.file_name_padding;
+ }
is_local_file = true;
if (t_opt.__isset.broker_addresses) {
diff --git a/be/src/exec/sink/writer/vfile_result_writer.cpp
b/be/src/exec/sink/writer/vfile_result_writer.cpp
index ce9e73d2eba..9ccf2f2dcbf 100644
--- a/be/src/exec/sink/writer/vfile_result_writer.cpp
+++ b/be/src/exec/sink/writer/vfile_result_writer.cpp
@@ -162,10 +162,11 @@ Status VFileResultWriter::_create_file_writer(const
std::string& file_name) {
Status VFileResultWriter::_get_next_file_name(std::string* file_name) {
std::string suffix =
_file_opts->file_suffix.empty() ? _file_format_to_name() :
_file_opts->file_suffix;
- std::stringstream ss;
- ss << _file_opts->file_path << print_id(_fragment_instance_id) << "_" <<
(_file_idx++) << "."
- << suffix;
- *file_name = ss.str();
+ // Padding is a minimum width: larger indices must retain every digit.
+ const int padding = _file_opts->file_name_padding;
+ *file_name =
+ fmt::format("{}{}_{:0{}}.{}", _file_opts->file_path,
print_id(_fragment_instance_id),
+ _file_idx++, padding < 0 ? 0 : padding, suffix);
if (_storage_type == TStorageBackendType::LOCAL) {
// For local file writer, the file_path is a local dir.
// Here we do a simple security verification by checking whether the
file exists.
diff --git a/be/src/exec/sink/writer/vtvf_table_writer.cpp
b/be/src/exec/sink/writer/vtvf_table_writer.cpp
index c4111c7c2d3..70765f51d81 100644
--- a/be/src/exec/sink/writer/vtvf_table_writer.cpp
+++ b/be/src/exec/sink/writer/vtvf_table_writer.cpp
@@ -48,6 +48,7 @@ Status VTVFTableWriter::open(RuntimeState* state,
RuntimeProfile* profile) {
_file_path = _tvf_sink.file_path;
_max_file_size_bytes =
_tvf_sink.__isset.max_file_size_bytes ?
_tvf_sink.max_file_size_bytes : 0;
+ _file_name_padding = _tvf_sink.__isset.file_name_padding ?
_tvf_sink.file_name_padding : 0;
VLOG_DEBUG << "TVF table writer open, query_id=" <<
print_id(_state->query_id())
<< ", tvf_name=" << _tvf_sink.tvf_name << ", file_path=" <<
_tvf_sink.file_path
@@ -164,8 +165,9 @@ Status VTVFTableWriter::_get_next_file_name(std::string*
file_name) {
// file_path is a prefix, generate: {prefix}{query_id}_{idx}.{ext}
std::string query_id_str = print_id(_state->query_id());
- *file_name = fmt::format("{}{}_{}.{}", _file_path, query_id_str,
_file_idx, ext);
- _file_idx++;
+ // Keep the same minimum-width index format as SELECT INTO OUTFILE.
+ *file_name = fmt::format("{}{}_{:0{}}.{}", _file_path, query_id_str,
_file_idx++,
+ _file_name_padding < 0 ? 0 : _file_name_padding,
ext);
return Status::OK();
}
diff --git a/be/src/exec/sink/writer/vtvf_table_writer.h
b/be/src/exec/sink/writer/vtvf_table_writer.h
index 54fabeecd5f..a215a21c4b5 100644
--- a/be/src/exec/sink/writer/vtvf_table_writer.h
+++ b/be/src/exec/sink/writer/vtvf_table_writer.h
@@ -72,6 +72,7 @@ private:
int64_t _current_written_bytes = 0;
int64_t _max_file_size_bytes = 0;
int _file_idx = 0;
+ int _file_name_padding = 0;
std::string _file_path;
// profile counters
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
b/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
index 7bca73b3bee..54b2bea6180 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
@@ -122,11 +122,15 @@ public class OutFileClause {
public static final String PROP_DELETE_EXISTING_FILES =
"delete_existing_files";
public static final String PROP_FILE_SUFFIX = "file_suffix";
public static final String PROP_WITH_BOM = "with_bom";
+ public static final String PROP_FILE_NAME_PADDING = "file_name_padding";
private static final String SCHEMA = "schema";
private static final long DEFAULT_MAX_FILE_SIZE_BYTES = 1 * 1024 * 1024 *
1024; // 1GB
private static final long MIN_FILE_SIZE_BYTES = 5 * 1024 * 1024L; // 5MB
+ // Upper bound on the zero-padding width for the per-file index in output
file
+ // names. _file_idx on the BE is a 32-bit int, so 10 digits covers its
full range.
+ private static final int MAX_FILE_NAME_PADDING = 10;
private String filePath;
private Map<String, String> properties;
@@ -135,6 +139,7 @@ public class OutFileClause {
private boolean deleteExistingFiles = false;
private String fileSuffix = "";
private boolean withBom = false;
+ private int fileNamePadding = 0;
private BrokerDesc brokerDesc = null;
// True if result is written to local disk.
// If set to true, the brokerDesc must be null.
@@ -559,6 +564,20 @@ public class OutFileClause {
copiedProps.remove(PROP_WITH_BOM);
}
+ if (copiedProps.containsKey(PROP_FILE_NAME_PADDING)) {
+ try {
+ fileNamePadding =
Integer.parseInt(copiedProps.get(PROP_FILE_NAME_PADDING));
+ } catch (NumberFormatException e) {
+ throw new AnalysisException(PROP_FILE_NAME_PADDING + " must be
an integer. Given: "
+ + copiedProps.get(PROP_FILE_NAME_PADDING));
+ }
+ if (fileNamePadding < 0 || fileNamePadding >
MAX_FILE_NAME_PADDING) {
+ throw new AnalysisException(PROP_FILE_NAME_PADDING + " must be
between 0 and "
+ + MAX_FILE_NAME_PADDING + ". Given: " +
fileNamePadding);
+ }
+ copiedProps.remove(PROP_FILE_NAME_PADDING);
+ }
+
if (copiedProps.containsKey(PROP_SUCCESS_FILE_NAME)) {
successFileName = copiedProps.get(PROP_SUCCESS_FILE_NAME);
FeNameFormat.checkOutfileSuccessFileName("file name",
successFileName);
@@ -763,6 +782,7 @@ public class OutFileClause {
sinkOptions.setDeleteExistingFiles(deleteExistingFiles);
sinkOptions.setFileSuffix(fileSuffix);
sinkOptions.setWithBom(withBom);
+ sinkOptions.setFileNamePadding(fileNamePadding);
if (brokerDesc != null) {
sinkOptions.setBrokerProperties(brokerDesc.getBackendConfigProperties());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/TVFTableSink.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/TVFTableSink.java
index b8c184ca401..102da9af2e4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/TVFTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/TVFTableSink.java
@@ -47,6 +47,11 @@ import java.util.Map;
* infrastructure as the read-side TVF (SELECT * FROM s3/hdfs/local(...)).
*/
public class TVFTableSink extends DataSink {
+ // Upper bound on the zero-padding width for the per-file index in output
file names
+ // (the "file_name_padding" property). _file_idx on the BE is a 32-bit
int, so 10 digits
+ // covers its full range.
+ private static final int MAX_FILE_NAME_PADDING = 10;
+
private final PlanNodeId exchNodeId;
private final String tvfName;
private final Map<String, String> properties;
@@ -155,6 +160,22 @@ public class TVFTableSink extends DataSink {
tSink.setMaxFileSizeBytes(Long.parseLong(maxFileSizeStr));
}
+ // Zero-pad the per-file index in output file names so they sort
lexicographically.
+ String fileNamePaddingStr = properties.get("file_name_padding");
+ if (fileNamePaddingStr != null) {
+ int fileNamePadding;
+ try {
+ fileNamePadding = Integer.parseInt(fileNamePaddingStr);
+ } catch (NumberFormatException e) {
+ throw new AnalysisException("file_name_padding must be an
integer. Given: " + fileNamePaddingStr);
+ }
+ if (fileNamePadding < 0 || fileNamePadding >
MAX_FILE_NAME_PADDING) {
+ throw new AnalysisException("file_name_padding must be between
0 and "
+ + MAX_FILE_NAME_PADDING + ". Given: " +
fileNamePadding);
+ }
+ tSink.setFileNamePadding(fileNamePadding);
+ }
+
// Delete existing files is handled by FE (InsertIntoTVFCommand),
always tell BE not to delete
tSink.setDeleteExistingFiles(false);
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index ccb3629455f..cc29b7811ee 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -160,6 +160,12 @@ struct TResultFileSinkOptions {
// currently only for csv
// TODO: merge with parquet_compression_type and orc_compression_type
22: optional PlanNodes.TFileCompressType compression_type
+
+ // Zero-pad the per-file index in the output file name to this many digits
+ // (e.g. my_prefix_{fragment_instance_id}_00000.csv). 0 or unset means no
+ // padding, preserving the historical name format. Used to make rollover
+ // files sort lexicographically (so _00011 sorts after _00002).
+ 23: optional i32 file_name_padding
}
struct TMemoryScratchSink {
@@ -601,6 +607,9 @@ struct TTVFTableSink {
13: optional i64 backend_id // local TVF: specify BE
14: optional TTVFWriterType writer_type // NATIVE or JNI
15: optional string writer_class // Java class name (required
when writer_type=JNI)
+ // Zero-pad the per-file index in the output file name to this many digits.
+ // 0 or unset means no padding. Mirrors
TResultFileSinkOptions.file_name_padding.
+ 16: optional i32 file_name_padding
}
struct TMCCommitData {
diff --git
a/regression-test/suites/export_p0/test_result_file_name_padding.groovy
b/regression-test/suites/export_p0/test_result_file_name_padding.groovy
new file mode 100644
index 00000000000..7250c9f4662
--- /dev/null
+++ b/regression-test/suites/export_p0/test_result_file_name_padding.groovy
@@ -0,0 +1,111 @@
+// 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.
+
+suite("test_result_file_name_padding", "p0,external") {
+ def bucket = context.config.otherConfigs.get("s3BucketName")
+ def prefix = "regression/result_file_name_padding/${UUID.randomUUID()}/"
+ def client = getS3Client()
+ def storage = """
+ "s3.endpoint" = "${getS3Endpoint()}",
+ "s3.region" = "${getS3Region()}",
+ "s3.access_key" = "${getS3AK()}",
+ "s3.secret_key" = "${getS3SK()}"
+ """
+ def listKeys = { String path ->
+ def keys = []
+ def listing = client.listObjects(bucket, path)
+ while (true) {
+ keys.addAll(listing.objectSummaries.collect { it.key })
+ if (!listing.isTruncated()) {
+ return keys.sort()
+ }
+ listing = client.listNextBatchOfObjects(listing)
+ }
+ }
+ def writeQuery = { String writer, String path, String format, String
extra, String query ->
+ if (writer == "outfile") {
+ return """${query} INTO OUTFILE "s3://${bucket}/${path}" FORMAT AS
${format}
+ PROPERTIES (${storage} ${extra})"""
+ } else {
+ return """INSERT INTO s3("file_path" = "s3://${bucket}/${path}",
+ "format" = "${format}", ${storage} ${extra}) ${query}"""
+ }
+ }
+ def readFiles = { String path, String format, String projection ->
+ sql """SELECT ${projection} FROM s3("uri" = "s3://${bucket}/${path}",
+ "format" = "${format}", ${storage})"""
+ }
+
+ sql "set parallel_pipeline_task_num = 1"
+ sql "set enable_parallel_outfile = false"
+ sql "set batch_size = 512"
+ try {
+ ["outfile", "tvf"].each { writer ->
+ ["csv", "parquet", "orc"].each { format ->
+ [null, 0, 1, 5, 10].each { padding ->
+ def path = "${prefix}${writer}_${format}_${padding}/part_"
+ def extra = padding == null ? "" : ",
\"file_name_padding\" = \"${padding}\""
+ sql writeQuery(writer, path, format, extra, "SELECT 7 AS
id")
+ def keys = listKeys(path)
+ assertEquals(1, keys.size())
+ def index = "0".padLeft(padding == null ? 0 : padding, '0')
+ assertTrue(keys[0].endsWith("_${index}.${format}"),
keys[0])
+ assertEquals("7", readFiles(keys[0], format,
"*")[0][0].toString())
+ }
+ }
+ ["-1", "11", "abc", "2147483648"].each { padding ->
+ test {
+ sql writeQuery(writer, "${prefix}invalid/part_", "csv",
+ ", \"file_name_padding\" = \"${padding}\"", "SELECT 7
AS id")
+ exception "file_name_padding"
+ }
+ }
+ }
+
+ def suffixPath = "${prefix}suffix/part_"
+ sql writeQuery("outfile", suffixPath, "csv",
+ ', "file_name_padding" = "5", "file_suffix" = "txt"', "SELECT 7 AS
id")
+ def suffixKeys = listKeys(suffixPath)
+ assertEquals(1, suffixKeys.size())
+ assertTrue(suffixKeys[0].endsWith("_00000.txt"))
+ assertEquals("7", readFiles(suffixKeys[0], "csv",
"*")[0][0].toString())
+
+ // Cross index 9 using real file rotation: width is a minimum, never a
truncation limit.
+ ["outfile": 1, "tvf": 5].each { writer, padding ->
+ def path = "${prefix}${writer}_rotation/part_"
+ def maxFileSize = writer == "outfile" ? "5MB" : "5242880"
+ sql writeQuery(writer, path, "csv",
+ ", \"file_name_padding\" = \"${padding}\", \"max_file_size\" =
\"${maxFileSize}\"",
+ "SELECT number, repeat('x', 4096) FROM numbers(\"number\" =
\"18000\") ORDER BY number")
+ def keys = listKeys(path)
+ assertTrue(keys.size() > 10, "Expected rotation past index 9:
${keys}")
+ def indices = keys.collect { key ->
+ def digits = key.substring(key.lastIndexOf('_') + 1,
key.lastIndexOf('.'))
+ def index = digits.toInteger()
+ assertEquals(index.toString().padLeft(padding, '0'), digits)
+ index
+ }
+ assertEquals((0..<keys.size()).toList(), indices.sort(false))
+ if (padding == 5) {
+ assertEquals(indices.sort(false), indices)
+ }
+ assertEquals("18000", readFiles("${path}*", "csv",
"count(*)")[0][0].toString())
+ }
+ } finally {
+ listKeys(prefix).each { client.deleteObject(bucket, it) }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
