feat(bulk-load): add meta_bulk_load_service and some structures (#445)

apache · Apr 30, 2020 · b9e2bbe · b9e2bbe
1 parent 966c04d
commit b9e2bbe
Show file tree

Hide file tree

Showing 18 changed files with 1,344 additions and 581 deletions.
diff --git a/include/dsn/cpp/json_helper.h b/include/dsn/cpp/json_helper.h
@@ -86,6 +86,10 @@
 #define JSON_ENCODE_ENTRIES13(out, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13) \
     JSON_ENCODE_ENTRIES12(out, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12);         \
     JSON_ENCODE_ENTRY(out, prefix, T13)
+#define JSON_ENCODE_ENTRIES14(                                                                     \
+    out, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14)                      \
+    JSON_ENCODE_ENTRIES13(out, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13);    \
+    JSON_ENCODE_ENTRY(out, prefix, T14)
 
 #define JSON_DECODE_ENTRY(in, prefix, T)                                                           \
     do {                                                                                           \
@@ -140,16 +144,21 @@
 #define JSON_DECODE_ENTRIES13(in, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13)  \
     JSON_DECODE_ENTRIES12(in, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12);          \
     JSON_TRY_DECODE_ENTRY(in, prefix, T13)
+#define JSON_DECODE_ENTRIES14(                                                                     \
+    in, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14)                       \
+    JSON_DECODE_ENTRIES13(in, prefix, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13);     \
+    JSON_TRY_DECODE_ENTRY(in, prefix, T14)
 
 #define JSON_ENTRIES_GET_MACRO(                                                                    \
-    ph1, ph2, ph3, ph4, ph5, ph6, ph7, ph8, ph9, ph10, ph11, ph12, ph13, NAME, ...)                \
+    ph1, ph2, ph3, ph4, ph5, ph6, ph7, ph8, ph9, ph10, ph11, ph12, ph13, ph14, NAME, ...)          \
     NAME
 // workaround due to the way VC handles "..."
 #define JSON_ENTRIES_GET_MACRO_(tuple) JSON_ENTRIES_GET_MACRO tuple
 
 #define JSON_ENCODE_ENTRIES(out, prefix, ...)                                                      \
     out.StartObject();                                                                             \
     JSON_ENTRIES_GET_MACRO_((__VA_ARGS__,                                                          \
+                             JSON_ENCODE_ENTRIES14,                                                \
                              JSON_ENCODE_ENTRIES13,                                                \
                              JSON_ENCODE_ENTRIES12,                                                \
                              JSON_ENCODE_ENTRIES11,                                                \
@@ -171,6 +180,7 @@
     int arguments_count = 0;                                                                       \
     int parsed_count = 0;                                                                          \
     JSON_ENTRIES_GET_MACRO_((__VA_ARGS__,                                                          \
+                             JSON_DECODE_ENTRIES14,                                                \
                              JSON_DECODE_ENTRIES13,                                                \
                              JSON_DECODE_ENTRIES12,                                                \
                              JSON_DECODE_ENTRIES11,                                                \
@@ -325,6 +335,8 @@ UINT_TYPE_SERIALIZATION(uint64_t)
 ENUM_TYPE_SERIALIZATION(dsn::replication::partition_status::type,
                         dsn::replication::partition_status::PS_INVALID)
 ENUM_TYPE_SERIALIZATION(dsn::app_status::type, dsn::app_status::AS_INVALID)
+ENUM_TYPE_SERIALIZATION(dsn::replication::bulk_load_status::type,
+                        dsn::replication::bulk_load_status::BLS_INVALID)
 
 // json serialization for gpid, we treat it as string: "app_id.partition_id"
 inline void json_encode(JsonWriter &out, const dsn::gpid &pid)
@@ -357,6 +369,10 @@ inline void json_encode(JsonWriter &out, const dsn::partition_configuration &con
 inline bool json_decode(const JsonObject &in, dsn::partition_configuration &config);
 inline void json_encode(JsonWriter &out, const dsn::app_info &info);
 inline bool json_decode(const JsonObject &in, dsn::app_info &info);
+inline void json_encode(JsonWriter &out, const dsn::replication::file_meta &f_meta);
+inline bool json_decode(const JsonObject &in, dsn::replication::file_meta &f_meta);
+inline void json_encode(JsonWriter &out, const dsn::replication::bulk_load_metadata &metadata);
+inline bool json_decode(const JsonObject &in, dsn::replication::bulk_load_metadata &metadata);
 
 template <typename T>
 inline void json_encode_iterable(JsonWriter &out, const T &t)
@@ -620,6 +636,11 @@ NON_MEMBER_JSON_SERIALIZATION(dsn::app_info,
                               create_second,
                               drop_second,
                               duplicating,
-                              init_partition_count)
+                              init_partition_count,
+                              is_bulk_loading)
+
+NON_MEMBER_JSON_SERIALIZATION(dsn::replication::file_meta, name, size, md5)
+
+NON_MEMBER_JSON_SERIALIZATION(dsn::replication::bulk_load_metadata, files, file_total_size)
 } // namespace json
 } // namespace dsn
diff --git a/include/dsn/cpp/serialization_helper/dsn.layer2_types.h b/include/dsn/cpp/serialization_helper/dsn.layer2_types.h
diff --git a/include/dsn/dist/replication/replication.codes.h b/include/dsn/dist/replication/replication.codes.h
@@ -105,6 +105,7 @@ MAKE_EVENT_CODE_RPC(RPC_CM_UPDATE_APP_ENV, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE_RPC(RPC_CM_DDD_DIAGNOSE, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE_RPC(RPC_CM_APP_PARTITION_SPLIT, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE_RPC(RPC_CM_REGISTER_CHILD_REPLICA, TASK_PRIORITY_COMMON)
+MAKE_EVENT_CODE_RPC(RPC_CM_START_BULK_LOAD, TASK_PRIORITY_COMMON)
 #undef CURRENT_THREAD_POOL
 
 #define CURRENT_THREAD_POOL THREAD_POOL_META_STATE

diff --git a/include/dsn/dist/replication/replication_enums.h b/include/dsn/dist/replication/replication_enums.h
@@ -65,4 +65,18 @@ ENUM_BEGIN2(replication::node_status::type, node_status, replication::node_statu
 ENUM_REG(replication::node_status::NS_ALIVE)
 ENUM_REG(replication::node_status::NS_UNALIVE)
 ENUM_END2(replication::node_status::type, node_status)
+
+ENUM_BEGIN2(replication::bulk_load_status::type,
+            bulk_load_status,
+            replication::bulk_load_status::BLS_INVALID)
+ENUM_REG(replication::bulk_load_status::BLS_INVALID)
+ENUM_REG(replication::bulk_load_status::BLS_DOWNLOADING)
+ENUM_REG(replication::bulk_load_status::BLS_DOWNLOADED)
+ENUM_REG(replication::bulk_load_status::BLS_INGESTING)
+ENUM_REG(replication::bulk_load_status::BLS_SUCCEED)
+ENUM_REG(replication::bulk_load_status::BLS_FAILED)
+ENUM_REG(replication::bulk_load_status::BLS_PAUSING)
+ENUM_REG(replication::bulk_load_status::BLS_PAUSED)
+ENUM_REG(replication::bulk_load_status::BLS_CANCELED)
+ENUM_END2(replication::bulk_load_status::type, bulk_load_status)
 }
diff --git a/include/dsn/dist/replication/replication_types.h b/include/dsn/dist/replication/replication_types.h