started json_cursor functionality; some fixes to json-parser function…

…ality

libconfluo/confluo/atomic_multilog.h

@@ -22,6 +22,7 @@
 #include "conf/configuration_params.h"
 #include "container/data_log.h"
 #include "container/cursor/record_cursors.h"
+#include "container/cursor/json_cursors.h"
 #include "container/cursor/alert_cursor.h"
 #include "container/monolog/monolog.h"
 #include "container/radix_tree.h"
@@ -252,7 +253,7 @@ class atomic_multilog {
    * @param json_data The json-formatted data to be stored
    * @return The offset of where the data is located
    */
-  size_t append_json(std::string json_data);
+  size_t append_json(const std::string &json_data);
 
   // TODO: Add a std::tuple based variant
   // TODO: Add a JSON based variant
@@ -333,6 +334,13 @@ class atomic_multilog {
    */
   std::unique_ptr<record_cursor> execute_filter(const std::string &expr) const;
 
+  /**
+   * Executes the filter expression
+   * @param expr The filter expression
+   * @return The result of applying the filter to the atomic multilog
+   */
+//  std::unique_ptr<json_cursor> execute_filter_json(const std::string &expr) const;
+
   // TODO: Add tests
   /**
    * Executes an aggregate
@@ -355,6 +363,17 @@ class atomic_multilog {
                                               uint64_t begin_ms,
                                               uint64_t end_ms) const;
 
+  /**
+   * Queries an existing filter
+   * @param filter_name Name of the filter
+   * @param begin_ms Beginning of time-range in ms
+   * @param end_ms End of time-range in ms
+   * @return A stream containing the results of the filter
+   */
+//  std::unique_ptr<json_cursor> query_filter_json(const std::string &filter_name,
+//                                                 uint64_t begin_ms,
+//                                                 uint64_t end_ms) const;
+
   /**
    * Queries an existing filter
    * @param filter_name The name of the filter
@@ -366,6 +385,17 @@ class atomic_multilog {
   std::unique_ptr<record_cursor> query_filter(const std::string &filter_name, uint64_t begin_ms, uint64_t end_ms,
                                               const std::string &additional_filter_expr) const;
 
+  /**
+   * Queries an existing filter
+   * @param filter_name The name of the filter
+   * @param begin_ms Beginning of time-range in ms
+   * @param end_ms End of time-range in ms
+   * @param additional_filter_expr Additional filter expression
+   * @return A stream containing the results of the filter
+   */
+//  std::unique_ptr<json_cursor> query_filter_json(const std::string &filter_name, uint64_t begin_ms, uint64_t end_ms,
+//                                              const std::string &additional_filter_expr) const;
+
   /**
    * Query a stored aggregate.
    * @param aggregate_name The name of the aggregate

libconfluo/confluo/container/cursor/json_cursors.h

@@ -0,0 +1,51 @@
+#ifndef CONFLUO_CONTAINER_CURSOR_JSON_CURSOR_H_
+#define CONFLUO_CONTAINER_CURSOR_JSON_CURSOR_H_
+
+#include <unordered_set>
+
+#include "batched_cursor.h"
+#include "offset_cursors.h"
+#include "schema/record.h"
+#include "parser/expression_compiler.h"
+#include "container/data_log.h"
+
+namespace confluo {
+
+typedef batched_cursor<std::string> json_cursor;
+
+/**
+ * A json cursor that make records into a json formatted string
+ */
+class aggregated_json_cursor : public json_cursor {
+ public:
+  /**
+   * Initializes the filter record
+   *
+   * @param o_cursor The offset cursor
+   * @param dlog The data log pointer
+   * @param schema The schema
+   * @param cexpr The filter expression
+   * @param batch_size The number of records in the batch
+   */
+  aggregated_json_cursor(std::unique_ptr<offset_cursor> o_cursor,
+                       const data_log *dlog, const schema_t *schema,
+                       const parser::compiled_expression &cexpr,
+                       size_t batch_size = 64);
+
+  /**
+   * Loads the next batch from the cursor
+   *
+   * @return The size of the batch
+   */
+  virtual size_t load_next_batch() override;
+
+ private:
+  std::unique_ptr<offset_cursor> o_cursor_;
+  const data_log *dlog_;
+  const schema_t *schema_;
+  const parser::compiled_expression &cexpr_;
+};
+
+}
+
+#endif /* CONFLUO_CONTAINER_CURSOR_JSON_CURSOR_H_ */

libconfluo/confluo/schema/schema.h

@@ -169,14 +169,13 @@ class schema_t {
    *
    * @return A pointer to the record data
    */
-  void *json_string_to_data(const std::string json_record) const;
+  void *json_string_to_data(const std::string &json_record) const;
 
   /**
    * Converts the records into a pointer to the record data
    *
+   * @param ret The json string that is filled up with the data
    * @param record The records used for conversion
-   *
-   * @return A pointer to the record data
    */
   void data_to_json_string(std::string &ret, const void *data) const;
 

libconfluo/src/atomic_multilog.cc

@@ -228,7 +228,7 @@ size_t atomic_multilog::append(void *data) {
   return offset;
 }
 
-size_t atomic_multilog::append_json(std::string json_record) {
+size_t atomic_multilog::append_json(const std::string &json_record) {
   void *buf = schema_.json_string_to_data(json_record);
   size_t off = append(buf);
   delete[] reinterpret_cast<uint8_t *>(buf);
@@ -299,6 +299,15 @@ std::unique_ptr<record_cursor> atomic_multilog::execute_filter(const std::string
   return plan.execute(version);
 }
 
+//std::unique_ptr<json_cursor> atomic_multilog::execute_filter_json(const std::string &expr) const {
+//  uint64_t version = rt_.get();
+//  auto t = parser::parse_expression(expr);
+//  auto cexpr = parser::compile_expression(t, schema_);
+//  query_plan plan = planner_.plan(cexpr);
+//  return plan.execute(version);
+//  // TODO: figure out return for this
+//}
+
 numeric atomic_multilog::execute_aggregate(const std::string &aggregate_expr, const std::string &filter_expr) {
   auto pa = parser::parse_aggregate(aggregate_expr);
   aggregator agg = aggregate_manager::get_aggregator(pa.agg);
@@ -331,6 +340,27 @@ std::unique_ptr<record_cursor> atomic_multilog::query_filter(const std::string &
                                parser::compiled_expression()));
 }
 
+//std::unique_ptr<json_cursor> atomic_multilog::query_filter_json(const std::string &filter_name,
+//                                                             uint64_t begin_ms,
+//                                                             uint64_t end_ms) const {
+//  filter_id_t filter_id;
+//  if (filter_map_.get(filter_name, filter_id) == -1) {
+//    throw invalid_operation_exception(
+//        "Filter " + filter_name + " does not exist.");
+//  }
+//
+//  filter::range_result res = filters_.at(filter_id)->lookup_range(begin_ms,
+//                                                                  end_ms);
+//  uint64_t version = rt_.get();
+//  std::unique_ptr<offset_cursor> o_cursor(
+//      new offset_iterator_cursor<filter::range_result::iterator>(res.begin(),
+//                                                                 res.end(),
+//                                                                 version));
+//  return std::unique_ptr<json_cursor>(
+//      new json_cursor(std::move(o_cursor), &data_log_, &schema_,
+//                               parser::compiled_expression()));
+//}
+
 std::unique_ptr<record_cursor> atomic_multilog::query_filter(const std::string &filter_name,
                                                              uint64_t begin_ms,
                                                              uint64_t end_ms,
@@ -350,6 +380,25 @@ std::unique_ptr<record_cursor> atomic_multilog::query_filter(const std::string &
   return std::unique_ptr<record_cursor>(new filter_record_cursor(std::move(o_cursor), &data_log_, &schema_, e));
 }
 
+//std::unique_ptr<json_cursor> atomic_multilog::query_filter_json(const std::string &filter_name,
+//                                                             uint64_t begin_ms,
+//                                                             uint64_t end_ms,
+//                                                             const std::string &additional_filter_expr) const {
+//  auto t = parser::parse_expression(additional_filter_expr);
+//  auto e = parser::compile_expression(t, schema_);
+//  filter_id_t filter_id;
+//  if (filter_map_.get(filter_name, filter_id) == -1) {
+//    throw invalid_operation_exception(
+//        "Filter " + filter_name + " does not exist.");
+//  }
+//
+//  filter::range_result res = filters_.at(filter_id)->lookup_range(begin_ms, end_ms);
+//  uint64_t version = rt_.get();
+//  std::unique_ptr<offset_cursor> o_cursor(
+//      new offset_iterator_cursor<filter::range_result::iterator>(res.begin(), res.end(), version));
+//  return std::unique_ptr<json_cursor>(new json_cursor(std::move(o_cursor), &data_log_, &schema_, e));
+//}
+
 numeric atomic_multilog::get_aggregate(const std::string &aggregate_name, uint64_t begin_ms, uint64_t end_ms) {
   aggregate_id_t aggregate_id;
   if (aggregate_map_.get(aggregate_name, aggregate_id) == -1) {

libconfluo/src/container/cursor/json_cursors.cc

@@ -0,0 +1,33 @@
+#include "container/cursor/json_cursors.h"
+
+namespace confluo {
+
+aggregated_json_cursor::aggregated_json_cursor(std::unique_ptr<offset_cursor> o_cursor,
+                                           const data_log *dlog,
+                                           const schema_t *schema,
+                                           const parser::compiled_expression &cexpr,
+                                           size_t batch_size)
+    : json_cursor(batch_size),
+      o_cursor_(std::move(o_cursor)),
+      dlog_(dlog),
+      schema_(schema),
+      cexpr_(cexpr) {
+  init();
+}
+
+size_t aggregated_json_cursor::load_next_batch() {
+  // TODO add functionality to make into a json_string
+  size_t i = 0;
+  for (; i < current_batch_.size() && o_cursor_->has_more();
+         ++i, o_cursor_->advance()) {
+    uint64_t o = o_cursor_->get();
+    read_only_data_log_ptr ptr;
+    dlog_->cptr(o, ptr);
+    if (!cexpr_.test(current_batch_[i] = schema_->apply(o, ptr))) {
+      i--;
+    }
+  }
+  return i;
+}
+
+}

libconfluo/src/schema/schema.cc

@@ -94,7 +94,7 @@ std::string schema_t::to_string() const {
   return str;
 }
 
-void *schema_t::json_string_to_data(const std::string json_record) const {
+void *schema_t::json_string_to_data(const std::string &json_record) const {
   // need to convert json_data into a record vector
   std::stringstream ss;
   // putting the json data into a stream

librpc/src/rpc_server.cc

@@ -178,9 +178,11 @@ void rpc_service_handler::read(std::string &_return, int64_t id, const int64_t o
   _return.assign(data, size);
 }
 void rpc_service_handler::read_json(std::string &_return, int64_t id, const int64_t offset, const int64_t nrecords) {
+  if (nrecords > 1) {
+    THROW(unsupported_exception, "Reading more than 1 JSON records at a time is unsupported!");
+  }
   atomic_multilog *mlog = store_->get_atomic_multilog(id);
   _return = mlog->read_json((uint64_t) offset);
-  // TODO: put in functionality for nrecords to be read
 }
 void rpc_service_handler::query_aggregate(std::string &_return,
                                           int64_t id,