This is an automated email from the ASF dual-hosted git repository.
morningman 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 b78c656c0cb [fix](iceberg) Commit every Iceberg file under the table's
own location (#69392)
b78c656c0cb is described below
commit b78c656c0cbcc255b28509c5dc79a9d9d4b282f2
Author: Chenjunwei <[email protected]>
AuthorDate: Wed Nov 4 09:59:11 2026 +0800
[fix](iceberg) Commit every Iceberg file under the table's own location
(#69392)
### What problem does this PR solve?
Issue Number: None
Related PR: #69082
Problem Summary:
**In short.** DELETE, UPDATE and MERGE on an Iceberg table recorded the
delete files they write (v2 position delete files and v3 deletion vector
files) under the internal `s3://` address BE writes them at, instead of
the table's own location (`oss://`, `s3a://`, `abfss://`, ...). Doris
could not always read such a table back, and other engines and
maintenance procedures resolved those files with the wrong file system.
This PR commits delete files under the table's location. It also makes
every Iceberg file Doris writes take both of its paths from one place in
BE and pass one check in FE, so the next writer cannot repeat the
mistake without failing.
#### Background
Every Iceberg sink FE sends carries two forms of the table's data
location:
- a normalized one BE can address the storage with (`output_path`): FE
rewrites `oss://`, `cos://`, `cosn://`, `obs://`, `bos://`, `gs://`,
`s3a://`, `s3n://` and Azure `abfs[s]://` / `wasb[s]://` locations
(except OneLake) to `s3://`;
- the raw one Iceberg metadata must record: `original_output_path` on
`TIcebergTableSink`, `table_location` on `TIcebergDeleteSink`, both on
`TIcebergMergeSink`.
BE writes each file under the first form and reports it in
`TIcebergCommitData.file_path`, which FE writes into the manifests as it
is. So the reported path must be in the second form.
#### The problem, and what it cost
Reproduce on an Iceberg table located at `oss://bucket/...`:
```sql
CREATE TABLE t (id INT, v STRING) PROPERTIES ('format-version' = '2');
INSERT INTO t VALUES (1, 'a'), (2, 'b'), (3, 'c');
DELETE FROM t WHERE id = 1;
SELECT file_path FROM t$delete_files; --
s3://bucket/.../data/delete_pos_xxx.parquet
SELECT file_path FROM t$data_files; -- oss://bucket/.../data/...
```
- Whether Doris can still read the table depends on the catalog and
storage binding. In a real Azure ADLS Hadoop catalog test with explicit
SharedKey, Doris SELECT failed with `No storage properties found for
schema: s3`, and Spark failed with `UnsupportedFileSystemException: No
FileSystem for scheme "s3"`.
- Other engines and maintenance operations such as `expire_snapshots` /
`remove_orphan_files` may resolve the wrong file system from the
recorded scheme, depending on the engine and FileIO. Doris's own
`remove_orphan_files` refuses to run on such a table (`Cannot determine
whether listed and reachable file locations are equivalent`).
- Cause: data files were written under `write_path` and committed under
`original_write_path` (`VIcebergPartitionWriter`), but
`VIcebergDeleteSink` had no such split; it reported the
`output_path`-based path it wrote.
- Why it went unnoticed: the two writers mapped the paths in different
ways, `VIcebergDeleteFileWriter` reported its write path by default, and
FE accepted any path. CI did run on such a location: the Iceberg REST
fixture's warehouse is `s3a://`, which FE normalizes to `s3://`. But the
DML suites only checked that delete file paths contain
`/data/delete_pos_`, and a stray
`test_iceberg_update_delete_basic.out.org` in the repository records the
wrong `s3://warehouse/wh/...` paths.
#### How this PR fixes it
1. `[fix](be) Commit Iceberg delete files under the table's own
location`, `[test](be) Verify Iceberg delete file commit paths` and
`[test](be) Initialize the Arrow memory pool in the Iceberg position
delete write test`: `VIcebergDeleteSink` writes under `output_path` and
commits position delete files and deletion vectors under
`table_location`. An FE unit test pins that both sinks carrying
`table_location` name the same directory as `output_path`, and
`DataSinks.thrift` documents the contract. The position delete write
test sets up the Arrow memory pool the Parquet writer allocates from, so
it also passes when run alone (CI review bot finding on this PR).
2. `[refactor](be) Take both paths of every Iceberg file from one output
location`:
- The new `IcebergOutputLocation` (header-only) holds one directory in
both forms and gives the write path and the commit path of a file name,
and of a partition subdirectory.
- The data writer (`VIcebergTableWriter` / `VIcebergPartitionWriter`),
the delete file writer and the Puffin path all take their paths from it.
`VIcebergDeleteFileWriter` now reports the commit path itself instead of
its write path.
- A table or delete sink without both location fields is rejected in
`open()` / `init_properties()` instead of committing a write path or a
relative path. The FE of every branch (3.1, 4.0, 4.1, 4.2, master) sets
both fields on every Iceberg sink.
- Directories join file names with exactly one `/`.
3. `[improvement](fe) Reject Iceberg commit data reported outside the
sink's location`:
- `planWrite` registers on the transaction the raw location(s) each sink
it builds carries, read back from the built sink.
- `IcebergConnectorTransaction.addCommitData`, where every report from
every writer enters FE, rejects a file that is not under a registered
location, or has no path. FE then answers the report as not accepted,
and BE deletes the files under the paths it wrote them at and cancels
the query, so a wrong path fails the statement and leaves no orphan
files.
- Thrift comments state the contract, including that "the same
directory" can differ in more than the scheme (Azure moves the account
out of the authority). Comments only, no protocol change.
4. `[test](regression) Check that every Iceberg write records its files
under the table location`:
- The new `test_iceberg_write_file_locations` runs INSERT, INSERT
OVERWRITE PARTITION, branch INSERT, DELETE, UPDATE, MERGE and
`rewrite_data_files` on a v2 table, and INSERT, DELETE, UPDATE and MERGE
on a v3 table. After each operation every file in `$all_files` must
start with the table's LOCATION plus `/data/`. A new write operation
belongs in this suite too.
- `test_iceberg_update_delete_basic` checks the prefix of its delete
files as well, and the stray `.out.org` is removed.
| Scenario | Before | After |
| --- | --- | --- |
| DELETE / UPDATE / MERGE on a v2 table at `oss://`, `s3a://`,
`abfss://`, ... | position delete files recorded as `s3://...` |
recorded under the table's location |
| The same on a v3 table | deletion vector (Puffin) files recorded as
`s3://...` | recorded under the table's location |
| A BE writer reports a file outside the location its sink carries |
committed as reported | the statement fails; BE deletes the file |
| A sink plan without `original_output_path` / `table_location` (no
supported FE sends one) | data files committed as `/p=1/x.parquet`,
delete files as `s3://...` | rejected before any file is written |
| `write.data.path` ending with `/` | data files at
`.../data//p=1/x.parquet` | `.../data/p=1/x.parquet`, as Iceberg's own
location provider places them |
Delete files that an older version already committed with `s3://` keep
that address until they are rewritten or expired.
Upgrade BE before FE, as usual. An upgraded FE rejects the `s3://`
delete file paths that a BE without this fix reports, so until that BE
is upgraded, DELETE / UPDATE / MERGE on a table whose location FE
normalizes fails instead of committing the wrong path.
#### Classes, and how they call each other
- `IcebergWritePlanProvider.planWrite` (FE) builds the sink and
registers the raw location(s) it carries with the transaction
(`commitLocationsOf`).
- `IcebergConnectorTransaction.addCommitData` (FE) accepts a reported
file only under a registered location; `IcebergWriterHelper` then turns
the fragments into `DataFile` / `DeleteFile`.
- `IcebergOutputLocation` (BE, new) is one directory in its write form
and its commit form.
- `VIcebergTableWriter` builds the data location in `open()` and hands
each `VIcebergPartitionWriter` its partition directory
(`WriteInfo.location`). `VIcebergDeleteSink` writes position delete
files through `VIcebergDeleteFileWriter` and deletion vectors into one
Puffin file, both from its location. `VIcebergMergeSink` builds one of
each.
```
FE planWrite ── sink: output_path + raw location ──────────────► BE
│ addCommitLocation(raw location)
IcebergOutputLocation(output_path, raw location)
▼ ├─
VIcebergTableWriter ─► VIcebergPartitionWriter data files
IcebergConnectorTransaction ├─
VIcebergDeleteSink ──► VIcebergDeleteFileWriter position deletes
▲ addCommitData: file_path under a registered │
└─► Puffin writer deletion vectors
│ location, else reject the report └─
VIcebergMergeSink ──► both of the above
└──────────── TIcebergCommitData.file_path = commit path ◄───────┘
```
Not changed: Thrift fields (comments only), the scan side, and how FE
normalizes locations.
---
be/src/exec/sink/viceberg_delete_sink.cpp | 75 ++++-----
be/src/exec/sink/viceberg_delete_sink.h | 16 +-
.../sink/writer/iceberg/iceberg_output_location.h | 81 ++++++++++
.../writer/iceberg/viceberg_delete_file_writer.cpp | 17 ++-
.../writer/iceberg/viceberg_delete_file_writer.h | 18 ++-
.../writer/iceberg/viceberg_partition_writer.cpp | 5 +-
.../sink/writer/iceberg/viceberg_table_writer.cpp | 48 +++---
.../sink/writer/iceberg/viceberg_table_writer.h | 5 +
.../sink/writer/iceberg/vpartition_writer_base.h | 7 +-
be/test/exec/sink/viceberg_delete_sink_test.cpp | 167 ++++++++++++++++++---
be/test/exec/sink/viceberg_merge_sink_test.cpp | 38 +++++
.../iceberg/iceberg_output_location_test.cpp | 56 +++++++
.../iceberg/iceberg_partition_writer_test.cpp | 27 +++-
.../writer/iceberg/viceberg_table_writer_test.cpp | 60 ++++++++
.../format/transformer/vparquet_writer_test.cpp | 4 +-
.../iceberg/IcebergConnectorTransaction.java | 40 +++++
.../iceberg/IcebergWritePlanProvider.java | 43 +++++-
.../iceberg/IcebergConnectorTransactionTest.java | 72 ++++++++-
.../iceberg/IcebergWriteKnownBugsTest.java | 8 +-
.../iceberg/IcebergWritePlanProviderTest.java | 66 ++++++++
.../IcebergWriteTransactionCoverageTest.java | 9 +-
.../java/org/apache/doris/qe/QeProcessorImpl.java | 14 +-
.../doris/qe/QeProcessorImplReportAckTest.java | 27 +++-
gensrc/thrift/DataSinks.thrift | 17 +++
.../dml/test_iceberg_update_delete_basic.out.org | 77 ----------
.../dml/test_iceberg_update_delete_basic.groovy | 16 +-
.../iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md | 5 +-
.../write/test_iceberg_write_file_locations.groovy | 161 ++++++++++++++++++++
28 files changed, 956 insertions(+), 223 deletions(-)
diff --git a/be/src/exec/sink/viceberg_delete_sink.cpp
b/be/src/exec/sink/viceberg_delete_sink.cpp
index 55b732d6527..3cb11884898 100644
--- a/be/src/exec/sink/viceberg_delete_sink.cpp
+++ b/be/src/exec/sink/viceberg_delete_sink.cpp
@@ -148,14 +148,16 @@ Status VIcebergDeleteSink::init_properties(ObjectPool*
pool) {
_compress_type = delete_sink.compress_type;
}
- // Get output path and table location
- if (delete_sink.__isset.output_path) {
- _output_path = delete_sink.output_path;
- }
-
- if (delete_sink.__isset.table_location) {
- _table_location = delete_sink.table_location;
+ // Every FE version sends both forms of the data location (see
IcebergOutputLocation). Without
+ // the table's own form the commit could only record the address BE wrote
the file at, e.g.
+ // s3://b/db/t/data/delete_pos_x.parquet for an oss:// table, which other
engines resolve with
+ // the wrong file system.
+ if (delete_sink.output_path.empty() || delete_sink.table_location.empty())
{
+ return Status::InternalError(
+ "Iceberg delete sink must carry output_path and
table_location, got '{}' and '{}'",
+ delete_sink.output_path, delete_sink.table_location);
}
+ _output_location.emplace(delete_sink.output_path,
delete_sink.table_location);
// Get Hadoop configuration
if (delete_sink.__isset.hadoop_config) {
@@ -222,7 +224,7 @@ Status VIcebergDeleteSink::open(RuntimeState* state,
RuntimeProfile* profile) {
LOG(INFO) << fmt::format(
"VIcebergDeleteSink opened: delete_type={}, output_path={},
format_version={}",
- to_string(_delete_type), _output_path, _format_version);
+ to_string(_delete_type), _t_sink.iceberg_delete_sink.output_path,
_format_version);
return Status::OK();
}
@@ -481,18 +483,16 @@ Status VIcebergDeleteSink::_collect_position_deletes(
Status VIcebergDeleteSink::_write_position_delete_files(
const std::map<std::string, IcebergFileDeletion>& file_deletions) {
+ DORIS_CHECK(_output_location.has_value());
constexpr size_t kBatchSize = 4096;
for (const auto& [data_file_path, deletion] : file_deletions) {
if (deletion.rows_to_delete.isEmpty()) {
continue;
}
- // Generate unique delete file path
- std::string delete_file_path =
_generate_delete_file_path(data_file_path);
-
// Create delete file writer
auto writer = VIcebergDeleteFileWriterFactory::create_writer(
- TFileContent::POSITION_DELETES, delete_file_path,
_file_format_type,
- _compress_type);
+ TFileContent::POSITION_DELETES, *_output_location,
+ _delete_file_name(data_file_path), _file_format_type,
_compress_type);
// Build column names for position delete
std::vector<std::string> column_names = {"file_path", "pos"};
@@ -506,7 +506,7 @@ Status VIcebergDeleteSink::_write_position_delete_files(
writer->open(_state, _state->runtime_profile(),
_position_delete_output_expr_ctxs,
column_names, _hadoop_conf, _file_type,
_broker_addresses);
if (writer->file_system() != nullptr) {
- _created_files.emplace_back(writer->file_system(),
delete_file_path);
+ _created_files.emplace_back(writer->file_system(),
writer->write_path());
}
RETURN_IF_ERROR(open_status);
@@ -543,7 +543,7 @@ Status VIcebergDeleteSink::_write_position_delete_files(
_delete_file_count++;
VLOG(1) << fmt::format("Written position delete file: path={},
rows={}, referenced_file={}",
- delete_file_path, commit_data.row_count,
data_file_path);
+ writer->write_path(), commit_data.row_count,
data_file_path);
}
return Status::OK();
@@ -643,6 +643,7 @@ std::string VIcebergDeleteSink::_get_file_extension() const
{
Status VIcebergDeleteSink::_write_deletion_vector_files(
const std::map<std::string, IcebergFileDeletion>& file_deletions) {
+ DORIS_CHECK(_output_location.has_value());
std::vector<DeletionVectorBlob> blobs;
for (const auto& [data_file_path, deletion] : file_deletions) {
if (deletion.rows_to_delete.isEmpty()) {
@@ -687,13 +688,15 @@ Status VIcebergDeleteSink::_write_deletion_vector_files(
return Status::OK();
}
- std::string puffin_path = _generate_puffin_file_path();
+ const std::string puffin_file_name = _puffin_file_name();
int64_t puffin_file_size = 0;
- RETURN_IF_ERROR(_write_puffin_file(puffin_path, &blobs,
&puffin_file_size));
+
RETURN_IF_ERROR(_write_puffin_file(_output_location->write_path(puffin_file_name),
&blobs,
+ &puffin_file_size));
+ const std::string commit_path =
_output_location->commit_path(puffin_file_name);
for (const auto& blob : blobs) {
TIcebergCommitData commit_data;
- commit_data.__set_file_path(puffin_path);
+ commit_data.__set_file_path(commit_path);
commit_data.__set_row_count(blob.merged_count);
commit_data.__set_affected_rows(blob.delete_count);
commit_data.__set_file_size(puffin_file_size);
@@ -803,37 +806,15 @@ std::string VIcebergDeleteSink::_build_puffin_footer_json(
return {buffer.GetString(), buffer.GetSize()};
}
-std::string VIcebergDeleteSink::_generate_delete_file_path(
- const std::string& referenced_data_file) {
- // Generate unique delete file name using UUID
- std::string uuid = generate_uuid_string();
- std::string file_name;
-
- std::string file_extension = _get_file_extension();
- file_name =
- fmt::format("delete_pos_{}_{}{}", uuid,
- std::hash<std::string> {}(referenced_data_file) %
10000000, file_extension);
-
- // Combine with output path or table location
- std::string base_path = _output_path.empty() ? _table_location :
_output_path;
-
- // Ensure base path ends with /
- if (!base_path.empty() && base_path.back() != '/') {
- base_path += '/';
- }
-
- // Delete files are data files in Iceberg, write under data location
- return fmt::format("{}{}", base_path, file_name);
+std::string VIcebergDeleteSink::_delete_file_name(const std::string&
referenced_data_file) const {
+ // Delete files live in the data location itself, next to the partition
directories.
+ return fmt::format("delete_pos_{}_{}{}", generate_uuid_string(),
+ std::hash<std::string> {}(referenced_data_file) %
10000000,
+ _get_file_extension());
}
-std::string VIcebergDeleteSink::_generate_puffin_file_path() {
- std::string uuid = generate_uuid_string();
- std::string file_name = fmt::format("delete_dv_{}.puffin", uuid);
- std::string base_path = _output_path.empty() ? _table_location :
_output_path;
- if (!base_path.empty() && base_path.back() != '/') {
- base_path += '/';
- }
- return fmt::format("{}{}", base_path, file_name);
+std::string VIcebergDeleteSink::_puffin_file_name() const {
+ return fmt::format("delete_dv_{}.puffin", generate_uuid_string());
}
} // namespace doris
diff --git a/be/src/exec/sink/viceberg_delete_sink.h
b/be/src/exec/sink/viceberg_delete_sink.h
index 625134ba3a3..a4bb50d20f7 100644
--- a/be/src/exec/sink/viceberg_delete_sink.h
+++ b/be/src/exec/sink/viceberg_delete_sink.h
@@ -24,12 +24,14 @@
#include <cstdint>
#include <map>
#include <memory>
+#include <optional>
#include <string>
#include <vector>
#include "common/status.h"
#include "core/block/block.h"
#include "exec/sink/writer/async_result_writer.h"
+#include "exec/sink/writer/iceberg/iceberg_output_location.h"
#include "exec/sink/writer/iceberg/viceberg_delete_file_writer.h"
#include "exprs/vexpr_fwd.h"
#include "roaring/roaring64map.hh"
@@ -114,12 +116,9 @@ private:
std::string _build_puffin_footer_json(const
std::vector<DeletionVectorBlob>& blobs);
- std::string _generate_puffin_file_path();
-
- /**
- * Generate unique delete file path
- */
- std::string _generate_delete_file_path(const std::string&
referenced_data_file = "");
+ // Unique file names, relative to the data location (_output_location).
+ std::string _puffin_file_name() const;
+ std::string _delete_file_name(const std::string& referenced_data_file)
const;
/**
* Get $row_id column index from block
@@ -163,9 +162,8 @@ private:
TFileFormatType::type _file_format_type = TFileFormatType::FORMAT_PARQUET;
TFileCompressType::type _compress_type = TFileCompressType::SNAPPYBLOCK;
- // Output directory for delete files
- std::string _output_path;
- std::string _table_location;
+ // The data location delete files are written to and committed under, set
by init_properties().
+ std::optional<IcebergOutputLocation> _output_location;
TFileType::type _file_type = TFileType::FILE_HDFS;
std::vector<TNetworkAddress> _broker_addresses;
diff --git a/be/src/exec/sink/writer/iceberg/iceberg_output_location.h
b/be/src/exec/sink/writer/iceberg/iceberg_output_location.h
new file mode 100644
index 00000000000..58982f73156
--- /dev/null
+++ b/be/src/exec/sink/writer/iceberg/iceberg_output_location.h
@@ -0,0 +1,81 @@
+// 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 <string>
+#include <string_view>
+#include <utility>
+
+#include "common/check.h"
+
+namespace doris {
+
+// The directory an Iceberg writer puts its files in, in both forms of the
same location. BE
+// creates, writes and deletes a file under write_dir, the form BE addresses
the storage with.
+// Iceberg metadata records the file under commit_dir, the table's own form,
which every engine
+// reading the table resolves. FE sends both: TIcebergTableSink output_path /
original_output_path
+// and TIcebergDeleteSink output_path / table_location. Example: an OSS table
with data location
+// oss://b/db/t/data reaches BE as write_dir s3://b/db/t/data and commit_dir
oss://b/db/t/data, so
+// delete_pos_1.parquet is written to s3://b/db/t/data/delete_pos_1.parquet
and committed as
+// oss://b/db/t/data/delete_pos_1.parquet.
+//
+// Every file a writer reports takes both of its paths from one
IcebergOutputLocation, so it
+// cannot commit the form it wrote: delete files used to be committed as
s3://..., and other
+// engines resolved them with a file system the table is not on.
+class IcebergOutputLocation {
+public:
+ IcebergOutputLocation(std::string write_dir, std::string commit_dir)
+ : _write_dir(std::move(write_dir)),
_commit_dir(std::move(commit_dir)) {
+ DORIS_CHECK(!_write_dir.empty());
+ DORIS_CHECK(!_commit_dir.empty());
+ }
+
+ // The same subdirectory in both forms, e.g. the partition directory
"p=1/q=2".
+ IcebergOutputLocation subdirectory(std::string_view relative_dir) const {
+ DORIS_CHECK(!relative_dir.empty());
+ return {_join(_write_dir, relative_dir), _join(_commit_dir,
relative_dir)};
+ }
+
+ // The path BE writes file_name at.
+ std::string write_path(std::string_view file_name) const {
+ return _join(_write_dir, file_name);
+ }
+
+ // The path Iceberg metadata records file_name at.
+ std::string commit_path(std::string_view file_name) const {
+ return _join(_commit_dir, file_name);
+ }
+
+private:
+ // FE takes write.data.path verbatim, so a directory may end with '/'.
Both forms join with
+ // exactly one separator, as Iceberg's own location provider does:
oss://b/t/data and
+ // oss://b/t/data/ both join "p=1" into oss://b/t/data/p=1.
+ static std::string _join(const std::string& dir, std::string_view name) {
+ std::string path = dir;
+ if (path.back() != '/') {
+ path.push_back('/');
+ }
+ path.append(name);
+ return path;
+ }
+
+ std::string _write_dir;
+ std::string _commit_dir;
+};
+
+} // namespace doris
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 7f7f2c6f2f4..1233d09dae5 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
@@ -44,11 +44,13 @@ std::unique_ptr<iceberg::Schema>
build_position_delete_schema() {
}
VIcebergDeleteFileWriter::VIcebergDeleteFileWriter(TFileContent::type
delete_type,
- const std::string&
output_path,
+ const
IcebergOutputLocation& location,
+ const std::string&
file_name,
TFileFormatType::type
file_format,
TFileCompressType::type
compress_type)
: _delete_type(delete_type),
- _output_path(output_path),
+ _write_path(location.write_path(file_name)),
+ _commit_path(location.commit_path(file_name)),
_file_format(file_format),
_compress_type(compress_type) {}
@@ -74,7 +76,7 @@ Status VIcebergDeleteFileWriter::open(RuntimeState* state,
RuntimeProfile* profi
fs_properties.broker_addresses = &broker_addresses;
}
- io::FileDescription file_description = {.path = _output_path, .fs_name {}};
+ io::FileDescription file_description = {.path = _write_path, .fs_name {}};
_fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
io::FileWriterOptions file_writer_options = {.used_by_s3_committer =
false};
@@ -131,7 +133,7 @@ Status VIcebergDeleteFileWriter::close(TIcebergCommitData&
commit_data) {
}
// Fill commit data (use __set_ to mark optional fields as present)
- commit_data.__set_file_path(_output_path);
+ commit_data.__set_file_path(_commit_path);
commit_data.__set_row_count(_written_rows);
commit_data.__set_affected_rows(_written_rows);
commit_data.__set_file_size(_file_size);
@@ -152,9 +154,10 @@ Status VIcebergDeleteFileWriter::close(TIcebergCommitData&
commit_data) {
// Factory method
std::unique_ptr<VIcebergDeleteFileWriter>
VIcebergDeleteFileWriterFactory::create_writer(
- TFileContent::type delete_type, const std::string& output_path,
- TFileFormatType::type file_format, TFileCompressType::type
compress_type) {
- return std::make_unique<VIcebergDeleteFileWriter>(delete_type,
output_path, file_format,
+ TFileContent::type delete_type, const IcebergOutputLocation& location,
+ const std::string& file_name, TFileFormatType::type file_format,
+ TFileCompressType::type compress_type) {
+ return std::make_unique<VIcebergDeleteFileWriter>(delete_type, location,
file_name, file_format,
compress_type);
}
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h
b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h
index 4ceb7080a3c..b216b3ba16f 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h
@@ -23,6 +23,7 @@
#include "common/status.h"
#include "core/block/block.h"
+#include "exec/sink/writer/iceberg/iceberg_output_location.h"
#include "exprs/vexpr_fwd.h"
#include "gen_cpp/DataSinks_types.h"
#include "io/fs/file_writer.h"
@@ -50,8 +51,10 @@ class VFileFormatTransformer;
*/
class VIcebergDeleteFileWriter {
public:
- VIcebergDeleteFileWriter(TFileContent::type delete_type, const
std::string& output_path,
- TFileFormatType::type file_format,
+ // Writes file_name under location; the commit data reports the file under
the table's own form
+ // of that directory, not the form it was written at.
+ VIcebergDeleteFileWriter(TFileContent::type delete_type, const
IcebergOutputLocation& location,
+ const std::string& file_name,
TFileFormatType::type file_format,
TFileCompressType::type compress_type);
~VIcebergDeleteFileWriter();
@@ -96,9 +99,13 @@ public:
std::shared_ptr<io::FileSystem> file_system() const { return _fs; }
+ // The path the file is written at, which is what cleanup must delete.
+ const std::string& write_path() const { return _write_path; }
+
private:
TFileContent::type _delete_type;
- std::string _output_path;
+ std::string _write_path;
+ std::string _commit_path;
TFileFormatType::type _file_format;
TFileCompressType::type _compress_type = TFileCompressType::SNAPPYBLOCK;
@@ -122,8 +129,9 @@ private:
class VIcebergDeleteFileWriterFactory {
public:
static std::unique_ptr<VIcebergDeleteFileWriter> create_writer(
- TFileContent::type delete_type, const std::string& output_path,
- TFileFormatType::type file_format, TFileCompressType::type
compress_type);
+ TFileContent::type delete_type, const IcebergOutputLocation&
location,
+ const std::string& file_name, TFileFormatType::type file_format,
+ TFileCompressType::type compress_type);
};
} // 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 363776dcf17..249cd15dcf1 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
@@ -93,7 +93,7 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
if (!_write_info.broker_addresses.empty()) {
fs_properties.broker_addresses = &(_write_info.broker_addresses);
}
- _path = fmt::format("{}/{}", _write_info.write_path,
_get_target_file_name());
+ _path = _write_info.location.write_path(_get_target_file_name());
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};
@@ -210,8 +210,7 @@ Status
VIcebergPartitionWriter::_build_iceberg_commit_data(TIcebergCommitData* c
DCHECK(commit_data != nullptr);
DCHECK(_file_format_transformer != nullptr);
- commit_data->__set_file_path(
- fmt::format("{}/{}", _write_info.original_write_path,
_get_target_file_name()));
+
commit_data->__set_file_path(_write_info.location.commit_path(_get_target_file_name()));
commit_data->__set_row_count(_row_count);
commit_data->__set_affected_rows(_row_count);
commit_data->__set_file_size(_file_format_transformer->written_len());
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 61cb105460d..d20eb8a005f 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
@@ -61,6 +61,18 @@ Status VIcebergTableWriter::open(RuntimeState* state,
RuntimeProfile* profile) {
RETURN_IF_ERROR(validate_external_file_report_ack(state->query_options()));
+ // Every FE version sends both forms of the data location (see
IcebergOutputLocation). A plan
+ // without the table's own form leaves no path to commit the files under:
the commit used to
+ // record a relative path such as /p=1/xxx.parquet that no engine can open.
+ const auto& iceberg_sink = _t_sink.iceberg_table_sink;
+ if (iceberg_sink.output_path.empty() ||
iceberg_sink.original_output_path.empty()) {
+ return Status::InternalError(
+ "Iceberg table sink must carry output_path and
original_output_path, got '{}' "
+ "and '{}'",
+ iceberg_sink.output_path, iceberg_sink.original_output_path);
+ }
+ _output_location.emplace(iceberg_sink.output_path,
iceberg_sink.original_output_path);
+
// Get target file size from query options
// If value is 0 or not set, use config::iceberg_sink_max_file_size
_target_file_size_bytes = config::iceberg_sink_max_file_size;
@@ -822,47 +834,29 @@ std::vector<std::string>
VIcebergTableWriter::_partition_values(
std::shared_ptr<IPartitionWriterBase>
VIcebergTableWriter::_create_partition_writer(
Block* transformed_block, int position, const std::string* file_name,
int file_name_index) {
+ DORIS_CHECK(_output_location.has_value());
auto& iceberg_table_sink = _t_sink.iceberg_table_sink;
std::vector<std::string> partition_values;
- const std::string& output_path = iceberg_table_sink.output_path;
- std::string write_path;
- std::string original_write_path;
- std::string target_path;
+ // Relative to the data location, e.g. "year=2023/month=12"; empty when
the file goes to the
+ // data location itself.
+ std::string partition_path;
// Case 1: Full static partition - use pre-computed static partition path
and values
if (_is_full_static_partition) {
partition_values = _static_partition_value_list;
- if (!_static_partition_path.empty()) {
- original_write_path = fmt::format("{}/{}",
iceberg_table_sink.original_output_path,
- _static_partition_path);
- target_path = fmt::format("{}/{}", output_path,
_static_partition_path);
- write_path = fmt::format("{}/{}", output_path,
_static_partition_path);
- } else {
- original_write_path = iceberg_table_sink.original_output_path;
- target_path = output_path;
- write_path = output_path;
- }
+ partition_path = _static_partition_path;
} else if (transformed_block != nullptr) {
// Case 2: Dynamic partition or Hybrid mode (partial static + partial
dynamic)
// _partition_to_path and _partition_values already handle hybrid mode
internally
IcebergPartitionData partition_data =
_get_partition_data(transformed_block, position);
- std::string partition_path = _partition_to_path(partition_data);
+ partition_path = _partition_to_path(partition_data);
partition_values = _partition_values(partition_data);
- original_write_path =
- fmt::format("{}/{}", iceberg_table_sink.original_output_path,
partition_path);
- target_path = fmt::format("{}/{}", output_path, partition_path);
- write_path = fmt::format("{}/{}", output_path, partition_path);
- } else {
- // Case 3: Non-partitioned table
- original_write_path = iceberg_table_sink.original_output_path;
- target_path = output_path;
- write_path = output_path;
}
+ // Case 3: Non-partitioned table, written to the data location itself
IPartitionWriterBase::WriteInfo write_info = {
- .write_path = std::move(write_path),
- .original_write_path = std::move(original_write_path),
- .target_path = std::move(target_path),
+ .location = partition_path.empty() ? *_output_location
+ :
_output_location->subdirectory(partition_path),
.file_type = iceberg_table_sink.file_type,
.broker_addresses = {}};
if (iceberg_table_sink.__isset.broker_addresses) {
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 36eb5c08aa0..5c9a8dddb9f 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
@@ -20,10 +20,13 @@
#include <gen_cpp/DataSinks_types.h>
#include <gtest/gtest_prod.h>
+#include <optional>
+
#include "common/atomic_shared_ptr.h"
#include "common/status.h"
#include "core/block/block.h"
#include "core/column/column.h"
+#include "exec/sink/writer/iceberg/iceberg_output_location.h"
#include "exec/sink/writer/iceberg/partition_data.h"
#include "exec/sink/writer/iceberg/partition_transformers.h"
#include "exprs/vexpr_fwd.h"
@@ -164,6 +167,8 @@ private:
TDataSink _t_sink;
RuntimeState* _state = nullptr;
RuntimeProfile* _operator_profile = nullptr;
+ // The table's data location, set by open(); partition writers write under
its subdirectories.
+ std::optional<IcebergOutputLocation> _output_location;
// Target file size in bytes for controlling when to split files
int64_t _target_file_size_bytes = 0;
diff --git a/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h
b/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h
index dc70f646795..0c1980ca252 100644
--- a/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h
+++ b/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h
@@ -20,6 +20,7 @@
#include <string>
#include "common/status.h"
+#include "exec/sink/writer/iceberg/iceberg_output_location.h"
#include "runtime/descriptors.h"
namespace doris {
@@ -34,9 +35,9 @@ class RuntimeProfile;
class IPartitionWriterBase {
public:
struct WriteInfo {
- std::string write_path;
- std::string original_write_path;
- std::string target_path;
+ // The directory of the partition this writer writes, in the form BE
writes the file in
+ // and the form the commit records it in.
+ IcebergOutputLocation location;
TFileType::type file_type;
std::vector<TNetworkAddress> broker_addresses;
};
diff --git a/be/test/exec/sink/viceberg_delete_sink_test.cpp
b/be/test/exec/sink/viceberg_delete_sink_test.cpp
index e2897125e49..2eaa762350b 100644
--- a/be/test/exec/sink/viceberg_delete_sink_test.cpp
+++ b/be/test/exec/sink/viceberg_delete_sink_test.cpp
@@ -17,7 +17,10 @@
#include "exec/sink/viceberg_delete_sink.h"
+#include <arrow/api.h>
+#include <arrow/io/api.h>
#include <gtest/gtest.h>
+#include <parquet/arrow/reader.h>
#include <rapidjson/document.h>
#include <filesystem>
@@ -36,7 +39,9 @@
#include "format/table/deletion_vector.h"
#include "gen_cpp/DataSinks_types.h"
#include "gen_cpp/Types_types.h"
+#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
#include "util/uid_util.h"
namespace doris {
@@ -471,26 +476,57 @@ TEST_F(VIcebergDeleteSinkTest,
TestBuildPositionDeleteBlock) {
ASSERT_EQ(40, pos_column->get_element(3));
}
-TEST_F(VIcebergDeleteSinkTest, TestGenerateDeleteFilePath) {
+TEST_F(VIcebergDeleteSinkTest, TestDeleteFileNames) {
VExprContextSPtrs output_exprs;
auto sink = std::make_shared<VIcebergDeleteSink>(_t_data_sink,
output_exprs, nullptr, nullptr);
ObjectPool pool;
- Status status = sink->init_properties(&pool);
- ASSERT_TRUE(status.ok());
+ ASSERT_TRUE(sink->init_properties(&pool).ok());
- std::string data_file_path = "data/file1.parquet";
- std::string delete_file_path =
sink->_generate_delete_file_path(data_file_path);
+ // Names are relative to the data location; the location supplies both
forms of the directory.
+ std::string delete_file_name =
sink->_delete_file_name("data/file1.parquet");
+ EXPECT_TRUE(delete_file_name.starts_with("delete_pos_")) <<
delete_file_name;
+ EXPECT_TRUE(delete_file_name.ends_with(".snappy.parquet")) <<
delete_file_name;
+ EXPECT_EQ(std::string::npos, delete_file_name.find('/'));
+ EXPECT_NE(delete_file_name, sink->_delete_file_name("data/file1.parquet"));
+
+ std::string puffin_file_name = sink->_puffin_file_name();
+ EXPECT_TRUE(puffin_file_name.starts_with("delete_dv_")) <<
puffin_file_name;
+ EXPECT_TRUE(puffin_file_name.ends_with(".puffin")) << puffin_file_name;
+ EXPECT_EQ(std::string::npos, puffin_file_name.find('/'));
+}
- // Verify the path format
- ASSERT_FALSE(delete_file_path.empty());
- const auto& delete_sink = _t_data_sink.iceberg_delete_sink;
- std::string expected_base =
- delete_sink.__isset.output_path ? delete_sink.output_path :
delete_sink.table_location;
- if (!expected_base.empty() && expected_base.back() != '/') {
- expected_base += '/';
- }
- ASSERT_TRUE(delete_file_path.rfind(expected_base, 0) == 0);
- ASSERT_NE(std::string::npos, delete_file_path.find("delete_pos_"));
+TEST_F(VIcebergDeleteSinkTest,
TestInitPropertiesTakesBothFormsOfTheDataLocation) {
+ // FE writes through the normalized output_path but the table records its
own location.
+ TDataSink t_data_sink = _t_data_sink;
+ t_data_sink.iceberg_delete_sink.__set_output_path("s3://b/db/t/data");
+ t_data_sink.iceberg_delete_sink.__set_table_location("oss://b/db/t/data/");
+ VExprContextSPtrs output_exprs;
+ auto sink = std::make_shared<VIcebergDeleteSink>(t_data_sink,
output_exprs, nullptr, nullptr);
+ ObjectPool pool;
+ ASSERT_TRUE(sink->init_properties(&pool).ok());
+ ASSERT_TRUE(sink->_output_location.has_value());
+ EXPECT_EQ("s3://b/db/t/data/delete_dv_1.puffin",
+ sink->_output_location->write_path("delete_dv_1.puffin"));
+ EXPECT_EQ("oss://b/db/t/data/delete_dv_1.puffin",
+ sink->_output_location->commit_path("delete_dv_1.puffin"));
+
+ // Without the table's own form the commit could only record the s3://
address BE wrote.
+ t_data_sink.iceberg_delete_sink.__isset.table_location = false;
+ t_data_sink.iceberg_delete_sink.table_location.clear();
+ auto no_table_location =
+ std::make_shared<VIcebergDeleteSink>(t_data_sink, output_exprs,
nullptr, nullptr);
+ Status status = no_table_location->init_properties(&pool);
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>()) << status;
+ EXPECT_NE(std::string::npos, status.to_string().find("table_location"));
+
+ t_data_sink.iceberg_delete_sink.__set_table_location("oss://b/db/t/data/");
+ t_data_sink.iceberg_delete_sink.__isset.output_path = false;
+ t_data_sink.iceberg_delete_sink.output_path.clear();
+ auto no_output_path =
+ std::make_shared<VIcebergDeleteSink>(t_data_sink, output_exprs,
nullptr, nullptr);
+ status = no_output_path->init_properties(&pool);
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>()) << status;
+ EXPECT_NE(std::string::npos, status.to_string().find("output_path"));
}
TEST_F(VIcebergDeleteSinkTest, TestUnsupportedDeleteType) {
@@ -533,12 +569,88 @@ TEST_F(VIcebergDeleteSinkTest,
TestValidateDeletionVectorContentSize) {
ASSERT_TRUE(unsupported_status.is<ErrorCode::NOT_IMPLEMENTED_ERROR>()) <<
unsupported_status;
}
+TEST_F(VIcebergDeleteSinkTest, TestWritePositionDeletesCommitsTableLocation) {
+ // The Parquet writer's dictionary encoder allocates from ExecEnv's Arrow
pool, which this
+ // fixture does not otherwise initialize.
+ auto* env = ExecEnv::GetInstance();
+ auto* previous_pool = env->_arrow_memory_pool;
+ env->_arrow_memory_pool = arrow::default_memory_pool();
+ Defer restore_pool([&] { env->_arrow_memory_pool = previous_pool; });
+
+ std::filesystem::path temp_dir = std::filesystem::temp_directory_path() /
+ ("iceberg_delete_sink_test_" +
generate_uuid_string());
+ ASSERT_TRUE(std::filesystem::create_directories(temp_dir));
+
+ // Write locally, but record the table's original URI in the commit
metadata.
+ const std::string table_location = "oss://bucket/warehouse/table/data/";
+ const std::string data_file_path = table_location + "file1.parquet";
+ TDataSink t_data_sink = build_local_delete_sink(temp_dir.string(), 2);
+ t_data_sink.iceberg_delete_sink.__set_table_location(table_location);
+ RuntimeState state;
+ VExprContextSPtrs output_exprs;
+ auto sink = std::make_shared<VIcebergDeleteSink>(t_data_sink,
output_exprs, nullptr, nullptr);
+ ObjectPool pool;
+ ASSERT_TRUE(sink->init_properties(&pool).ok());
+ sink->_state = &state;
+
+ std::map<std::string, IcebergFileDeletion> file_deletions;
+ auto [it, inserted] =
+ file_deletions.emplace(data_file_path, IcebergFileDeletion(1,
"[\"p=1\"]"));
+ ASSERT_TRUE(inserted);
+ it->second.rows_to_delete.add(uint32_t {10});
+ it->second.rows_to_delete.add(uint32_t {20});
+
+ const auto status = sink->_write_position_delete_files(file_deletions);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(1, sink->_commit_data_list.size());
+ ASSERT_EQ(1, sink->_created_files.size());
+ ASSERT_EQ(1, sink->_delete_file_count);
+
+ const std::filesystem::path write_path = sink->_created_files[0].second;
+ ASSERT_EQ(temp_dir, write_path.parent_path());
+ ASSERT_TRUE(std::filesystem::exists(write_path));
+ const auto& commit_data = sink->_commit_data_list[0];
+ EXPECT_EQ(table_location + write_path.filename().string(),
commit_data.file_path);
+ EXPECT_EQ(data_file_path, commit_data.referenced_data_file_path);
+ EXPECT_EQ(TFileContent::POSITION_DELETES, commit_data.file_content);
+ EXPECT_EQ(2, commit_data.row_count);
+ EXPECT_EQ(2, commit_data.affected_rows);
+ EXPECT_EQ(1, commit_data.partition_spec_id);
+ EXPECT_EQ("[\"p=1\"]", commit_data.partition_data_json);
+ EXPECT_EQ(std::filesystem::file_size(write_path), commit_data.file_size);
+
+ // The referenced data-file URI inside the position-delete file must stay
unchanged too.
+ auto input = arrow::io::ReadableFile::Open(write_path.string());
+ ASSERT_TRUE(input.ok()) << input.status();
+ auto reader = ::parquet::arrow::OpenFile(*input,
arrow::default_memory_pool());
+ ASSERT_TRUE(reader.ok()) << reader.status();
+ auto table = (*reader)->ReadTable();
+ ASSERT_TRUE(table.ok()) << table.status();
+ ASSERT_EQ(2, (*table)->num_rows());
+ ASSERT_EQ(2, (*table)->num_columns());
+ for (int64_t row = 0; row < 2; ++row) {
+ auto file_path = (*table)->column(0)->GetScalar(row);
+ ASSERT_TRUE(file_path.ok()) << file_path.status();
+ EXPECT_EQ(data_file_path, (*file_path)->ToString());
+ auto position = (*table)->column(1)->GetScalar(row);
+ ASSERT_TRUE(position.ok()) << position.status();
+ EXPECT_EQ(std::to_string((row + 1) * 10), (*position)->ToString());
+ }
+
+ std::filesystem::remove_all(temp_dir);
+}
+
TEST_F(VIcebergDeleteSinkTest, TestWriteDeletionVectorsToSingleSharedPuffin) {
std::filesystem::path temp_dir = std::filesystem::temp_directory_path() /
("iceberg_delete_sink_test_" +
generate_uuid_string());
ASSERT_TRUE(std::filesystem::create_directories(temp_dir));
- TDataSink t_data_sink = build_local_delete_sink(temp_dir.string(), 3);
+ const std::string table_location =
+
"abfss://[email protected]/warehouse/table/data";
+ const std::string first_data_file = table_location + "/file1.parquet";
+ const std::string second_data_file = table_location + "/file2.parquet";
+ TDataSink t_data_sink = build_local_delete_sink(temp_dir.string() + "/",
3);
+ t_data_sink.iceberg_delete_sink.__set_table_location(table_location);
VExprContextSPtrs output_exprs;
auto sink = std::make_shared<VIcebergDeleteSink>(t_data_sink,
output_exprs, nullptr, nullptr);
ObjectPool pool;
@@ -546,23 +658,38 @@ TEST_F(VIcebergDeleteSinkTest,
TestWriteDeletionVectorsToSingleSharedPuffin) {
std::map<std::string, IcebergFileDeletion> file_deletions;
auto [file1_it, file1_inserted] =
- file_deletions.emplace("file1.parquet", IcebergFileDeletion(1,
"[\"p=1\"]"));
+ file_deletions.emplace(first_data_file, IcebergFileDeletion(1,
"[\"p=1\"]"));
ASSERT_TRUE(file1_inserted);
file1_it->second.rows_to_delete.add((uint32_t)10);
file1_it->second.rows_to_delete.add((uint32_t)20);
auto [file2_it, file2_inserted] =
- file_deletions.emplace("file2.parquet", IcebergFileDeletion(2,
"[\"p=2\"]"));
+ file_deletions.emplace(second_data_file, IcebergFileDeletion(2,
"[\"p=2\"]"));
ASSERT_TRUE(file2_inserted);
file2_it->second.rows_to_delete.add((uint32_t)30);
- ASSERT_TRUE(sink->_write_deletion_vector_files(file_deletions).ok());
+ const auto status = sink->_write_deletion_vector_files(file_deletions);
+ ASSERT_TRUE(status.ok()) << status;
ASSERT_EQ(2, sink->_commit_data_list.size());
ASSERT_EQ(2, sink->_delete_file_count);
+ ASSERT_EQ(1, sink->_created_files.size());
+
+ const std::filesystem::path write_path = sink->_created_files[0].second;
+ ASSERT_EQ(temp_dir, write_path.parent_path());
+ ASSERT_TRUE(std::filesystem::exists(write_path));
const auto& first_commit = sink->_commit_data_list[0];
const auto& second_commit = sink->_commit_data_list[1];
+ EXPECT_EQ(table_location + "/" + write_path.filename().string(),
first_commit.file_path);
ASSERT_EQ(first_commit.file_path, second_commit.file_path);
+ EXPECT_EQ(first_data_file, first_commit.referenced_data_file_path);
+ EXPECT_EQ(second_data_file, second_commit.referenced_data_file_path);
+ EXPECT_EQ(TFileContent::DELETION_VECTOR, first_commit.file_content);
+ EXPECT_EQ(TFileContent::DELETION_VECTOR, second_commit.file_content);
+ EXPECT_EQ(1, first_commit.partition_spec_id);
+ EXPECT_EQ(2, second_commit.partition_spec_id);
+ EXPECT_EQ("[\"p=1\"]", first_commit.partition_data_json);
+ EXPECT_EQ("[\"p=2\"]", second_commit.partition_data_json);
ASSERT_EQ(first_commit.file_size, second_commit.file_size);
ASSERT_LT(first_commit.content_offset, second_commit.content_offset);
@@ -574,7 +701,7 @@ TEST_F(VIcebergDeleteSinkTest,
TestWriteDeletionVectorsToSingleSharedPuffin) {
}
ASSERT_EQ(1, puffin_file_count);
- std::ifstream input(first_commit.file_path, std::ios::binary);
+ std::ifstream input(write_path, std::ios::binary);
ASSERT_TRUE(input.good());
std::string file_bytes((std::istreambuf_iterator<char>(input)),
std::istreambuf_iterator<char>());
diff --git a/be/test/exec/sink/viceberg_merge_sink_test.cpp
b/be/test/exec/sink/viceberg_merge_sink_test.cpp
index 991f7bc2650..a262e06d8f2 100644
--- a/be/test/exec/sink/viceberg_merge_sink_test.cpp
+++ b/be/test/exec/sink/viceberg_merge_sink_test.cpp
@@ -211,6 +211,44 @@ TEST_F(VIcebergMergeSinkTest,
TestUpdateProducesDeleteAndInsert) {
ASSERT_TRUE(sink->close(Status::OK()).ok());
}
+TEST_F(VIcebergMergeSinkTest,
TestInnerSinksWriteAndCommitUnderBothFormsOfTheDataLocation) {
+ ObjectPool pool;
+ IcebergWriteMockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ // UPDATE / MERGE write data files and delete files through one normalized
location, and both
+ // kinds must be committed under the table's own form of it.
+ TDataSink t_sink = build_sink();
+ t_sink.iceberg_merge_sink.__set_output_path("s3://b/db/t/data");
+ t_sink.iceberg_merge_sink.__set_original_output_path("oss://b/db/t/data");
+ t_sink.iceberg_merge_sink.__set_table_location("oss://b/db/t/data");
+
+ auto sink = std::make_shared<VIcebergMergeSink>(t_sink, output_exprs,
nullptr, nullptr);
+ sink->set_skip_io(true);
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("iceberg_merge_sink_locations");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+
+ ASSERT_TRUE(sink->_table_writer->_output_location.has_value());
+ EXPECT_EQ("s3://b/db/t/data/f.parquet",
+ sink->_table_writer->_output_location->write_path("f.parquet"));
+ EXPECT_EQ("oss://b/db/t/data/f.parquet",
+ sink->_table_writer->_output_location->commit_path("f.parquet"));
+ ASSERT_TRUE(sink->_delete_writer->_output_location.has_value());
+ EXPECT_EQ("s3://b/db/t/data/f.parquet",
+ sink->_delete_writer->_output_location->write_path("f.parquet"));
+ EXPECT_EQ("oss://b/db/t/data/f.parquet",
+
sink->_delete_writer->_output_location->commit_path("f.parquet"));
+
+ ASSERT_TRUE(sink->close(Status::OK()).ok());
+}
+
TEST_F(VIcebergMergeSinkTest, TestDeleteOnlySkipsVariantDataWriter) {
ObjectPool pool;
// Opening the retained delete writer still requires the coordinator
ownership handshake.
diff --git a/be/test/exec/sink/writer/iceberg/iceberg_output_location_test.cpp
b/be/test/exec/sink/writer/iceberg/iceberg_output_location_test.cpp
new file mode 100644
index 00000000000..1abd5d71481
--- /dev/null
+++ b/be/test/exec/sink/writer/iceberg/iceberg_output_location_test.cpp
@@ -0,0 +1,56 @@
+// 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_output_location.h"
+
+#include <gtest/gtest.h>
+
+namespace doris {
+
+TEST(IcebergOutputLocationTest,
WritesUnderTheNormalizedFormAndCommitsUnderTheTableForm) {
+ IcebergOutputLocation location("s3://b/db/t/data", "oss://b/db/t/data");
+
+ EXPECT_EQ("s3://b/db/t/data/delete_pos_1.parquet",
location.write_path("delete_pos_1.parquet"));
+ EXPECT_EQ("oss://b/db/t/data/delete_pos_1.parquet",
+ location.commit_path("delete_pos_1.parquet"));
+}
+
+TEST(IcebergOutputLocationTest,
JoinsDirectoriesWithAndWithoutTrailingSlashOnce) {
+ // write.data.path is taken verbatim, so either form may end with '/'.
+ IcebergOutputLocation location("s3://b/db/t/data/", "oss://b/db/t/data");
+
+ EXPECT_EQ("s3://b/db/t/data/f.parquet", location.write_path("f.parquet"));
+ EXPECT_EQ("oss://b/db/t/data/f.parquet",
location.commit_path("f.parquet"));
+
+ IcebergOutputLocation partition = location.subdirectory("p=1/q=2");
+ EXPECT_EQ("s3://b/db/t/data/p=1/q=2/f.parquet",
partition.write_path("f.parquet"));
+ EXPECT_EQ("oss://b/db/t/data/p=1/q=2/f.parquet",
partition.commit_path("f.parquet"));
+}
+
+TEST(IcebergOutputLocationTest,
KeepsTheTableFormWhenNormalizationRewritesTheAuthority) {
+ // FE rewrites more than the scheme for Azure: the account moves out of
the authority.
+ IcebergOutputLocation location(
+ "s3://container/warehouse/t/data",
+ "abfss://[email protected]/warehouse/t/data");
+
+ IcebergOutputLocation partition = location.subdirectory("dt=2024-01-01");
+ EXPECT_EQ("s3://container/warehouse/t/data/dt=2024-01-01/f.orc",
partition.write_path("f.orc"));
+
EXPECT_EQ("abfss://[email protected]/warehouse/t/data/dt=2024-01-01/f.orc",
+ partition.commit_path("f.orc"));
+}
+
+} // namespace doris
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 590e1953c67..ba3f13b6086 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
@@ -82,9 +82,12 @@ protected:
const TDataSink& sink, const VExprContextSPtrs& output_exprs,
const iceberg::Schema& schema, const std::string* schema_json,
const std::map<std::string, std::string>& hadoop_conf,
- TFileFormatType::type format = TFileFormatType::FORMAT_ORC) {
- IPartitionWriterBase::WriteInfo write_info;
- write_info.file_type = TFileType::FILE_LOCAL;
+ TFileFormatType::type format = TFileFormatType::FORMAT_ORC,
+ IcebergOutputLocation location = IcebergOutputLocation(
+ "/tmp/iceberg_partition_writer_test",
"/tmp/iceberg_partition_writer_test")) {
+ IPartitionWriterBase::WriteInfo write_info {.location =
std::move(location),
+ .file_type =
TFileType::FILE_LOCAL,
+ .broker_addresses = {}};
return std::make_unique<VIcebergPartitionWriter>(
sink, std::vector<std::string> {}, output_exprs, schema,
schema_json,
std::vector<std::string> {}, std::move(write_info), "data", 0,
format,
@@ -128,6 +131,24 @@ TEST_F(VIcebergPartitionWriterTest,
OrcSkipsFooterCollectionWhenMetricsAreDisabl
EXPECT_FALSE(commit_data.__isset.column_stats);
}
+TEST_F(VIcebergPartitionWriterTest,
CommitDataRecordsTheTableFormOfTheLocation) {
+ // The file is written through FE's normalized form; the commit must keep
the table's own.
+ VExprContextSPtrs output_exprs;
+ iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+ std::string schema_json;
+ std::map<std::string, std::string> hadoop_conf;
+ auto writer =
+ make_writer(make_table_sink(false), output_exprs, schema,
&schema_json, hadoop_conf,
+ TFileFormatType::FORMAT_ORC,
+ IcebergOutputLocation("s3://b/db/t/data/p=1",
"oss://b/db/t/data/p=1"));
+ install_fake_transformer(writer.get(), output_exprs);
+
+ TIcebergCommitData commit_data;
+ ASSERT_TRUE(build_commit_data(writer.get(), &commit_data).ok());
+ EXPECT_EQ("oss://b/db/t/data/p=1/data-0.zlib.orc", commit_data.file_path);
+ EXPECT_EQ(TFileContent::DATA, commit_data.file_content);
+}
+
TEST_F(VIcebergPartitionWriterTest,
ParquetKeepsSplitOffsetsWhenColumnMetricsAreDisabled) {
TimezoneUtils::load_timezones_to_cache();
auto* env = ExecEnv::GetInstance();
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 2f136adcdae..41a126a9cbf 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
@@ -32,6 +32,7 @@
#include "core/data_type/data_type_timestamptz.h"
#include "core/data_type/data_type_varbinary.h"
#include "exec/sink/writer/iceberg/partition_transformers.h"
+#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
#include "exec/sink/writer/iceberg/vpartition_writer_base.h"
#include "exprs/vexpr_context.h"
#include "exprs/vslot_ref.h"
@@ -327,6 +328,65 @@ TEST(VIcebergTableWriterTest,
BinaryIdentityStaticAndDynamicRouting) {
EXPECT_EQ(std::vector<std::string>({"0xdead", "7"}),
writer._partition_values(hybrid));
}
+TEST(VIcebergTableWriterTest,
OpenRejectsASinkWithoutTheTableFormOfTheDataLocation) {
+ // Without original_output_path the commit used to record relative paths
such as
+ // /p=1/xxx.parquet; now the write fails before any file is created.
+ TIcebergTableSink sink;
+ sink.__set_output_path("s3://b/db/t/data");
+ TDataSink data_sink;
+ data_sink.__set_iceberg_table_sink(sink);
+ VIcebergTableWriter writer(data_sink, {});
+ RuntimeState state;
+ TQueryOptions query_options;
+ query_options.__set_supports_external_file_report_ack(true);
+ state.set_query_options(query_options);
+ RuntimeProfile profile("test");
+
+ Status status = writer.open(&state, &profile);
+
+ EXPECT_TRUE(status.is<ErrorCode::INTERNAL_ERROR>()) << status;
+ EXPECT_NE(std::string::npos,
status.to_string().find("original_output_path"));
+ EXPECT_FALSE(writer._output_location.has_value());
+}
+
+TEST(VIcebergTableWriterTest,
PartitionWritersWriteAndCommitUnderBothFormsOfTheDataLocation) {
+ // FE normalizes the location BE writes to (oss:// becomes s3://); the
commit must keep the
+ // table's own form in the data location and in every partition directory
under it.
+ std::vector<iceberg::NestedField> columns;
+ columns.emplace_back(false, 1, "p",
std::make_unique<iceberg::IntegerType>(), std::nullopt);
+ TIcebergTableSink sink;
+ sink.__set_output_path("s3://b/db/t/data");
+ sink.__set_original_output_path("oss://b/db/t/data/");
+ sink.__set_file_format(TFileFormatType::FORMAT_PARQUET);
+ sink.__set_compression_type(TFileCompressType::SNAPPYBLOCK);
+ TDataSink data_sink;
+ data_sink.__set_iceberg_table_sink(sink);
+ VIcebergTableWriter writer(data_sink, {});
+ writer._schema = std::make_shared<iceberg::Schema>(std::move(columns));
+ // What open() derives from the two location fields.
+ writer._output_location.emplace(sink.output_path,
sink.original_output_path);
+
+ const std::string file_name = "f";
+ auto expect_paths = [&](const std::string& partition_dir) {
+ auto writer_base = writer._create_partition_writer(nullptr, -1,
&file_name, 0);
+ auto* partition_writer =
dynamic_cast<VIcebergPartitionWriter*>(writer_base.get());
+ ASSERT_NE(nullptr, partition_writer);
+ const auto& location = partition_writer->_write_info.location;
+ const std::string target_file =
partition_writer->_get_target_file_name();
+ EXPECT_EQ("s3://b/db/t/data/" + partition_dir + target_file,
+ location.write_path(target_file));
+ EXPECT_EQ("oss://b/db/t/data/" + partition_dir + target_file,
+ location.commit_path(target_file));
+ };
+ // A non-partitioned table writes into the data location itself.
+ expect_paths("");
+ // A partition writes into its directory under the data location.
+ writer._is_full_static_partition = true;
+ writer._static_partition_path = "p=1";
+ writer._static_partition_value_list = {"1"};
+ expect_paths("p=1/");
+}
+
TEST(VIcebergTableWriterTest, RejectMissingPartitionSource) {
std::vector<iceberg::NestedField> columns;
columns.emplace_back(false, 3, "id",
std::make_unique<iceberg::IntegerType>(), std::nullopt);
diff --git a/be/test/format/transformer/vparquet_writer_test.cpp
b/be/test/format/transformer/vparquet_writer_test.cpp
index 6591ca24f57..1c53c4b6916 100644
--- a/be/test/format/transformer/vparquet_writer_test.cpp
+++ b/be/test/format/transformer/vparquet_writer_test.cpp
@@ -594,9 +594,7 @@ TEST_F(VParquetWriterTest,
IcebergNanosecondsRoundTripWithoutCoercion) {
const auto name = "iceberg_ns_" + UniqueId::gen_uid().to_string();
_file_path = "./" + name + "-0.parquet";
VIcebergPartitionWriter writer(sink, {}, exprs, *schema, &json, {"ts"},
- {.write_path = ".",
- .original_write_path = ".",
- .target_path = ".",
+ {.location = IcebergOutputLocation(".",
"."),
.file_type = TFileType::FILE_LOCAL,
.broker_addresses = {}},
name, 0, TFileFormatType::FORMAT_PARQUET,
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 ad17c6d353e..b6ca61f312f 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
@@ -144,6 +144,10 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
private final IcebergCatalogResourceTracker resourceTracker;
private final String catalogFlavor;
private final List<TIcebergCommitData> commitDataList = new ArrayList<>();
+ // The raw data locations the sinks of this write commit their files
under, registered by planWrite for
+ // every sink it builds (several for a rewrite_data_files whose groups
each plan one). Guarded by this,
+ // like commitDataList.
+ private final Set<String> commitLocations = new HashSet<>();
// The single SDK transaction / table, opened lazily by beginWrite (the
write plan binds the target
// table only when the sink is planned). volatile: addCommitData / commit
may run on different threads.
@@ -626,10 +630,46 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
throw new DorisConnectorException("failed to deserialize Iceberg
commit data", e);
}
synchronized (this) {
+ checkCommitLocation(data);
commitDataList.add(data);
}
}
+ /**
+ * Registers a raw data location a sink of this write carries ({@code
TIcebergTableSink.original_output_path}
+ * or {@code TIcebergDeleteSink.table_location}): BE writes the sink's
files under the normalized form of it
+ * and must report them under it. Called by planWrite for every sink it
builds.
+ */
+ void addCommitLocation(String location) {
+ if (location == null || location.isEmpty()) {
+ throw new DorisConnectorException("An iceberg sink must carry the
raw location it commits files under");
+ }
+ synchronized (this) {
+ commitLocations.add(location);
+ }
+ }
+
+ /**
+ * Rejects a fragment whose file is not recorded under a location a sink
of this write carries. BE writes
+ * through the normalized location (s3://b/db/t/data for an oss:// table)
and must report the table's own
+ * form (oss://b/db/t/data/x.parquet): Iceberg metadata, other engines and
{@link #rollback} resolve the
+ * reported path with the table's file system, so the s3:// form names a
file no other engine can open, and
+ * nothing failed when delete files were committed that way. Throwing here
makes FE reject the report: the
+ * statement fails and BE deletes the files under the paths it wrote them
at.
+ */
+ private void checkCommitLocation(TIcebergCommitData data) {
+ String path = data.getFilePath();
+ if (path != null) {
+ for (String location : commitLocations) {
+ if (path.startsWith(location.endsWith("/") ? location :
location + "/")) {
+ return;
+ }
+ }
+ }
+ throw new DorisConnectorException("Iceberg commit data reports file "
+ path
+ + " outside the locations its sinks commit files under: " +
commitLocations);
+ }
+
/**
* O5-2: stashes the engine-extracted target-only write constraint
(neutral form). It is converted to an
* iceberg {@link Expression} lazily at commit time ({@link
#buildWriteConstraintExpression}) — the table
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
index 80a5e3e429c..53794216318 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
@@ -236,21 +236,32 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
Map<String, List<TIcebergDeleteFileDesc>> rewritableDeletes =
IcebergStatementScope.rewritableDeleteSupply(session);
+ TDataSink dataSink = buildDataSink(writeContext.getWriteOperation(),
table, tableHandle, handle,
+ rewritableDeletes, schemaContext);
+ // BE reports every file of the sink under the raw location the sink
carries, whichever writer
+ // produced it; the transaction rejects a reported file anywhere else.
+ commitLocationsOf(dataSink).forEach(transaction::addCommitLocation);
+ return new ConnectorSinkPlan(dataSink);
+ }
+
+ private TDataSink buildDataSink(WriteOperation writeOperation, Table
table, IcebergTableHandle tableHandle,
+ ConnectorWriteHandle handle, Map<String,
List<TIcebergDeleteFileDesc>> rewritableDeletes,
+ IcebergWriteSchemaContext schemaContext) {
// Dispatch on the write operation to the matching BE sink dialect
(each is a distinct TDataSinkType,
// byte-identical to the legacy fe-core planner sink). OVERWRITE
shares TIcebergTableSink with INSERT
// (the overwrite flag is read from the handle); UPDATE shares
TIcebergMergeSink with MERGE.
- switch (writeContext.getWriteOperation()) {
+ switch (writeOperation) {
case INSERT:
case OVERWRITE: {
TDataSink dataSink = new
TDataSink(TDataSinkType.ICEBERG_TABLE_SINK);
dataSink.setIcebergTableSink(buildSink(table, tableHandle,
handle, schemaContext));
- return new ConnectorSinkPlan(dataSink);
+ return dataSink;
}
case DELETE: {
TDataSink dataSink = new
TDataSink(TDataSinkType.ICEBERG_DELETE_SINK);
dataSink.setIcebergDeleteSink(
buildDeleteSink(table, tableHandle, rewritableDeletes,
schemaContext));
- return new ConnectorSinkPlan(dataSink);
+ return dataSink;
}
case UPDATE:
case MERGE: {
@@ -260,7 +271,7 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
handle.getBoundTargetColumns().stream()
.anyMatch(column ->
containsVariant(column.getType())),
schemaContext));
- return new ConnectorSinkPlan(dataSink);
+ return dataSink;
}
case REWRITE: {
// Compaction rewrite (ALTER TABLE ... EXECUTE
rewrite_data_files): same TIcebergTableSink
@@ -269,11 +280,29 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
// is already evicted above.
TDataSink dataSink = new
TDataSink(TDataSinkType.ICEBERG_TABLE_SINK);
dataSink.setIcebergTableSink(buildRewriteSink(table,
tableHandle, handle));
- return new ConnectorSinkPlan(dataSink);
+ return dataSink;
}
default:
- throw new DorisConnectorException(
- "Unsupported iceberg write operation: " +
writeContext.getWriteOperation());
+ throw new DorisConnectorException("Unsupported iceberg write
operation: " + writeOperation);
+ }
+ }
+
+ /**
+ * The raw data locations BE commits the files of {@code sink} under, read
from the field each dialect
+ * carries them in: data files under {@code original_output_path}, delete
files and deletion vectors under
+ * {@code table_location}. The merge sink writes both kinds and carries
both fields.
+ */
+ private static List<String> commitLocationsOf(TDataSink sink) {
+ switch (sink.getType()) {
+ case ICEBERG_TABLE_SINK:
+ return
Collections.singletonList(sink.getIcebergTableSink().getOriginalOutputPath());
+ case ICEBERG_DELETE_SINK:
+ return
Collections.singletonList(sink.getIcebergDeleteSink().getTableLocation());
+ case ICEBERG_MERGE_SINK:
+ return
Arrays.asList(sink.getIcebergMergeSink().getOriginalOutputPath(),
+ sink.getIcebergMergeSink().getTableLocation());
+ default:
+ throw new DorisConnectorException("Unsupported iceberg sink
type: " + sink.getType());
}
}
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 ba5ce1cc586..771588ecca5 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
@@ -131,8 +131,14 @@ public class IcebergConnectorTransactionTest {
return ops;
}
+ // Stands in for the raw location planWrite registers for the sink it
builds; every fragment fed to a
+ // transaction from txnFor lives under it.
+ private static final String COMMIT_LOCATION = "s3://b";
+
private static IcebergConnectorTransaction
txnFor(RecordingIcebergCatalogOps ops, RecordingConnectorContext ctx) {
- return new IcebergConnectorTransaction(42L, ops, ctx);
+ IcebergConnectorTransaction txn = new IcebergConnectorTransaction(42L,
ops, ctx);
+ txn.addCommitLocation(COMMIT_LOCATION);
+ return txn;
}
// The overwrite tests below predate the static default
(iceberg_partition_overwrite_mode defaults to
@@ -249,6 +255,7 @@ public class IcebergConnectorTransactionTest {
private static TIcebergCommitData dataItem(long affectedRows, long
rowCount, TFileContent content) {
TIcebergCommitData d = new TIcebergCommitData();
+ d.setFilePath(COMMIT_LOCATION + "/db/t/f.parquet");
d.setRowCount(rowCount);
if (affectedRows >= 0) {
d.setAffectedRows(affectedRows);
@@ -396,6 +403,63 @@ public class IcebergConnectorTransactionTest {
() -> txn.addCommitData(new byte[] {(byte) 0xDE, (byte) 0xAD,
(byte) 0xBE, (byte) 0xEF}));
}
+ // ─────────────────── addCommitData: files must be reported under a
location a sink carries ───────────────────
+
+ private static IcebergConnectorTransaction txnCommittingUnder(String...
locations) {
+ IcebergConnectorTransaction txn =
+ new IcebergConnectorTransaction(42L, opsReturning(null), new
RecordingConnectorContext());
+ for (String location : locations) {
+ txn.addCommitLocation(location);
+ }
+ return txn;
+ }
+
+ @Test
+ public void addCommitDataAcceptsFilesUnderEveryRegisteredLocation() {
+ // One location without and one with a trailing '/': write.data.path
is taken verbatim.
+ IcebergConnectorTransaction txn =
txnCommittingUnder("oss://b/db/t/data", "oss://b/db/t2/data/");
+
txn.addCommitData(commitBytes(dataFileItem("oss://b/db/t/data/id=1/f.parquet",
1L, 64L)));
+ txn.addCommitData(commitBytes(
+ positionDeleteItem("oss://b/db/t2/data/delete_pos_1.parquet",
1L, "oss://b/db/t2/data/f.parquet")));
+ Assertions.assertEquals(2, txn.getCommitDataList().size());
+ }
+
+ @Test
+ public void addCommitDataRejectsTheNormalizedAddressBeWroteTheFileAt() {
+ // BE writes an oss:// table through s3://; reporting that form
records a file no other engine opens.
+ IcebergConnectorTransaction txn =
txnCommittingUnder("oss://b/db/t/data");
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> txn.addCommitData(commitBytes(positionDeleteItem(
+ "s3://b/db/t/data/delete_pos_1.parquet", 1L,
"oss://b/db/t/data/f.parquet"))));
+
Assertions.assertTrue(ex.getMessage().contains("s3://b/db/t/data/delete_pos_1.parquet"),
ex.getMessage());
+ Assertions.assertTrue(ex.getMessage().contains("oss://b/db/t/data"),
ex.getMessage());
+ Assertions.assertTrue(txn.getCommitDataList().isEmpty(), "a rejected
fragment must not be accumulated");
+ }
+
+ @Test
+ public void
addCommitDataRejectsASiblingDirectorySharingTheLocationPrefix() {
+ IcebergConnectorTransaction txn =
txnCommittingUnder("oss://b/db/t/data");
+ Assertions.assertThrows(DorisConnectorException.class,
+ () ->
txn.addCommitData(commitBytes(dataFileItem("oss://b/db/t/data_old/f.parquet",
1L, 64L))));
+ }
+
+ @Test
+ public void
addCommitDataRejectsAFragmentWithoutAPathOrBeforeAnySinkWasPlanned() {
+ IcebergConnectorTransaction planned =
txnCommittingUnder("oss://b/db/t/data");
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> planned.addCommitData(commitBytes(dataItem(1L, 1L,
TFileContent.DATA).setFilePath(null))));
+ IcebergConnectorTransaction unplanned = txnCommittingUnder();
+ Assertions.assertThrows(DorisConnectorException.class,
+ () ->
unplanned.addCommitData(commitBytes(dataFileItem("oss://b/db/t/data/f.parquet",
1L, 64L))));
+ }
+
+ @Test
+ public void addCommitLocationRejectsAnEmptyLocation() {
+ IcebergConnectorTransaction txn = txnCommittingUnder();
+ Assertions.assertThrows(DorisConnectorException.class, () ->
txn.addCommitLocation(""));
+ Assertions.assertThrows(DorisConnectorException.class, () ->
txn.addCommitLocation(null));
+ }
+
// ─────────────────── getUpdateCnt: data/delete split, affectedRows
priority (T03, regression) ───────────────────
@Test
@@ -1180,7 +1244,7 @@ public class IcebergConnectorTransactionTest {
long snapshotId = table.currentSnapshot().snapshotId();
IcebergConnectorTransaction txn = txnFor(opsReturning(table), new
RecordingConnectorContext());
txn.beginWrite(SESSION, "db1", "t1", overwriteStaticCtx(table,
Collections.singletonMap("ts", "2024-01-02")));
-
txn.addCommitData(commitBytes(dataFileItem("s3://bucket/replacement.parquet",
1L, 100L,
+
txn.addCommitData(commitBytes(dataFileItem("s3://b/replacement.parquet", 1L,
100L,
Collections.singletonList("2024-01-02 00:00:00.000000000"))));
// DATE metrics are days, so silently committing a nanos equality
would retain the old file.
DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class, txn::commit);
@@ -1206,7 +1270,7 @@ public class IcebergConnectorTransactionTest {
}
IcebergConnectorTransaction txn = txnFor(opsReturning(table), new
RecordingConnectorContext());
txn.beginWrite(SESSION, "db1", "t1", overwriteStaticCtx(table,
Collections.singletonMap("ts", "2024-01-01")));
-
txn.addCommitData(commitBytes(dataFileItem("s3://bucket/replacement.parquet",
1L, 1024L,
+
txn.addCommitData(commitBytes(dataFileItem("s3://b/replacement.parquet", 1L,
1024L,
Collections.singletonList("2024-01-01 00:00:00.000000000"))));
txn.commit();
List<DataFile> remaining = currentDataFiles(catalog.loadTable(id));
@@ -1233,7 +1297,7 @@ public class IcebergConnectorTransactionTest {
IcebergConnectorTransaction txn = txnFor(opsReturning(table), new
RecordingConnectorContext());
txn.beginWrite(SESSION, "db1", "t1",
overwriteStaticCtx(table, Collections.singletonMap("ts",
"2024-01-01 00:00:00+08:00")));
-
txn.addCommitData(commitBytes(dataFileItem("s3://bucket/replacement.parquet",
1L, 1024L,
+
txn.addCommitData(commitBytes(dataFileItem("s3://b/replacement.parquet", 1L,
1024L,
Collections.singletonList("2023-12-31 16:00:00.000000000"))));
txn.commit();
List<DataFile> remaining = currentDataFiles(catalog.loadTable(id));
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java
index f18a79a4e95..a1ae89bc1ef 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java
@@ -45,6 +45,9 @@ public class IcebergWriteKnownBugsTest {
private static final String SEED = "s3://bucket/events/seed.parquet";
private static final String ADDED = "s3://bucket/events/added.parquet";
private static final String DELETED = "s3://bucket/events/deleted.parquet";
+ // Stands in for the raw location planWrite registers for the sink it
builds; every reported file
+ // above lives under it.
+ private static final String COMMIT_LOCATION = "s3://bucket/events";
@Test
@Disabled("Known issue: DELETE ignores the standard snapshot isolation
property.")
@@ -113,7 +116,10 @@ public class IcebergWriteKnownBugsTest {
private static IcebergConnectorTransaction transaction(Table table) {
RecordingIcebergCatalogOps operations = new
RecordingIcebergCatalogOps();
operations.table = table;
- return new IcebergConnectorTransaction(9, operations, new
RecordingConnectorContext());
+ IcebergConnectorTransaction transaction =
+ new IcebergConnectorTransaction(9, operations, new
RecordingConnectorContext());
+ transaction.addCommitLocation(COMMIT_LOCATION);
+ return transaction;
}
private static TIcebergCommitData fragment(String path, long count,
TFileContent content) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
index b0eb9bee169..26ae3e6fd06 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
@@ -44,6 +44,7 @@ import org.apache.doris.thrift.TFileCompressType;
import org.apache.doris.thrift.TFileContent;
import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TFileType;
+import org.apache.doris.thrift.TIcebergCommitData;
import org.apache.doris.thrift.TIcebergDeleteFileDesc;
import org.apache.doris.thrift.TIcebergDeleteSink;
import org.apache.doris.thrift.TIcebergMergeSink;
@@ -83,6 +84,9 @@ import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.DateTimeUtil;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TBinaryProtocol;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -2491,6 +2495,68 @@ public class IcebergWritePlanProviderTest {
Assertions.assertTrue(update.isWritesDataFiles());
}
+ // BE writes position-delete and deletion-vector files under output_path
and commits them under
+ // table_location (VIcebergDeleteSink::_commit_path_of), so both must name
the same directory, the
+ // data location, differing only in its normalized vs raw form. If
table_location became the table
+ // root, BE would record delete files without the data directory, i.e.
files that do not exist,
+ // and nothing would fail. See TIcebergDeleteSink.table_location in
DataSinks.thrift.
+ @Test
+ public void
planWriteDeleteAndMergeSinksKeepTableLocationInTheOutputDirectory() {
+ Table table = partitionedSortedTable(freshCatalog());
+ TIcebergDeleteSink deleteSink = planDeleteSink(table,
contextWithStorage(),
+ new WriteHandle(new IcebergTableHandle("db1",
"t1")).writeOperation(WriteOperation.DELETE));
+ TIcebergMergeSink mergeSink = planMergeSink(table,
contextWithStorage(),
+ new WriteHandle(new IcebergTableHandle("db1",
"t1")).writeOperation(WriteOperation.UPDATE));
+
+ Assertions.assertEquals(pathAfterScheme(deleteSink.getOutputPath()),
+ pathAfterScheme(deleteSink.getTableLocation()),
+ "delete table_location must be the raw form of output_path's
directory");
+ Assertions.assertEquals(pathAfterScheme(mergeSink.getOutputPath()),
+ pathAfterScheme(mergeSink.getTableLocation()),
+ "merge table_location must be the raw form of output_path's
directory");
+ }
+
+ // "s3://bucket/wh/db1/t1/data" and "oss://bucket/wh/db1/t1/data" ->
"bucket/wh/db1/t1/data".
+ private static String pathAfterScheme(String location) {
+ return location.substring(location.indexOf("://") + 3);
+ }
+
+ // planWrite registers the raw location each sink dialect carries with the
transaction, so a file BE
+ // reports under the normalized address it wrote (the s3:// form of this
oss:// table, as delete files once
+ // were) fails the statement instead of being recorded where other engines
cannot resolve it.
+ @Test
+ public void planWriteRegistersTheCommitLocationOfEverySinkDialect() {
+ for (WriteOperation operation : Arrays.asList(WriteOperation.INSERT,
WriteOperation.REWRITE,
+ WriteOperation.DELETE, WriteOperation.UPDATE)) {
+ Table table = partitionedSortedTable(freshCatalog());
+ RecordingConnectorContext ctx = contextWithStorage();
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = table;
+ IcebergConnectorTransaction txn = new
IcebergConnectorTransaction(42L, ops, ctx);
+ providerFor(table, ctx).planWrite(new WriteSession(txn),
+ new WriteHandle(new IcebergTableHandle("db1",
"t1")).writeOperation(operation));
+
+
txn.addCommitData(commitFragment("oss://bucket/wh/db1/t1/data/id=1/f.parquet"));
+ DorisConnectorException rejected =
Assertions.assertThrows(DorisConnectorException.class,
+ () ->
txn.addCommitData(commitFragment("s3://bucket/wh/db1/t1/data/id=1/f.parquet")),
+ operation + " must reject a file reported under the
normalized location");
+
Assertions.assertTrue(rejected.getMessage().contains("s3://bucket/wh/db1/t1/data/id=1/f.parquet"),
+ rejected.getMessage());
+ Assertions.assertEquals(1, txn.getCommitDataList().size(),
operation.toString());
+ }
+ }
+
+ private static byte[] commitFragment(String filePath) {
+ TIcebergCommitData data = new TIcebergCommitData();
+ data.setFilePath(filePath);
+ data.setFileContent(TFileContent.DATA);
+ try {
+ return new TSerializer(new
TBinaryProtocol.Factory()).serialize(data);
+ } catch (TException e) {
+ throw new AssertionError(e);
+ }
+ }
+
@Test
public void planWriteBuildsMergeSinkWithTableDerivedFields() {
Table table = partitionedSortedTable(freshCatalog());
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java
index 83412d30c0c..b29e8e3848d 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java
@@ -58,6 +58,9 @@ public class IcebergWriteTransactionCoverageTest {
private static final String SEED = "s3://bucket/events/seed.parquet";
private static final String ADDED = "s3://bucket/events/added.parquet";
private static final String DELETED = "s3://bucket/events/deleted.parquet";
+ // Stands in for the raw location planWrite registers for the sink it
builds; every reported file
+ // above lives under it.
+ private static final String COMMIT_LOCATION = "s3://bucket/events";
@Test
public void appendV1PreservesHistoricalSnapshot() throws IOException {
@@ -252,6 +255,7 @@ public class IcebergWriteTransactionCoverageTest {
operations.throwOnLoadTable = true;
IcebergConnectorTransaction transaction = new
IcebergConnectorTransaction(
7, operations, new RecordingConnectorContext());
+ transaction.addCommitLocation(COMMIT_LOCATION);
Assertions.assertThrows(DorisConnectorException.class,
() -> transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.INSERT)));
Assertions.assertNull(transaction.getTransaction());
@@ -321,7 +325,10 @@ public class IcebergWriteTransactionCoverageTest {
private static IcebergConnectorTransaction transaction(Table table) {
RecordingIcebergCatalogOps operations = new
RecordingIcebergCatalogOps();
operations.table = table;
- return new IcebergConnectorTransaction(7, operations, new
RecordingConnectorContext());
+ IcebergConnectorTransaction transaction =
+ new IcebergConnectorTransaction(7, operations, new
RecordingConnectorContext());
+ transaction.addCommitLocation(COMMIT_LOCATION);
+ return transaction;
}
private static IcebergWriteContext context(WriteOperation operation) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java
index babfb4b2fcb..06f13007bcb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java
@@ -324,13 +324,15 @@ public final class QeProcessorImpl implements QeProcessor
{
try {
boolean accepted =
info.getCoord().updateFragmentExecStatus(params);
if (hasExternalCommitData && !accepted) {
- return rejectedExternalFileReport(result, "FE has not accepted
the external-file report");
+ return rejectExternalFileReport(info.getCoord(), result,
+ "FE has not accepted the external-file report");
}
} catch (Exception e) {
LOG.warn("Exception during handle report, response: {}, query: {},
instance: {}", result.toString(),
DebugUtil.printId(params.query_id),
DebugUtil.printId(params.fragment_instance_id), e);
return hasExternalCommitData
- ? rejectedExternalFileReport(result, "FE did not accept
the external-file report")
+ ? rejectExternalFileReport(info.getCoord(), result,
+ "FE did not accept the external-file report: " +
e.getMessage())
: result;
}
result.setStatus(new TStatus(TStatusCode.OK));
@@ -354,6 +356,14 @@ public final class QeProcessorImpl implements QeProcessor {
+ params.getFragmentId() + ":" + params.getBackendId();
}
+ // A rejected final report is terminal: BE drops its files and never
resends it, so the statement
+ // must fail now instead of waiting for a fragment that can no longer
finish.
+ private static TReportExecStatusResult rejectExternalFileReport(
+ CoordInterface coord, TReportExecStatusResult result, String
message) {
+ coord.cancel(new Status(TStatusCode.INTERNAL_ERROR, message));
+ return rejectedExternalFileReport(result, message);
+ }
+
private static TReportExecStatusResult rejectedExternalFileReport(
TReportExecStatusResult result, String message) {
TStatus status = new TStatus(TStatusCode.INTERNAL_ERROR);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java
index c4abc116445..103fffa297f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.qe;
+import org.apache.doris.common.Status;
import org.apache.doris.common.profile.ExecutionProfile;
import org.apache.doris.planner.PlanFragmentId;
import org.apache.doris.system.Backend;
@@ -34,6 +35,7 @@ import com.google.common.cache.Cache;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import java.lang.reflect.Field;
@@ -82,17 +84,34 @@ class QeProcessorImplReportAckTest {
Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
result.getStatus().getStatusCode());
Assertions.assertFalse(result.isExternalFileCommitDataAccepted());
+ // BE drops the files of a rejected final report, so the statement
must fail instead of waiting.
+ Status cancelReason = verifyCancelled(coordinator);
+ Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
cancelReason.getErrorCode());
+ Assertions.assertTrue(cancelReason.getErrorMsg().contains("injected
acceptance failure"));
}
@Test
void rejectsExternalReportWhenHandlerDoesNotAcceptIt() throws Exception {
TUniqueId queryId = new TUniqueId(12345, 4);
- register(queryId);
+ Coordinator coordinator = register(queryId);
TReportExecStatusResult result = report(params(queryId));
Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
result.getStatus().getStatusCode());
Assertions.assertFalse(result.isExternalFileCommitDataAccepted());
+ Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
verifyCancelled(coordinator).getErrorCode());
+ }
+
+ @Test
+ void acceptedExternalReportDoesNotCancelCoordinator() throws Exception {
+ TUniqueId queryId = new TUniqueId(12345, 8);
+ Coordinator coordinator = register(queryId);
+
Mockito.when(coordinator.updateFragmentExecStatus(Mockito.any())).thenReturn(true);
+
+ TReportExecStatusResult result = report(params(queryId));
+
+ Assertions.assertTrue(result.isExternalFileCommitDataAccepted());
+ Mockito.verify(coordinator, Mockito.never()).cancel(Mockito.any());
}
@Test
@@ -186,6 +205,12 @@ class QeProcessorImplReportAckTest {
return coordinator;
}
+ private static Status verifyCancelled(Coordinator coordinator) {
+ ArgumentCaptor<Status> reason = ArgumentCaptor.forClass(Status.class);
+ Mockito.verify(coordinator, Mockito.times(1)).cancel(reason.capture());
+ return reason.getValue();
+ }
+
private static TReportExecStatusParams params(TUniqueId queryId) {
return new TReportExecStatusParams()
.setQueryId(queryId)
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index dd0dbabc51a..c4f303c118b 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -439,6 +439,9 @@ struct TIcebergColumnStats {
}
struct TIcebergCommitData {
+ // The path Iceberg metadata records for the file: under the table's own
form of the location
+ // the sink carries (TIcebergTableSink.original_output_path,
TIcebergDeleteSink.table_location),
+ // never under the normalized output_path BE wrote it at. FE rejects a
report outside it.
1: optional string file_path
// File-level record count used to construct Iceberg data/delete file
metadata.
2: optional i64 row_count
@@ -493,6 +496,12 @@ struct TIcebergTableSink {
9: optional map<string, string> hadoop_config
10: optional bool overwrite
11: optional Types.TFileType file_type
+ // The raw (un-normalized) data location: the same directory as
output_path, in the table's
+ // own form. BE writes data files under output_path and commits them under
+ // original_output_path, e.g. written to s3://b/db/t/data/p=1/x.parquet,
committed as
+ // oss://b/db/t/data/p=1/x.parquet. "Same directory" is not "same text
after the scheme":
+ // normalization can also rewrite the authority, e.g.
abfss://[email protected]/t/data
+ // becomes s3://c/t/data.
12: optional string original_output_path
13: optional PlanNodes.TFileCompressType compression_type
14: optional list<Types.TNetworkAddress> broker_addresses;
@@ -530,6 +539,11 @@ struct TIcebergDeleteSink {
5: optional PlanNodes.TFileFormatType file_format
6: optional PlanNodes.TFileCompressType compress_type
7: optional string output_path
+ // The raw (un-normalized) data location: the same directory as
output_path, in the table's
+ // own form. BE writes delete files under output_path and commits them
under table_location,
+ // e.g. written to s3://b/db/t/data/delete_x.parquet, committed as
oss://b/db/t/data/delete_x.parquet.
+ // It must not be the table root, or committed paths would miss the data
directory. See
+ // TIcebergTableSink.original_output_path for what "same directory" means.
8: optional string table_location
9: optional map<string, string> hadoop_config
10: optional Types.TFileType file_type
@@ -554,6 +568,8 @@ struct TIcebergMergeSink {
7: optional PlanNodes.TFileFormatType file_format
8: optional PlanNodes.TFileCompressType compression_type
9: optional string output_path
+ // Same contract as TIcebergTableSink.original_output_path, for the data
files; FE sets it to
+ // the same value as table_location, which the delete files are committed
under.
10: optional string original_output_path
11: optional map<string, string> hadoop_config
12: optional Types.TFileType file_type
@@ -573,6 +589,7 @@ struct TIcebergMergeSink {
// delete side (position delete only)
20: optional TFileContent delete_type
+ // Same contract as TIcebergDeleteSink.table_location: the raw form of
output_path's directory.
21: optional string table_location
22: optional i32 partition_spec_id_for_delete
23: optional string partition_data_json_for_delete
diff --git
a/regression-test/data/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.out.org
b/regression-test/data/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.out.org
deleted file mode 100644
index c42c6138a81..00000000000
---
a/regression-test/data/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.out.org
+++ /dev/null
@@ -1,77 +0,0 @@
--- This file is automatically generated. You should know what you did if you
want to edit this
--- !parquet_q01 --
-1
-
--- !order_parquet_q02 --
-1 Updated 25
-2 Bob 30
-3 Charlie 35
-
--- !parquet_q03 --
-1
-
--- !order_parquet_q04 --
-3 Charlie 35
-1 Updated 25
-
--- !order_parquet_q05 --
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_parquet/data/delete_pos_5066988_0553b2d5.zstd.parquet
PARQUET 0 1 2078 \N \N \N \N \N
\N \N \N \N \N \N \N \N \N
{"age":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"id":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"name":{"column_size":null, [...]
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_parquet/data/delete_pos_5066988_3638addc.zstd.parquet
PARQUET 0 1 2078 \N \N \N \N \N
\N \N \N \N \N \N \N \N \N
{"age":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"id":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"name":{"column_size":null, [...]
-
--- !parquet_q06 --
-1
-
--- !order_parquet_q07 --
-12 Cat 22 2024-01-03
-11 Ben 21 2024-01-02
-10 UpdatedP 20 2024-01-01
-
--- !parquet_q08 --
-1
-
--- !order_parquet_q09 --
-12 Cat 22 2024-01-03
-10 UpdatedP 20 2024-01-01
-
--- !order_parquet_q10 --
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_parquet/data/delete_pos_5092880_cd626510.zstd.parquet
PARQUET 0 {"dt_day":"2024-01-01"} 1 2226 \N \N \N
\N \N \N \N \N \N \N \N \N \N
\N {"age":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, "dt":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, [...]
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_parquet/data/delete_pos_7425526_6dd1173c.zstd.parquet
PARQUET 0 {"dt_day":"2024-01-02"} 1 2225 \N \N \N
\N \N \N \N \N \N \N \N \N \N
\N {"age":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, "dt":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, [...]
-
--- !orc_q01 --
-1
-
--- !order_orc_q02 --
-1 Updated 25
-2 Bob 30
-3 Charlie 35
-
--- !orc_q03 --
-1
-
--- !order_orc_q04 --
-3 Charlie 35
-1 Updated 25
-
--- !order_orc_q05 --
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_orc/data/delete_pos_4451312_c1ffb646.zlib.orc
ORC 0 1 926 \N \N \N \N \N
\N \N \N \N \N \N \N \N \N
{"age":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"id":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"name":{"column_size":null, "value_count" [...]
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_orc/data/delete_pos_4451312_c5f91315.zlib.orc
ORC 0 1 924 \N \N \N \N \N
\N \N \N \N \N \N \N \N \N
{"age":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"id":{"column_size":null, "value_count":null, "null_value_count":null,
"nan_value_count":null, "lower_bound":null, "upper_bound":null},
"name":{"column_size":null, "value_count" [...]
-
--- !orc_q06 --
-1
-
--- !order_orc_q07 --
-11 Ben 21 2024-01-02
-12 Cat 22 2024-01-03
-10 UpdatedP 20 2024-01-01
-
--- !orc_q08 --
-1
-
--- !order_orc_q09 --
-12 Cat 22 2024-01-03
-10 UpdatedP 20 2024-01-01
-
--- !order_orc_q10 --
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_orc/data/delete_pos_7724320_d7587f50.zlib.orc
ORC 0 {"dt_day":"2024-01-01"} 1 986 \N \N \N
\N \N \N \N \N \N \N \N \N \N
\N {"age":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, "dt":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, "id":{"column [...]
-1
s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_orc/data/delete_pos_817404_2d2aa46c.zlib.orc
ORC 0 {"dt_day":"2024-01-02"} 1 990 \N \N \N
\N \N \N \N \N \N \N \N \N \N
\N {"age":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, "dt":{"column_size":null, "value_count":null,
"null_value_count":null, "nan_value_count":null, "lower_bound":null,
"upper_bound":null}, "id":{"column_ [...]
-
diff --git
a/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy
b/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy
index 21384862153..9ef8c30b1c8 100644
---
a/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy
@@ -47,6 +47,15 @@ suite("test_iceberg_update_delete_basic",
"p0,external,iceberg,external_docker,e
sql """create database if not exists ${dbName}"""
sql """use ${dbName}"""
+ // Delete files must be recorded under the table's own data location
(s3a:// in this fixture), not
+ // under the s3:// address BE writes them at.
+ def dataLocationOf = { String table ->
+ String ddl = sql("""show create table ${table}""")[0][1]
+ def matcher = ddl =~ /LOCATION '([^']+)'/
+ assert matcher.find() : "no LOCATION in ${ddl}"
+ return matcher.group(1) + "/data/"
+ }
+
def formats = ["parquet", "orc"]
for (String format : formats) {
logger.info("Run update/delete test with format ${format}")
@@ -94,10 +103,11 @@ suite("test_iceberg_update_delete_basic",
"p0,external,iceberg,external_docker,e
FROM ${catalogName}.${dbName}.${formatTableName}\$delete_files
"""
assert deleteFiles.size() > 0 : "Delete files should be created for
${formatTableName}"
+ String dataLocation = dataLocationOf(formatTableName)
for (def row : deleteFiles) {
String filePath = row[0].toString()
String fileFormat = row[1].toString()
- assert filePath.contains("/data/delete_pos_")
+ assert filePath.startsWith(dataLocation + "delete_pos_") :
"${filePath} is not under ${dataLocation}"
assert filePath.endsWith(format == "parquet" ? ".parquet" : ".orc")
assert fileFormat.equalsIgnoreCase(format)
}
@@ -142,10 +152,12 @@ suite("test_iceberg_update_delete_basic",
"p0,external,iceberg,external_docker,e
FROM
${catalogName}.${dbName}.${formatTableNamePartition}\$delete_files
"""
assert partitionDeleteFiles.size() > 0 : "Delete files should be
created for ${formatTableNamePartition}"
+ String partitionDataLocation = dataLocationOf(formatTableNamePartition)
for (def row : partitionDeleteFiles) {
String filePath = row[0].toString()
String fileFormat = row[1].toString()
- assert filePath.contains("/data/delete_pos_")
+ assert filePath.startsWith(partitionDataLocation + "delete_pos_") :
+ "${filePath} is not under ${partitionDataLocation}"
assert filePath.endsWith(format == "parquet" ? ".parquet" : ".orc")
assert fileFormat.equalsIgnoreCase(format)
}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md
b/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md
index 06a7281bfb9..f740652d960 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md
+++
b/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md
@@ -53,6 +53,7 @@ Coverage status:
| R16 | CTAS behaves inconsistently for complex types, NULL, partition
transforms, file formats and failure cleanup | Black box: DDL and writer
committed together | Schema, file format, atomicity | P0 |
| R17 | Combining sort order, distribution mode, multiple file flushes and
concurrent commits causes misordering, lost rows or duplicate commits | White
box: exchange, sort writer, optimistic commit | Distributed correctness,
stability | P0 |
| R18 | STRING identity/bucket/truncate computes wrong physical partition
values for empty strings, Chinese text, emoji, combining characters and NULL |
Black box: UTF-8 transform metadata | Partition routing, pruning | P0 |
+| R19 | Written data files, position deletes or deletion vectors are recorded
in metadata under the normalized address BE actually wrote them at (such as
s3://) instead of the table's own location (such as s3a:// or oss://) | White
box: sink write path versus commit path | Whether other engines and maintenance
operations can open the files | P0 |
## Combined coverage
@@ -87,6 +88,7 @@ Coverage status:
| Sort and distribution properties | Multi-column sort order, NULL ordering,
none/hash/range distribution, forced multi-file flush | Covered (verified) |
`test_iceberg_write_order_distribution_properties` |
| Concurrent writes | Serializability invariant of conflicting MERGEs on the
same row, non-conflicting distributed appends | Covered (verified) |
`test_iceberg_write_concurrent_merge_invariants` |
| Distributed execution | Multi-bucket source tables, multi-partition Iceberg
sinks, multiple BE writers, no catalog/database shared between suites | Covered
(verified) | All suites added by this work |
+| File path recording | Every file written by INSERT, partition OVERWRITE,
branch INSERT, DELETE, UPDATE, MERGE and rewrite_data_files (v2 position
deletes, v3 deletion vectors) is recorded under the table's data location |
Covered (regular regression) | `test_iceberg_write_file_locations` |
| Spark cross-check | Fifteen positive suites compare row by row after the
last write; the STRING transform suite also compares physical partition
metadata before and after evolution | Covered (verified) | The fifteen positive
suites; physical metadata in `test_iceberg_write_string_transform_metadata` |
## Design of the added cases
@@ -112,9 +114,10 @@ Coverage status:
| W17 | Verify sort order, distribution mode and multi-file flush | R03, R11,
R17 | Correctness, stability | Multi-BE Doris | NULL sort keys, multi-column
ascending/descending order, none/hash/range, low target file size | The plan
contains the declared sort; the total row count across files is correct; the
three distribution modes agree |
| W18 | Verify commit invariants of concurrent MERGE and append | R11, R13,
R17 | Concurrency, atomicity | Multi-BE Doris | A readiness barrier makes two
independent sessions ready to dispatch at the same time; updates of the same
row and non-conflicting appends | Commits on the same row are serializable with
cardinality one; only Iceberg validation/commit conflicts are accepted;
non-conflicting writes are neither lost nor duplicated |
| W19 | Verify type safety of the MERGE source projection entering the
truncate transform | R12, R18 | Regular regression, stability | Iceberg v2 MOR
| Matched updates and unmatched inserts on a required STRING truncate column |
MERGE accepts non-NULL values in the nullable physical projection and produces
correct physical partitions |
+| W20 | Verify that every write operation records its file paths under the
table's own location | R19 | Correctness, compatibility | The REST fixture's
warehouse is s3a://, which BE normalizes to s3:// | INSERT, partition
OVERWRITE, branch INSERT, DELETE, UPDATE, MERGE and rewrite_data_files in turn
on v2 and v3 tables | Every file in `$all_files` starts with the LOCATION from
`SHOW CREATE TABLE` plus `/data/`; new write operations are added to this case
too |
## P0 coverage check
-Each of R01-R18 maps to at least one P0 regression. After the last successful
write, all fifteen positive suites cross-check the logical result of the same
table between Spark and Doris; transforms, rollover, CTAS properties and
failure atomicity additionally have physical metadata or state oracles. W08,
W10 and W19 run as regular regressions; W07, which may still write unreadable
required-null files, keeps its isolation switch so that it does not damage a
shared cluster.
+Each of R01-R19 maps to at least one P0 regression. After the last successful
write, all fifteen positive suites cross-check the logical result of the same
table between Spark and Doris; transforms, rollover, CTAS properties and
failure atomicity additionally have physical metadata or state oracles. W08,
W10 and W19 run as regular regressions; W07, which may still write unreadable
required-null files, keeps its isolation switch so that it does not damage a
shared cluster.
This matrix has no uncovered items. COW row-level DML, branch row-level DML,
tag writes and Avro writes are explicit current capability boundaries, each
pinned by an expected-rejection case that fixes the error semantics and failure
atomicity; only the known required-null defect keeps using an isolated negative
regression.
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_file_locations.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_file_locations.groovy
new file mode 100644
index 00000000000..b8cc5203d6c
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_file_locations.groovy
@@ -0,0 +1,161 @@
+// 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.
+
+// Every file Doris writes into an Iceberg table must be recorded under the
table's own location, not
+// under the address BE wrote it at. The REST fixture's warehouse is
s3a://warehouse/wh/, which FE
+// normalizes to s3:// for BE, so a writer that records the s3:// address is
caught here: delete files
+// and deletion vectors were once committed that way. Each write operation
Doris supports runs below;
+// a new one belongs here too.
+suite("test_iceberg_write_file_locations",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test")
+ 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")
+ String catalogName = "test_iceberg_write_file_locations"
+ String dbName = "iceberg_write_file_locations_db"
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1"
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """drop database if exists ${dbName} force"""
+ sql """create database ${dbName}"""
+ sql """use ${dbName}"""
+
+ // The table's data location, taken from its metadata rather than from any
file Doris wrote.
+ def dataLocationOf = { String table ->
+ String ddl = sql("""show create table ${table}""")[0][1]
+ def matcher = ddl =~ /LOCATION '([^']+)'/
+ assertTrue(matcher.find(), "no LOCATION in ${ddl}")
+ return matcher.group(1) + "/data/"
+ }
+
+ // Every file of every snapshot, branches included: data files, position
deletes and deletion vectors.
+ def assertFilesUnderDataLocation = { String table, String operation ->
+ String dataLocation = dataLocationOf(table)
+ def files = sql """select content, file_format, file_path from
${table}\$all_files"""
+ assertTrue(files.size() > 0, "no files in ${table} after ${operation}")
+ for (def file : files) {
+ String path = file[2].toString()
+ assertTrue(path.startsWith(dataLocation),
+ "${operation} recorded ${file[1]} file ${path} (content
${file[0]}) outside ${dataLocation}")
+ }
+ }
+
+ def countFiles = { String table, String predicate ->
+ return (sql """select count(*) from ${table}\$all_files where
${predicate}""")[0][0] as long
+ }
+
+ // Format v2: data files and position delete files.
+ sql """drop table if exists file_locations_v2"""
+ sql """
+ create table file_locations_v2 (
+ id int,
+ region string,
+ payload string
+ )
+ partition by list (region) ()
+ properties (
+ "format-version" = "2",
+ "write.delete.mode" = "merge-on-read",
+ "write.update.mode" = "merge-on-read",
+ "write.merge.mode" = "merge-on-read"
+ )
+ """
+ sql """insert into file_locations_v2 values (1, 'A', 'a1'), (2, 'B',
'b1'), (3, 'C', 'c1')"""
+ assertFilesUnderDataLocation("file_locations_v2", "INSERT")
+ sql """insert into file_locations_v2 values (4, 'A', 'a2'), (5, 'B',
'b2')"""
+ assertFilesUnderDataLocation("file_locations_v2", "second INSERT")
+ sql """insert overwrite table file_locations_v2 partition (region = 'C')
values (6, 'c2')"""
+ assertFilesUnderDataLocation("file_locations_v2", "INSERT OVERWRITE
PARTITION")
+ sql """alter table file_locations_v2 create branch location_branch"""
+ sql """insert into file_locations_v2@branch(location_branch) values (7,
'A', 'branch')"""
+ assertFilesUnderDataLocation("file_locations_v2", "branch INSERT")
+ sql """delete from file_locations_v2 where id = 1"""
+ assertFilesUnderDataLocation("file_locations_v2", "DELETE")
+ sql """update file_locations_v2 set payload = 'b1-updated' where id = 2"""
+ assertFilesUnderDataLocation("file_locations_v2", "UPDATE")
+ sql """
+ merge into file_locations_v2 t
+ using (
+ select 4 as id, 'A' as region, 'a2-merged' as payload
+ union all
+ select 8, 'D', 'd1'
+ ) s
+ on t.id = s.id
+ when matched then update set payload = s.payload
+ when not matched then insert (id, region, payload) values (s.id,
s.region, s.payload)
+ """
+ assertFilesUnderDataLocation("file_locations_v2", "MERGE")
+ sql """
+ alter table ${catalogName}.${dbName}.file_locations_v2
+ execute rewrite_data_files("target-file-size-bytes" = "10485760",
"min-input-files" = "2")
+ """
+ assertFilesUnderDataLocation("file_locations_v2", "rewrite_data_files")
+ assertTrue(countFiles("file_locations_v2", "content = 1") > 0, "DELETE /
UPDATE / MERGE wrote no delete files")
+
+ // Format v3: row-level operations write deletion vectors into Puffin
files.
+ sql """drop table if exists file_locations_v3"""
+ sql """
+ create table file_locations_v3 (
+ id int,
+ region string,
+ payload string
+ )
+ partition by list (region) ()
+ properties (
+ "format-version" = "3",
+ "write.delete.mode" = "merge-on-read",
+ "write.update.mode" = "merge-on-read",
+ "write.merge.mode" = "merge-on-read"
+ )
+ """
+ sql """insert into file_locations_v3 values (1, 'A', 'a1'), (2, 'B',
'b1'), (3, 'C', 'c1')"""
+ assertFilesUnderDataLocation("file_locations_v3", "INSERT")
+ sql """delete from file_locations_v3 where id = 1"""
+ assertFilesUnderDataLocation("file_locations_v3", "DELETE")
+ sql """update file_locations_v3 set payload = 'b1-updated' where id = 2"""
+ assertFilesUnderDataLocation("file_locations_v3", "UPDATE")
+ sql """
+ merge into file_locations_v3 t
+ using (
+ select 3 as id, 'C' as region, 'c1-merged' as payload
+ union all
+ select 4, 'D', 'd1'
+ ) s
+ on t.id = s.id
+ when matched then update set payload = s.payload
+ when not matched then insert (id, region, payload) values (s.id,
s.region, s.payload)
+ """
+ assertFilesUnderDataLocation("file_locations_v3", "MERGE")
+ assertTrue(countFiles("file_locations_v3", "upper(file_format) =
'PUFFIN'") > 0,
+ "DELETE / UPDATE / MERGE wrote no deletion vectors")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
