[fix](io) Fix iceberg delete file file_size propagation and add HDFS read path bvar

Author: cambyzju

be/src/format/table/iceberg_delete_file_reader_helper.cpp

@@ -241,8 +241,7 @@ TFileRangeDesc build_iceberg_delete_file_range(const std::string& path, int64_t
     range.path = path;
     range.start_offset = 0;
     range.size = -1;
-    // thrift optional defaults to 0 when unset; treat 0 as unknown (-1)
-    range.file_size = (file_size <= 0) ? -1 : file_size;
+    range.file_size = file_size;
     return range;
 }
 
@@ -277,8 +276,9 @@ 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, delete_file.file_size);
+    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);
     }
@@ -350,8 +350,9 @@ 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, delete_file.file_size);
+    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);
     }

be/src/format_v2/table/iceberg_reader.cpp

@@ -1165,7 +1165,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 = deletion_vector->file_size;
+    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();
@@ -1521,7 +1521,9 @@ 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, delete_file.file_size);
+    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);

be/src/io/fs/file_handle_cache.cpp

@@ -27,13 +27,16 @@
 
 #include "common/cast_set.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 {
 
 HdfsFileHandle::~HdfsFileHandle() {
     if (_hdfs_file != nullptr && _fs != nullptr) {
         VLOG_FILE << "hdfsCloseFile() fid=" << _hdfs_file;
+        SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_close_latency);
         hdfsCloseFile(_fs, _hdfs_file); // TODO: check return code
     }
     _fs = nullptr;
@@ -57,6 +60,7 @@ Status HdfsFileHandle::init(int64_t file_size) {
 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 = hdfsOpenFile(_fs, _fname.c_str(), O_RDONLY, 0, 0, 0);
     });
     if (_hdfs_file == nullptr) {

be/src/io/fs/hdfs_file_reader.cpp

@@ -38,6 +38,7 @@
 #include "runtime/workload_management/io_throttle.h"
 #include "runtime/workload_management/resource_context.h"
 #include "service/backend_options.h"
+#include "util/bvar_helper.h"
 
 namespace doris::io {
 
@@ -164,8 +165,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);
@@ -230,8 +235,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

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"
 
@@ -120,7 +121,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());
     }

be/test/format_v2/table/iceberg_reader_test.cpp

@@ -1203,14 +1203,14 @@ 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);
-    // Set file_size to actual file size, simulating FE propagation from iceberg manifest
-    delete_file.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(path)));
+    delete_file.__set_file_size(file_size);
     return delete_file;
 }
 
@@ -5062,7 +5062,9 @@ TEST(IcebergV2ReaderTest, IcebergEqualityDeleteFileSizePropagatedToReader) {
     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})}));
+            file_path, {make_iceberg_equality_delete_file(delete_file_path, {0},
+                                                            TFileFormatType::FORMAT_PARQUET,
+                                                            delete_file_size)}));
     ASSERT_TRUE(reader.prepare_split(split_options).ok());
 
     Block block = build_table_block(projected_columns);

be/test/io/fs/file_handle_cache_test.cpp

@@ -55,17 +55,4 @@ TEST(FileHandleCacheTest, InitWithKnownFileSizeDoesNotOpenFile) {
     EXPECT_EQ(handle.file(), nullptr);
 }
 
-// Verify that build_iceberg_delete_file_range treats file_size <= 0 as unknown (-1).
-// This covers the case where thrift optional file_size defaults to 0.
-TEST(FileHandleCacheTest, BuildDeleteFileRangeTreatsZeroAsUnknown) {
-    auto range_known = build_iceberg_delete_file_range("s3://b/f.parquet", 1024);
-    EXPECT_EQ(range_known.file_size, 1024);
-
-    auto range_zero = build_iceberg_delete_file_range("s3://b/f.parquet", 0);
-    EXPECT_EQ(range_zero.file_size, -1);
-
-    auto range_neg = build_iceberg_delete_file_range("s3://b/f.parquet", -1);
-    EXPECT_EQ(range_neg.file_size, -1);
-}
-
 } // namespace doris::io