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 1052fc31b1f [fix](iceberg) Validate required NULLs before file
creation (#69327)
1052fc31b1f is described below
commit 1052fc31b1f3bff28577f553598ee9d9aa85f757
Author: daidai <[email protected]>
AuthorDate: Thu Oct 22 10:33:39 2026 +0800
[fix](iceberg) Validate required NULLs before file creation (#69327)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
Iceberg required fields could be written as NULL for ORC and Parquet,
and failed writes could leave data files behind. This PR adds a shared
BE Block validator for required nested fields and MAP keys, runs
validation before writers/files are created, and cleans up failed files
without publishing a snapshot. FE rollback now also removes files
already reported by completed BE fragments.
### Release note
Fix Iceberg ORC/Parquet writes to reject required NULL values and clean
up failed files without publishing a snapshot.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- BE validator, TableWriter, and cleanup tests pass.
- FE rollback tests pass.
- [x] Manual test
- Verified ORC/Parquet failures leave 0 snapshots, 0 physical objects,
and 0 rows; optional parent NULL writes successfully.
- [ ] 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:
- [ ] No.
- [x] 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/iceberg_write_validator.cpp | 355 +++++++++++++++++++++
.../sink/writer/iceberg/iceberg_write_validator.h | 35 ++
.../writer/iceberg/viceberg_partition_writer.cpp | 44 ++-
.../writer/iceberg/viceberg_partition_writer.h | 8 +-
.../sink/writer/iceberg/viceberg_table_writer.cpp | 2 +
.../sink/writer/iceberg/viceberg_table_writer.h | 2 +
.../iceberg/iceberg_partition_writer_test.cpp | 74 ++++-
.../iceberg/iceberg_write_validator_test.cpp | 235 ++++++++++++++
.../writer/iceberg/viceberg_table_writer_test.cpp | 67 ++++
.../iceberg/IcebergConnectorTransaction.java | 50 ++-
.../iceberg/IcebergConnectorTransactionTest.java | 32 +-
...berg_write_required_null_select_negative.groovy | 74 +++--
...berg_write_required_null_values_negative.groovy | 173 ++++++++--
13 files changed, 1062 insertions(+), 89 deletions(-)
diff --git a/be/src/exec/sink/writer/iceberg/iceberg_write_validator.cpp
b/be/src/exec/sink/writer/iceberg/iceberg_write_validator.cpp
new file mode 100644
index 00000000000..c1eb2a97b41
--- /dev/null
+++ b/be/src/exec/sink/writer/iceberg/iceberg_write_validator.cpp
@@ -0,0 +1,355 @@
+// 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.
+
+#include "exec/sink/writer/iceberg/iceberg_write_validator.h"
+
+#include <fmt/format.h>
+
+#include <algorithm>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_array.h"
+#include "core/column/column_map.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_struct.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/table/iceberg/schema.h"
+#include "format/table/iceberg/types.h"
+
+namespace doris::iceberg {
+namespace {
+
+using Path = std::vector<std::string_view>;
+
+struct ValidationLocation {
+ size_t row = 0;
+ std::optional<size_t> ordinal;
+ std::string_view ordinal_kind;
+};
+
+enum class NonNullReason {
+ SCHEMA_REQUIRED,
+ MAP_KEY,
+};
+
+std::string path_to_string(const Path& path) {
+ std::string result;
+ for (size_t i = 0; i < path.size(); ++i) {
+ if (i > 0) {
+ result.push_back('.');
+ }
+ result.append(path[i]);
+ }
+ return result;
+}
+
+Status null_violation(const Path& path, const ValidationLocation& location,
NonNullReason reason) {
+ std::string position = fmt::format("row {}", location.row);
+ if (location.ordinal.has_value()) {
+ position += fmt::format(", {} {}", location.ordinal_kind,
*location.ordinal);
+ }
+ return Status::DataQualityError("Iceberg {} '{}' contains NULL at {}",
+ reason == NonNullReason::MAP_KEY ? "map
key" : "required field",
+ path_to_string(path), position);
+}
+
+bool requires_validation(const NestedField& field) {
+ if (field.is_required()) {
+ return true;
+ }
+ switch (field.field_type()->type_id()) {
+ case TypeID::STRUCT: {
+ const auto& fields = field.field_type()->as_struct_type()->fields();
+ return std::any_of(fields.begin(), fields.end(),
+ [](const NestedField& child) { return
requires_validation(child); });
+ }
+ case TypeID::LIST:
+ return
requires_validation(field.field_type()->as_list_type()->element_field());
+ case TypeID::MAP:
+ // Iceberg map keys are required independently of the map field's
nullability.
+ return true;
+ default:
+ return false;
+ }
+}
+
+template <typename Offsets>
+Status validate_collection_offsets(const Offsets& offsets, size_t position,
size_t row,
+ std::string_view container_name, const
Path& path, size_t* start,
+ size_t* end) {
+ if (position >= offsets.size()) {
+ return Status::InternalError(
+ "Iceberg {} field '{}' has {} offsets, but position {} is
required", container_name,
+ path_to_string(path), offsets.size(), position);
+ }
+ *start = position == 0 ? 0 : offsets[position - 1];
+ *end = offsets[position];
+ if (*start > *end) {
+ return Status::InternalError(
+ "Iceberg {} field '{}' has invalid offsets at row {}: [{},
{}]", container_name,
+ path_to_string(path), row, *start, *end);
+ }
+ return Status::OK();
+}
+
+Status validate_value(const NestedField& field, const IColumn& column, const
DataTypePtr& type,
+ size_t position, const ValidationLocation& location,
Path& path,
+ NonNullReason reason) {
+ if (position >= column.size()) {
+ return Status::InternalError(
+ "Iceberg write column '{}' has {} values, but position {} "
+ "is required",
+ path_to_string(path), column.size(), position);
+ }
+
+ if (column.is_null_at(position)) {
+ if (reason == NonNullReason::MAP_KEY || field.is_required()) {
+ return null_violation(path, location, reason);
+ }
+ // An optional STRUCT/LIST/MAP is a NULL value in its own right. Its
children and
+ // elements are masked by the parent NULL and must not be inspected.
+ return Status::OK();
+ }
+
+ const IColumn* data_column = &column;
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(column)) {
+ data_column = &nullable->get_nested_column();
+ }
+
+ const auto primitive_type = remove_nullable(type)->get_primitive_type();
+ switch (field.field_type()->type_id()) {
+ case TypeID::STRUCT: {
+ if (primitive_type != TYPE_STRUCT) {
+ return Status::InternalError("Iceberg STRUCT field '{}' does not
match Doris type {}",
+ path_to_string(path),
type->get_name());
+ }
+ const auto* struct_column =
check_and_get_column<ColumnStruct>(*data_column);
+ const auto* struct_type = assert_cast<const
DataTypeStruct*>(remove_nullable(type).get());
+ if (struct_column == nullptr) {
+ return Status::InternalError("Iceberg STRUCT field '{}' is not
stored as a STRUCT",
+ path_to_string(path));
+ }
+ const auto& fields = field.field_type()->as_struct_type()->fields();
+ if (struct_column->tuple_size() != fields.size() ||
+ struct_type->get_elements().size() != fields.size()) {
+ return Status::InternalError(
+ "Iceberg STRUCT field '{}' has {} schema fields, {} Doris
type fields, and "
+ "{} column fields",
+ path_to_string(path), fields.size(),
struct_type->get_elements().size(),
+ struct_column->tuple_size());
+ }
+ for (size_t i = 0; i < fields.size(); ++i) {
+ if (!requires_validation(fields[i])) {
+ continue;
+ }
+ path.emplace_back(fields[i].field_name());
+ Status status = validate_value(fields[i],
struct_column->get_column(i),
+ struct_type->get_element(i),
position, location, path,
+ NonNullReason::SCHEMA_REQUIRED);
+ path.pop_back();
+ RETURN_IF_ERROR(status);
+ }
+ return Status::OK();
+ }
+ case TypeID::LIST: {
+ if (primitive_type != TYPE_ARRAY) {
+ return Status::InternalError("Iceberg LIST field '{}' does not
match Doris type {}",
+ path_to_string(path),
type->get_name());
+ }
+ const auto* array_column =
check_and_get_column<ColumnArray>(*data_column);
+ const auto* array_type = assert_cast<const
DataTypeArray*>(remove_nullable(type).get());
+ if (array_column == nullptr) {
+ return Status::InternalError("Iceberg LIST field '{}' is not
stored as an ARRAY",
+ path_to_string(path));
+ }
+ const auto& offsets = array_column->get_offsets();
+ size_t start = 0;
+ size_t end = 0;
+ RETURN_IF_ERROR(validate_collection_offsets(offsets, position,
location.row, "LIST", path,
+ &start, &end));
+ if (end > array_column->get_data().size()) {
+ return Status::InternalError(
+ "Iceberg LIST field '{}' has invalid offsets at row {}:
[{}, {}], "
+ "element count {}",
+ path_to_string(path), location.row, start, end,
+ array_column->get_data().size());
+ }
+ const auto& element =
field.field_type()->as_list_type()->element_field();
+ if (!requires_validation(element)) {
+ return Status::OK();
+ }
+ path.emplace_back(element.field_name());
+ for (size_t i = start; i < end; ++i) {
+ ValidationLocation element_location {
+ .row = location.row,
+ .ordinal = i - start,
+ .ordinal_kind = "element",
+ };
+ Status status =
+ validate_value(element, array_column->get_data(),
array_type->get_nested_type(),
+ i, element_location, path,
NonNullReason::SCHEMA_REQUIRED);
+ if (!status.ok()) {
+ path.pop_back();
+ return status;
+ }
+ }
+ path.pop_back();
+ return Status::OK();
+ }
+ case TypeID::MAP: {
+ if (primitive_type != TYPE_MAP) {
+ return Status::InternalError("Iceberg MAP field '{}' does not
match Doris type {}",
+ path_to_string(path),
type->get_name());
+ }
+ const auto* map_column = check_and_get_column<ColumnMap>(*data_column);
+ const auto* map_type = assert_cast<const
DataTypeMap*>(remove_nullable(type).get());
+ if (map_column == nullptr) {
+ return Status::InternalError("Iceberg MAP field '{}' is not stored
as a MAP",
+ path_to_string(path));
+ }
+ const auto& offsets = map_column->get_offsets();
+ size_t start = 0;
+ size_t end = 0;
+ RETURN_IF_ERROR(validate_collection_offsets(offsets, position,
location.row, "MAP", path,
+ &start, &end));
+ if (end > map_column->get_keys().size() || end >
map_column->get_values().size()) {
+ return Status::InternalError(
+ "Iceberg MAP field '{}' has invalid offsets at row {}:
[{}, {}], "
+ "key count {}, value count {}",
+ path_to_string(path), location.row, start, end,
map_column->get_keys().size(),
+ map_column->get_values().size());
+ }
+
+ const auto& map_type_desc = field.field_type()->as_map_type();
+ path.emplace_back(map_type_desc->key_field().field_name());
+ for (size_t i = start; i < end; ++i) {
+ ValidationLocation entry_location {
+ .row = location.row,
+ .ordinal = i - start,
+ .ordinal_kind = "map entry",
+ };
+ Status status = validate_value(map_type_desc->key_field(),
map_column->get_keys(),
+ map_type->get_key_type(), i,
entry_location, path,
+ NonNullReason::MAP_KEY);
+ if (!status.ok()) {
+ path.pop_back();
+ return status;
+ }
+ }
+ path.pop_back();
+
+ const auto& value = map_type_desc->value_field();
+ if (!requires_validation(value)) {
+ return Status::OK();
+ }
+ path.emplace_back(value.field_name());
+ for (size_t i = start; i < end; ++i) {
+ ValidationLocation entry_location {
+ .row = location.row,
+ .ordinal = i - start,
+ .ordinal_kind = "map entry",
+ };
+ Status status =
+ validate_value(value, map_column->get_values(),
map_type->get_value_type(), i,
+ entry_location, path,
NonNullReason::SCHEMA_REQUIRED);
+ if (!status.ok()) {
+ path.pop_back();
+ return status;
+ }
+ }
+ path.pop_back();
+ return Status::OK();
+ }
+ default:
+ return Status::OK();
+ }
+}
+
+} // namespace
+
+Status validate_write_block(const Schema& schema, const Block& block) {
+ if (block.columns() != schema.columns().size()) {
+ return Status::InternalError(
+ "Iceberg write block has {} columns, but pinned schema has {}
fields",
+ block.columns(), schema.columns().size());
+ }
+
+ Path path;
+ path.reserve(8);
+ for (size_t field_index = 0; field_index < schema.columns().size();
++field_index) {
+ const auto& field = schema.columns()[field_index];
+ if (!requires_validation(field)) {
+ continue;
+ }
+ const auto& column_with_name = block.get_by_position(field_index);
+ ColumnPtr column =
column_with_name.column->convert_to_full_column_if_const();
+ if (!column) {
+ return Status::InternalError("Iceberg write block column {} is
null", field_index);
+ }
+ if (column->size() < block.rows()) {
+ return Status::InternalError(
+ "Iceberg write block column '{}' has {} values, but block
has {} rows",
+ field.field_name(), column->size(), block.rows());
+ }
+
+ path.emplace_back(field.field_name());
+ if (field.field_type()->is_primitive_type() && field.is_required()) {
+ if (column->has_null()) {
+ for (size_t row = 0; row < block.rows(); ++row) {
+ if (!column->is_null_at(row)) {
+ continue;
+ }
+ ValidationLocation location {
+ .row = row,
+ .ordinal = std::nullopt,
+ .ordinal_kind = {},
+ };
+ Status status = null_violation(path, location,
NonNullReason::SCHEMA_REQUIRED);
+ path.pop_back();
+ return status;
+ }
+ }
+ path.pop_back();
+ continue;
+ }
+ for (size_t row = 0; row < block.rows(); ++row) {
+ ValidationLocation location {
+ .row = row,
+ .ordinal = std::nullopt,
+ .ordinal_kind = {},
+ };
+ Status status = validate_value(field, *column,
column_with_name.type, row, location,
+ path,
NonNullReason::SCHEMA_REQUIRED);
+ if (!status.ok()) {
+ path.pop_back();
+ return status;
+ }
+ }
+ path.pop_back();
+ }
+ return Status::OK();
+}
+
+} // namespace doris::iceberg
diff --git a/be/src/exec/sink/writer/iceberg/iceberg_write_validator.h
b/be/src/exec/sink/writer/iceberg/iceberg_write_validator.h
new file mode 100644
index 00000000000..30d55e1f52a
--- /dev/null
+++ b/be/src/exec/sink/writer/iceberg/iceberg_write_validator.h
@@ -0,0 +1,35 @@
+// 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.
+
+#pragma once
+
+#include "common/status.h"
+
+namespace doris {
+
+class Block;
+
+namespace iceberg {
+
+class Schema;
+
+// Validates the data-dependent Iceberg write contract for one materialized
Block.
+// Schema identity, field IDs, partition specs, and sort orders are pinned and
checked in FE.
+Status validate_write_block(const Schema& schema, const Block& block);
+
+} // namespace iceberg
+} // namespace doris
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 0be4d1d371d..5410b67a673 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
@@ -138,13 +138,34 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
}
}
if (!open_status.ok()) {
- // A transformer failure happens after object creation, so remove any
published file.
- WARN_IF_ERROR(_fs->delete_file(_path), "failed to delete Iceberg file
after open error");
+ // A transformer failure happens after writer creation. Defer deletion
until the output
+ // stream has finished closing so an asynchronous object upload cannot
publish it later.
+ _defer_file_cleanup();
}
return open_status;
}
+void VIcebergPartitionWriter::_defer_file_cleanup() {
+ if (_fs == nullptr || _file_cleanup_deferred) {
+ return;
+ }
+ _file_cleanup_deferred = true;
+ // Destroy the format writer and its output stream before deleting the
physical object. In
+ // particular, a failed close may leave an asynchronous upload running,
and the object must not
+ // be visible after the rollback has completed.
+ _file_format_transformer.reset();
+ _file_writer.reset();
+ if (_closed_file_callback) {
+ _closed_file_callback(_fs, _path);
+ } else {
+ WARN_IF_ERROR(_fs->delete_file(_path), "failed to delete Iceberg
file");
+ }
+}
+
Status VIcebergPartitionWriter::close(const Status& status) {
+ if (_file_cleanup_deferred) {
+ return _terminal_close_status;
+ }
Status result_status;
if (_file_format_transformer != nullptr) {
result_status = _file_format_transformer->close();
@@ -155,28 +176,21 @@ Status VIcebergPartitionWriter::close(const Status&
status) {
}
bool status_ok = result_status.ok() && status.ok();
if (!status_ok && _fs != nullptr) {
- // delete the actual created file, otherwise an orphan file is left
behind
- Status st = _fs->delete_file(_path);
- if (!st.ok()) {
- LOG(WARNING) << fmt::format("Delete file {} failed, reason: {}",
_path, st.to_string());
- }
+ _terminal_close_status = result_status;
+ _defer_file_cleanup();
}
if (status_ok) {
TIcebergCommitData commit_data;
Status commit_status = _build_iceberg_commit_data(&commit_data);
if (!commit_status.ok()) {
- // A closed object without commit data can never be published, so
remove it immediately.
- Status delete_status = _fs->delete_file(_path);
- if (!delete_status.ok()) {
- LOG(WARNING) << fmt::format("Delete file {} failed, reason:
{}", _path,
- delete_status.to_string());
- }
+ _terminal_close_status = commit_status;
+ _defer_file_cleanup();
return commit_status;
}
Status report_status =
_state->add_iceberg_commit_datas(std::move(commit_data));
if (!report_status.ok()) {
- // A closed object that cannot be reported can never be committed,
so remove it immediately.
- WARN_IF_ERROR(_fs->delete_file(_path), "failed to delete
unreportable Iceberg file");
+ _terminal_close_status = report_status;
+ _defer_file_cleanup();
return report_status;
}
if (_closed_file_callback) {
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
index 954d458538c..b0f5f4eefd2 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
@@ -70,7 +70,9 @@ public:
inline int file_name_index() const override { return _file_name_index; }
- inline size_t written_len() const override { return
_file_format_transformer->written_len(); }
+ inline size_t written_len() const override {
+ return _file_format_transformer == nullptr ? 0 :
_file_format_transformer->written_len();
+ }
private:
friend class VIcebergPartitionWriterTest;
@@ -79,6 +81,8 @@ private:
Status _build_iceberg_commit_data(TIcebergCommitData* commit_data);
+ void _defer_file_cleanup();
+
std::string _get_file_extension(TFileFormatType::type file_format_type,
TFileCompressType::type
write_compress_type);
@@ -100,6 +104,8 @@ private:
TFileCompressType::type _compress_type;
const std::map<std::string, std::string>& _hadoop_conf;
ClosedFileCallback _closed_file_callback;
+ bool _file_cleanup_deferred = false;
+ Status _terminal_close_status = Status::OK();
bool _collect_column_stats = true;
// FE's metrics policy for NaN counting; empty (including an older FE that
sends nothing) means the
// parquet writer skips the extra pass entirely. See
IcebergWriterHelper#nanCountFieldIds.
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
index c041ac9e662..bdfed84ee5d 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
@@ -33,6 +33,7 @@
#include "core/data_type_serde/data_type_serde.h"
#include "exec/sink/writer/external_file_writer_compatibility.h"
#include "exec/sink/writer/iceberg/iceberg_partition_path.h"
+#include "exec/sink/writer/iceberg/iceberg_write_validator.h"
#include "exec/sink/writer/iceberg/partition_transformers.h"
#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
#include "exec/sink/writer/iceberg/viceberg_sort_writer.h"
@@ -402,6 +403,7 @@ Status
VIcebergTableWriter::_process_row_lineage_columns(Block& block) {
Status VIcebergTableWriter::_write_prepared_block(Block& output_block) {
RETURN_IF_ERROR(_process_row_lineage_columns(output_block));
+ RETURN_IF_ERROR(iceberg::validate_write_block(*_schema, output_block));
std::unordered_map<std::shared_ptr<IPartitionWriterBase>,
IColumn::Permutation>
writer_positions;
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
index 742e9625905..48b7d24e126 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
@@ -79,6 +79,8 @@ public:
private:
FRIEND_TEST(VIcebergTableWriterTest, RejectMissingPartitionSource);
FRIEND_TEST(VIcebergTableWriterTest, ResolvesNestedPartitionSource);
+ FRIEND_TEST(VIcebergTableWriterTest,
RejectsRequiredNullBeforeWriterCreation);
+ FRIEND_TEST(VIcebergTableWriterTest,
RejectsNestedRequiredNullBeforeWriterCreation);
// The lifecycle fixture inspects snapshots to verify that cross-thread
writer ownership stays stable.
friend class VIcebergTableWriterLifecycleTest;
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 d5a7e3f9ab3..9b10ff0e909 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 "exec/sink/writer/iceberg/viceberg_partition_writer.h"
#include "exec/sink/writer/iceberg/viceberg_sort_writer.h"
+#include "io/fs/local_file_system.h"
#include "testutil/mock/mock_descriptors.h"
#include "testutil/mock/mock_runtime_state.h"
#include "testutil/mock/mock_slot_ref.h"
@@ -32,9 +33,17 @@ namespace {
class FakeFileFormatTransformer final : public VFileFormatTransformer {
public:
explicit FakeFileFormatTransformer(const VExprContextSPtrs& output_exprs,
- Status close_status = Status::OK())
+ Status close_status = Status::OK(),
+ bool* destroyed = nullptr)
: VFileFormatTransformer(nullptr, output_exprs, false),
- _close_status(std::move(close_status)) {}
+ _close_status(std::move(close_status)),
+ _destroyed(destroyed) {}
+
+ ~FakeFileFormatTransformer() override {
+ if (_destroyed != nullptr) {
+ *_destroyed = true;
+ }
+ }
Status open() override { return Status::OK(); }
Status write(const Block&) override { return Status::OK(); }
@@ -43,6 +52,7 @@ public:
private:
Status _close_status;
+ bool* _destroyed;
};
TDataSink make_table_sink(std::optional<bool> collect_column_stats) {
@@ -74,9 +84,10 @@ protected:
static void install_fake_transformer(VIcebergPartitionWriter* writer,
const VExprContextSPtrs& output_exprs,
- Status close_status = Status::OK()) {
- writer->_file_format_transformer =
- std::make_unique<FakeFileFormatTransformer>(output_exprs,
std::move(close_status));
+ Status close_status = Status::OK(),
+ bool* destroyed = nullptr) {
+ writer->_file_format_transformer =
std::make_unique<FakeFileFormatTransformer>(
+ output_exprs, std::move(close_status), destroyed);
}
static Status build_commit_data(VIcebergPartitionWriter* writer,
@@ -133,6 +144,59 @@ TEST_F(VIcebergPartitionWriterTest,
SortWriterPropagatesUnderlyingCloseFailure)
EXPECT_NE(status.to_string().find("injected close failure"),
std::string::npos);
}
+TEST_F(VIcebergPartitionWriterTest,
FailedCloseDefersFileCleanupToOuterCallback) {
+ VExprContextSPtrs output_exprs;
+ iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+ std::string schema_json;
+ std::map<std::string, std::string> hadoop_conf;
+ auto writer = std::shared_ptr<VIcebergPartitionWriter>(
+ make_writer(make_table_sink(false), output_exprs, schema,
&schema_json, hadoop_conf));
+ bool transformer_destroyed = false;
+ install_fake_transformer(writer.get(), output_exprs, Status::OK(),
&transformer_destroyed);
+ writer->_fs = io::global_local_filesystem();
+ writer->_path = "/tmp/doris-iceberg-failed-close.orc";
+ int cleanup_calls = 0;
+ writer->_closed_file_callback = [&](std::shared_ptr<io::FileSystem> fs,
+ const std::string& path) {
+ EXPECT_EQ(writer->_fs, fs);
+ EXPECT_EQ(writer->_path, path);
+ EXPECT_TRUE(transformer_destroyed);
+ ++cleanup_calls;
+ };
+
+ EXPECT_TRUE(writer->close(Status::Cancelled("injected write
failure")).ok());
+ EXPECT_EQ(1, cleanup_calls);
+ EXPECT_EQ(0, writer->written_len());
+
+ EXPECT_TRUE(writer->close(Status::Cancelled("injected write
failure")).ok());
+ EXPECT_EQ(1, cleanup_calls);
+}
+
+TEST_F(VIcebergPartitionWriterTest, RepeatedClosePreservesTerminalError) {
+ VExprContextSPtrs output_exprs;
+ iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+ std::string schema_json;
+ std::map<std::string, std::string> hadoop_conf;
+ auto writer = std::shared_ptr<VIcebergPartitionWriter>(
+ make_writer(make_table_sink(false), output_exprs, schema,
&schema_json, hadoop_conf));
+ install_fake_transformer(writer.get(), output_exprs,
Status::IOError("injected close failure"));
+ writer->_fs = io::global_local_filesystem();
+ writer->_path = "/tmp/doris-iceberg-close-failure.orc";
+ int cleanup_calls = 0;
+ writer->_closed_file_callback = [&](std::shared_ptr<io::FileSystem>, const
std::string&) {
+ ++cleanup_calls;
+ };
+
+ Status first_status = writer->close(Status::OK());
+ Status second_status = writer->close(Status::OK());
+
+ EXPECT_FALSE(first_status.ok());
+ EXPECT_FALSE(second_status.ok());
+ EXPECT_NE(first_status.to_string().find("injected close failure"),
std::string::npos);
+ EXPECT_EQ(first_status, second_status);
+ EXPECT_EQ(1, cleanup_calls);
+}
+
TEST_F(VIcebergPartitionWriterTest, EosReservationIncludesActualSpillFanIn) {
VExprContextSPtrs output_exprs;
iceberg::Schema schema(std::vector<iceberg::NestedField> {});
diff --git a/be/test/exec/sink/writer/iceberg/iceberg_write_validator_test.cpp
b/be/test/exec/sink/writer/iceberg/iceberg_write_validator_test.cpp
new file mode 100644
index 00000000000..a547321fa5a
--- /dev/null
+++ b/be/test/exec/sink/writer/iceberg/iceberg_write_validator_test.cpp
@@ -0,0 +1,235 @@
+// 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.
+
+#include "exec/sink/writer/iceberg/iceberg_write_validator.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <optional>
+#include <vector>
+
+#include "core/block/block.h"
+#include "core/column/column_array.h"
+#include "core/column/column_map.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/table/iceberg/schema.h"
+#include "format/table/iceberg/types.h"
+
+namespace doris::iceberg {
+namespace {
+
+DataTypePtr nullable_int() {
+ return make_nullable(std::make_shared<DataTypeInt32>());
+}
+
+ColumnPtr nullable_int_column(std::vector<int32_t> values,
std::vector<uint8_t> nulls) {
+ auto data = ColumnInt32::create();
+ data->get_data().assign(values.begin(), values.end());
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign(nulls.begin(), nulls.end());
+ return ColumnNullable::create(std::move(data), std::move(null_map));
+}
+
+void expect_data_quality_error(const Status& status, std::string_view path,
+ std::string_view location) {
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find(path), std::string::npos);
+ EXPECT_NE(status.to_string().find(location), std::string::npos);
+}
+
+} // namespace
+
+TEST(IcebergWriteValidatorTest, RejectsTopLevelRequiredNullWithLocation) {
+ std::vector<NestedField> fields;
+ fields.emplace_back(false, 1, "payload", std::make_unique<StringType>(),
std::nullopt);
+ Schema schema(std::move(fields));
+ auto type = make_nullable(std::make_shared<DataTypeString>());
+ auto data = ColumnString::create();
+ data->insert_data("A", 1);
+ data->insert_default();
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign({0, 1});
+ Block block({ColumnWithTypeAndName(ColumnNullable::create(std::move(data),
std::move(null_map)),
+ type, "payload")});
+
+ expect_data_quality_error(validate_write_block(schema, block), "payload",
"row 1");
+}
+
+TEST(IcebergWriteValidatorTest, SkipsRequiredChildWhenOptionalParentIsNull) {
+ std::vector<NestedField> child_fields;
+ child_fields.emplace_back(false, 2, "id", std::make_unique<IntegerType>(),
std::nullopt);
+ std::vector<NestedField> fields;
+ fields.emplace_back(true, 1, "payload",
std::make_unique<StructType>(std::move(child_fields)),
+ std::nullopt);
+ Schema schema(std::move(fields));
+ auto child_type = nullable_int();
+ auto struct_type = std::make_shared<DataTypeStruct>(DataTypes
{child_type}, Strings {"id"});
+ auto parent_type = make_nullable(struct_type);
+ Columns children {nullable_int_column({7}, {1})};
+ auto parent_null_map = ColumnUInt8::create();
+ parent_null_map->get_data().push_back(1);
+ Block block(
+
{ColumnWithTypeAndName(ColumnNullable::create(ColumnStruct::create(std::move(children)),
+
std::move(parent_null_map)),
+ parent_type, "payload")});
+
+ EXPECT_TRUE(validate_write_block(schema, block).ok());
+}
+
+TEST(IcebergWriteValidatorTest, RejectsRequiredStructChildNull) {
+ std::vector<NestedField> child_fields;
+ child_fields.emplace_back(false, 2, "id", std::make_unique<IntegerType>(),
std::nullopt);
+ std::vector<NestedField> fields;
+ fields.emplace_back(false, 1, "payload",
std::make_unique<StructType>(std::move(child_fields)),
+ std::nullopt);
+ Schema schema(std::move(fields));
+ auto child_type = nullable_int();
+ auto struct_type = std::make_shared<DataTypeStruct>(DataTypes
{child_type}, Strings {"id"});
+ auto parent_type = make_nullable(struct_type);
+ Columns children {nullable_int_column({7, 8}, {0, 1})};
+ auto parent_null_map = ColumnUInt8::create();
+ parent_null_map->get_data().assign({0, 0});
+ Block block(
+
{ColumnWithTypeAndName(ColumnNullable::create(ColumnStruct::create(std::move(children)),
+
std::move(parent_null_map)),
+ parent_type, "payload")});
+
+ expect_data_quality_error(validate_write_block(schema, block),
"payload.id", "row 1");
+}
+
+TEST(IcebergWriteValidatorTest, RejectsRequiredListElementNull) {
+ std::vector<NestedField> fields;
+ fields.emplace_back(true, 1, "items", ListType::of_required(2,
std::make_unique<IntegerType>()),
+ std::nullopt);
+ Schema schema(std::move(fields));
+ auto element_type = nullable_int();
+ auto array_type = std::make_shared<DataTypeArray>(element_type);
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->get_data().push_back(3);
+ auto array = ColumnArray::create(nullable_int_column({1, 2, 3}, {0, 1,
0}), std::move(offsets));
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().push_back(0);
+ Block block(
+ {ColumnWithTypeAndName(ColumnNullable::create(std::move(array),
std::move(null_map)),
+ make_nullable(array_type), "items")});
+
+ expect_data_quality_error(validate_write_block(schema, block),
"items.element", "element 1");
+}
+
+TEST(IcebergWriteValidatorTest, RejectsNullMapKey) {
+ std::vector<NestedField> fields;
+ fields.emplace_back(true, 1, "attrs",
+ MapType::of_required(2, 3,
std::make_unique<IntegerType>(),
+ std::make_unique<IntegerType>()),
+ std::nullopt);
+ Schema schema(std::move(fields));
+ auto key_type = std::make_shared<DataTypeInt32>();
+ auto value_type = nullable_int();
+ auto map_type = std::make_shared<DataTypeMap>(key_type, value_type);
+ auto offsets = ColumnMap::COffsets::create();
+ offsets->get_data().push_back(2);
+ auto map = ColumnMap::create(nullable_int_column({1, 2}, {0, 1}),
+ nullable_int_column({10, 20}, {0, 0}),
std::move(offsets));
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().push_back(0);
+ Block block({ColumnWithTypeAndName(ColumnNullable::create(std::move(map),
std::move(null_map)),
+ make_nullable(map_type), "attrs")});
+
+ expect_data_quality_error(validate_write_block(schema, block),
"attrs.key", "map entry 1");
+}
+
+TEST(IcebergWriteValidatorTest, SkipsMapKeyWhenOptionalMapIsNull) {
+ std::vector<NestedField> fields;
+ fields.emplace_back(true, 1, "attrs",
+ MapType::of_required(2, 3,
std::make_unique<IntegerType>(),
+ std::make_unique<IntegerType>()),
+ std::nullopt);
+ Schema schema(std::move(fields));
+ auto key_type = std::make_shared<DataTypeInt32>();
+ auto value_type = nullable_int();
+ auto map_type = std::make_shared<DataTypeMap>(key_type, value_type);
+ auto offsets = ColumnMap::COffsets::create();
+ offsets->get_data().push_back(1);
+ // Keep a physically materialized NULL key under the NULL parent to prove
that the parent mask,
+ // rather than an empty map, is what suppresses the map-key check.
+ auto map = ColumnMap::create(nullable_int_column({1}, {1}),
nullable_int_column({10}, {0}),
+ std::move(offsets));
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().push_back(1);
+ Block block({ColumnWithTypeAndName(ColumnNullable::create(std::move(map),
std::move(null_map)),
+ make_nullable(map_type), "attrs")});
+
+ EXPECT_TRUE(validate_write_block(schema, block).ok());
+}
+
+TEST(IcebergWriteValidatorTest, RejectsRequiredMapValueNull) {
+ std::vector<NestedField> fields;
+ fields.emplace_back(true, 1, "attrs",
+ MapType::of_required(2, 3,
std::make_unique<IntegerType>(),
+ std::make_unique<IntegerType>()),
+ std::nullopt);
+ Schema schema(std::move(fields));
+ auto key_type = std::make_shared<DataTypeInt32>();
+ auto value_type = nullable_int();
+ auto map_type = std::make_shared<DataTypeMap>(key_type, value_type);
+ auto offsets = ColumnMap::COffsets::create();
+ offsets->get_data().push_back(2);
+ auto map = ColumnMap::create(nullable_int_column({1, 2}, {0, 0}),
+ nullable_int_column({10, 20}, {0, 1}),
std::move(offsets));
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().push_back(0);
+ Block block({ColumnWithTypeAndName(ColumnNullable::create(std::move(map),
std::move(null_map)),
+ make_nullable(map_type), "attrs")});
+
+ expect_data_quality_error(validate_write_block(schema, block),
"attrs.value", "map entry 1");
+}
+
+TEST(IcebergWriteValidatorTest, AcceptsValidRequiredNestedValues) {
+ std::vector<NestedField> fields;
+ fields.emplace_back(false, 1, "id", std::make_unique<IntegerType>(),
std::nullopt);
+ fields.emplace_back(true, 2, "items", ListType::of_required(3,
std::make_unique<IntegerType>()),
+ std::nullopt);
+ Schema schema(std::move(fields));
+ auto id_type = nullable_int();
+ auto id_column = nullable_int_column({1, 2}, {0, 0});
+ auto element_type = nullable_int();
+ auto array_type =
make_nullable(std::make_shared<DataTypeArray>(element_type));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->get_data().assign({2, 3});
+ auto array =
+ ColumnArray::create(nullable_int_column({10, 20, 30}, {0, 0, 0}),
std::move(offsets));
+ auto array_nulls = ColumnUInt8::create();
+ array_nulls->get_data().assign({0, 0});
+ Block block({
+ ColumnWithTypeAndName(std::move(id_column), id_type, "id"),
+ ColumnWithTypeAndName(ColumnNullable::create(std::move(array),
std::move(array_nulls)),
+ array_type, "items"),
+ });
+
+ EXPECT_TRUE(validate_write_block(schema, block).ok());
+}
+
+} // namespace doris::iceberg
diff --git a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
index 21397daa141..c16ba9a2164 100644
--- a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
@@ -64,6 +64,73 @@ public:
};
} // namespace
+TEST(VIcebergTableWriterTest, RejectsRequiredNullBeforeWriterCreation) {
+ std::vector<iceberg::NestedField> fields;
+ fields.emplace_back(false, 1, "id",
std::make_unique<iceberg::IntegerType>(), std::nullopt);
+ auto schema = std::make_shared<iceberg::Schema>(std::move(fields));
+ TIcebergTableSink sink;
+ TDataSink data_sink;
+ data_sink.__set_iceberg_table_sink(sink);
+ VIcebergTableWriter writer(data_sink, {}, nullptr, nullptr);
+ writer._schema = schema;
+
+ auto type = make_nullable(std::make_shared<DataTypeInt32>());
+ auto data = ColumnInt32::create();
+ data->insert_value(1);
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().push_back(1);
+ Block block({ColumnWithTypeAndName(ColumnNullable::create(std::move(data),
std::move(null_map)),
+ type, "id")});
+
+ Status status = writer._write_prepared_block(block);
+
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("required field 'id'"),
std::string::npos);
+ EXPECT_TRUE(writer._partitions_to_writers.empty());
+ EXPECT_EQ(0, writer._row_count);
+}
+
+TEST(VIcebergTableWriterTest, RejectsNestedRequiredNullBeforeWriterCreation) {
+ std::vector<iceberg::NestedField> child_fields;
+ child_fields.emplace_back(false, 2, "required_metric",
std::make_unique<iceberg::IntegerType>(),
+ std::nullopt);
+ std::vector<iceberg::NestedField> fields;
+ fields.emplace_back(true, 1, "info",
+
std::make_unique<iceberg::StructType>(std::move(child_fields)),
+ std::nullopt);
+ auto schema = std::make_shared<iceberg::Schema>(std::move(fields));
+ TIcebergTableSink sink;
+ TDataSink data_sink;
+ data_sink.__set_iceberg_table_sink(sink);
+ VIcebergTableWriter writer(data_sink, {}, nullptr, nullptr);
+ writer._schema = schema;
+
+ auto child_type = make_nullable(std::make_shared<DataTypeInt32>());
+ auto child_data = ColumnInt32::create();
+ child_data->insert_value(7);
+ auto child_nulls = ColumnUInt8::create();
+ child_nulls->get_data().push_back(1);
+ Columns children {
+ ColumnNullable::create(std::move(child_data),
std::move(child_nulls)),
+ };
+ auto struct_type =
+ std::make_shared<DataTypeStruct>(DataTypes {child_type}, Strings
{"required_metric"});
+ auto parent_nulls = ColumnUInt8::create();
+ parent_nulls->get_data().push_back(0);
+ Block block({
+
ColumnWithTypeAndName(ColumnNullable::create(ColumnStruct::create(std::move(children)),
+
std::move(parent_nulls)),
+ make_nullable(struct_type), "info"),
+ });
+
+ Status status = writer._write_prepared_block(block);
+
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("required field
'info.required_metric'"), std::string::npos);
+ EXPECT_TRUE(writer._partitions_to_writers.empty());
+ EXPECT_EQ(0, writer._row_count);
+}
+
TEST(VIcebergTableWriterTest, TimestampIdentityPreservesUtcCommitAndNull) {
std::vector<iceberg::NestedField> fields;
fields.emplace_back(false, 1, "event_time",
std::make_unique<iceberg::TimestampTzType>(),
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java
index 42c3118a9bd..ddfa9d9fae1 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java
@@ -1469,11 +1469,51 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
@Override
public void rollback() {
- // Insert-mode: nothing to undo on the FE side — an uncommitted SDK
transaction simply discards
- // its pending manifests (legacy IcebergTransaction.rollback no-ops
the insert path). The rewrite
- // path's file-list cleanup is a P6.4 procedure concern, out of scope
here.
- LOG.info("Iceberg transaction {} rollback called; uncommitted
manifests will be discarded.",
- transactionId);
+ List<TIcebergCommitData> pending;
+ synchronized (this) {
+ pending = new ArrayList<>(commitDataList);
+ commitDataList.clear();
+ }
+ Table rollbackTable = table;
+ if (rollbackTable == null || pending.isEmpty()) {
+ LOG.info("Iceberg transaction {} rollback called; no external
files were reported.",
+ transactionId);
+ return;
+ }
+
+ Set<String> paths = new HashSet<>();
+ for (TIcebergCommitData data : pending) {
+ if (data != null && data.isSetFilePath() && data.getFilePath() !=
null
+ && !data.getFilePath().isEmpty()) {
+ paths.add(data.getFilePath());
+ }
+ }
+ if (paths.isEmpty()) {
+ LOG.info("Iceberg transaction {} rollback called; no external
files were reported.",
+ transactionId);
+ return;
+ }
+
+ try {
+ context.executeAuthenticated(() -> {
+ for (String path : paths) {
+ try {
+ rollbackTable.io().deleteFile(path);
+ } catch (Exception e) {
+ // Rollback is best-effort. A missing file is already
clean, while an IO failure
+ // must be logged for orphan-file maintenance rather
than masking the query error.
+ LOG.warn("Failed to delete uncommitted Iceberg file {}
for transaction {}: {}",
+ path, transactionId, e.getMessage());
+ }
+ }
+ return null;
+ });
+ LOG.info("Iceberg transaction {} rollback attempted to delete {}
reported external file(s).",
+ transactionId, paths.size());
+ } catch (Exception e) {
+ LOG.warn("Failed to clean reported Iceberg files for transaction
{}: {}", transactionId,
+ e.getMessage());
+ }
}
@Override
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
index 6489c1d317e..ddd1ac2c7bf 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
@@ -51,10 +51,13 @@ import org.apache.iceberg.Table;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NotFoundException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.PositionOutputStream;
import org.apache.iceberg.types.Types;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
@@ -1274,7 +1277,7 @@ public class IcebergConnectorTransactionTest {
}
@Test
- public void rollbackAndCloseAreNoOps() {
+ public void rollbackWithoutBeginAndCloseAreNoOps() {
IcebergConnectorTransaction txn = txnFor(opsReturning(null), new
RecordingConnectorContext());
Assertions.assertDoesNotThrow(() -> {
txn.rollback();
@@ -1282,6 +1285,33 @@ public class IcebergConnectorTransactionTest {
});
}
+ @Test
+ public void rollbackDeletesReportedFilesAndClearsCommitData() throws
IOException {
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table table = catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned());
+ String path = "s3://b/db1/t1/uncommitted.parquet";
+ OutputFile output = table.io().newOutputFile(path);
+ try (PositionOutputStream stream = output.create()) {
+ stream.write(1);
+ }
+ Assertions.assertTrue(table.io().newInputFile(path).exists());
+
+ IcebergConnectorTransaction txn =
+ txnFor(opsReturning(table), new RecordingConnectorContext());
+ txn.beginWrite(SESSION, "db1", "t1", insertCtx());
+ txn.addCommitData(commitBytes(dataFileItem(path, 1L, 1L)));
+
+ txn.rollback();
+
+ Assertions.assertThrows(NotFoundException.class,
+ () -> table.io().newInputFile(path).exists(),
+ "rollback must delete data files reported by completed BE
fragments");
+ Assertions.assertTrue(txn.getCommitDataList().isEmpty(),
+ "rollback must not leave fragments that a later lifecycle call
could commit");
+ txn.close();
+ }
+
@Test
public void beginWriteRejectsReplacementExposedByTransactionRefresh() {
InMemoryCatalog catalog = freshCatalog();
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_select_negative.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_select_negative.groovy
index 765e5c58345..cf1441369f0 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_select_negative.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_select_negative.groovy
@@ -23,13 +23,6 @@ suite("test_iceberg_write_required_null_select_negative",
return
}
- // This opt-in switch isolates a write that can publish an unreadable
Iceberg data file.
- String knownBugTestEnabled =
context.config.otherConfigs.get("enableIcebergKnownBugTest")
- if (knownBugTestEnabled == null ||
!knownBugTestEnabled.equalsIgnoreCase("true")) {
- logger.info("skip isolated Iceberg known-bug regression")
- return
- }
-
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
@@ -71,33 +64,48 @@ suite("test_iceberg_write_required_null_select_negative",
sql """create database ${dbName}"""
sql """use ${dbName}"""
- sql """
- create table required_select (
- id int not null,
- required_text string not null
- )
- partition by list (bucket(8, id)) ()
- properties ("format-version" = "2")
- """
-
- long snapshotsBefore = (sql """select count(*) from
required_select\$snapshots""")[0][0] as long
- long filesBefore = (sql """select count(*) from
required_select\$files""")[0][0] as long
- long rowsBefore = (sql """select count(*) from required_select""")[0][0]
as long
-
- // W07-S02: The metadata and row baselines distinguish an early rejection
- // from a late failure that has already published a snapshot or data file.
- test {
+ for (String fileFormat : ["orc", "parquet"]) {
+ String tableName = "required_select_${fileFormat}"
sql """
- insert into required_select
- select id, required_text
- from internal.${internalDb}.nullable_source
+ create table ${tableName} (
+ id int not null,
+ required_text string not null
+ )
+ partition by list (bucket(8, id)) ()
+ properties (
+ "format-version" = "2",
+ "write.format.default" = "${fileFormat}"
+ )
"""
- exception "null"
+
+ long snapshotsBefore = (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long
+ long filesBefore = (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long
+ long rowsBefore = (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long
+
+ // W07-S02: The metadata and row baselines distinguish an early
rejection
+ // from a late failure that has already published a snapshot or data
file.
+ test {
+ sql """
+ insert into ${tableName}
+ select id, required_text
+ from internal.${internalDb}.nullable_source
+ """
+ exception "required field 'required_text' contains NULL"
+ }
+ assertEquals(snapshotsBefore, (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long)
+ assertEquals(filesBefore, (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long)
+ assertEquals(rowsBefore, (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long)
}
- assertEquals(snapshotsBefore,
- (sql """select count(*) from required_select\$snapshots""")[0][0]
as long)
- assertEquals(filesBefore,
- (sql """select count(*) from required_select\$files""")[0][0] as
long)
- assertEquals(rowsBefore,
- (sql """select count(*) from required_select""")[0][0] as long)
}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_values_negative.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_values_negative.groovy
index 86c433dc634..2842e3d8f63 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_values_negative.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_values_negative.groovy
@@ -23,13 +23,6 @@ suite("test_iceberg_write_required_null_values_negative",
return
}
- // This opt-in switch isolates a write that can publish an unreadable
Iceberg data file.
- String knownBugTestEnabled =
context.config.otherConfigs.get("enableIcebergKnownBugTest")
- if (knownBugTestEnabled == null ||
!knownBugTestEnabled.equalsIgnoreCase("true")) {
- logger.info("skip isolated Iceberg known-bug regression")
- return
- }
-
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
@@ -53,29 +46,151 @@ suite("test_iceberg_write_required_null_values_negative",
sql """create database ${dbName}"""
sql """use ${dbName}"""
- sql """
- create table required_values (
- id int not null,
- required_text string not null
- )
- partition by list (bucket(8, id)) ()
- properties ("format-version" = "2")
- """
+ for (String fileFormat : ["orc", "parquet"]) {
+ String tableName = "required_values_${fileFormat}"
+ sql """
+ create table ${tableName} (
+ id int not null,
+ required_text string not null
+ )
+ partition by list (bucket(8, id)) ()
+ properties (
+ "format-version" = "2",
+ "write.format.default" = "${fileFormat}"
+ )
+ """
+
+ long snapshotsBefore = (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long
+ long filesBefore = (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long
+ long rowsBefore = (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long
+
+ // W07-S01: A thrown NULL error is insufficient proof of atomicity; all
+ // visible and metadata state must remain at the pre-statement
baseline.
+ test {
+ sql """insert into ${tableName} values (1, null)"""
+ exception "required field 'required_text' contains NULL"
+ }
+ assertEquals(snapshotsBefore, (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long)
+ assertEquals(filesBefore, (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long)
+ assertEquals(rowsBefore, (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long)
+ }
+
+ // Spark can express required nested fields that Doris DDL cannot declare
directly.
+ for (String fileFormat : ["orc", "parquet"]) {
+ String tableName = "required_nested_${fileFormat}"
+ spark_iceberg """
+ drop table if exists demo.${dbName}.${tableName}
+ """
+ spark_iceberg """
+ create table demo.${dbName}.${tableName} (
+ id int,
+ info struct<required_metric: int not null, required_label:
string not null>
+ )
+ using iceberg
+ tblproperties (
+ 'format-version' = '2',
+ 'write.format.default' = '${fileFormat}'
+ )
+ """
+ spark_iceberg """
+ insert into demo.${dbName}.${tableName}
+ values (1, named_struct('required_metric', 10, 'required_label',
'old-label'))
+ """
+ sql """refresh table ${tableName}"""
+
+ // The STRUCT itself is optional. A NULL parent masks its required
children and is a valid row.
+ sql """insert into ${tableName} values (2, null)"""
+ sql """refresh table ${tableName}"""
+
+ long snapshotsBefore = (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long
+ long filesBefore = (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long
+ long rowsBefore = (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long
+
+ test {
+ sql """
+ insert into ${tableName}
+ values (
+ 3,
+ named_struct(
+ 'required_metric', cast(null as int),
+ 'required_label', 'new-label'
+ )
+ )
+ """
+ // Current FE versions may reject the nullable-to-required cast
during planning; later
+ // versions may let the block reach BE. Both paths must identify
the required child.
+ exception "required_metric"
+ }
+ assertEquals(snapshotsBefore, (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long)
+ assertEquals(filesBefore, (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long)
+ assertEquals(rowsBefore, (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long)
+ }
+
+ // Iceberg MAP keys are always required, including when the MAP parent is
optional.
+ for (String fileFormat : ["orc", "parquet"]) {
+ String tableName = "required_map_key_${fileFormat}"
+ sql """
+ create table ${tableName} (
+ id int not null,
+ attrs map<int, int>
+ )
+ partition by list (bucket(8, id)) ()
+ properties (
+ "format-version" = "2",
+ "write.format.default" = "${fileFormat}"
+ )
+ """
+
+ // A NULL MAP parent masks all entries, so it must remain writable.
+ sql """insert into ${tableName} values (1, null)"""
+ sql """refresh table ${tableName}"""
- long snapshotsBefore = (sql """select count(*) from
required_values\$snapshots""")[0][0] as long
- long filesBefore = (sql """select count(*) from
required_values\$files""")[0][0] as long
- long rowsBefore = (sql """select count(*) from required_values""")[0][0]
as long
+ long snapshotsBefore = (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long
+ long filesBefore = (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long
+ long rowsBefore = (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long
- // W07-S01: A thrown NULL error is insufficient proof of atomicity; all
- // visible and metadata state must remain at the pre-statement baseline.
- test {
- sql """insert into required_values values (1, null)"""
- exception "null"
+ test {
+ sql """insert into ${tableName} values (2, map(cast(null as int),
2))"""
+ exception "map key 'attrs.key' contains NULL"
+ }
+ assertEquals(snapshotsBefore, (sql """
+ select count(*) from ${tableName}\$snapshots
+ """)[0][0] as long)
+ assertEquals(filesBefore, (sql """
+ select count(*) from ${tableName}\$files
+ """)[0][0] as long)
+ assertEquals(rowsBefore, (sql """
+ select count(*) from ${tableName}
+ """)[0][0] as long)
}
- assertEquals(snapshotsBefore,
- (sql """select count(*) from required_values\$snapshots""")[0][0]
as long)
- assertEquals(filesBefore,
- (sql """select count(*) from required_values\$files""")[0][0] as
long)
- assertEquals(rowsBefore,
- (sql """select count(*) from required_values""")[0][0] as long)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
