This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 70bf96e8b0b [feat](iceberg) HDFS lazy open + iceberg delete file
file_size propagation (#66773) (#67842)
70bf96e8b0b is described below
commit 70bf96e8b0bed673a8048585160a27f39eca88cc
Author: camby <[email protected]>
AuthorDate: Mon Sep 21 16:30:28 2026 +0800
[feat](iceberg) HDFS lazy open + iceberg delete file file_size propagation
(#66773) (#67842)
Cherry-picked from #66773
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/exec/scan/file_scanner.cpp | 10 +-
.../table/iceberg_delete_file_reader_helper.cpp | 10 +-
.../table/iceberg_delete_file_reader_helper.h | 2 +-
be/src/format/table/iceberg_reader.cpp | 24 +-
be/src/format/table/iceberg_reader.h | 4 +
be/src/format_v2/orc/orc_reader.cpp | 9 +-
be/src/format_v2/table/iceberg_reader.cpp | 5 +-
be/src/io/fs/file_handle_cache.cpp | 68 +++-
be/src/io/fs/file_handle_cache.h | 10 +-
be/src/io/fs/hdfs_file_reader.cpp | 47 ++-
be/src/io/fs/hdfs_file_system.cpp | 9 +-
be/src/io/fs/local_file_reader.cpp | 2 +-
be/src/io/hdfs_util.cpp | 1 +
be/src/io/hdfs_util.h | 1 +
be/src/storage/index/index_file_reader.cpp | 10 +
.../index/inverted/inverted_index_fs_directory.cpp | 6 +-
be/test/exec/scan/vfile_scanner_exception_test.cpp | 131 ++++++-
.../iceberg_delete_file_reader_helper_test.cpp | 16 +-
.../format/table/iceberg/iceberg_reader_test.cpp | 65 ++++
.../format_v2/orc/orc_file_input_stream_test.cpp | 33 ++
be/test/format_v2/orc/orc_reader_test.cpp | 34 ++
be/test/format_v2/table/iceberg_reader_test.cpp | 120 ++++++-
be/test/io/fs/file_handle_cache_test.cpp | 391 +++++++++++++++++++++
be/test/io/fs/hdfs_file_system_test.cpp | 70 ++++
.../segment/inverted_index_file_reader_test.cpp | 27 ++
.../datasource/iceberg/source/IcebergScanNode.java | 4 +
.../iceberg/source/IcebergScanNodeTest.java | 77 ++++
gensrc/thrift/PlanNodes.thrift | 1 +
28 files changed, 1122 insertions(+), 65 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index 7bc042d8e97..34cf0928553 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -575,8 +575,14 @@ Status FileScanner::_get_block_wrapped(RuntimeState*
state, Block* block, bool*
// Read next block.
// Some of column in block may not be filled (column not exist in
file)
- RETURN_IF_ERROR(
- _cur_reader->get_next_block(_src_block_ptr, &read_rows,
&_cur_reader_eof));
+ Status st = _cur_reader->get_next_block(_src_block_ptr,
&read_rows, &_cur_reader_eof);
+ // Lazy open may surface NOT_FOUND on the first read; skip as
above.
+ if (st.is<ErrorCode::NOT_FOUND>() &&
config::ignore_not_found_file_in_external_table) {
+ _cur_reader_eof = true;
+ COUNTER_UPDATE(_not_found_file_counter, 1);
+ continue;
+ }
+ RETURN_IF_ERROR(st);
}
// use read_rows instead of _src_block_ptr->rows(), because the first
column of _src_block_ptr
// may not be filled after calling `get_next_block()`, so
_src_block_ptr->rows() may return wrong result.
diff --git a/be/src/format/table/iceberg_delete_file_reader_helper.cpp
b/be/src/format/table/iceberg_delete_file_reader_helper.cpp
index 7ae563d963a..0ee8fab5792 100644
--- a/be/src/format/table/iceberg_delete_file_reader_helper.cpp
+++ b/be/src/format/table/iceberg_delete_file_reader_helper.cpp
@@ -243,12 +243,12 @@ TFileScanRangeParams
build_iceberg_delete_scan_range_params(
return params;
}
-TFileRangeDesc build_iceberg_delete_file_range(const std::string& path) {
+TFileRangeDesc build_iceberg_delete_file_range(const std::string& path,
int64_t file_size) {
TFileRangeDesc range;
range.path = path;
range.start_offset = 0;
range.size = -1;
- range.file_size = -1;
+ range.file_size = file_size;
return range;
}
@@ -283,7 +283,8 @@ Status read_iceberg_position_delete_file(const
TIcebergDeleteFileDesc& delete_fi
return Status::InvalidArgument("invalid position delete reader
options");
}
- TFileRangeDesc delete_range =
build_iceberg_delete_file_range(delete_file.path);
+ TFileRangeDesc delete_range = build_iceberg_delete_file_range(
+ delete_file.path, delete_file.__isset.file_size ?
delete_file.file_size : -1);
if (options.fs_name != nullptr && !options.fs_name->empty()) {
delete_range.__set_fs_name(*options.fs_name);
}
@@ -362,7 +363,8 @@ Status read_iceberg_deletion_vector(const
TIcebergDeleteFileDesc& delete_file,
DBUG_EXECUTE_IF("IcebergDeleteFileReader.read_deletion_vector.should_stop",
{ return Status::EndOfFile("stop read."); });
- TFileRangeDesc delete_range =
build_iceberg_delete_file_range(delete_file.path);
+ TFileRangeDesc delete_range = build_iceberg_delete_file_range(
+ delete_file.path, delete_file.__isset.file_size ?
delete_file.file_size : -1);
if (options.fs_name != nullptr && !options.fs_name->empty()) {
delete_range.__set_fs_name(*options.fs_name);
}
diff --git a/be/src/format/table/iceberg_delete_file_reader_helper.h
b/be/src/format/table/iceberg_delete_file_reader_helper.h
index adc0ef196f4..d43443ecd8e 100644
--- a/be/src/format/table/iceberg_delete_file_reader_helper.h
+++ b/be/src/format/table/iceberg_delete_file_reader_helper.h
@@ -67,7 +67,7 @@ TFileScanRangeParams build_iceberg_delete_scan_range_params(
const std::map<std::string, std::string>& hadoop_conf, TFileType::type
file_type,
const std::vector<TNetworkAddress>& broker_addresses);
-TFileRangeDesc build_iceberg_delete_file_range(const std::string& path);
+TFileRangeDesc build_iceberg_delete_file_range(const std::string& path,
int64_t file_size);
bool is_iceberg_deletion_vector(const TIcebergDeleteFileDesc& delete_file);
diff --git a/be/src/format/table/iceberg_reader.cpp
b/be/src/format/table/iceberg_reader.cpp
index 4481ce5b24d..56076d39bc9 100644
--- a/be/src/format/table/iceberg_reader.cpp
+++ b/be/src/format/table/iceberg_reader.cpp
@@ -1415,14 +1415,14 @@ Status IcebergTableReader::_position_delete_base(
Status create_status = Status::OK();
auto* delete_file_cache = _kv_cache->get<DeleteFile>(
_delet_file_cache_key(delete_file.path), [&]() -> DeleteFile* {
- auto* position_delete = new DeleteFile;
- create_status = _read_position_delete_file(delete_file,
position_delete);
+ auto position_delete = std::make_unique<DeleteFile>();
+ create_status = _read_position_delete_file(delete_file,
position_delete.get());
if (!create_status) {
return nullptr;
}
- return position_delete;
+ return position_delete.release();
});
if (create_status.is<ErrorCode::END_OF_FILE>()) {
continue;
@@ -1463,6 +1463,17 @@ Status IcebergTableReader::_position_delete_base(
return Status::OK();
}
+Status IcebergTableReader::TEST_position_delete_base(
+ const std::string& data_file_path,
+ const std::vector<TIcebergDeleteFileDesc>& delete_files) {
+ return _position_delete_base(data_file_path, delete_files);
+}
+
+Status IcebergTableReader::TEST_read_equality_delete_file(
+ const TIcebergDeleteFileDesc& delete_file) {
+ return _process_equality_delete({delete_file});
+}
+
Status IcebergTableReader::_read_position_delete_file(const
TIcebergDeleteFileDesc& delete_file,
DeleteFile*
position_delete) {
GroupedDeleteRowsVisitor visitor(position_delete);
@@ -2060,7 +2071,8 @@ Status IcebergTableReader::read_deletion_vector(const
std::string& data_file_pat
delete_range.path = delete_file_desc.path;
delete_range.start_offset = delete_file_desc.content_offset;
delete_range.size = delete_file_desc.content_size_in_bytes;
- delete_range.file_size = -1;
+ delete_range.file_size =
+ delete_file_desc.__isset.file_size ?
delete_file_desc.file_size : -1;
// We may consider caching the DeletionVectorReader when reading
Puffin files,
// where the underlying reader is an `InMemoryFileReader` and a single
data file is
@@ -2155,7 +2167,7 @@ Status IcebergParquetReader::_process_equality_delete(
delete_desc.path = delete_file.path;
delete_desc.start_offset = 0;
delete_desc.size = -1;
- delete_desc.file_size = -1;
+ delete_desc.file_size = delete_file.__isset.file_size ?
delete_file.file_size : -1;
auto delete_reader = ParquetReader::create_unique(
_profile, _params, delete_desc, READ_DELETE_FILE_BATCH_SIZE,
@@ -2302,7 +2314,7 @@ Status IcebergOrcReader::_process_equality_delete(
delete_desc.path = delete_file.path;
delete_desc.start_offset = 0;
delete_desc.size = -1;
- delete_desc.file_size = -1;
+ delete_desc.file_size = delete_file.__isset.file_size ?
delete_file.file_size : -1;
auto delete_reader = OrcReader::create_unique(_profile, _state,
_params, delete_desc,
READ_DELETE_FILE_BATCH_SIZE,
diff --git a/be/src/format/table/iceberg_reader.h
b/be/src/format/table/iceberg_reader.h
index a723a244172..9d9a206b08a 100644
--- a/be/src/format/table/iceberg_reader.h
+++ b/be/src/format/table/iceberg_reader.h
@@ -100,6 +100,10 @@ public:
enum { DATA, POSITION_DELETE, EQUALITY_DELETE, DELETION_VECTOR };
enum Fileformat { NONE, PARQUET, ORC, AVRO };
+ Status TEST_position_delete_base(const std::string& data_file_path,
+ const
std::vector<TIcebergDeleteFileDesc>& delete_files);
+ Status TEST_read_equality_delete_file(const TIcebergDeleteFileDesc&
delete_file);
+
virtual void set_delete_rows() = 0;
Status read_deletion_vector(const std::string& data_file_path,
diff --git a/be/src/format_v2/orc/orc_reader.cpp
b/be/src/format_v2/orc/orc_reader.cpp
index 85754b9997a..f08e221c1cb 100644
--- a/be/src/format_v2/orc/orc_reader.cpp
+++ b/be/src/format_v2/orc/orc_reader.cpp
@@ -930,8 +930,15 @@ Status OrcReader::init(RuntimeState* state) {
if (is_orc_stop(_io_ctx.get(), e)) {
return Status::EndOfFile("stop");
}
+ // invoker maybe just skip Status.NotFound and continue
+ // so we need distinguish between it and other kinds of errors
+ const std::string err_msg = e.what();
+ if (err_msg.find("No such file or directory") != std::string::npos
||
+ err_msg.find("NoSuchKey") != std::string::npos) {
+ return Status::NotFound(err_msg);
+ }
return Status::InternalError("Failed to open ORC file {}: {}",
_file_description->path,
- e.what());
+ err_msg);
}
return Status::OK();
};
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index c18cec8108f..1ba1686e037 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -1144,7 +1144,7 @@ Status
IcebergTableReader::_parse_deletion_vector_file(const TTableFormatFileDes
desc->path = deletion_vector->path;
desc->start_offset = deletion_vector->content_offset;
desc->size = static_cast<int64_t>(bytes_read);
- desc->file_size = -1;
+ desc->file_size = deletion_vector->__isset.file_size ?
deletion_vector->file_size : -1;
desc->format = DeleteFileDesc::Format::ICEBERG;
*has_delete_file = true;
return Status::OK();
@@ -1500,7 +1500,8 @@ Status
IcebergTableReader::_create_delete_file_reader(const TIcebergDeleteFileDe
return Status::NotSupported("Unsupported Iceberg delete file format
{}",
delete_file.file_format);
}
- auto delete_range = build_iceberg_delete_file_range(delete_file.path);
+ auto delete_range = build_iceberg_delete_file_range(
+ delete_file.path, delete_file.__isset.file_size ?
delete_file.file_size : -1);
if (_current_task != nullptr && _current_task->data_file != nullptr &&
!_current_task->data_file->fs_name.empty()) {
delete_range.__set_fs_name(_current_task->data_file->fs_name);
diff --git a/be/src/io/fs/file_handle_cache.cpp
b/be/src/io/fs/file_handle_cache.cpp
index 556d168041d..5213c4efcfc 100644
--- a/be/src/io/fs/file_handle_cache.cpp
+++ b/be/src/io/fs/file_handle_cache.cpp
@@ -26,7 +26,11 @@
#include <tuple>
#include "common/cast_set.h"
+#include "common/metrics/doris_metrics.h"
+#include "cpp/sync_point.h"
#include "io/fs/err_utils.h"
+#include "io/hdfs_util.h"
+#include "util/bvar_helper.h"
#include "util/hash_util.hpp"
#include "util/time.h"
namespace doris::io {
@@ -36,29 +40,30 @@ namespace doris::io {
HdfsFileHandle::~HdfsFileHandle() {
if (_hdfs_file != nullptr && _fs != nullptr) {
VLOG_FILE << "hdfsCloseFile() fid=" << _hdfs_file;
- hdfsCloseFile(_fs, _hdfs_file); // TODO: check return code
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_close_latency);
+ SYNC_POINT_HOOK_RETURN_VALUE(hdfsCloseFile(_fs, _hdfs_file),
+ "HdfsFileHandle::close::hdfsCloseFile");
+ DorisMetrics::instance()->hdfs_file_open_reading->increment(-1);
}
_fs = nullptr;
_hdfs_file = nullptr;
}
Status HdfsFileHandle::init(int64_t file_size) {
- _hdfs_file = hdfsOpenFile(_fs, _fname.c_str(), O_RDONLY, 0, 0, 0);
- if (_hdfs_file == nullptr) {
- std::string _err_msg = hdfs_error();
- // invoker maybe just skip Status.NotFound and continue
- // so we need distinguish between it and other kinds of errors
- if (_err_msg.find("No such file or directory") != std::string::npos) {
- return Status::NotFound(_err_msg);
- }
- return Status::InternalError("failed to open {}: {}", _fname,
_err_msg);
- }
-
_file_size = file_size;
if (_file_size <= 0) {
- hdfsFileInfo* file_info = hdfsGetPathInfo(_fs, _fname.c_str());
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_get_path_info_latency);
+ auto* file_info = SYNC_POINT_HOOK_RETURN_VALUE(hdfsGetPathInfo(_fs,
_fname.c_str()),
+
"HdfsFileHandle::init::hdfsGetPathInfo");
if (file_info == nullptr) {
- return Status::InternalError("failed to get file size of {}: {}",
_fname, hdfs_error());
+ std::string err_msg =
+ SYNC_POINT_HOOK_RETURN_VALUE(hdfs_error(),
"HdfsFileHandle::init::hdfs_error");
+ // invoker maybe just skip Status.NotFound and continue
+ // so we need distinguish between it and other kinds of errors
+ if (err_msg.find("No such file or directory") !=
std::string::npos) {
+ return Status::NotFound(err_msg);
+ }
+ return Status::InternalError("failed to get file size of {}: {}",
_fname, err_msg);
}
_file_size = file_info->mSize;
hdfsFreeFileInfo(file_info, 1);
@@ -66,6 +71,31 @@ Status HdfsFileHandle::init(int64_t file_size) {
return Status::OK();
}
+Status HdfsFileHandle::ensure_open() {
+ std::call_once(_open_once, [this]() {
+ VLOG_DEBUG << "lazy open hdfs file: " << _fname;
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_open_latency);
+ _hdfs_file =
+ SYNC_POINT_HOOK_RETURN_VALUE(hdfsOpenFile(_fs, _fname.c_str(),
O_RDONLY, 0, 0, 0),
+
"HdfsFileHandle::ensure_open::hdfsOpenFile");
+ if (_hdfs_file != nullptr) {
+ _open_status = Status::OK();
+ DorisMetrics::instance()->hdfs_file_open_reading->increment(1);
+ DorisMetrics::instance()->hdfs_file_reader_total->increment(1);
+ } else {
+ // Capture error inside the opening thread (libhdfs last-error is
thread-local).
+ std::string _err_msg = SYNC_POINT_HOOK_RETURN_VALUE(
+ hdfs_error(), "HdfsFileHandle::ensure_open::hdfs_error");
+ if (_err_msg.find("No such file or directory") !=
std::string::npos) {
+ _open_status = Status::NotFound(_err_msg);
+ } else {
+ _open_status = Status::InternalError("failed to open {}: {}",
_fname, _err_msg);
+ }
+ }
+ });
+ return _open_status;
+}
+
CachedHdfsFileHandle::CachedHdfsFileHandle(const hdfsFS& fs, const
std::string& fname,
int64_t mtime)
: HdfsFileHandle(fs, fname, mtime) {}
@@ -100,8 +130,16 @@ void FileHandleCache::Accessor::destroy() {
FileHandleCache::Accessor::~Accessor() {
if (_cache_accessor.get()) {
+ auto* handle = get();
+ if (handle->file() == nullptr) {
+ // Not opened or open failed (call_once won't retry), destroy to
avoid cache pollution
+ destroy();
+ return;
+ }
#ifdef USE_HADOOP_HDFS
- if (hdfsUnbufferFile(get()->file()) != 0) {
+ int unbuffer_ret =
SYNC_POINT_HOOK_RETURN_VALUE(hdfsUnbufferFile(handle->file()),
+
"HdfsFileHandle::close::hdfsUnbufferFile");
+ if (unbuffer_ret != 0) {
VLOG_FILE << "FS does not support file handle unbuffering, closing
file="
<< _cache_accessor.get_key()->second.first;
destroy();
diff --git a/be/src/io/fs/file_handle_cache.h b/be/src/io/fs/file_handle_cache.h
index ce3c708ba99..46e91035746 100644
--- a/be/src/io/fs/file_handle_cache.h
+++ b/be/src/io/fs/file_handle_cache.h
@@ -26,6 +26,7 @@
#include <list>
#include <map>
#include <memory>
+#include <mutex>
#include <string>
#include <utility>
@@ -49,9 +50,12 @@ public:
/// Destructor will close the file handle
~HdfsFileHandle();
- /// Init opens the file handle
+ /// Init only sets file_size (from param or hdfsGetPathInfo), does NOT
open the file.
Status init(int64_t file_size);
+ /// Lazily opens the file handle on first read. Thread-safe via
std::call_once.
+ Status ensure_open();
+
hdfsFS fs() const { return _fs; }
hdfsFile file() const { return _hdfs_file; }
int64_t mtime() const { return _mtime; }
@@ -66,7 +70,9 @@ private:
const std::string _fname;
hdfsFile _hdfs_file = nullptr;
int64_t _mtime;
- int64_t _file_size;
+ int64_t _file_size = -1;
+ std::once_flag _open_once;
+ Status _open_status;
};
/// CachedHdfsFileHandles are owned by the file handle cache and are used for
no
diff --git a/be/src/io/fs/hdfs_file_reader.cpp
b/be/src/io/fs/hdfs_file_reader.cpp
index db8214c12bb..c77494ad7c9 100644
--- a/be/src/io/fs/hdfs_file_reader.cpp
+++ b/be/src/io/fs/hdfs_file_reader.cpp
@@ -25,13 +25,13 @@
#include "bvar/latency_recorder.h"
#include "bvar/reducer.h"
#include "common/compiler_util.h" // IWYU pragma: keep
-#include "common/metrics/doris_metrics.h"
#include "cpp/sync_point.h"
#include "io/fs/err_utils.h"
#include "io/hdfs_util.h"
#include "runtime/thread_context.h"
#include "runtime/workload_management/io_throttle.h"
#include "service/backend_options.h"
+#include "util/bvar_helper.h"
namespace doris::io {
#include "common/compile_check_begin.h"
@@ -73,9 +73,6 @@ HdfsFileReader::HdfsFileReader(Path path, std::string
fs_name, FileHandleCache::
_accessor(std::move(accessor)),
_mtime(mtime) {
_handle = _accessor.get();
-
- DorisMetrics::instance()->hdfs_file_open_reading->increment(1);
- DorisMetrics::instance()->hdfs_file_reader_total->increment(1);
}
HdfsFileReader::~HdfsFileReader() {
@@ -83,15 +80,20 @@ HdfsFileReader::~HdfsFileReader() {
}
Status HdfsFileReader::close() {
- bool expected = false;
- if (_closed.compare_exchange_strong(expected, true,
std::memory_order_acq_rel)) {
- DorisMetrics::instance()->hdfs_file_open_reading->increment(-1);
- }
+ _closed = true;
return Status::OK();
}
Status HdfsFileReader::read_at_impl(size_t offset, Slice result, size_t*
bytes_read,
const IOContext* io_ctx) {
+ if (closed()) [[unlikely]] {
+ return Status::InternalError("read closed file: {}", _path.native());
+ }
+ if (_handle == nullptr) [[unlikely]] {
+ return Status::InternalError("cached hdfs file handle has been
destroyed: {}",
+ _path.native());
+ }
+ RETURN_IF_ERROR(_handle->ensure_open());
auto st = do_read_at_impl(offset, result, bytes_read, io_ctx);
if (!st.ok()) {
_handle = nullptr;
@@ -103,15 +105,6 @@ Status HdfsFileReader::read_at_impl(size_t offset, Slice
result, size_t* bytes_r
#ifdef USE_HADOOP_HDFS
Status HdfsFileReader::do_read_at_impl(size_t offset, Slice result, size_t*
bytes_read,
const IOContext* /*io_ctx*/) {
- if (closed()) [[unlikely]] {
- return Status::InternalError("read closed file: {}", _path.native());
- }
-
- if (_handle == nullptr) [[unlikely]] {
- return Status::InternalError("cached hdfs file handle has been
destroyed: {}",
- _path.native());
- }
-
if (offset > _handle->file_size()) {
return Status::IOError("offset exceeds file size(offset: {}, file
size: {}, path: {})",
offset, _handle->file_size(), _path.native());
@@ -132,8 +125,12 @@ Status HdfsFileReader::do_read_at_impl(size_t offset,
Slice result, size_t* byte
int64_t max_to_read = bytes_req - has_read;
tSize to_read = static_cast<tSize>(
std::min(max_to_read,
static_cast<int64_t>(std::numeric_limits<tSize>::max())));
- tSize loop_read = hdfsPread(_handle->fs(), _handle->file(), offset +
has_read,
- to + has_read, to_read);
+ tSize loop_read;
+ {
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_read_latency);
+ loop_read = hdfsPread(_handle->fs(), _handle->file(), offset +
has_read, to + has_read,
+ to_read);
+ }
{
[[maybe_unused]] Status error_ret;
TEST_INJECTION_POINT_RETURN_WITH_VALUE("HdfsFileReader:read_error", error_ret);
@@ -165,10 +162,6 @@ Status HdfsFileReader::do_read_at_impl(size_t offset,
Slice result, size_t* byte
// TODO: rethink here to see if there are some difference between hdfsPread()
and hdfsRead()
Status HdfsFileReader::do_read_at_impl(size_t offset, Slice result, size_t*
bytes_read,
const IOContext* /*io_ctx*/) {
- if (closed()) [[unlikely]] {
- return Status::InternalError("read closed file: ", _path.native());
- }
-
if (offset > _handle->file_size()) {
return Status::IOError("offset exceeds file size(offset: {}, file
size: {}, path: {})",
offset, _handle->file_size(), _path.native());
@@ -198,8 +191,12 @@ Status HdfsFileReader::do_read_at_impl(size_t offset,
Slice result, size_t* byte
size_t has_read = 0;
while (has_read < bytes_req) {
- int64_t loop_read = hdfsRead(_handle->fs(), _handle->file(), to +
has_read,
- static_cast<int32_t>(bytes_req -
has_read));
+ int64_t loop_read;
+ {
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_read_latency);
+ loop_read = hdfsRead(_handle->fs(), _handle->file(), to + has_read,
+ static_cast<int32_t>(bytes_req - has_read));
+ }
if (loop_read < 0) {
// invoker maybe just skip Status.NotFound and continue
// so we need distinguish between it and other kinds of errors
diff --git a/be/src/io/fs/hdfs_file_system.cpp
b/be/src/io/fs/hdfs_file_system.cpp
index e85b582e24f..5043c9eaaa6 100644
--- a/be/src/io/fs/hdfs_file_system.cpp
+++ b/be/src/io/fs/hdfs_file_system.cpp
@@ -40,6 +40,7 @@
#include "io/hdfs_builder.h"
#include "io/hdfs_util.h"
#include "runtime/exec_env.h"
+#include "util/bvar_helper.h"
#include "util/obj_lru_cache.h"
#include "util/slice.h"
@@ -117,7 +118,11 @@ Status HdfsFileSystem::open_file_internal(const Path&
file, FileReaderSPtr* read
Status HdfsFileSystem::create_directory_impl(const Path& dir, bool
failed_if_exists) {
CHECK_HDFS_HANDLER(_fs_handler);
Path real_path = convert_path(dir, _fs_name);
- int res = hdfsCreateDirectory(_fs_handler->hdfs_fs,
real_path.string().c_str());
+ int res;
+ {
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_create_dir_latency);
+ res = hdfsCreateDirectory(_fs_handler->hdfs_fs,
real_path.string().c_str());
+ }
if (res == -1) {
return Status::IOError("failed to create directory {}: {}",
dir.native(), hdfs_error());
}
@@ -180,6 +185,7 @@ Status HdfsFileSystem::exists_impl(const Path& path, bool*
res) const {
Status HdfsFileSystem::file_size_impl(const Path& path, int64_t* file_size)
const {
CHECK_HDFS_HANDLER(_fs_handler);
Path real_path = convert_path(path, _fs_name);
+ SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_get_path_info_latency);
hdfsFileInfo* file_info = hdfsGetPathInfo(_fs_handler->hdfs_fs,
real_path.string().c_str());
if (file_info == nullptr) {
return Status::IOError("failed to get file size of {}: {}",
path.native(), hdfs_error());
@@ -222,6 +228,7 @@ Status HdfsFileSystem::list_impl(const Path& path, bool
only_file, std::vector<F
}
Status HdfsFileSystem::rename_impl(const Path& orig_name, const Path&
new_name) {
+ CHECK_HDFS_HANDLER(_fs_handler);
Path normal_orig_name = convert_path(orig_name, _fs_name);
Path normal_new_name = convert_path(new_name, _fs_name);
int ret = hdfsRename(_fs_handler->hdfs_fs, normal_orig_name.c_str(),
normal_new_name.c_str());
diff --git a/be/src/io/fs/local_file_reader.cpp
b/be/src/io/fs/local_file_reader.cpp
index 8cdc1e67663..ce1be5d9529 100644
--- a/be/src/io/fs/local_file_reader.cpp
+++ b/be/src/io/fs/local_file_reader.cpp
@@ -174,7 +174,7 @@ Status LocalFileReader::read_at_impl(size_t offset, Slice
result, size_t* bytes_
if ((sub_path.empty() && _path.filename().compare(kTestFilePath))
||
(!sub_path.empty() && _path.native().find(sub_path) !=
std::string::npos)) {
res = -1;
- errno = EIO;
+ errno = dp->param<int>("errno", EIO);
LOG(WARNING) << Status::IOError("debug read io error: {}",
_path.native());
}
});
diff --git a/be/src/io/hdfs_util.cpp b/be/src/io/hdfs_util.cpp
index c7060108282..fb54f82b6a1 100644
--- a/be/src/io/hdfs_util.cpp
+++ b/be/src/io/hdfs_util.cpp
@@ -41,6 +41,7 @@ bvar::LatencyRecorder hdfs_close_latency("hdfs_close");
bvar::LatencyRecorder hdfs_flush_latency("hdfs_flush");
bvar::LatencyRecorder hdfs_hflush_latency("hdfs_hflush");
bvar::LatencyRecorder hdfs_hsync_latency("hdfs_hsync");
+bvar::LatencyRecorder hdfs_get_path_info_latency("hdfs_get_path_info");
}; // namespace hdfs_bvar
Path convert_path(const Path& path, const std::string& namenode) {
diff --git a/be/src/io/hdfs_util.h b/be/src/io/hdfs_util.h
index 8d63a19ec92..d7a74134db6 100644
--- a/be/src/io/hdfs_util.h
+++ b/be/src/io/hdfs_util.h
@@ -80,6 +80,7 @@ extern bvar::LatencyRecorder hdfs_close_latency;
extern bvar::LatencyRecorder hdfs_flush_latency;
extern bvar::LatencyRecorder hdfs_hflush_latency;
extern bvar::LatencyRecorder hdfs_hsync_latency;
+extern bvar::LatencyRecorder hdfs_get_path_info_latency;
}; // namespace hdfs_bvar
// if the format of path is hdfs://ip:port/path, replace it to /path.
diff --git a/be/src/storage/index/index_file_reader.cpp
b/be/src/storage/index/index_file_reader.cpp
index 348e1399421..ce2a4f1347f 100644
--- a/be/src/storage/index/index_file_reader.cpp
+++ b/be/src/storage/index/index_file_reader.cpp
@@ -129,6 +129,11 @@ Status IndexFileReader::_init_from(int32_t
read_buffer_size, const io::IOContext
index_file_full_path, err.what());
}
}
+ // Lazy open can surface a missing file as a read error; keep NotFound
distinguishable
+ if (err.number() == CL_ERR_FileNotFound) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>(
+ "inverted index file {} is not found.",
index_file_full_path);
+ }
return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
"CLuceneError occur when init idx file {}, error msg: {}",
index_file_full_path,
err.what());
@@ -200,6 +205,11 @@ Result<std::unique_ptr<DorisCompoundReader,
DirectoryDeleter>> IndexFileReader::
// 3. read file in DorisCompoundReader
compound_reader.reset(new DorisCompoundReader(index_input,
_read_buffer_size));
} catch (CLuceneError& err) {
+ // Lazy open can surface a missing file as a read error; keep
NotFound distinguishable
+ if (err.number() == CL_ERR_FileNotFound) {
+ return
ResultError(Status::Error<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>(
+ "inverted index file {} is not found.",
index_file_path));
+ }
return
ResultError(Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
"CLuceneError occur when open idx file {}, error msg: {}",
index_file_path,
err.what()));
diff --git a/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
b/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
index 30f168e8b14..e3a604a4f40 100644
--- a/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
+++ b/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
@@ -237,7 +237,11 @@ void DorisFSDirectory::FSIndexInput::readInternal(uint8_t*
b, const int32_t len)
"DorisFSDirectory::FSIndexInput::readInternal_reader_read_at_error");
})
if (!st.ok()) {
- _CLTHROWA(CL_ERR_IO, "read past EOF");
+ // Carry NotFound across the CLucene boundary so index callers can
downgrade
+ if (st.is<ErrorCode::NOT_FOUND>()) {
+ _CLTHROWA(CL_ERR_FileNotFound,
st.to_string_no_stack().c_str());
+ }
+ _CLTHROWA(CL_ERR_IO, st.to_string_no_stack().c_str());
}
bufferLength = len;
DBUG_EXECUTE_IF("DorisFSDirectory::FSIndexInput::readInternal_bytes_read_error",
diff --git a/be/test/exec/scan/vfile_scanner_exception_test.cpp
b/be/test/exec/scan/vfile_scanner_exception_test.cpp
index 3a489a7d52e..a68002c24c6 100644
--- a/be/test/exec/scan/vfile_scanner_exception_test.cpp
+++ b/be/test/exec/scan/vfile_scanner_exception_test.cpp
@@ -23,6 +23,7 @@
#include <utility>
#include <vector>
+#include "common/config.h"
#include "common/object_pool.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
@@ -31,13 +32,17 @@
#include "exec/scan/file_scanner.h"
#include "exec/scan/split_source_connector.h"
#include "format_v2/table/hive_reader.h"
+#include "io/fs/hdfs/hdfs_mgr.h"
#include "io/fs/local_file_system.h"
+#include "io/hdfs_util.h"
#include "load/group_commit/wal/wal_manager.h"
#include "runtime/cluster_info.h"
#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
#include "runtime/memory/mem_tracker.h"
#include "runtime/runtime_state.h"
#include "runtime/user_function_cache.h"
+#include "util/defer_op.h"
namespace doris {
class TestSplitSourceConnectorStub : public SplitSourceConnector {
@@ -65,6 +70,47 @@ public:
TFileScanRangeParams* get_params() override { return &_scan_range.params; }
};
+// Returns a fake HDFS handler so reader construction never touches the
network.
+class FakeHdfsMgr final : public io::HdfsMgr {
+public:
+ Status _create_hdfs_fs_impl(const THdfsParams& hdfs_params, const
std::string& fs_name,
+ std::shared_ptr<io::HdfsHandler>* fs_handler)
override {
+ *fs_handler = std::make_shared<io::HdfsHandler>(
+ reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1)), false,
"", "", fs_name);
+ return Status::OK();
+ }
+};
+
+// Registers a SyncPoint callback that returns a fixed value.
+template <typename T>
+static void set_mock_return(const std::string& point, T value,
SyncPoint::CallbackGuard* guard) {
+ SyncPoint::get_instance()->set_call_back(
+ point,
+ [value = std::move(value)](auto&& args) {
+ auto* ret = try_any_cast_ret<T>(args);
+ ret->first = std::move(value);
+ ret->second = true;
+ },
+ guard);
+}
+
+// Injects a read-stage NOT_FOUND through the HDFS lazy-open sync points.
+struct HdfsNotFoundGuard {
+ SyncPoint::CallbackGuard open_guard;
+ SyncPoint::CallbackGuard err_guard;
+ SyncPoint::CallbackGuard close_guard;
+ HdfsNotFoundGuard() {
+ auto* sp = SyncPoint::get_instance();
+ sp->enable_processing();
+ set_mock_return<hdfsFile>("HdfsFileHandle::ensure_open::hdfsOpenFile",
nullptr,
+ &open_guard);
+ set_mock_return<std::string>("HdfsFileHandle::ensure_open::hdfs_error",
+ "No such file or directory", &err_guard);
+ set_mock_return<int>("HdfsFileHandle::close::hdfsCloseFile", 0,
&close_guard);
+ }
+ ~HdfsNotFoundGuard() { SyncPoint::get_instance()->disable_processing(); }
+};
+
class VfileScannerExceptionTest : public testing::Test {
public:
VfileScannerExceptionTest()
@@ -77,17 +123,26 @@ public:
}
void init();
void generate_scanner(std::shared_ptr<FileScanner>& scanner);
+ // Fake HDFS CSV range with known size, so open is deferred to the first
read.
+ void prepare_hdfs_csv_range();
void TearDown() override {
WARN_IF_ERROR(_scan_node->close(&_runtime_state), "fail to close
scan_node")
+ // Avoid leaving a dangling _hdfs_mgr after _fake_hdfs_mgr is
destroyed.
+ ExecEnv::GetInstance()->_hdfs_mgr = _old_hdfs_mgr;
}
protected:
- virtual void SetUp() override {}
+ void SetUp() override {
+ _old_hdfs_mgr = ExecEnv::GetInstance()->_hdfs_mgr;
+ ExecEnv::GetInstance()->_hdfs_mgr = &_fake_hdfs_mgr;
+ }
private:
void _init_desc_table();
+ FakeHdfsMgr _fake_hdfs_mgr;
+ io::HdfsMgr* _old_hdfs_mgr = nullptr;
ExecEnv* _env = nullptr;
int64_t _backend_id = 1001;
std::string _label_1 = "test1";
@@ -286,6 +341,34 @@ void
VfileScannerExceptionTest::generate_scanner(std::shared_ptr<FileScanner>& s
WARN_IF_ERROR(scanner->init(&_runtime_state, _conjuncts), "fail to prepare
scanner");
}
+void VfileScannerExceptionTest::prepare_hdfs_csv_range() {
+ _range_desc.path = "hdfs://fake-nn:8020/not_found/data.csv";
+ _range_desc.start_offset = 0;
+ _range_desc.size = 100;
+ _range_desc.__set_file_size(100);
+ _ranges[0] = _range_desc;
+ _scan_range.ranges = _ranges;
+ auto& params = _scan_range.params;
+ params.format_type = TFileFormatType::FORMAT_CSV_PLAIN;
+ params.file_type = TFileType::FILE_HDFS;
+ params.hdfs_params.__set_fs_name("hdfs://fake-nn:8020");
+ params.__isset.file_attributes = true;
+ params.file_attributes.__isset.text_params = true;
+ params.file_attributes.text_params.column_separator = ",";
+ params.file_attributes.text_params.line_delimiter = "\n";
+ params.__isset.column_idxs = true;
+ params.column_idxs = {0, 1, 2};
+ params.__set_num_of_columns_from_file(3);
+ params.__isset.required_slots = true;
+ params.required_slots.clear();
+ for (int32_t slot_id = 1; slot_id <= 3; ++slot_id) {
+ TFileScanSlotInfo slot_info;
+ slot_info.__set_slot_id(slot_id);
+ slot_info.__set_is_file_slot(true);
+ params.required_slots.push_back(slot_info);
+ }
+}
+
TEST_F(VfileScannerExceptionTest, failure_case) {
std::shared_ptr<FileScanner> scanner = nullptr;
generate_scanner(scanner);
@@ -341,6 +424,52 @@ TEST_F(VfileScannerExceptionTest,
process_late_arrival_conjuncts_retain) {
WARN_IF_ERROR(scanner->close(&_runtime_state), "fail to close scanner");
}
+// A lazy-open NOT_FOUND on the first HDFS read must be skipped and counted,
not fail the scan.
+TEST_F(VfileScannerExceptionTest, read_stage_not_found_skipped_when_enabled) {
+ HdfsNotFoundGuard guard;
+ const bool old_ignore = config::ignore_not_found_file_in_external_table;
+ config::ignore_not_found_file_in_external_table = true;
+ Defer restore_ignore {[&]() {
config::ignore_not_found_file_in_external_table = old_ignore; }};
+
+ prepare_hdfs_csv_range();
+ std::shared_ptr<FileScanner> scanner = nullptr;
+ generate_scanner(scanner);
+
+ std::unique_ptr<Block> block(new Block());
+ bool eof = false;
+ auto st = scanner->get_block(&_runtime_state, block.get(), &eof);
+ EXPECT_TRUE(st.ok()) << st;
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(block->rows(), 0);
+
+ auto* local_state =
&(_runtime_state.get_local_state(0)->cast<FileScanLocalState>());
+ auto* counter =
local_state->scanner_profile()->get_counter("NotFoundFileNum");
+ ASSERT_NE(counter, nullptr);
+ EXPECT_EQ(counter->value(), 1);
+
+ WARN_IF_ERROR(scanner->close(&_runtime_state), "fail to close scanner");
+}
+
+// With the skip config off, a lazy-open NOT_FOUND must surface as an error,
not be wrapped.
+TEST_F(VfileScannerExceptionTest, read_stage_not_found_errors_when_disabled) {
+ HdfsNotFoundGuard guard;
+ const bool old_ignore = config::ignore_not_found_file_in_external_table;
+ config::ignore_not_found_file_in_external_table = false;
+ Defer restore_ignore {[&]() {
config::ignore_not_found_file_in_external_table = old_ignore; }};
+
+ prepare_hdfs_csv_range();
+ std::shared_ptr<FileScanner> scanner = nullptr;
+ generate_scanner(scanner);
+
+ std::unique_ptr<Block> block(new Block());
+ bool eof = false;
+ auto st = scanner->get_block(&_runtime_state, block.get(), &eof);
+ EXPECT_FALSE(st.ok());
+ EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>()) << st;
+
+ WARN_IF_ERROR(scanner->close(&_runtime_state), "fail to close scanner");
+}
+
TEST(HiveReaderPositionMappingTest, PositionMappingUsesColumnIdxsForFileSlots)
{
TQueryOptions query_options;
query_options.hive_parquet_use_column_names = false;
diff --git
a/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
b/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
index 841e01f9cc9..72b519e2191 100644
--- a/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
+++ b/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
@@ -214,11 +214,14 @@ IcebergDeleteFileReaderOptions
delete_reader_options(RuntimeState* runtime_state
} // namespace
TEST(IcebergDeleteFileReaderHelperTest, BuildDeleteFileRange) {
- auto range = build_iceberg_delete_file_range("s3://bucket/delete.parquet");
+ auto range = build_iceberg_delete_file_range("s3://bucket/delete.parquet",
-1);
EXPECT_EQ(range.path, "s3://bucket/delete.parquet");
EXPECT_EQ(range.start_offset, 0);
EXPECT_EQ(range.size, -1);
EXPECT_EQ(range.file_size, -1);
+
+ auto range2 =
build_iceberg_delete_file_range("s3://bucket/delete.parquet", 1024);
+ EXPECT_EQ(range2.file_size, 1024);
}
TEST(IcebergDeleteFileReaderHelperTest, IsDeletionVector) {
@@ -384,7 +387,7 @@ TEST(IcebergDeleteFileReaderHelperTest,
DeletionVectorReaderValidatesOpenedFileR
IcebergDeleteFileIOContext io_context(&state);
{
- TFileRangeDesc exact_range = build_iceberg_delete_file_range(dv_path);
+ TFileRangeDesc exact_range = build_iceberg_delete_file_range(dv_path,
-1);
exact_range.start_offset = 4;
exact_range.size = dv_size - exact_range.start_offset;
DeletionVectorReader exact_reader(&state, &profile, scan_params,
exact_range,
@@ -392,6 +395,15 @@ TEST(IcebergDeleteFileReaderHelperTest,
DeletionVectorReaderValidatesOpenedFileR
const auto exact_status = exact_reader.open();
EXPECT_TRUE(exact_status.ok()) << exact_status;
+ // Manifest-style range: exact file_size provided by the FE instead of
-1.
+ TFileRangeDesc manifest_range =
build_iceberg_delete_file_range(dv_path, dv_size);
+ manifest_range.start_offset = 4;
+ manifest_range.size = dv_size - manifest_range.start_offset;
+ DeletionVectorReader manifest_reader(&state, &profile, scan_params,
manifest_range,
+ &io_context.io_ctx);
+ const auto manifest_status = manifest_reader.open();
+ EXPECT_TRUE(manifest_status.ok()) << manifest_status;
+
TFileRangeDesc oversized_range = exact_range;
oversized_range.size = MAX_ICEBERG_DELETION_VECTOR_BYTES;
DeletionVectorReader oversized_reader(&state, &profile, scan_params,
oversized_range,
diff --git a/be/test/format/table/iceberg/iceberg_reader_test.cpp
b/be/test/format/table/iceberg/iceberg_reader_test.cpp
index 6472bb6dcd5..3f5f3883c8a 100644
--- a/be/test/format/table/iceberg/iceberg_reader_test.cpp
+++ b/be/test/format/table/iceberg/iceberg_reader_test.cpp
@@ -48,6 +48,7 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
+#include "format/format_common.h"
#include "format/parquet/vparquet_column_chunk_reader.h"
#include "format/parquet/vparquet_reader.h"
#include "format/table/iceberg_default_value.h"
@@ -56,6 +57,7 @@
#include "io/fs/file_reader_writer_fwd.h"
#include "io/fs/file_system.h"
#include "io/fs/local_file_system.h"
+#include "io/io_common.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
#include "storage/olap_scan_common.h"
@@ -1650,6 +1652,69 @@ TEST_F(IcebergReaderTest,
rejects_missing_required_nested_field_before_parquet_l
EXPECT_NE(status.to_string().find("has no initial default"),
std::string::npos);
}
+// An inflated size breaks footer reads, proving the v1 position-delete path
consumes the FE file_size.
+TEST_F(IcebergReaderTest, v1_position_delete_consumes_delete_file_size) {
+ RuntimeState runtime_state = RuntimeState(TQueryOptions(),
TQueryGlobals());
+ TFileScanRangeParams scan_params;
+ scan_params.__set_file_type(TFileType::FILE_LOCAL);
+ scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+
+ TFileRangeDesc scan_range;
+ scan_range.__set_fs_name("");
+ scan_range.__set_path("data.parquet");
+ scan_range.__set_start_offset(0);
+ scan_range.__set_size(0);
+
+ RuntimeProfile profile("test_profile");
+ io::IOContext io_ctx;
+ ShardedKVCache kv_cache(8);
+
+ IcebergParquetReader iceberg_reader(nullptr, &profile, &runtime_state,
scan_params, scan_range,
+ &kv_cache, &io_ctx, cache.get());
+
+ TIcebergDeleteFileDesc delete_file;
+ delete_file.__set_content(IcebergTableReader::POSITION_DELETE);
+ delete_file.__set_path(mixed_position_delete_file());
+ delete_file.__set_file_size(1 << 30); // wrong on purpose: must reach the
reader
+
+ const auto status =
+
iceberg_reader.TEST_position_delete_base("file:///tmp/data.parquet",
{delete_file});
+
+ ASSERT_FALSE(status.ok());
+}
+
+// An inflated size must break the read, proving the v1 equality-delete path
consumes the FE file_size.
+TEST_F(IcebergReaderTest, v1_equality_delete_consumes_delete_file_size) {
+ RuntimeState runtime_state = RuntimeState(TQueryOptions(),
TQueryGlobals());
+ TFileScanRangeParams scan_params;
+ scan_params.__set_file_type(TFileType::FILE_LOCAL);
+ scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+
+ TFileRangeDesc scan_range;
+ scan_range.__set_fs_name("");
+ scan_range.__set_path("data.parquet");
+ scan_range.__set_start_offset(0);
+ scan_range.__set_size(0);
+
+ RuntimeProfile profile("test_profile");
+ io::IOContext io_ctx;
+ ShardedKVCache kv_cache(8);
+
+ IcebergParquetReader iceberg_reader(nullptr, &profile, &runtime_state,
scan_params, scan_range,
+ &kv_cache, &io_ctx, cache.get());
+
+ TIcebergDeleteFileDesc delete_file;
+ delete_file.__set_content(IcebergTableReader::EQUALITY_DELETE);
+ delete_file.__set_path(mixed_position_delete_file());
+ delete_file.__set_field_ids({0});
+ delete_file.__set_file_format(TFileFormatType::FORMAT_PARQUET);
+ delete_file.__set_file_size(1 << 30); // wrong on purpose: must reach the
reader
+
+ const auto status =
iceberg_reader.TEST_read_equality_delete_file(delete_file);
+
+ ASSERT_FALSE(status.ok());
+}
+
// Test reading real Iceberg Orc file using IcebergTableReader
TEST_F(IcebergReaderTest, read_iceberg_orc_file) {
// Read only: name, profile.address.coordinates.lat,
profile.address.coordinates.lng, profile.contact.email
diff --git a/be/test/format_v2/orc/orc_file_input_stream_test.cpp
b/be/test/format_v2/orc/orc_file_input_stream_test.cpp
index 8d63a4ab21a..bfc56f0349f 100644
--- a/be/test/format_v2/orc/orc_file_input_stream_test.cpp
+++ b/be/test/format_v2/orc/orc_file_input_stream_test.cpp
@@ -75,6 +75,25 @@ private:
io::Path _path = "/tmp/orc_v2_input_stream";
};
+// FileReader whose reads always fail with NotFound, mimicking a missing HDFS
file under lazy open.
+class NotFoundFileReader final : public io::FileReader {
+public:
+ Status close() override { return Status::OK(); }
+ const io::Path& path() const override { return _path; }
+ size_t size() const override { return 4096; }
+ bool closed() const override { return false; }
+ int64_t mtime() const override { return 0; }
+
+protected:
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const io::IOContext* io_ctx) override {
+ return Status::NotFound("failed to open /test/missing.orc: No such
file or directory");
+ }
+
+private:
+ io::Path _path = "/test/missing.orc";
+};
+
class TestStreamInformation final : public ::orc::StreamInformation {
public:
TestStreamInformation(TestStream stream, uint64_t offset) :
_stream(stream), _offset(offset) {}
@@ -374,5 +393,19 @@ TEST(OrcFileInputStreamTest,
MergedIoChildrenStayIsolatedInBothInitializationOrd
}
}
+// A failed read must carry the status text so the ORC reader can restore
NotFound (parity with v1).
+TEST(OrcFileInputStreamTest, ReadFailureCarriesNotFoundText) {
+ auto reader = std::make_shared<NotFoundFileReader>();
+ OrcFileInputStream input("missing.orc", reader, nullptr, nullptr, {});
+ std::array<char, 16> buf {};
+ try {
+ input.read(buf.data(), buf.size(), 0);
+ FAIL() << "expected orc::ParseError";
+ } catch (const ::orc::ParseError& e) {
+ const std::string msg = e.what();
+ EXPECT_NE(msg.find("No such file or directory"), std::string::npos);
+ }
+}
+
} // namespace
} // namespace doris::format::orc
diff --git a/be/test/format_v2/orc/orc_reader_test.cpp
b/be/test/format_v2/orc/orc_reader_test.cpp
index 10f3a524aeb..7d9f9604943 100644
--- a/be/test/format_v2/orc/orc_reader_test.cpp
+++ b/be/test/format_v2/orc/orc_reader_test.cpp
@@ -18,6 +18,7 @@
#include "format_v2/orc/orc_reader.h"
#include <cctz/time_zone.h>
+#include <errno.h>
#include <gtest/gtest.h>
#include <unistd.h>
@@ -4662,6 +4663,39 @@ TEST_F(NewOrcReaderTest,
AggregatePushdownReturnsCountFromFileMetadata) {
EXPECT_TRUE(aggregate_result.columns.empty());
}
+// Only ENOENT-style errors map to NotFound so FileScannerV2 does not silently
skip unhealthy splits.
+TEST_F(NewOrcReaderTest, InitKeepsInternalErrorForDirectory) {
+ auto system_properties = std::make_shared<io::FileSystemProperties>();
+ system_properties->system_type = TFileType::FILE_LOCAL;
+ auto file_description = std::make_unique<io::FileDescription>();
+ file_description->path = _test_dir; // open() on a directory succeeds,
read fails with EISDIR
+ file_description->file_size = 4096;
+ format::orc::OrcReader reader(system_properties, file_description,
nullptr, nullptr,
+ std::nullopt);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto st = reader.init(&state);
+ ASSERT_FALSE(st.ok());
+ EXPECT_TRUE(st.is<ErrorCode::INTERNAL_ERROR>()) << st;
+}
+
+// ENOENT injected at the file layer surfaces as NotFound so FileScannerV2 can
skip the split.
+TEST_F(NewOrcReaderTest, InitRestoresNotFoundFromReadFailure) {
+ const auto old_enable = config::enable_debug_points;
+ config::enable_debug_points = true;
+ const std::string point = "LocalFileReader::read_at_impl.io_error";
+ DebugPoints::instance()->add_with_params(point, {{"errno",
std::to_string(ENOENT)}});
+
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto st = reader->init(&state);
+
+ DebugPoints::instance()->remove(point);
+ config::enable_debug_points = old_enable;
+
+ ASSERT_FALSE(st.ok());
+ EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>()) << st;
+}
+
TEST_F(NewOrcReaderTest, AggregatePushdownCountUsesOnlySplitStripes) {
const auto multi_stripe_file_path = (_test_dir /
"aggregate_count_split.orc").string();
write_multi_stripe_orc_int_file(multi_stripe_file_path);
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 7585ef73568..c9b033dcb00 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -1268,12 +1268,17 @@ TIcebergDeleteFileDesc
make_iceberg_position_delete_file(const std::string& path
TIcebergDeleteFileDesc make_iceberg_equality_delete_file(
const std::string& path, const std::vector<int32_t>& field_ids,
- TFileFormatType::type file_format = TFileFormatType::FORMAT_PARQUET) {
+ TFileFormatType::type file_format = TFileFormatType::FORMAT_PARQUET,
+ int64_t file_size = -1) {
TIcebergDeleteFileDesc delete_file;
delete_file.__set_content(2);
delete_file.__set_path(path);
delete_file.__set_field_ids(field_ids);
delete_file.__set_file_format(file_format);
+ // Default callers emulate an old FE that sends no file_size.
+ if (file_size >= 0) {
+ delete_file.__set_file_size(file_size);
+ }
return delete_file;
}
@@ -5375,5 +5380,118 @@ TEST(IcebergV2ReaderTest,
DataFileIsMarkedImmutableForPageCache) {
EXPECT_TRUE(reader.current_data_file_is_immutable());
}
+TEST(IcebergV2ReaderTest, IcebergEqualityDeleteFileSizePropagatedToReader) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_eq_delete_file_size_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+
+ const auto file_path = (test_dir / "split.parquet").string();
+ const auto delete_file_path = (test_dir /
"equality-delete.parquet").string();
+ write_int_pair_parquet_file(file_path, {1, 2, 3}, {10, 20, 30}, {"one",
"two", "three"});
+ write_iceberg_equality_delete_parquet_file(delete_file_path, 0, 2);
+
+ const auto delete_file_size =
+ static_cast<int64_t>(std::filesystem::file_size(delete_file_path));
+ ASSERT_GT(delete_file_size, 0) << "delete file should not be empty";
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ // A truncated size must break the read; a correct size is
indistinguishable from the stat fallback.
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+
split_options.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ file_path, {make_iceberg_equality_delete_file(delete_file_path,
{0},
+
TFileFormatType::FORMAT_PARQUET,
+ delete_file_size -
16)}));
+
+ const bool prepare_ok = reader.prepare_split(split_options).ok();
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ const bool read_ok = prepare_ok && reader.get_block(&block, &eos).ok();
+ ASSERT_FALSE(read_ok) << "truncated file_size must fail the delete-file
read";
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+TEST(IcebergV2ReaderTest, IcebergEqualityDeleteFileSizeUnknownFallsBackToStat)
{
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_eq_delete_no_file_size_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+
+ const auto file_path = (test_dir / "split.parquet").string();
+ const auto delete_file_path = (test_dir /
"equality-delete.parquet").string();
+ write_int_pair_parquet_file(file_path, {1, 2, 3}, {10, 20, 30}, {"one",
"two", "three"});
+ write_iceberg_equality_delete_parquet_file(delete_file_path, 0, 2);
+
+ TIcebergDeleteFileDesc delete_file;
+ delete_file.__set_content(2);
+ delete_file.__set_path(delete_file_path);
+ delete_file.__set_field_ids({0});
+ delete_file.__set_file_format(TFileFormatType::FORMAT_PARQUET);
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {delete_file}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ ASSERT_EQ(block.rows(), 2);
+ const auto& id_column = assert_cast<const
ColumnInt32&>(expect_not_null_table_column(block, 0));
+ EXPECT_EQ(id_column.get_element(0), 1);
+ EXPECT_EQ(id_column.get_element(1), 3);
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
} // namespace
} // namespace doris::format
diff --git a/be/test/io/fs/file_handle_cache_test.cpp
b/be/test/io/fs/file_handle_cache_test.cpp
index 5c1f7d1d9e0..a7da38ad91a 100644
--- a/be/test/io/fs/file_handle_cache_test.cpp
+++ b/be/test/io/fs/file_handle_cache_test.cpp
@@ -19,8 +19,19 @@
#include <gtest/gtest.h>
+#include <atomic>
#include <cstdint>
+#include <cstdlib>
+#include <cstring>
#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "cpp/sync_point.h"
+#include "gen_cpp/Status_types.h"
+#include "io/fs/hdfs_file_reader.h"
+#include "util/defer_op.h"
namespace doris::io {
@@ -40,4 +51,384 @@ TEST(FileHandleCacheTest, CacheKeyIncludesHdfsFs) {
mtime + 1));
}
+// init(file_size>0) does not open the file.
+TEST(FileHandleCacheTest, InitWithKnownFileSizeDoesNotOpenFile) {
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/nonexistent/file.parquet",
12345);
+ auto st = handle.init(4096);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_EQ(handle.file_size(), 4096);
+ EXPECT_EQ(handle.file(), nullptr);
+}
+
+// Destructor is safe when file was never opened.
+TEST(FileHandleCacheTest, DestructorSafeWithoutOpen) {
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ {
+ ExclusiveHdfsFileHandle handle(mock_fs, "/nonexistent/file.parquet",
12345);
+ ASSERT_TRUE(handle.init(4096).ok());
+ EXPECT_EQ(handle.file(), nullptr);
+ }
+}
+
+// Register a SyncPoint callback that returns a fixed value.
+template <typename T>
+static void set_mock_return(const std::string& point, T value,
SyncPoint::CallbackGuard* guard) {
+ SyncPoint::get_instance()->set_call_back(
+ point,
+ [value = std::move(value)](auto&& args) {
+ auto* ret = try_any_cast_ret<T>(args);
+ ret->first = std::move(value);
+ ret->second = true;
+ },
+ guard);
+}
+
+// Mocks hdfsOpenFile/hdfsCloseFile/hdfsGetPathInfo/hdfsUnbufferFile via
SyncPoint to avoid JNI.
+struct MockHandleGuard {
+ SyncPoint::CallbackGuard open_guard;
+ SyncPoint::CallbackGuard close_guard;
+ SyncPoint::CallbackGuard info_guard;
+ SyncPoint::CallbackGuard unbuffer_guard;
+ MockHandleGuard(hdfsFile mock_file, int64_t file_size = 4096) {
+ auto* sp = SyncPoint::get_instance();
+ sp->enable_processing();
+ set_mock_return<hdfsFile>("HdfsFileHandle::ensure_open::hdfsOpenFile",
mock_file,
+ &open_guard);
+ set_mock_return<int>("HdfsFileHandle::close::hdfsCloseFile", 0,
&close_guard);
+ // Each hit allocates a fresh heap info so init()'s real
hdfsFreeFileInfo can free it safely.
+ sp->set_call_back(
+ "HdfsFileHandle::init::hdfsGetPathInfo",
+ [file_size](auto&& args) {
+ auto* ret = try_any_cast_ret<hdfsFileInfo*>(args);
+ auto* info = static_cast<hdfsFileInfo*>(calloc(1,
sizeof(hdfsFileInfo)));
+ info->mSize = file_size;
+ info->mName = strdup("/test/mock-file.parquet");
+ ret->first = info;
+ ret->second = true;
+ },
+ &info_guard);
+ set_mock_return<int>("HdfsFileHandle::close::hdfsUnbufferFile", 0,
&unbuffer_guard);
+ }
+ ~MockHandleGuard() { SyncPoint::get_instance()->disable_processing(); }
+};
+
+// ensure_open() succeeds via SyncPoint mock.
+TEST(FileHandleCacheTest, EnsureOpenSucceedsWithMock) {
+ MockHandleGuard
mg(reinterpret_cast<hdfsFile>(static_cast<uintptr_t>(0xdeadbeef)));
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/file.parquet", 12345);
+ ASSERT_TRUE(handle.init(4096).ok());
+ EXPECT_EQ(handle.file(), nullptr);
+
+ ASSERT_TRUE(handle.ensure_open().ok());
+ EXPECT_NE(handle.file(), nullptr);
+}
+
+// ensure_open() fails when mock returns nullptr.
+TEST(FileHandleCacheTest, EnsureOpenFailsWithMock) {
+ MockHandleGuard mg(nullptr);
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/file.parquet", 12345);
+ ASSERT_TRUE(handle.init(4096).ok());
+
+ auto st = handle.ensure_open();
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(handle.file(), nullptr);
+}
+
+// ensure_open() is idempotent via call_once.
+TEST(FileHandleCacheTest, EnsureOpenIsIdempotentWithMock) {
+ MockHandleGuard
mg(reinterpret_cast<hdfsFile>(static_cast<uintptr_t>(0xdeadbeef)));
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/file.parquet", 12345);
+ ASSERT_TRUE(handle.init(4096).ok());
+
+ ASSERT_TRUE(handle.ensure_open().ok());
+ ASSERT_TRUE(handle.ensure_open().ok());
+}
+
+// init(-1) fetches file_size via mocked hdfsGetPathInfo.
+TEST(FileHandleCacheTest, InitWithUnknownFileSizeWithMock) {
+ MockHandleGuard mg(nullptr, 8192);
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/file.parquet", 12345);
+
+ ASSERT_TRUE(handle.init(-1).ok());
+ EXPECT_EQ(handle.file_size(), 8192);
+ EXPECT_EQ(handle.file(), nullptr);
+}
+
+// init(-1) fails when hdfsGetPathInfo returns nullptr.
+TEST(FileHandleCacheTest, InitFailsWhenGetPathInfoReturnsNull) {
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/file.parquet", 12345);
+
+ SyncPoint::get_instance()->enable_processing();
+ Defer defer {[&]() { SyncPoint::get_instance()->disable_processing(); }};
+ SyncPoint::CallbackGuard guard;
+ set_mock_return<hdfsFileInfo*>("HdfsFileHandle::init::hdfsGetPathInfo",
nullptr, &guard);
+ SyncPoint::CallbackGuard err_guard;
+ set_mock_return<std::string>("HdfsFileHandle::init::hdfs_error",
"connection refused",
+ &err_guard);
+
+ auto st = handle.init(-1);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::INTERNAL_ERROR);
+}
+
+// init(-1) returns NotFound when the file is missing.
+TEST(FileHandleCacheTest, InitReturnsNotFoundForMissingFile) {
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/missing.parquet", 12345);
+
+ SyncPoint::get_instance()->enable_processing();
+ Defer defer {[&]() { SyncPoint::get_instance()->disable_processing(); }};
+ SyncPoint::CallbackGuard guard;
+ set_mock_return<hdfsFileInfo*>("HdfsFileHandle::init::hdfsGetPathInfo",
nullptr, &guard);
+ SyncPoint::CallbackGuard err_guard;
+ set_mock_return<std::string>("HdfsFileHandle::init::hdfs_error", "No such
file or directory",
+ &err_guard);
+
+ auto st = handle.init(-1);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::NOT_FOUND);
+}
+
+// ensure_open() returns NotFound when error contains "No such file or
directory".
+TEST(FileHandleCacheTest, EnsureOpenReturnsNotFoundForMissingFile) {
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/missing.parquet", 12345);
+ ASSERT_TRUE(handle.init(4096).ok());
+
+ SyncPoint::get_instance()->enable_processing();
+ Defer defer {[&]() { SyncPoint::get_instance()->disable_processing(); }};
+ SyncPoint::CallbackGuard open_guard;
+ set_mock_return<hdfsFile>("HdfsFileHandle::ensure_open::hdfsOpenFile",
nullptr, &open_guard);
+ SyncPoint::CallbackGuard err_guard;
+ set_mock_return<std::string>("HdfsFileHandle::ensure_open::hdfs_error",
+ "No such file or directory", &err_guard);
+
+ auto st = handle.ensure_open();
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::NOT_FOUND);
+}
+
+// --- Cache lifecycle tests ---
+
+// Helper: create a FileHandleCache with small capacity for testing.
+static std::unique_ptr<FileHandleCache> make_test_cache() {
+ return std::make_unique<FileHandleCache>(4, 1, 0);
+}
+
+// Helper: get a file handle from cache, asserting success.
+static void get_handle(FileHandleCache& cache, const hdfsFS& fs, const
std::string& fname,
+ int64_t mtime, FileHandleCache::Accessor* accessor,
bool* cache_hit) {
+ ASSERT_TRUE(cache.get_file_handle(fs, fname, mtime, 4096, false, accessor,
cache_hit).ok());
+}
+
+// ensure_open succeeds → ~Accessor releases (unbuffer OK) → next
get_file_handle hits cache.
+TEST(FileHandleCacheTest, OpenedHandleReleasedBackToCache) {
+ MockHandleGuard
mg(reinterpret_cast<hdfsFile>(static_cast<uintptr_t>(0xdeadbeef)));
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ auto cache = make_test_cache();
+ const std::string fname = "/test/opened_release.parquet";
+ constexpr int64_t mtime = 12345;
+
+ bool cache_hit = false;
+ {
+ FileHandleCache::Accessor accessor;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor, &cache_hit);
+ EXPECT_FALSE(cache_hit);
+ ASSERT_TRUE(accessor.get()->ensure_open().ok());
+ EXPECT_NE(accessor.get()->file(), nullptr);
+ }
+
+ FileHandleCache::Accessor accessor2;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor2, &cache_hit);
+#ifdef USE_HADOOP_HDFS
+ EXPECT_TRUE(cache_hit);
+ EXPECT_NE(accessor2.get()->file(), nullptr);
+#else
+ // libhdfs3: ~Accessor() destroys the opened handle, so the next lookup
must miss.
+ EXPECT_FALSE(cache_hit);
+ EXPECT_EQ(accessor2.get()->file(), nullptr);
+#endif
+}
+
+// ensure_open fails → ~Accessor destroys → next get_file_handle misses cache.
+TEST(FileHandleCacheTest, OpenFailedHandleDestroyedNotCached) {
+ MockHandleGuard mg(nullptr);
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ auto cache = make_test_cache();
+ const std::string fname = "/test/open_fail.parquet";
+ constexpr int64_t mtime = 12345;
+
+ bool cache_hit = false;
+ {
+ FileHandleCache::Accessor accessor;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor, &cache_hit);
+ EXPECT_FALSE(cache_hit);
+ auto st = accessor.get()->ensure_open();
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(accessor.get()->file(), nullptr);
+ }
+
+ FileHandleCache::Accessor accessor2;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor2, &cache_hit);
+ EXPECT_FALSE(cache_hit);
+}
+
+// read_at triggers ensure_open failure → reader destroyed → ~Accessor
destroys → cache miss.
+TEST(FileHandleCacheTest, ReadAtOpenFailedHandleDestroyedNotCached) {
+ MockHandleGuard mg(nullptr);
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ auto cache = make_test_cache();
+ const std::string fname = "/test/read_at_fail.parquet";
+ constexpr int64_t mtime = 12345;
+
+ bool cache_hit = false;
+ {
+ FileHandleCache::Accessor accessor;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor, &cache_hit);
+ EXPECT_FALSE(cache_hit);
+ auto reader =
+ std::make_shared<HdfsFileReader>(Path(fname), "hdfs",
std::move(accessor), mtime);
+ char buf[16];
+ size_t bytes_read = 0;
+ auto st = reader->read_at(0, {buf, sizeof(buf)}, &bytes_read, nullptr);
+ ASSERT_FALSE(st.ok());
+ }
+
+ FileHandleCache::Accessor accessor2;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor2, &cache_hit);
+ EXPECT_FALSE(cache_hit);
+}
+
+// Serial ensure_open() preserves Status inside call_once.
+TEST(FileHandleCacheTest, OpenFailurePreservesStatusInsideCallOnce) {
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/serial_fail.parquet",
12345);
+ ASSERT_TRUE(handle.init(4096).ok());
+
+ auto* sp = SyncPoint::get_instance();
+ sp->enable_processing();
+ Defer defer {[&]() { sp->disable_processing(); }};
+ SyncPoint::CallbackGuard open_guard;
+ set_mock_return<hdfsFile>("HdfsFileHandle::ensure_open::hdfsOpenFile",
nullptr, &open_guard);
+ SyncPoint::CallbackGuard err_guard;
+ set_mock_return<std::string>("HdfsFileHandle::ensure_open::hdfs_error",
+ "No such file or directory", &err_guard);
+
+ // First call opens and fails -> NotFound.
+ auto st1 = handle.ensure_open();
+ ASSERT_FALSE(st1.ok());
+ EXPECT_EQ(st1.code(), TStatusCode::NOT_FOUND);
+
+ // Change hdfs_error mock to a different value; _open_status must be
preserved.
+ sp->clear_call_back("HdfsFileHandle::ensure_open::hdfs_error");
+ set_mock_return<std::string>("HdfsFileHandle::ensure_open::hdfs_error",
"Permission denied",
+ &err_guard);
+
+ auto st2 = handle.ensure_open();
+ ASSERT_FALSE(st2.ok());
+ EXPECT_EQ(st2.code(), TStatusCode::NOT_FOUND);
+}
+
+// Concurrent ensure_open() must return the same Status to all callers.
+TEST(FileHandleCacheTest, ConcurrentOpenFailureReturnsSameStatusToAllCallers) {
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ ExclusiveHdfsFileHandle handle(mock_fs, "/test/concurrent_fail.parquet",
12345);
+ ASSERT_TRUE(handle.init(4096).ok());
+
+ auto* sp = SyncPoint::get_instance();
+ sp->enable_processing();
+ Defer defer {[&]() { sp->disable_processing(); }};
+ SyncPoint::CallbackGuard open_guard;
+ set_mock_return<hdfsFile>("HdfsFileHandle::ensure_open::hdfsOpenFile",
nullptr, &open_guard);
+
+ // Simulate libhdfs thread-local last-error: only first call returns real
message.
+ std::atomic<int> err_call_count {0};
+ SyncPoint::CallbackGuard err_guard;
+ sp->set_call_back(
+ "HdfsFileHandle::ensure_open::hdfs_error",
+ [&err_call_count](auto&& args) {
+ auto* ret = try_any_cast_ret<std::string>(args);
+ if (err_call_count.fetch_add(1) == 0) {
+ ret->first = "No such file or directory";
+ } else {
+ ret->first = "";
+ }
+ ret->second = true;
+ },
+ &err_guard);
+
+ constexpr int kNumThreads = 8;
+ std::vector<std::thread> threads;
+ std::vector<TStatusCode::type> codes(kNumThreads);
+ for (int i = 0; i < kNumThreads; ++i) {
+ threads.emplace_back([&handle, &codes, i]() {
+ auto st = handle.ensure_open();
+ codes[i] = static_cast<TStatusCode::type>(st.code());
+ });
+ }
+ for (auto& t : threads) {
+ t.join();
+ }
+
+ for (int i = 0; i < kNumThreads; ++i) {
+ EXPECT_EQ(codes[i], TStatusCode::NOT_FOUND) << "thread " << i << " got
code " << codes[i];
+ }
+}
+
+// A read after close must fail before lazy open: no hdfsOpenFile call is
allowed.
+TEST(FileHandleCacheTest, ReadAfterCloseSkipsLazyOpen) {
+ MockHandleGuard
mg(reinterpret_cast<hdfsFile>(static_cast<uintptr_t>(0xdeadbeef)));
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ auto cache = make_test_cache();
+ const std::string fname = "/test/close_before_read.parquet";
+ constexpr int64_t mtime = 12345;
+
+ bool cache_hit = false;
+ FileHandleCache::Accessor accessor;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor, &cache_hit);
+ auto reader = std::make_shared<HdfsFileReader>(Path(fname), "hdfs",
std::move(accessor), mtime);
+
+ ASSERT_TRUE(reader->close().ok());
+ char buf[16];
+ size_t bytes_read = 0;
+ auto st = reader->read_at(0, {buf, sizeof(buf)}, &bytes_read, nullptr);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::INTERNAL_ERROR);
+ // "read closed file" proves the closed guard fired before lazy open.
+ EXPECT_NE(st.to_string().find("read closed file"), std::string::npos);
+}
+
+// Second read_at after a failed read must not dereference null _handle.
+TEST(FileHandleCacheTest, SecondReadAfterFailureDoesNotCrash) {
+ MockHandleGuard
mg(reinterpret_cast<hdfsFile>(static_cast<uintptr_t>(0xdeadbeef)));
+ auto mock_fs = reinterpret_cast<hdfsFS>(static_cast<uintptr_t>(0x1));
+ auto cache = make_test_cache();
+ const std::string fname = "/test/second_read.parquet";
+ constexpr int64_t mtime = 12345;
+
+ bool cache_hit = false;
+ FileHandleCache::Accessor accessor;
+ get_handle(*cache, mock_fs, fname, mtime, &accessor, &cache_hit);
+ auto reader = std::make_shared<HdfsFileReader>(Path(fname), "hdfs",
std::move(accessor), mtime);
+
+ char buf[16];
+ size_t bytes_read = 0;
+ // offset > file_size(4096) -> IOError -> read_at_impl sets
_handle=nullptr.
+ auto st1 = reader->read_at(5000, {buf, sizeof(buf)}, &bytes_read, nullptr);
+ ASSERT_FALSE(st1.ok());
+ // Confirm error came from do_read_at_impl's offset guard (sets
_handle=nullptr).
+ EXPECT_EQ(st1.code(), TStatusCode::IO_ERROR);
+
+ // Second read must not crash; should return InternalError about destroyed
handle.
+ auto st2 = reader->read_at(0, {buf, sizeof(buf)}, &bytes_read, nullptr);
+ ASSERT_FALSE(st2.ok());
+ EXPECT_EQ(st2.code(), TStatusCode::INTERNAL_ERROR);
+}
+
} // namespace doris::io
diff --git a/be/test/io/fs/hdfs_file_system_test.cpp
b/be/test/io/fs/hdfs_file_system_test.cpp
index db443951a26..5d6cef97f2e 100644
--- a/be/test/io/fs/hdfs_file_system_test.cpp
+++ b/be/test/io/fs/hdfs_file_system_test.cpp
@@ -15,14 +15,22 @@
// specific language governing permissions and limitations
// under the License.
+#include "io/fs/hdfs_file_system.h"
+
#include <gtest/gtest.h>
+#include <map>
+#include <string>
+
#include "common/config.h"
#include "cpp/sync_point.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Status_types.h"
#include "io/fs/file_reader.h"
#include "io/fs/file_writer.h"
#include "io/fs/hdfs_file_writer.h"
#include "io/fs/local_file_system.h"
+#include "util/defer_op.h"
namespace doris {
@@ -151,4 +159,66 @@ TEST(HdfsFileSystemTest, Write) {
st = local_fs->delete_directory(test_dir);
}
+// Guarded: the enable_java_support check is compiled out on libhdfs3 builds.
+#ifdef USE_HADOOP_HDFS
+// create() returns error when java support is disabled.
+TEST(HdfsFileSystemTest, CreateFailsWhenJavaSupportDisabled) {
+ const bool old_enable_java_support = config::enable_java_support;
+ config::enable_java_support = false;
+ Defer defer {[&]() { config::enable_java_support =
old_enable_java_support; }};
+ std::map<std::string, std::string> properties;
+ auto res = io::HdfsFileSystem::create(properties, "hdfs://namenode:8020",
"test_id", "/");
+ ASSERT_FALSE(res.has_value());
+ EXPECT_NE(res.error().to_string().find("enable_java_support"),
std::string::npos);
+}
+#endif
+
+// open_file_internal returns IOError when _fs_handler is null.
+TEST(HdfsFileSystemTest, OpenFileFailsWithoutHandler) {
+ THdfsParams params;
+ io::HdfsFileSystem fs(params, "hdfs://namenode:8020", "test_id", "/");
+ io::FileReaderSPtr reader;
+ auto st = fs.open_file_internal("/test/file.parquet", &reader,
io::FileReaderOptions::DEFAULT);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::IO_ERROR);
+}
+
+// file_size_impl returns IOError when _fs_handler is null.
+TEST(HdfsFileSystemTest, FileSizeFailsWithoutHandler) {
+ THdfsParams params;
+ io::HdfsFileSystem fs(params, "hdfs://namenode:8020", "test_id", "/");
+ int64_t file_size = 0;
+ auto st = fs.file_size_impl("/test/file.parquet", &file_size);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::IO_ERROR);
+}
+
+// exists_impl returns IOError when _fs_handler is null.
+TEST(HdfsFileSystemTest, ExistsFailsWithoutHandler) {
+ THdfsParams params;
+ io::HdfsFileSystem fs(params, "hdfs://namenode:8020", "test_id", "/");
+ bool exists = false;
+ auto st = fs.exists_impl("/test/file.parquet", &exists);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::IO_ERROR);
+}
+
+// create_directory_impl returns IOError when _fs_handler is null.
+TEST(HdfsFileSystemTest, CreateDirectoryFailsWithoutHandler) {
+ THdfsParams params;
+ io::HdfsFileSystem fs(params, "hdfs://namenode:8020", "test_id", "/");
+ auto st = fs.create_directory_impl("/test/dir", false);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::IO_ERROR);
+}
+
+// rename_impl returns IOError when _fs_handler is null.
+TEST(HdfsFileSystemTest, RenameFailsWithoutHandler) {
+ THdfsParams params;
+ io::HdfsFileSystem fs(params, "hdfs://namenode:8020", "test_id", "/");
+ auto st = fs.rename_impl("/test/old.parquet", "/test/new.parquet");
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(st.code(), TStatusCode::IO_ERROR);
+}
+
} // namespace doris
diff --git a/be/test/storage/segment/inverted_index_file_reader_test.cpp
b/be/test/storage/segment/inverted_index_file_reader_test.cpp
index ba14fbb8359..8af14c095ab 100644
--- a/be/test/storage/segment/inverted_index_file_reader_test.cpp
+++ b/be/test/storage/segment/inverted_index_file_reader_test.cpp
@@ -16,6 +16,7 @@
// under the License.
#include <CLucene.h>
+#include <errno.h>
#include <gtest/gtest.h>
#include <unistd.h>
@@ -23,6 +24,7 @@
#include <memory>
#include <string>
+#include "common/config.h"
#include "io/fs/local_file_system.h"
#include "runtime/exec_env.h"
#include "storage/data_dir.h"
@@ -34,6 +36,7 @@
#include "storage/options.h"
#include "storage/storage_engine.h"
#include "storage/tablet/tablet_schema.h"
+#include "util/debug_points.h"
namespace doris::segment_v2 {
@@ -254,6 +257,30 @@ TEST_F(InvertedIndexFileReaderTest,
TestUnknownIndexFormatError) {
status.msg().find("CLuceneError") != std::string::npos);
}
+// A read-stage NotFound must surface as INVERTED_INDEX_FILE_NOT_FOUND so
callers can downgrade.
+TEST_F(InvertedIndexFileReaderTest, TestV2ReadNotFoundReturnsFileNotFound) {
+ std::string index_path = kTestDir + "/read_not_found_index_file";
+ create_invalid_version_file(index_path + ".idx", 1);
+
+ InvertedIndexFileInfo file_info;
+ file_info.set_index_size(1024); // size known: init() skips stat, open
succeeds
+
+ IndexFileReader reader(io::global_local_filesystem(), index_path,
+ InvertedIndexStorageFormatPB::V2, file_info);
+
+ const auto old_enable = config::enable_debug_points;
+ config::enable_debug_points = true;
+ const std::string point = "LocalFileReader::read_at_impl.io_error";
+ DebugPoints::instance()->add_with_params(
+ point, {{"errno", std::to_string(ENOENT)}, {"sub_path",
"read_not_found"}});
+ Status status = reader.init(4096);
+ DebugPoints::instance()->remove(point);
+ config::enable_debug_points = old_enable;
+
+ EXPECT_FALSE(status.ok());
+ EXPECT_EQ(status.code(), ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND);
+}
+
// Test case for V1 format file not found error
TEST_F(InvertedIndexFileReaderTest, TestV1FileNotFoundError) {
std::string index_path = kTestDir + "/non_existent";
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
index 10b119eda81..3b7ba8150ad 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
@@ -414,6 +414,7 @@ public class IcebergScanNode extends FileQueryScanNode {
String deleteFilePath = filter.getDeleteFilePath();
LocationPath locationPath = LocationPath.of(deleteFilePath,
icebergSplit.getConfig());
deleteFileDesc.setPath(locationPath.toStorageLocation().toString());
+ deleteFileDesc.setFileSize(filter.getFilesize());
setDeleteFileFormat(deleteFileDesc, filter.getFileformat());
if (filter instanceof IcebergDeleteFileFilter.PositionDelete) {
IcebergDeleteFileFilter.PositionDelete positionDelete =
@@ -549,6 +550,9 @@ public class IcebergScanNode extends FileQueryScanNode {
deleteFileDesc.setOriginalPath(icebergSplit.getPositionDeleteOriginalPath());
deleteFileDesc.setFileFormat(icebergSplit.getPositionDeleteFileFormat());
deleteFileDesc.setContent(icebergSplit.getPositionDeleteContent());
+ if (rangeDesc.isSetFileSize()) {
+ deleteFileDesc.setFileSize(rangeDesc.getFileSize());
+ }
if (icebergSplit.getPositionDeleteContentOffset() != null) {
deleteFileDesc.setContentOffset(icebergSplit.getPositionDeleteContentOffset());
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
index 036c63569ba..c0bbc4b91ff 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
@@ -1855,6 +1855,83 @@ public class IcebergScanNodeTest {
.isSetEqualityDeleteSchema());
}
+ @Test
+ public void testDeleteFileSizePropagatedToThrift() throws Exception {
+ Types.NestedField id = Types.NestedField.required(1, "id",
Types.LongType.get());
+ Schema schema = new Schema(1, ImmutableList.of(id));
+ Snapshot snapshot = mockSnapshot(1001L, schema, null);
+ TableMetadata metadata = Mockito.mock(TableMetadata.class);
+ Mockito.when(metadata.schemas()).thenReturn(ImmutableList.of(schema));
+ Mockito.when(metadata.schemasById()).thenReturn(ImmutableMap.of(1,
schema));
+ Mockito.when(metadata.snapshot(1001L)).thenReturn(snapshot);
+ TableOperations operations = Mockito.mock(TableOperations.class);
+ Mockito.when(operations.current()).thenReturn(metadata);
+ BaseTable table = new BaseTable(operations, "test");
+ TableScan tableScan = Mockito.mock(TableScan.class);
+ Mockito.when(tableScan.snapshot()).thenReturn(snapshot);
+
+ TestIcebergScanNode node = new TestIcebergScanNode(new
SessionVariable());
+ setIcebergTable(node, table);
+ node.setTableScan(tableScan);
+ setPrivateField(node, "plannedScanSchema", schema);
+ setPrivateField(node, "storagePropertiesMap", Collections.emptyMap());
+ setPrivateField(node, "formatVersion", 2);
+ setPrivateField(node, "orderedPathPartitionKeys",
Collections.emptyList());
+ setPrivateField(node, "orderedPartitionMetadataKeys",
Collections.emptyList());
+
+ DeleteFile positionDelete = Mockito.mock(DeleteFile.class);
+
Mockito.when(positionDelete.content()).thenReturn(FileContent.POSITION_DELETES);
+ Mockito.when(positionDelete.recordCount()).thenReturn(1L);
+
Mockito.when(positionDelete.path()).thenReturn("file:///tmp/pos-delete.parquet");
+ Mockito.when(positionDelete.fileSizeInBytes()).thenReturn(96L);
+ Mockito.when(positionDelete.format()).thenReturn(FileFormat.PARQUET);
+
+ DeleteFile deletionVector = Mockito.mock(DeleteFile.class);
+
Mockito.when(deletionVector.content()).thenReturn(FileContent.POSITION_DELETES);
+ Mockito.when(deletionVector.recordCount()).thenReturn(1L);
+
Mockito.when(deletionVector.path()).thenReturn("file:///tmp/dv.puffin");
+ Mockito.when(deletionVector.fileSizeInBytes()).thenReturn(256L);
+ Mockito.when(deletionVector.format()).thenReturn(FileFormat.PUFFIN);
+ Mockito.when(deletionVector.contentOffset()).thenReturn(16L);
+ Mockito.when(deletionVector.contentSizeInBytes()).thenReturn(64L);
+
Mockito.when(deletionVector.referencedDataFile()).thenReturn("file:///tmp/data.parquet");
+
+ DeleteFile equalityDelete = equalityDeleteFile(1,
"file:///tmp/eq-delete.parquet");
+
+ DataFile dataFile = Mockito.mock(DataFile.class);
+ Mockito.when(dataFile.path()).thenReturn("file:///tmp/data.parquet");
+ Mockito.when(dataFile.fileSizeInBytes()).thenReturn(128L);
+ Mockito.when(dataFile.format()).thenReturn(FileFormat.PARQUET);
+ FileScanTask task = Mockito.mock(FileScanTask.class);
+ Mockito.when(task.file()).thenReturn(dataFile);
+ Mockito.when(task.start()).thenReturn(0L);
+ Mockito.when(task.length()).thenReturn(128L);
+ Mockito.when(task.deletes())
+ .thenReturn(ImmutableList.of(positionDelete, deletionVector,
equalityDelete));
+
+ IcebergSplit split = createIcebergSplit(node, task);
+ TFileRangeDesc rangeDesc = new TFileRangeDesc();
+ setIcebergParams(node, rangeDesc, split);
+
+ List<TIcebergDeleteFileDesc> deletes =
rangeDesc.getTableFormatParams().getIcebergParams()
+ .getDeleteFiles();
+ Assert.assertEquals(3, deletes.size());
+ TIcebergDeleteFileDesc posDesc = deletes.get(0);
+ Assert.assertEquals(1, posDesc.getContent());
+ Assert.assertTrue(posDesc.isSetFileSize());
+ Assert.assertEquals(96L, posDesc.getFileSize());
+ TIcebergDeleteFileDesc dvDesc = deletes.get(1);
+ Assert.assertEquals(3, dvDesc.getContent());
+ Assert.assertEquals(16L, dvDesc.getContentOffset());
+ Assert.assertEquals(64L, dvDesc.getContentSizeInBytes());
+ Assert.assertTrue(dvDesc.isSetFileSize());
+ Assert.assertEquals(256L, dvDesc.getFileSize());
+ TIcebergDeleteFileDesc eqDesc = deletes.get(2);
+ Assert.assertEquals(2, eqDesc.getContent());
+ Assert.assertTrue(eqDesc.isSetFileSize());
+ Assert.assertEquals(64L, eqDesc.getFileSize());
+ }
+
@Test
public void testSchemaCarrierKeepsDroppedNestedEqualityFieldPath() throws
Exception {
Types.NestedField id = Types.NestedField.required(1, "id",
Types.LongType.get());
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 04c859a30d8..8eeef28017b 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -317,6 +317,7 @@ struct TIcebergDeleteFileDesc {
9: optional string original_path;
// Referenced data file path. Required to materialize rows from deletion
vectors.
10: optional string referenced_data_file_path;
+ 11: optional i64 file_size;
}
struct TIcebergFileDesc {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
