[refactor](libhdfs) introduce hadoop libhdfs

.gitignore

@@ -100,4 +100,3 @@ data_test
 lru_cache_test
 
 /conf/log4j2-spring.xml
-/fe/fe-core/src/test/resources/real-help-resource.zip

be/CMakeLists.txt

@@ -420,9 +420,6 @@ set_target_properties(k5crypto PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/li
 add_library(gssapi_krb5 STATIC IMPORTED)
 set_target_properties(gssapi_krb5 PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/libgssapi_krb5.a)
 
-add_library(hdfs3 STATIC IMPORTED)
-set_target_properties(hdfs3 PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/libhdfs3.a)
-
 find_program(THRIFT_COMPILER thrift ${CMAKE_SOURCE_DIR}/bin)
 
 if (OS_MACOSX)
@@ -685,7 +682,6 @@ set(DORIS_LINK_LIBS
     ${WL_START_GROUP}
     Agent
     Common
-    Env
     Exec
     Exprs
     Gutil
@@ -771,12 +767,37 @@ set(COMMON_THIRDPARTY
     # put this after lz4 to avoid using lz4 lib in librdkafka
     librdkafka_cpp
     librdkafka
-    hdfs3
     xml2
     lzma
     simdjson
 )
 
+if (ARCH_AMD64)
+    add_library(hadoop_hdfs STATIC IMPORTED)
+    set_target_properties(hadoop_hdfs PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/hadoop_hdfs/native/libhdfs.a)
+
+    add_library(jvm SHARED IMPORTED)
+    FILE(GLOB_RECURSE LIB_JVM $ENV{JAVA_HOME}/jre/lib/*/libjvm.so)
+    set_target_properties(jvm PROPERTIES IMPORTED_LOCATION ${LIB_JVM})
+
+    set(COMMON_THIRDPARTY
+        ${COMMON_THIRDPARTY}
+        hadoop_hdfs
+        jvm
+    )
+    add_definitions(-DUSE_HADOOP_HDFS)
+else()
+    add_library(hdfs3 STATIC IMPORTED)
+    set_target_properties(hdfs3 PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/libhdfs3.a)
+
+    # TODO: use arm hadoop hdfs to replace this
+    set(COMMON_THIRDPARTY
+        ${COMMON_THIRDPARTY}
+        hdfs3
+    )
+    add_definitions(-DUSE_LIBHDFS3)
+endif()
+
 if (absl_FOUND)
     set(COMMON_THIRDPARTY
         ${COMMON_THIRDPARTY}
@@ -944,7 +965,6 @@ endif()
 
 add_subdirectory(${SRC_DIR}/agent)
 add_subdirectory(${SRC_DIR}/common)
-add_subdirectory(${SRC_DIR}/env)
 add_subdirectory(${SRC_DIR}/exec)
 add_subdirectory(${SRC_DIR}/exprs)
 add_subdirectory(${SRC_DIR}/gen_cpp)

be/src/agent/task_worker_pool.cpp

@@ -32,9 +32,9 @@
 #include "agent/utils.h"
 #include "common/logging.h"
 #include "common/status.h"
-#include "env/env.h"
 #include "gen_cpp/Types_types.h"
 #include "gutil/strings/substitute.h"
+#include "io/fs/local_file_system.h"
 #include "io/fs/s3_file_system.h"
 #include "olap/data_dir.h"
 #include "olap/olap_common.h"
@@ -53,7 +53,6 @@
 #include "runtime/snapshot_loader.h"
 #include "service/backend_options.h"
 #include "util/doris_metrics.h"
-#include "util/file_utils.h"
 #include "util/random.h"
 #include "util/scoped_cleanup.h"
 #include "util/stopwatch.hpp"
@@ -1519,10 +1518,16 @@ void TaskWorkerPool::_make_snapshot_thread_callback() {
             // list and save all snapshot files
             // snapshot_path like: data/snapshot/20180417205230.1.86400
             // we need to add subdir: tablet_id/schema_hash/
-            std::stringstream ss;
-            ss << snapshot_path << "/" << snapshot_request.tablet_id << "/"
-               << snapshot_request.schema_hash << "/";
-            status = FileUtils::list_files(Env::Default(), ss.str(), &snapshot_files);
+            std::vector<io::FileInfo> files;
+            bool exists = true;
+            io::Path path = fmt::format("{}/{}/{}/", snapshot_path, snapshot_request.tablet_id,
+                                        snapshot_request.schema_hash);
+            status = io::global_local_filesystem()->list(path, true, &files, &exists);
+            if (status.ok()) {
+                for (auto& file : files) {
+                    snapshot_files.push_back(file.file_name);
+                }
+            }
         }
         if (!status.ok()) {
             LOG_WARNING("failed to make snapshot")
@@ -1657,7 +1662,7 @@ Status TaskWorkerPool::_move_dir(const TTabletId tablet_id, const std::string& s
     return loader.move(src, tablet, overwrite);
 }
 
-void TaskWorkerPool::_handle_report(const TReportRequest& request, ReportType type) {
+void TaskWorkerPool::_handle_report(TReportRequest& request, ReportType type) {
     TMasterResult result;
     Status status = MasterServerClient::instance()->report(request, &result);
     bool is_report_success = false;

be/src/agent/task_worker_pool.h

@@ -203,7 +203,7 @@ class TaskWorkerPool {
 
     void _alter_tablet(const TAgentTaskRequest& alter_tablet_request, int64_t signature,
                        const TTaskType::type task_type, TFinishTaskRequest* finish_task_request);
-    void _handle_report(const TReportRequest& request, ReportType type);
+    void _handle_report(TReportRequest& request, ReportType type);
 
     Status _get_tablet_info(const TTabletId tablet_id, const TSchemaHash schema_hash,
                             int64_t signature, TTabletInfo* tablet_info);

be/src/common/config.h

@@ -874,8 +874,6 @@ CONF_Int32(segcompaction_threshold_segment_num, "10");
 // The segment whose row number above the threshold will be compacted during segcompaction
 CONF_Int32(segcompaction_small_threshold, "1048576");
 
-CONF_String(jvm_max_heap_size, "1024M");
-
 // enable java udf and jdbc scannode
 CONF_Bool(enable_java_support, "true");
 

be/src/common/configbase.cpp

@@ -29,7 +29,8 @@
 
 #include "common/status.h"
 #include "gutil/strings/substitute.h"
-#include "util/filesystem_util.h"
+#include "io/fs/file_writer.h"
+#include "io/fs/local_file_system.h"
 
 namespace doris {
 namespace config {
@@ -260,28 +261,23 @@ void Properties::set_force(const std::string& key, const std::string& val) {
     file_conf_map[key] = val;
 }
 
-bool Properties::dump(const std::string& conffile) {
-    std::vector<std::string> files = {conffile};
-    Status st = FileSystemUtil::remove_paths(files);
-    if (!st.ok()) {
-        return false;
-    }
-    st = FileSystemUtil::create_file(conffile);
-    if (!st.ok()) {
-        return false;
-    }
-
-    std::ofstream out(conffile);
-    out << "# THIS IS AN AUTO GENERATED CONFIG FILE.\n";
-    out << "# You can modify this file manually, and the configurations in this file\n";
-    out << "# will overwrite the configurations in be.conf\n";
-    out << "\n";
+Status Properties::dump(const std::string& conffile) {
+    RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(conffile));
+    io::FileWriterPtr file_writer;
+    RETURN_IF_ERROR(io::global_local_filesystem()->create_file(conffile, &file_writer));
+    RETURN_IF_ERROR(file_writer->append("# THIS IS AN AUTO GENERATED CONFIG FILE.\n"));
+    RETURN_IF_ERROR(file_writer->append(
+            "# You can modify this file manually, and the configurations in this file\n"));
+    RETURN_IF_ERROR(file_writer->append("# will overwrite the configurations in be.conf\n\n"));
 
     for (auto const& iter : file_conf_map) {
-        out << iter.first << " = " << iter.second << "\n";
+        RETURN_IF_ERROR(file_writer->append(iter.first));
+        RETURN_IF_ERROR(file_writer->append(" = "));
+        RETURN_IF_ERROR(file_writer->append(iter.second));
+        RETURN_IF_ERROR(file_writer->append("\n"));
     }
-    out.close();
-    return true;
+
+    return file_writer->close();
 }
 
 template <typename T>
@@ -383,14 +379,14 @@ bool init(const char* conf_file, bool fill_conf_map, bool must_exist, bool set_t
             (*full_conf_map)[(FIELD).name] = oss.str();                                           \
         }                                                                                         \
         if (PERSIST) {                                                                            \
-            persist_config(std::string((FIELD).name), VALUE);                                     \
+            RETURN_IF_ERROR(persist_config(std::string((FIELD).name), VALUE));                    \
         }                                                                                         \
         return Status::OK();                                                                      \
     }
 
 // write config to be_custom.conf
 // the caller need to make sure that the given config is valid
-bool persist_config(const std::string& field, const std::string& value) {
+Status persist_config(const std::string& field, const std::string& value) {
     // lock to make sure only one thread can modify the be_custom.conf
     std::lock_guard<std::mutex> l(custom_conf_lock);
 
@@ -399,7 +395,7 @@ bool persist_config(const std::string& field, const std::string& value) {
     Properties tmp_props;
     if (!tmp_props.load(conffile.c_str(), false)) {
         LOG(WARNING) << "failed to load " << conffile;
-        return false;
+        return Status::InternalError("failed to load conf file: {}", conffile);
     }
 
     tmp_props.set_force(field, value);

be/src/common/configbase.h

@@ -156,7 +156,7 @@ class Properties {
     void set_force(const std::string& key, const std::string& val);
 
     // dump props to conf file
-    bool dump(const std::string& conffile);
+    Status dump(const std::string& conffile);
 
 private:
     std::map<std::string, std::string> file_conf_map;
@@ -177,7 +177,7 @@ bool init(const char* conf_file, bool fill_conf_map = false, bool must_exist = t
 Status set_config(const std::string& field, const std::string& value, bool need_persist = false,
                   bool force = false);
 
-bool persist_config(const std::string& field, const std::string& value);
+Status persist_config(const std::string& field, const std::string& value);
 
 std::mutex* get_mutable_string_config_lock();