Merge pull request #7 from RunningXie/feature_support_cubefs_sdk

Feature support cubefs sdk

.gitmodules

@@ -262,3 +262,7 @@
 [submodule "contrib/minizip-ng"]
 	path = contrib/minizip-ng
 	url = https://github.com/zlib-ng/minizip-ng
+[submodule "contrib/cubefs"]
+	path = contrib/cubefs
+	url = https://github.com/RunningXie/cubefs.git
+	branch = fix_sdk_compilation

contrib/CMakeLists.txt

@@ -135,6 +135,7 @@ add_contrib (libpq-cmake libpq)
 add_contrib (nuraft-cmake NuRaft)
 add_contrib (fast_float-cmake fast_float)
 add_contrib (datasketches-cpp-cmake datasketches-cpp)
+add_contrib (cubefs-cmake cubefs)
 
 option(ENABLE_NLP "Enable NLP functions support" ${ENABLE_LIBRARIES})
 if (ENABLE_NLP)

contrib/cubefs-cmake/CMakeLists.txt

@@ -0,0 +1,23 @@
+cmake_minimum_required(VERSION 3.20)
+set(CUBEFS_DIR "${ClickHouse_SOURCE_DIR}/contrib/cubefs")
+execute_process(
+    COMMAND make libsdk
+    WORKING_DIRECTORY ${CUBEFS_DIR})
+set(FILE_PATH "${CUBEFS_DIR}/libsdk/libcfs.h")
+# Add configuration to ignore warnings in the header file
+file(STRINGS ${FILE_PATH} FILE_CONTENTS)
+list(FIND FILE_CONTENTS "#pragma GCC diagnostic push" PRAGMA_INDEX)
+if (PRAGMA_INDEX EQUAL -1)
+ execute_process(
+   COMMAND sh -c "echo '#pragma GCC diagnostic push\n#pragma GCC diagnostic ignored \"-Wreserved-identifier\"\n#pragma GCC diagnostic ignored \"-Wmacro-redefined\"\n' | cat - ${FILE_PATH}"
+   OUTPUT_FILE "${FILE_PATH}.tmp"
+ )
+ execute_process(
+   COMMAND sh -c "echo '#pragma GCC diagnostic pop' >> ${FILE_PATH}.tmp"
+ )
+ execute_process(
+   COMMAND mv "${FILE_PATH}.tmp" "${FILE_PATH}"
+ )
+endif()
+
+

programs/cubefs-test/CMakeLists.txt

@@ -0,0 +1,4 @@
+set (CLICKHOUSE_CUBEFS_TEST_SOURCES cubefs-test.cpp clickhouse-cubefs-test.cpp)
+set (CLICKHOUSE_CUBEFS_TEST_LINK dbms ch_contrib::cubefs)
+set (CLICKHOUSE_CUBEFS_TEST_INCLUDE PUBLIC ${ClickHouse_SOURCE_DIR}/contrib/cubefs/libsdk)
+clickhouse_program_add(cubefs-test)

programs/cubefs-test/clickhouse-cubefs-test.cpp

@@ -0,0 +1,5 @@
+int mainEntryClickHouseCubefsTest(int argc, char ** argv);
+int main(int argc_, char ** argv_)
+{
+    return mainEntryClickHouseCubefsTest(argc_, argv_);
+}

programs/cubefs-test/cubefs-dlopen.cpp

