fix core&&format

Author: hubgeter

be/src/vec/exec/format/table/deletion_vector_reader.cpp

@@ -15,70 +15,77 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "util/block_compression.h"
+#include "deletion_vector_reader.h"
+
 #include "rapidjson/document.h"
 #include "rapidjson/stringbuffer.h"
-
-#include "deletion_vector_reader.h"
+#include "util/block_compression.h"
 
 namespace doris {
-    namespace vectorized {
-        Status DeletionVectorReader::open() {
-            if (_is_opened) [[unlikely]] {
-                return Status::OK();
-            }
+namespace vectorized {
+Status DeletionVectorReader::open() {
+    if (_is_opened) [[unlikely]] {
+        return Status::OK();
+    }
+
+    _init_system_properties();
+    _init_file_description();
+    RETURN_IF_ERROR(_create_file_reader());
 
-            _init_system_properties();
-            _init_file_description();
-            RETURN_IF_ERROR(_create_file_reader());
+    _file_size = _file_reader->size();
+    _is_opened = true;
+    return Status::OK();
+}
 
-            _file_size = _file_reader->size();
-            _is_opened = true;
-            return Status::OK();
-        }
+Status DeletionVectorReader::read_at(size_t offset, Slice result) {
+    if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) {
+        return Status::EndOfFile("stop read.");
+    }
+    size_t bytes_read = 0;
+    RETURN_IF_ERROR(_file_reader->read_at(offset, result, &bytes_read, _io_ctx));
+    if (bytes_read != result.size) [[unlikely]] {
+        return Status::IOError("Failed to read fully at offset {}, expected {}, got {}", offset,
+                               result.size, bytes_read);
+    }
+    return Status::OK();
+}
 
-        Status DeletionVectorReader::read_at(size_t offset, Slice result) {
-            size_t bytes_read = 0;
-            RETURN_IF_ERROR(_file_reader->read_at(offset, result, &bytes_read));
-            if (bytes_read != result.size) [[unlikely]] {
-                return Status::IOError("Failed to read fully at offset {}, expected {}, got {}",
-                                       offset, result.size, bytes_read);
-            }
-            return Status::OK();
-        }
+Status DeletionVectorReader::_create_file_reader() {
+    if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) {
+        return Status::EndOfFile("stop read.");
+    }
 
-        Status DeletionVectorReader::_create_file_reader() {
-            _file_description.mtime = _range.__isset.modification_time ? _range.modification_time : 0;
-            io::FileReaderOptions reader_options =
-                    FileFactory::get_reader_options(_state, _file_description);
-            _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
-                    _profile, _system_properties, _file_description, reader_options,
-                    io::DelegateReader::AccessMode::RANDOM, _io_ctx));
-            return Status::OK();
-        }
+    _file_description.mtime = _range.__isset.modification_time ? _range.modification_time : 0;
+    io::FileReaderOptions reader_options =
+            FileFactory::get_reader_options(_state, _file_description);
+    _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
+            _profile, _system_properties, _file_description, reader_options,
+            io::DelegateReader::AccessMode::RANDOM, _io_ctx));
+    return Status::OK();
+}
 
-        void DeletionVectorReader::_init_file_description() {
-            _file_description.path = _range.path;
-            _file_description.file_size = _range.__isset.file_size ? _range.file_size : -1;
-            if (_range.__isset.fs_name) {
-                _file_description.fs_name = _range.fs_name;
-            }
-        }
+void DeletionVectorReader::_init_file_description() {
+    _file_description.path = _range.path;
+    _file_description.file_size = _range.__isset.file_size ? _range.file_size : -1;
+    if (_range.__isset.fs_name) {
+        _file_description.fs_name = _range.fs_name;
+    }
+}
 
