Add commits in spectrum log

Author: Peng Gu

libspectrum/include/spectrum.h

@@ -53,6 +53,8 @@ extern bool is_spectrum_storage_replica();
 extern void disable_spectrum_compute(THD *thd);
 extern void enable_spectrum_compute(THD *thd);
 
+extern spectrum::StorageNode::Stub* get_storage_primary_client();
+
 extern void spectrum_print_row(char* method, TABLE* table);
 extern void spectrum_print_row(char* method, TABLE* table, uchar* record);
 extern void spectrum_row_fill_fields(TABLE* table, ::spectrum::Row *spectrum_row);
@@ -94,12 +96,15 @@ extern int spectrum_compute_post_ddl(THD *thd);
 
 extern int spectrum_storage_init();
 
+extern int spectrum_log_init();
 extern int spectrum_log_create_table(THD *thd, const char* db_name, const char* table_name, uint64 handler_id);
 extern int spectrum_log_delete_table(THD *thd, const char* db_name, const char* table_name, const char* table_path);
 extern int spectrum_log_post_ddl(THD *thd);
 extern int spectrum_log_update_metadata(THD *thd, const char* table, dd::Object_id object_id, const char* object_name);
 extern int spectrum_log_add_row(THD *thd, TABLE *table, uchar *new_row, uchar *old_row);
 extern int spectrum_log_prepare(THD *thd, bool all);
 extern int spectrum_log_commit(THD *thd, bool all);
+extern int spectrum_log_read_commit(THD *thd, uint64 start_id_exclusive, spectrum::Commit *commit);
+extern int spectrum_log_read_events_by_xid(THD *thd, uint64 xid, spectrum::EventList *events);
 
 #endif

libspectrum/src/protobuf/spectrum.proto

@@ -383,4 +383,18 @@ message Event {
   uint64 id = 2;
   uint64 type = 3;
   string body = 4;
+}
+
+message EventList {
+  repeated Event event = 1;
+}
+
+message Commit {
+  uint64 id = 1;
+  uint64 xid = 2;
+  EventList events = 3;
+}
+
+message CommitList {
+  repeated Commit commit = 1;
 }

libspectrum/src/spectrum_common.cc

@@ -53,6 +53,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include <mysql/plugin.h>
 
 #include <grpc/grpc.h>
+#include <grpcpp/create_channel.h>
 #include "spectrum.h"
 #include "spectrum_config.h"
 
@@ -82,6 +83,16 @@ void enable_spectrum_compute(THD *thd) {
   thd->spectrum_compute_disabled = false;
 }
 
+std::unique_ptr<spectrum::StorageNode::Stub> storage_primary_client;
+spectrum::StorageNode::Stub* get_storage_primary_client() {
+  if (!storage_primary_client) {
+    node_config_t* node_config = find_storage_primary_node_config();
+    std::shared_ptr<grpc::Channel> channel = grpc::CreateChannel(node_config->address, grpc::InsecureChannelCredentials());
+    storage_primary_client = spectrum::StorageNode::NewStub(channel);
+  }
+  return storage_primary_client.get();
+}
+
 void spectrum_thread_fill_system_variables(THD *thd, spectrum::Thread *spectrum_thread) {
   spectrum_thread->mutable_system_variables()->set_option_bits(thd->variables.option_bits);
 }

libspectrum/src/spectrum_compute.cc

@@ -110,21 +110,17 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include <vector>
 
 #include <grpc/grpc.h>
-#include <grpcpp/create_channel.h>
 #include "spectrum.h"
 #include "spectrum_config.h"
 #include "spectrum.grpc.pb.h"
 
 bool spectrum_debug = false;
 
-std::unique_ptr<spectrum::StorageNode::Stub> storage_primary_client;
-spectrum::StorageNode::Stub* get_storage_primary_client() {
-  if (!storage_primary_client) {
-    node_config_t* node_config = find_storage_primary_node_config();
-    std::shared_ptr<grpc::Channel> channel = grpc::CreateChannel(node_config->address, grpc::InsecureChannelCredentials());
-    storage_primary_client = spectrum::StorageNode::NewStub(channel);
-  }
-  return storage_primary_client.get();
+std::time_t spectrum_compute_init_time;
+int spectrum_compute_init() {
+  auto now = std::chrono::system_clock::now();
+  spectrum_compute_init_time = std::chrono::system_clock::to_time_t(now);
+  return 0;
 }
 
 int spectrum_compute_create_table(THD *thd, TABLE *table) {