@@ -0,0 +1,165 @@
+#include <cstring>
+#include <iostream>
+#include <dlfcn.h>
+#include <libcfs.h>
+
+
+int main()
+{
+    void * handle = dlopen("./libcfs.so", RTLD_LAZY);
+    if (!handle)
+    {
+        std::cerr << "Failed to open the dynamic library: " << dlerror() << std::endl;
+        return 1;
+    }
+    typedef int64_t (*CfsNewClientFunction)();
+    typedef int (*CfsSetClientFunction)(int64_t, char *, char *);
+    typedef int (*CfsStartClientFunction)(int64_t);
+    typedef void (*CfsCloseClientFunction)(int64_t);
+    typedef int (*CfsChdirFunction)(int64_t, char *);
+    typedef char * (*CfsGetcwdFunction)(int64_t);
+    typedef int (*CfsGetattrFunction)(int64_t, char *, struct cfs_stat_info *);
+    typedef int (*CfsSetattrFunction)(int64_t, char *, struct cfs_stat_info *, int);
+    typedef int (*CfsOpenFunction)(int64_t, char *, int, mode_t);
+    typedef int (*CfsFlushFunction)(int64_t, int);
+    typedef void (*CfsCloseFunction)(int64_t, int);
+    typedef ssize_t (*CfsWriteFunction)(int64_t, int, void *, size_t, off_t);
+    typedef ssize_t (*CfsReadFunction)(int64_t, int, void *, size_t, off_t);
+    typedef int (*CfsBatchGetInodesFunction)(int64_t, int, void *, GoSlice, int);
+    typedef int (*CfsRefreshSummaryFunction)(int64_t, char *, int);
+    typedef int (*CfsReaddirFunction)(int64_t, int, GoSlice, int);
+    typedef int (*CfsLsdirFunction)(int64_t, int, GoSlice, int);
+    typedef int (*CfsMkdirsFunction)(int64_t, char *, mode_t);
+    typedef int (*CfsRmdirFunction)(int64_t, char *);
+    typedef int (*CfsUnlinkFunction)(int64_t, char *);
+    typedef int (*CfsRenameFunction)(int64_t, char *, char *);
+    typedef int (*CfsFchmodFunction)(int64_t, int, mode_t);
+    typedef int (*CfsGetsummaryFunction)(int64_t, char *, struct cfs_summary_info *, char *, int);
+    CfsNewClientFunction cfsNewClient = (CfsNewClientFunction)dlsym(handle, "cfs_new_client");
+    CfsSetClientFunction cfsSetClient = (CfsSetClientFunction)dlsym(handle, "cfs_set_client");
+    CfsStartClientFunction cfsStartClient = (CfsStartClientFunction)dlsym(handle, "cfs_start_client");
+    CfsCloseClientFunction cfsCloseClient = (CfsCloseClientFunction)dlsym(handle, "cfs_close_client");
+    CfsChdirFunction cfsChdir = (CfsChdirFunction)dlsym(handle, "cfs_chdir");
+    CfsGetcwdFunction cfsGetcwd = (CfsGetcwdFunction)dlsym(handle, "cfs_getcwd");
+    CfsGetattrFunction cfsGetattr = (CfsGetattrFunction)dlsym(handle, "cfs_getattr");
+    CfsSetattrFunction cfsSetattr = (CfsSetattrFunction)dlsym(handle, "cfs_setattr");
+    CfsOpenFunction cfsOpen = (CfsOpenFunction)dlsym(handle, "cfs_open");
+    CfsFlushFunction cfsFlush = (CfsFlushFunction)dlsym(handle, "cfs_flush");
+    CfsCloseFunction cfsClose = (CfsCloseFunction)dlsym(handle, "cfs_close");
+    CfsLsdirFunction cfsLsdir = (CfsLsdirFunction)dlsym(handle, "cfs_lsdir");
+    CfsMkdirsFunction cfsMkdirs = (CfsMkdirsFunction)dlsym(handle, "cfs_mkdirs");
+    CfsRmdirFunction cfsRmdir = (CfsRmdirFunction)dlsym(handle, "cfs_rmdir");
+    CfsUnlinkFunction cfsUnlink = (CfsUnlinkFunction)dlsym(handle, "cfs_unlink");
+    CfsRenameFunction cfsRename = (CfsRenameFunction)dlsym(handle, "cfs_rename");
+    CfsFchmodFunction cfsFchmod = (CfsFchmodFunction)dlsym(handle, "cfs_fchmod");
+    CfsGetsummaryFunction cfsGetsummary = (CfsGetsummaryFunction)dlsym(handle, "cfs_getsummary");
+    CfsWriteFunction cfsWrite = (CfsWriteFunction)dlsym(handle, "cfs_write");
+    CfsReadFunction cfsRead = (CfsReadFunction)dlsym(handle, "cfs_read");
+    if (!cfsNewClient || !cfsSetClient || !cfsStartClient || !cfsCloseClient || !cfsChdir || !cfsGetcwd || !cfsGetattr || !cfsSetattr
+        || !cfsOpen || !cfsFlush || !cfsClose || !cfsWrite || !cfsRead || !cfsLsdir || !cfsMkdirs || !cfsRmdir || !cfsUnlink || !cfsRename
+        || !cfsFchmod || !cfsGetsummary)
+    {
+        std::cerr << "Failed to find one or more functions: " << dlerror() << std::endl;
+        dlclose(handle);
+        return 1;
+    }
+    int64_t clientId = cfsNewClient();
+    if (clientId == 0)
+    {
+        std::cerr << "Failed to create client" << std::endl;
+        dlclose(handle);
+        return 1;
+    }
+    int result = cfsSetClient(clientId, strdup("volName"), strdup("xieyichen"));
+    if (result != 0)
+    {
+        std::cerr << "Failed to set client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    result = cfsSetClient(clientId, strdup("masterAddr"), strdup("cfs-south.oppo.local"));
+    if (result != 0)
+    {
+        std::cerr << "Failed to set client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    result = cfsSetClient(clientId, strdup("logDir"), strdup("/home/service/var/logs/cfs/test-log"));
+    if (result != 0)
+    {
+        std::cerr << "Failed to set client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    result = cfsSetClient(clientId, strdup("logLevel"), strdup("debug"));
+    if (result != 0)
+    {
+        std::cerr << "Failed to set client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    result = cfsSetClient(clientId, strdup("accessKey"), strdup("jRlZO65q7XlH5bnV"));
+    if (result != 0)
+    {
+        std::cerr << "Failed to set client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    result = cfsSetClient(clientId, strdup("secretKey"), strdup("V1m730UzREHaK1jCkC0kL0cewOX0kH3K"));
+    if (result != 0)
+    {
+        std::cerr << "Failed to set client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    result = cfsSetClient(clientId, strdup("pushAddr"), strdup("cfs.dg-push.wanyol.com"));
+    if (result != 0)
+    {
+        std::cerr << "Failed to set client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    result = cfsStartClient(clientId);
+    if (result != 0)
+    {
+        std::cerr << "Failed to start client" << std::endl;
+        cfsCloseClient(clientId);
+        dlclose(handle);
+        return 1;
+    }
+    int fd = cfsOpen(clientId, strdup("/test_dir/file.txt"), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
+    if (fd < 0)
+    {
+        printf("Failed to open file\n");
+        cfsCloseClient(clientId);
+        return -1;
+    }
+    const char * data = "xieyichen";
+    if (cfsWrite(clientId, fd, static_cast<void *>(const_cast<char *>(data)), strlen(data), 0) < 0)
+    {
+        printf("Failed to write file\n");
+        cfsClose(clientId, fd);
+        cfsCloseClient(clientId);
+        return -1;
+    }
+    char buffer[1024];
+    memset(buffer, 0, sizeof(buffer));
+    if (cfsRead(clientId, fd, static_cast<void *>(const_cast<char *>(buffer)), sizeof(buffer), 0) < 0)
+    {
+        printf("Failed to read file\n");
+        cfsClose(clientId, fd);
+        cfsCloseClient(clientId);
+        return -1;
+    }
+    printf("Read file content: %s\n", buffer);
+    cfsCloseClient(clientId);
+    dlclose(handle);
+    return 0;
+}

programs/cubefs-test/cubefs-test.cpp

@@ -0,0 +1,103 @@
+#include <iostream>
+#include <thread>
+#include <libcfs.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+void readWriteTest()
+{
+    int64_t id = cfs_new_client();
+    if (id <= 0)
+    {
+        printf("Failed to create client\n");
+        return;
+    }
+    if (cfs_set_client(id, strdup("volName"), strdup("xieyichen")) != 0)
+    {
+        printf("Failed to set client info\n");
+        cfs_close_client(id);
+        return;
+    }
+    if (cfs_set_client(id, strdup("masterAddr"), strdup("cfs-south.oppo.local")) != 0)
+    {
+        printf("Failed to set client info\n");
+        cfs_close_client(id);
+        return;
+    }
+    if (cfs_set_client(id, strdup("logDir"), strdup("/home/service/var/logs/cfs/test-log")) != 0)
+    {
+        printf("Failed to set client info\n");
+        cfs_close_client(id);
+        return;
+    }
+    if (cfs_set_client(id, strdup("logLevel"), strdup("debug")) != 0)
+    {
+        printf("Failed to set client info\n");
+        cfs_close_client(id);
+        return;
+    }
+    if (cfs_set_client(id, strdup("accessKey"), strdup("jRlZO65q7XlH5bnV")) != 0)
+    {
+        printf("Failed to set client info\n");
+        cfs_close_client(id);
+        return;
+    }
+    if (cfs_set_client(id, strdup("secretKey"), strdup("V1m730UzREHaK1jCkC0kL0cewOX0kH3K")) != 0)
+    {
+        printf("Failed to set client info\n");
+        cfs_close_client(id);
+        return;
+    }
+    if (cfs_set_client(id, strdup("pushAddr"), strdup("cfs.dg-push.wanyol.com")) != 0)
+    {
+        printf("Failed to set client info\n");
+        cfs_close_client(id);
+        return;
+    }
+    if (cfs_start_client(id) != 0)
+    {
+        printf("Failed to start client \n");
+        cfs_close_client(id);
+        return;
+    }
+    int fd = cfs_open(id, strdup("/test_dir/file.txt"), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
+    if (fd < 0)
+    {
+        printf("Failed to open file\n");
+        cfs_close_client(id);
+        return;
+    }
+
+    const char * data = "Hello, world!";
+    if (cfs_write(id, fd, static_cast<void *>(const_cast<char *>(data)), strlen(data), 0) < 0)
+    {
+        printf("Failed to write file\n");
+        cfs_close(id, fd);
+        cfs_close_client(id);
+        return;
+    }
+
+    char buffer[1024];
+    memset(buffer, 0, sizeof(buffer));
+    if (cfs_read(id, fd, static_cast<void *>(const_cast<char *>(buffer)), sizeof(buffer), 0) < 0)
+    {
+        printf("Failed to read file\n");
+        cfs_close(id, fd);
+        cfs_close_client(id);
+        return;
+    }
+
+    printf("Read file content: %s\n", buffer);
+    cfs_close(id, fd);
+    cfs_close_client(id);
+}
+
+int mainEntryClickHouseCubefsTest(int argc, char ** argv)
+{
+    std::thread worker(readWriteTest);
+    worker.join();
+    std::cout << "argc = " << argc << std::endl;
+    std::cout << "argv = " << argv << std::endl;
+    return 0;
+}

src/CMakeLists.txt

@@ -75,12 +75,13 @@ add_subdirectory (Coordination)
 
 set(dbms_headers)
 set(dbms_sources)
-
+message("ClickHouse_SOURCE_DIR: ${ClickHouse_SOURCE_DIR}")
 add_headers_and_sources(clickhouse_common_io Common)
 add_headers_and_sources(clickhouse_common_io Common/HashTable)
 add_headers_and_sources(clickhouse_common_io IO)
 add_headers_and_sources(clickhouse_common_io IO/Archives)
 add_headers_and_sources(clickhouse_common_io IO/S3)
+add_headers_and_sources(clickhouse_common_io ${ClickHouse_SOURCE_DIR}/contrib/cubefs/libsdk)
 list (REMOVE_ITEM clickhouse_common_io_sources Common/malloc.cpp Common/new_delete.cpp)
 
 add_headers_and_sources(dbms Disks/IO)
@@ -113,6 +114,11 @@ if (TARGET ch_contrib::aws_s3)
     add_headers_and_sources(dbms Disks/S3)
 endif()
 
+if (TARGET ch_contrib::cubefs)
+    add_headers_and_sources(dbms Disks/CubeFS)
+    add_headers_and_sources(dbms ${ClickHouse_SOURCE_DIR}/contrib/cubefs/libsdk)
+endif()
+
 if (TARGET ch_contrib::azure_sdk)
     add_headers_and_sources(dbms Disks/AzureBlobStorage)
 endif()
@@ -306,6 +312,7 @@ endmacro ()
 
 dbms_target_include_directories (PUBLIC "${ClickHouse_SOURCE_DIR}/src" "${ClickHouse_BINARY_DIR}/src")
 target_include_directories (clickhouse_common_io PUBLIC "${ClickHouse_SOURCE_DIR}/src" "${ClickHouse_BINARY_DIR}/src")
+target_include_directories(clickhouse_common_io PUBLIC ${ClickHouse_SOURCE_DIR}/contrib/cubefs/libsdk)
 
 if (TARGET ch_contrib::llvm)
     dbms_target_link_libraries (PUBLIC ch_contrib::llvm)
@@ -466,6 +473,10 @@ if (TARGET ch_contrib::aws_s3)
     target_link_libraries (clickhouse_common_io PUBLIC ch_contrib::aws_s3)
 endif()
 
+if (TARGET ch_contrib::cubefs)
+    target_link_libraries (clickhouse_common_io PUBLIC ch_contrib::cubefs)
+endif()
+
 if (TARGET ch_contrib::azure_sdk)
     target_link_libraries (clickhouse_common_io PUBLIC ch_contrib::azure_sdk)
 endif()

src/Core/Settings.h

@@ -34,6 +34,7 @@ class IColumn;
   */
 
 #define COMMON_SETTINGS(M) \
+    M(String, handle_remove_error_path, "", "When remove_all throws the directory not empty exception, the undeleted data will be moved to the the setting directory. Empty means that the exception will not be processed.", 0) \
     M(UInt64, min_compress_block_size, 65536, "The actual size of the block to compress, if the uncompressed data less than max_compress_block_size is no less than this value and no less than the volume of data for one mark.", 0) \
     M(UInt64, max_compress_block_size, 1048576, "The maximum size of blocks of uncompressed data before compressing for writing to a table.", 0) \
     M(UInt64, max_block_size, DEFAULT_BLOCK_SIZE, "Maximum block size for reading", 0) \