s3query/get: pull object to local

source/dnode/vnode/src/inc/vndCos.h

@@ -28,6 +28,9 @@ int32_t s3Init();
 void    s3CleanUp();
 void    s3PutObjectFromFile(const char *file, const char *object);
 void    s3DeleteObjects(const char *object_name[], int nobject);
+bool    s3Exists(const char *object_name);
+void    s3Get(const char *object_name, const char *path);
+void    s3EvictCache();
 
 #ifdef __cplusplus
 }

source/dnode/vnode/src/tsdb/tsdbReaderWriter.c

@@ -14,6 +14,7 @@
  */
 
 #include "tsdb.h"
+#include "vndCos.h"
 
 // =============== PAGE-WISE FILE ===============
 int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **ppFD) {
@@ -34,10 +35,24 @@ int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **p
   pFD->flag = flag;
   pFD->pFD = taosOpenFile(path, flag);
   if (pFD->pFD == NULL) {
+    const char *object_name = taosDirEntryBaseName((char *)path);
+    if (!strncmp(path + strlen(path) - 5, ".data", 5) && s3Exists(object_name)) {
+      s3EvictCache();
+      s3Get(object_name, path);
+
+      pFD->pFD = taosOpenFile(path, flag);
+
+      if (pFD->pFD == NULL) {
+        code = TAOS_SYSTEM_ERROR(errno);
+        taosMemoryFree(pFD);
+        goto _exit;
+      }
+    } else {
       code = TAOS_SYSTEM_ERROR(errno);
       taosMemoryFree(pFD);
       goto _exit;
     }
+  }
   pFD->szPage = szPage;
   pFD->pgno = 0;
   pFD->pBuf = taosMemoryCalloc(1, szPage);

source/dnode/vnode/src/vnd/vnodeCos.c

@@ -112,3 +112,78 @@ void s3DeleteObjects(const char *object_name[], int nobject) {
     cos_warn_log("delete objects failed\n");
   }
 }
+
+bool s3Exists(const char *object_name) {
+  bool                      ret = false;
+  cos_pool_t               *p = NULL;
+  int                       is_cname = 0;
+  cos_status_t             *s = NULL;
+  cos_request_options_t    *options = NULL;
+  cos_string_t              bucket;
+  cos_string_t              object;
+  cos_table_t              *resp_headers;
+  cos_table_t              *headers = NULL;
+  cos_object_exist_status_e object_exist;
+
+  cos_pool_create(&p, NULL);
+  options = cos_request_options_create(p);
+  s3InitRequestOptions(options, is_cname);
+  cos_str_set(&bucket, tsS3BucketName);
+  cos_str_set(&object, object_name);
+
+  s = cos_check_object_exist(options, &bucket, &object, headers, &object_exist, &resp_headers);
+  if (object_exist == COS_OBJECT_NON_EXIST) {
+    cos_warn_log("object: %.*s non exist.\n", object.len, object.data);
+  } else if (object_exist == COS_OBJECT_EXIST) {
+    ret = true;
+    cos_warn_log("object: %.*s exist.\n", object.len, object.data);
+  } else {
+    cos_warn_log("object: %.*s unknown status.\n", object.len, object.data);
+    log_status(s);
+  }
+
+  cos_pool_destroy(p);
+
+  return ret;
+}
+
+void s3Get(const char *object_name, const char *path) {
+  cos_pool_t            *p = NULL;
+  int                    is_cname = 0;
+  cos_status_t          *s = NULL;
+  cos_request_options_t *options = NULL;
+  cos_string_t           bucket;
+  cos_string_t           object;
+  cos_string_t           file;
+  cos_table_t           *resp_headers = NULL;
+  cos_table_t           *headers = NULL;
+  int                    traffic_limit = 0;
+
+  //创建内存池
+  cos_pool_create(&p, NULL);
+
+  //初始化请求选项
+  options = cos_request_options_create(p);
+  s3InitRequestOptions(options, is_cname);
+  cos_str_set(&bucket, tsS3BucketName);
+  if (traffic_limit) {
+    //限速值设置范围为819200 - 838860800，即100KB/s - 100MB/s，如果超出该范围将返回400错误
+    headers = cos_table_make(p, 1);
+    cos_table_add_int(headers, "x-cos-traffic-limit", 819200);
+  }
+
+  //下载对象
+  cos_str_set(&file, path);
+  cos_str_set(&object, object_name);
+  s = cos_get_object_to_file(options, &bucket, &object, headers, NULL, &file, &resp_headers);
+  if (cos_status_is_ok(s)) {
+    cos_warn_log("get object succeeded\n");
+  } else {
+    cos_warn_log("get object failed\n");
+  }
+
+  //销毁内存池
+  cos_pool_destroy(p);
+}
+
+void s3EvictCache() {}