This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 897bb254984 [fix](iceberg) Validate Parquet compression before file
creation (#69376)
897bb254984 is described below
commit 897bb25498454ad9f4b6f2386167dc8422b337c1
Author: daidai <[email protected]>
AuthorDate: Thu Nov 5 14:42:29 2026 +0800
[fix](iceberg) Validate Parquet compression before file creation (#69376)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #68854, #69338
Problem Summary:
The Iceberg data, position-delete, and Hive Parquet writers created
their output file before validating the requested Parquet compression
codec. For an unsupported codec, the Iceberg data writer then attempted
to delete an open file, while the other writers returned after creating
the file. This made error cleanup depend on filesystem behavior and
could leave an orphan file. Resolve the codec before calling
`create_file` so invalid codecs fail without creating any output file,
and add a regression test for the failure path.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test: added `UnsupportedParquetCodecDoesNotCreateFile`; CI
will run it.
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [x] No.
- [ ] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../writer/iceberg/viceberg_delete_file_writer.cpp | 5 ++-
.../writer/iceberg/viceberg_partition_writer.cpp | 17 ++++-----
be/src/exec/sink/writer/vhive_partition_writer.cpp | 7 +++-
.../iceberg/iceberg_partition_writer_test.cpp | 42 ++++++++++++++++++++++
4 files changed, 59 insertions(+), 12 deletions(-)
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
index 1233d09dae5..7295a6d4f64 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp
@@ -80,13 +80,14 @@ Status VIcebergDeleteFileWriter::open(RuntimeState* state,
RuntimeProfile* profi
_fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
io::FileWriterOptions file_writer_options = {.used_by_s3_committer =
false};
- RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
// Create file format transformer based on format type
switch (_file_format) {
case TFileFormatType::FORMAT_PARQUET: {
TParquetCompressionType::type parquet_compression_type =
DORIS_TRY(ParquetBuildHelper::to_parquet_compression_type(_compress_type));
+ RETURN_IF_ERROR(
+ _fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
ParquetFileOptions parquet_options = {parquet_compression_type,
TParquetVersion::PARQUET_1_0,
false, false};
@@ -96,6 +97,8 @@ Status VIcebergDeleteFileWriter::open(RuntimeState* state,
RuntimeProfile* profi
return _file_format_transformer->open();
}
case TFileFormatType::FORMAT_ORC: {
+ RETURN_IF_ERROR(
+ _fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
_file_format_transformer.reset(new VOrcTransformer(state,
_file_writer.get(), output_exprs,
"", column_names,
false, _compress_type,
_position_delete_schema.get()));
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
index 249cd15dcf1..c7ae188677b 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
@@ -97,22 +97,17 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
io::FileDescription file_description = {.path = _path, .fs_name {}};
_fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
io::FileWriterOptions file_writer_options = {.used_by_s3_committer =
false};
- RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
Status open_status;
switch (_file_format_type) {
case TFileFormatType::FORMAT_PARQUET: {
- auto parquet_compression_type =
-
ParquetBuildHelper::to_parquet_compression_type(_compress_type);
- if (!parquet_compression_type.has_value()) {
- // The target file is already created, so let the shared cleanup
below remove it
- // instead of returning early.
- open_status = parquet_compression_type.error();
- break;
- }
+ TParquetCompressionType::type parquet_compression_type =
+
DORIS_TRY(ParquetBuildHelper::to_parquet_compression_type(_compress_type));
+ RETURN_IF_ERROR(
+ _fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
// Parquet 1.0 coerces Arrow nanoseconds to microseconds, including
nested fields.
ParquetFileOptions parquet_options = {
- .compression_type = parquet_compression_type.value(),
+ .compression_type = parquet_compression_type,
.parquet_version =
contains_nanosecond_timestamp(_schema.root_struct())
? TParquetVersion::PARQUET_2_LATEST
: TParquetVersion::PARQUET_1_0,
@@ -125,6 +120,8 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
break;
}
case TFileFormatType::FORMAT_ORC: {
+ RETURN_IF_ERROR(
+ _fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
_file_format_transformer = std::make_unique<VOrcTransformer>(
state, _file_writer.get(), _write_output_expr_ctxs, "",
_write_column_names, false,
_compress_type, &_schema, _fs, _nan_count_field_ids);
diff --git a/be/src/exec/sink/writer/vhive_partition_writer.cpp
b/be/src/exec/sink/writer/vhive_partition_writer.cpp
index f28adde9004..93da3f93ec0 100644
--- a/be/src/exec/sink/writer/vhive_partition_writer.cpp
+++ b/be/src/exec/sink/writer/vhive_partition_writer.cpp
@@ -85,12 +85,13 @@ Status VHivePartitionWriter::open(RuntimeState* state,
RuntimeProfile* operator_
"completion");
}
io::FileWriterOptions file_writer_options = {.used_by_s3_committer = true};
- RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
switch (_file_format_type) {
case TFileFormatType::FORMAT_PARQUET: {
TParquetCompressionType::type parquet_compression_type =
DORIS_TRY(ParquetBuildHelper::to_parquet_compression_type(_hive_compress_type));
+ RETURN_IF_ERROR(
+ _fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
// TODO: INT96 is kept for Hive 2/3 compatibility. Add an explicit
option before
// changing the default Hive parquet timestamp encoding to standard
logical types.
ParquetFileOptions parquet_options = {parquet_compression_type,
@@ -107,12 +108,16 @@ Status VHivePartitionWriter::open(RuntimeState* state,
RuntimeProfile* operator_
return _file_format_transformer->open();
}
case TFileFormatType::FORMAT_ORC: {
+ RETURN_IF_ERROR(
+ _fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
_file_format_transformer = std::make_unique<VOrcTransformer>(
state, _file_writer.get(), _write_output_expr_ctxs, "",
_write_column_names, false,
_hive_compress_type);
return _file_format_transformer->open();
}
case TFileFormatType::FORMAT_CSV_PLAIN: {
+ RETURN_IF_ERROR(
+ _fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
// TODO(syt): support hive csv table, only hive text file is
supportted now
_file_format_transformer = std::make_unique<VCSVTransformer>(
state, _file_writer.get(), _write_output_expr_ctxs, false,
"csv", "",
diff --git a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
index ba3f13b6086..c0a83a111eb 100644
--- a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
@@ -21,6 +21,7 @@
#include <optional>
#include "core/data_type/data_type_number.h"
+#include "cpp/sync_point.h"
#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
#include "exec/sink/writer/iceberg/viceberg_sort_writer.h"
#include "format/table/iceberg/schema_parser.h"
@@ -205,6 +206,47 @@ TEST_F(VIcebergPartitionWriterTest,
MissingPolicyKeepsCollectionEnabledForRollin
EXPECT_TRUE(collect_column_stats(*writer));
}
+TEST_F(VIcebergPartitionWriterTest, UnsupportedParquetCodecDoesNotCreateFile) {
+ VExprContextSPtrs output_exprs;
+ iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+ std::string schema_json;
+ std::map<std::string, std::string> hadoop_conf;
+ IPartitionWriterBase::WriteInfo write_info {.location =
IcebergOutputLocation(".", "."),
+ .file_type =
TFileType::FILE_LOCAL,
+ .broker_addresses = {}};
+ auto writer = std::make_unique<VIcebergPartitionWriter>(
+ make_table_sink(std::nullopt), std::vector<std::string> {},
output_exprs, schema,
+ &schema_json, std::vector<std::string> {}, std::move(write_info),
+ "unsupported_parquet_codec_" + UniqueId::gen_uid().to_string(), 0,
+ TFileFormatType::FORMAT_PARQUET, TFileCompressType::ZLIB,
hadoop_conf);
+
+ auto* sync_point = SyncPoint::get_instance();
+ const bool sync_point_was_enabled = sync_point->get_enable();
+ int create_file_calls = 0;
+ SyncPoint::CallbackGuard callback_guard;
+ sync_point->set_call_back(
+ "LocalFileSystem::create_file_impl", [&](auto&&) {
++create_file_calls; },
+ &callback_guard);
+ sync_point->enable_processing();
+ Defer restore_sync_point([&] {
+ if (!sync_point_was_enabled) {
+ sync_point->disable_processing();
+ }
+ });
+
+ MockRuntimeState state;
+ Status status = writer->open(&state, nullptr, nullptr);
+
+ EXPECT_EQ(0, create_file_calls);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(std::string::npos, status.to_string().find("Unsupported compress
type"));
+ EXPECT_EQ(nullptr, writer->_file_writer);
+ ASSERT_NE(nullptr, writer->_fs);
+ bool exists = true;
+ ASSERT_TRUE(writer->_fs->exists(writer->_path, &exists).ok());
+ EXPECT_FALSE(exists);
+}
+
TEST_F(VIcebergPartitionWriterTest,
SortWriterPropagatesUnderlyingCloseFailure) {
VExprContextSPtrs output_exprs;
iceberg::Schema schema(std::vector<iceberg::NestedField> {});
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
