This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 17899d6080b704076c9adc7fc60c7fa4fb94d0b3 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]
