[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

@@ -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

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;

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

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);
 }
 

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

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());

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) {

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;

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;
     };

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"));
 
-    // 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_"));
+    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('/'));
+}
+
+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://container@account.dfs.core.windows.net/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>());

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.

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://container@account.dfs.core.windows.net/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://container@account.dfs.core.windows.net/warehouse/t/data/dt=2024-01-01/f.orc",
+              partition.commit_path("f.orc"));
+}
+
+} // namespace doris

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();

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);

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,

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

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());
         }
     }
 

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));

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) {

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());

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) {

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);

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)

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://c@acct.dfs.core.windows.net/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

regression-test/data/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.out.org

@@ -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, "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_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, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":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}, "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":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}, "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":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":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_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":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}}
-
--- !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_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":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_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_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":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}}
-

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)
         }

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.

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")
+}