-        void DeletionVectorReader::_init_system_properties() {
-            if (_range.__isset.file_type) {
-                // for compatibility
-                _system_properties.system_type = _range.file_type;
-            } else {
-                _system_properties.system_type = _params.file_type;
-            }
-            _system_properties.properties = _params.properties;
-            _system_properties.hdfs_params = _params.hdfs_params;
-            if (_params.__isset.broker_addresses) {
-                _system_properties.broker_addresses.assign(_params.broker_addresses.begin(),
-                                                           _params.broker_addresses.end());
-            }
-        }
+void DeletionVectorReader::_init_system_properties() {
+    if (_range.__isset.file_type) {
+        // for compatibility
+        _system_properties.system_type = _range.file_type;
+    } else {
+        _system_properties.system_type = _params.file_type;
+    }
+    _system_properties.properties = _params.properties;
+    _system_properties.hdfs_params = _params.hdfs_params;
+    if (_params.__isset.broker_addresses) {
+        _system_properties.broker_addresses.assign(_params.broker_addresses.begin(),
+                                                   _params.broker_addresses.end());
+    }
+}
 
-    } // namespace iceberg
+} // namespace vectorized
 } // namespace doris

be/src/vec/exec/format/table/deletion_vector_reader.h

@@ -16,57 +16,54 @@
 // under the License.
 
 #include <cstdint>
+#include <memory>
 #include <string>
 #include <vector>
-#include <memory>
 
 #include "common/status.h"
+#include "io/file_factory.h"
+#include "io/fs/buffered_reader.h"
 #include "io/fs/file_reader.h"
-#include "util/slice.h"
 #include "roaring/roaring64map.hh"
-#include "io/fs/file_reader.h"
 #include "util/profile_collector.h"
+#include "util/slice.h"
 #include "vec/exec/format/generic_reader.h"
-#include "common/status.h"
-#include "io/file_factory.h"
-#include "io/fs/buffered_reader.h"
 
 namespace io {
-    struct IOContext;
+struct IOContext;
 } // namespace io
 
-
 namespace doris {
-    namespace vectorized {
-        class DeletionVectorReader {
-            ENABLE_FACTORY_CREATOR(DeletionVectorReader);
-        public:
-            DeletionVectorReader(RuntimeState* state, RuntimeProfile* profile , const TFileScanRangeParams& params,
-                         const TFileRangeDesc& range,io::IOContext* io_ctx)
-                    :  _state(state), _profile(profile),_range(range),_params(params) ,_io_ctx(io_ctx) {
-            }
-            ~DeletionVectorReader() = default;
-            Status open();
-            Status read_at(size_t offset, Slice result);
-        private:
+namespace vectorized {
+class DeletionVectorReader {
+    ENABLE_FACTORY_CREATOR(DeletionVectorReader);
 
-            void _init_system_properties();
-            void _init_file_description();
-            Status _create_file_reader();
+public:
+    DeletionVectorReader(RuntimeState* state, RuntimeProfile* profile,
+                         const TFileScanRangeParams& params, const TFileRangeDesc& range,
+                         io::IOContext* io_ctx)
+            : _state(state), _profile(profile), _range(range), _params(params), _io_ctx(io_ctx) {}
+    ~DeletionVectorReader() = default;
+    Status open();
+    Status read_at(size_t offset, Slice result);
 
-        private:
-            RuntimeState* _state = nullptr;
-            RuntimeProfile* _profile = nullptr;
-            const TFileRangeDesc& _range;
-            const TFileScanRangeParams& _params;
-            io::IOContext* _io_ctx = nullptr;
+private:
+    void _init_system_properties();
+    void _init_file_description();
+    Status _create_file_reader();
 
-            io::FileSystemProperties _system_properties;
-            io::FileDescription _file_description;
-            io::FileReaderSPtr _file_reader;
-            int64_t _file_size = 0;
-            bool _is_opened = false;
-        };
-    } // namespace vectorized
-} // namespace doris
+private:
+    RuntimeState* _state = nullptr;
+    RuntimeProfile* _profile = nullptr;
+    const TFileRangeDesc& _range;
+    const TFileScanRangeParams& _params;
+    io::IOContext* _io_ctx = nullptr;
 
+    io::FileSystemProperties _system_properties;
+    io::FileDescription _file_description;
+    io::FileReaderSPtr _file_reader;
+    int64_t _file_size = 0;
+    bool _is_opened = false;
+};
+} // namespace vectorized
+} // namespace doris