path: root/src/s3select/include/s3select_json_parser.h

#ifndef S3SELECT_JSON_PARSER_H
#define S3SELECT_JSON_PARSER_H

//TODO add __FILE__ __LINE__ message
#define RAPIDJSON_ASSERT(x) s3select_json_parse_error(x)
bool s3select_json_parse_error(bool b);
bool s3select_json_parse_error(const char* error);

#include "rapidjson/reader.h"
#include "rapidjson/writer.h"
#include "rapidjson/filereadstream.h"
#include "rapidjson/filewritestream.h"
#include "rapidjson/error/en.h"
#include "rapidjson/document.h"
#include <cassert>
#include <sstream>
#include <vector>
#include <iostream>
#include <functional>
#include <boost/spirit/include/classic_core.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include "s3select_oper.h"//class value
#include <boost/algorithm/string/predicate.hpp>

#define JSON_PROCESSING_LIMIT_REACHED 2

//TODO missing s3selectEngine namespace

bool s3select_json_parse_error(bool b)
{
  if(!b)
  {
    	const char* error_str = "failure while processing JSON document";
	throw s3selectEngine::base_s3select_exception(error_str, s3selectEngine::base_s3select_exception::s3select_exp_en_t::FATAL);
  }
  return false;
}

bool s3select_json_parse_error(const char* error)
{
  if(!error)
  {
    	const char* error_str = "failure while processing JSON document";
	throw s3selectEngine::base_s3select_exception(error_str, s3selectEngine::base_s3select_exception::s3select_exp_en_t::FATAL);
  }
  return false;
}

static auto iequal_predicate = [](std::string& it1, std::string& it2)
			  {
			    return boost::iequals(it1,it2);
			  };


class ChunksStreamer : public rapidjson::MemoryStream {

  //purpose: adding a method `resetBuffer` that enables to parse chunk after chunk
  //per each new chunk it reset internal data members
  public:

    std::string internal_buffer;
    const Ch* next_src_;
    size_t next_size_;

    ChunksStreamer():rapidjson::MemoryStream(0,0){next_src_=0;next_size_=0;}

    ChunksStreamer(const Ch *src, size_t size) : rapidjson::MemoryStream(src,size){next_src_=0;next_size_=0;}

    //override Peek methode
    Ch Peek() //const 
    {
      if(RAPIDJSON_UNLIKELY(src_ == end_))
      {
	if(next_src_)//next chunk exist
	{//upon reaching to end of current buffer, to switch with next one
	  src_ = next_src_;
	  begin_ = src_;
	  size_ =next_size_;
	  end_ = src_ + size_;

	  next_src_ = 0;
	  next_size_ = 0;
	  return *src_;
	}
	else return 0;
      }
      return *src_;
    }

    //override Take method
    Ch Take() 
    {
      if(RAPIDJSON_UNLIKELY(src_ == end_))
      {
	if(next_src_)//next chunk exist
	{//upon reaching to end of current buffer, to switch with next one
	  src_ = next_src_;
	  begin_ = src_;
	  size_ = next_size_;
	  end_ = src_ + size_;

	  next_src_ = 0;
	  next_size_ = 0;
	  return *src_;
	}
	else return 0;
      }
      return *src_++;
    }

    void resetBuffer(char* buff, size_t size)
    {
      if(!src_)
      {//first time calling
	begin_ = buff;
	src_ = buff;
	size_ = size;
	end_= src_ + size_;
	return;
      }

      if(!next_src_)
      {//save the next-chunk that will be used upon parser reaches end of current buffer
	next_src_ = buff;
	next_size_ = size;
      }
      else
      {// should not happen
	std::cout << "can not replace pointers!!!" << std::endl;//TODO exception
	return;
      }
    }

    void saveRemainingBytes()
    {//this routine called per each new chunk
      //save the remaining bytes, before its overriden by the next-chunk.
      size_t copy_left_sz = getBytesLeft(); //should be very small
      internal_buffer.assign(src_,copy_left_sz);
      
      src_ = internal_buffer.data();
      begin_ = src_;
      size_ = copy_left_sz;
      end_= src_ + copy_left_sz;
    }

    size_t getBytesLeft() { return end_ - src_; }

};

enum class row_state
{
  NA,
  OBJECT_START_ROW,
  ARRAY_START_ROW
};

class json_variable_access {
//purpose: a state-machine for json-variables. 
//upon the syntax-parser accepts a variable (projection / where-clause) it create this object.
//this object get events (key,start-array ... etc) as the JSON reader scans the input, 
//these events are advancing the states until it reaches to the last one, result with pushing value into scratch-area.

private:

// to set the following. 
std::vector<std::string>* from_clause;
std::vector<std::string>* key_path;
//m_current_depth : trace the depth of the reader, including "anonymous"(meaning JSON may begin with array that has no name attached to it)
int* m_current_depth;
//m_current_depth_non_anonymous : trace the depth of the reader, NOT including "anonymous" array/object.
//upon user request the following _1.a[12].b, the key-name{a} may reside on some array with no-name, 
//the state machine that search for a[12].b, does NOT contain states for that "anonymous" array, 
//thus, the state-machine will fail to trace the user request for that specific key.path
int* m_current_depth_non_anonymous;
std::function <int(s3selectEngine::value&,int)>* m_exact_match_cb;
//  a state number : (_1).a.b.c[ 17 ].d.e (a.b)=1 (c[)=2  (17)=3 (.d.e)=4
size_t current_state;//contain the current state of the state machine for searching-expression (each JSON variable in SQL statement has a searching expression)
int nested_array_level;//in the case of array within array it contain the nesting level
int m_json_index;
s3selectEngine::value v_null;
size_t m_from_clause_size;

struct variable_state_md {
    std::vector<std::string> required_path;//set by the syntax-parser. in the case of array its empty
    int required_array_entry_no;//set by the syntax-parser, in the case of object-key its -1.
    int actual_array_entry_no;//upon scanning the JSON input, this value increased by 1 each new element 
    int required_depth_size;// depth of state, is aggregated (include the previous). it's the summary of key-elements and array-operator's.
    int required_key_depth_size;// same as the above, not including the array-operators.
    int last_array_start;//it actually mark the nested-array-level (array within array)
};

std::vector<struct variable_state_md> variable_states;//vector is populated upon syntax phase.

public:

json_variable_access():from_clause(nullptr),key_path(nullptr),m_current_depth(nullptr),m_current_depth_non_anonymous(nullptr),m_exact_match_cb(nullptr),current_state(-1),nested_array_level(0),m_json_index(-1),v_null(nullptr),m_from_clause_size(0)
{}

void init(
	  std::vector<std::string>* reader_from_clause,
	  std::vector<std::string>* reader_key_path,
	  int* reader_current_depth,
	  int* reader_m_current_depth_non_anonymous,
	  std::function <int(s3selectEngine::value&,int)>* excat_match_cb,
	  int json_index)
{//this routine should be called before scanning the JSON input
  from_clause = reader_from_clause;
  key_path = reader_key_path;
  m_exact_match_cb = excat_match_cb;
  //m_current_depth and m_current_depth_non_anonymous points to the JSON reader variables.
  m_current_depth = reader_current_depth;
  m_current_depth_non_anonymous = reader_m_current_depth_non_anonymous;
  current_state = 0;
  m_json_index = json_index;
  m_from_clause_size = from_clause->size();

  //loop on variable_states compute required_depth_size
}

void clear()
{
	variable_states.clear();
}

void debug_info()
{
  auto f = [](std::vector<std::string> x){std::string res;for(auto i : x){res.append(i);res.append(".");};return res;};

  std::cout << "m_current_depth=" << *m_current_depth << " required_depth_size= " << reader_position_state().required_depth_size << " ";
  std::cout << "variable_states[ current_state ].last_array_start=" << reader_position_state().last_array_start;
  std::cout << " current_state=" << current_state << " key_path=" << f(*key_path) << std::endl; 
}
#define DBG {std::cout << "event=" << __FUNCTION__ << std::endl; debug_info();}
#undef DBG
#define DBG

void compile_state_machine()
{
  size_t aggregated_required_depth_size = 0;
  size_t aggregated_required_key_depth_size = 0;
  for(auto& v : variable_states)
  {
    if(v.required_path.size())
    {
      v.required_depth_size = aggregated_required_depth_size + v.required_path.size();//depth size in general, including array 
      v.required_key_depth_size = aggregated_required_key_depth_size;//depth include ONLY key parts 
      aggregated_required_key_depth_size += v.required_path.size();
    }
    else
    {
      v.required_depth_size = aggregated_required_depth_size + 1;
    }
    aggregated_required_depth_size = v.required_depth_size;
  }
}

void push_variable_state(std::vector<std::string>& required_path,int required_array_entry_no)
{
  struct variable_state_md new_state={required_path,required_array_entry_no,-1,0,0,-1};
  variable_states.push_back(new_state);
  //TODO required_path.size() > 0 or required_path,required_array_entry_no>=0 : not both
  compile_state_machine();
}

struct variable_state_md& reader_position_state()
{
  if (current_state>=variable_states.size())
  {//in case the state-machine reached a "dead-end", should push a null for that JSON variable
	//going back one state.
	(*m_exact_match_cb)(v_null,m_json_index);
	decrease_current_state();
  }

  return variable_states[ current_state ];
}

bool is_array_state()
{
  return (reader_position_state().required_array_entry_no >= 0);
}

bool is_reader_located_on_required_depth()
{
  //upon user request `select _1.a.b from s3object[*].c.d;` the c.d sould "cut off" from m_current_depth_non_anonymous
  //to get the correct depth of the state-machine
  return ((*m_current_depth_non_anonymous - static_cast<int>(m_from_clause_size)) == reader_position_state().required_depth_size);
}

bool is_on_final_state()
{
  return (current_state == (variable_states.size()));  
	  //&& *m_current_depth == variable_states[ current_state -1 ].required_depth_size); 
	  
	  // NOTE: by ignoring the current-depth, the matcher gives precedence to key-path match, while not ignoring accessing using array
	  // meaning, upon requeting a.b[12] , the [12] is not ignored, the a<-->b distance should be calculated as key distance, i.e. not counting array/object with *no keys*.
	  // user may request 'select _1.phonearray.num'; the reader will traverse `num` exist in `phonearray`
}

bool is_reader_passed_required_array_entry()
{
  return (reader_position_state().actual_array_entry_no > reader_position_state().required_array_entry_no);
}

bool is_reader_located_on_array_according_to_current_state()
{
  return (nested_array_level == reader_position_state().last_array_start);
}

bool is_reader_position_depth_lower_than_required()
{
  //upon user request `select _1.a.b from s3object[*].c.d;` the c.d sould "cut off" from m_current_depth_non_anonymous
  //to have the correct depth of the state-machine
  return ((*m_current_depth_non_anonymous - static_cast<int>(m_from_clause_size)) < reader_position_state().required_depth_size);
}

bool is_reader_located_on_array_entry_according_to_current_state()
{
  return (reader_position_state().actual_array_entry_no == reader_position_state().required_array_entry_no);
}

void increase_current_state()
{
  DBG

  if(current_state >= variable_states.size()) return;
  current_state ++;
}

void decrease_current_state()
{ 
  DBG

  if(current_state == 0) return;
  current_state --;
}

void key()
{
  DBG

  if(reader_position_state().required_path.size())//current state is a key
  {
    std::vector<std::string>* filter = &reader_position_state().required_path;
    auto required_key_depth_size = reader_position_state().required_key_depth_size;
    if(std::equal((*key_path).begin()+(*from_clause).size() + required_key_depth_size, //key-path-start-point + from-clause-depth-size + key-depth
		  (*key_path).end(), 
		  (*filter).begin(),
		  (*filter).end(), iequal_predicate))
    {
      increase_current_state();//key match according to user request, advancing to the next state
    }
  }
}

void increase_array_index()
{
  if(is_reader_located_on_required_depth() && is_array_state())//TODO && is_array_state().  is it necessary? what about nesting level
  {
      DBG
      reader_position_state().actual_array_entry_no++;
  }
}

void dec_key()
{
  DBG

  if(is_reader_position_depth_lower_than_required())
  {//actual key-path is shorter than required
    decrease_current_state();
    return;
  }

  if(is_reader_located_on_required_depth() && is_array_state())//TODO && is_array_state().  is it necessary?; json_element_state.back() != ARRAY_STATE)
  {//key-path-depth matches, and it an array
    if(is_reader_located_on_array_entry_according_to_current_state())
    {//we reached the required array entry
      increase_current_state();
    } 
    else if(is_reader_passed_required_array_entry())
    {//had passed the array entry
      decrease_current_state();
    }
  }
}

void new_value(s3selectEngine::value& v)
{
  DBG

  if(is_on_final_state())
  {
    (*m_exact_match_cb)(v, m_json_index);
    decrease_current_state();//the state-machine reached its final destination, "going back" one state, upon another match condition the matched value will override the last one
  } 
  increase_array_index();//next-value in array
}

void end_object()
{
  increase_array_index();
}

void end_array()
{
  //init the correct array index
  DBG

  if(is_reader_located_on_array_according_to_current_state())
  {//it reached end of required array
    reader_position_state().actual_array_entry_no = 0;
    decrease_current_state();
  }
  nested_array_level --;

  // option 1. move out of one array, and enter a new one; option-2. enter an object
  increase_array_index();//increase only upon correct array //TODO move it into dec_key()?
  dec_key();
}

void start_array()
{
  DBG

  nested_array_level++;
  if(is_reader_located_on_required_depth())
  {//reader entered an array required by JSON variable 
    reader_position_state().actual_array_entry_no = 0;
    reader_position_state().last_array_start  = nested_array_level;
    
    if(is_reader_located_on_array_entry_according_to_current_state())
    {//we reached the required array entry -> next state
      increase_current_state();
    }
  }
}

}; //json_variable_access

class json_variables_operations {

	public:
    
	std::vector<std::pair<json_variable_access*,size_t>> json_statement_variables{};

	void init(std::vector<std::pair<json_variable_access*,size_t>>& jsv, //TODO init upon construction?
		  std::vector <std::string>* from_clause,
		  std::vector<std::string>* key_path,
		  int* current_depth,
		  int* current_depth_non_anonymous,
		  std::function <int(s3selectEngine::value&,int)>* exact_match_cb)
	{
	  json_statement_variables = jsv;
	  int i=0;//the index per JSON variable
	  for(auto& var : json_statement_variables)
	  {
	    var.first->init(from_clause,
		      key_path,
		      current_depth,
		      current_depth_non_anonymous,
		      exact_match_cb,i++);
	  }
	}

	void start_array()
	{
	    for(auto& j : json_statement_variables)
	    {
	      j.first->start_array();
	    }
	}
	void end_array()
	{
	    for(auto& j : json_statement_variables)
	    {
	      j.first->end_array();
	    }
	}
	void dec_key()
	{
	    for(auto& j : json_statement_variables)
	    {
	      j.first->dec_key();
	    }
	}
	void end_object()
	{
	    for(auto& j : json_statement_variables)
	    {
	      j.first->end_object();
	    }
	}
	void key()
	{
	    for(auto& j : json_statement_variables)
	    {
	      j.first->key();
	    }
	}
	void new_value(s3selectEngine::value& v)
	{
	    for(auto& j : json_statement_variables)
	    {
	      j.first->new_value(v);
	    }
	}
};//json_variables_operations

class JsonParserHandler : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>, JsonParserHandler> {

  public:

    typedef enum {OBJECT_STATE,ARRAY_STATE} en_json_elm_state_t;
    typedef std::pair<std::vector<std::string>, s3selectEngine::value> json_key_value_t;

    row_state state = row_state::NA;
    std::function <int(s3selectEngine::value&,int)> m_exact_match_cb;
    std::function <int(s3selectEngine::scratch_area::json_key_value_t&)> m_star_operation_cb;

    json_variables_operations variable_match_operations;
    int row_count{};
    std::vector <std::string> from_clause{};
    bool prefix_match{};
    s3selectEngine::value var_value;
    ChunksStreamer stream_buffer;
    bool init_buffer_stream;
    rapidjson::Reader reader;
    std::vector<en_json_elm_state_t> json_element_state;
    std::vector<std::string> key_path;
    std::function<int(void)> m_s3select_processing;
    int m_start_row_depth;   
    int m_current_depth;
    int m_current_depth_non_anonymous;
    bool m_star_operation;
    int m_sql_processing_status;
    bool m_fatal_initialization_ind = false;
    std::string m_fatal_initialization_description;

    JsonParserHandler() : prefix_match(false),init_buffer_stream(false),m_start_row_depth(-1),m_current_depth(0),m_current_depth_non_anonymous(0),m_star_operation(false),m_sql_processing_status(0)
    {
    } 

    std::string get_key_path()
    {//for debug
	  std::string res;
	  for(const auto & i: key_path)
	  {
	    res.append(i);
	    res.append(std::string("/"));
	  }
	  return res;
    }

    void dec_key_path()
    {
      if (json_element_state.size())  {
        if(json_element_state.back() != ARRAY_STATE)  {
	        if(key_path.size() != 0) {
	          key_path.pop_back();
          }
        }
      }
     
      variable_match_operations.dec_key();

      //TODO m_current_depth-- should done here 
      if(m_start_row_depth > m_current_depth)
      {
	  prefix_match = false;
      } else
      if (prefix_match) {
          if (state == row_state::ARRAY_START_ROW && m_start_row_depth == m_current_depth) {
	    m_sql_processing_status = m_s3select_processing(); //per each element in array
            ++row_count;
          }
      }
    }

    void push_new_key_value(s3selectEngine::value& v)
    { 
      if (m_star_operation && prefix_match)
      {
	json_key_value_t key_value(key_path,v);
	m_star_operation_cb(key_value);
      }
	if (prefix_match)
      		variable_match_operations.new_value(v);

      dec_key_path();
    }

    bool Null() {
      var_value.setnull();
      push_new_key_value(var_value);
      return true; }

    bool Bool(bool b) {
      var_value = b;
      push_new_key_value(var_value);
      return true; }

    bool Int(int i) { 
      var_value = i;
      push_new_key_value(var_value);
      return true; }

    bool Uint(unsigned u) {
      var_value = u;
      push_new_key_value(var_value);
      return true; }

    bool Int64(int64_t i) { 
      var_value = i;
      push_new_key_value(var_value);
      return true; }

    bool Uint64(uint64_t u) { 
      var_value = u;
      push_new_key_value(var_value);
      return true; }

    bool Double(double d) { 
      var_value = d;
      push_new_key_value(var_value);
      return true; }

    bool String(const char* str, rapidjson::SizeType length, bool copy) {
      //TODO use copy
      var_value = str;
      push_new_key_value(var_value);
      return true;
    }

    bool Key(const char* str, rapidjson::SizeType length, bool copy) {
      key_path.push_back(std::string(str));
     
      if(!m_current_depth_non_anonymous){
      //important: upon a key and m_current_depth_non_anonymous is ZERO
      //it should advance by 1. to get the correct current depth(for non anonymous counter).
	m_current_depth_non_anonymous++;
      } 

      if(from_clause.size() == 0 || std::equal(key_path.begin(), key_path.end(), from_clause.begin(), from_clause.end(), iequal_predicate)) {
        prefix_match = true;
      }

      variable_match_operations.key();

      return true;
    }

    bool is_already_row_started()
    {
      if(state == row_state::OBJECT_START_ROW || state == row_state::ARRAY_START_ROW)
	return true;
      else
	return false;
    }

    bool StartObject() { 
	json_element_state.push_back(OBJECT_STATE);
	m_current_depth++;
	if(key_path.size()){
	  //advancing the counter only upon there is a key.
	  m_current_depth_non_anonymous++;
	}

        if (prefix_match && !is_already_row_started()) {
          state = row_state::OBJECT_START_ROW;
	  m_start_row_depth = m_current_depth;
          ++row_count;
        }

      return true; 
    }
  
    bool EndObject(rapidjson::SizeType memberCount) {
      json_element_state.pop_back();
      m_current_depth --;
      m_current_depth_non_anonymous --;

      variable_match_operations.end_object();
      
      dec_key_path();
      if (state == row_state::OBJECT_START_ROW && (m_start_row_depth > m_current_depth)) {
	m_sql_processing_status = m_s3select_processing();
	state = row_state::NA;
      }
      return true; 
    }
 
    bool StartArray() {
      json_element_state.push_back(ARRAY_STATE);
      m_current_depth++;
      if(key_path.size()){
	  //advancing the counter only upon there is a key.
	  m_current_depth_non_anonymous++;
      }

      if (prefix_match && !is_already_row_started()) {
          state = row_state::ARRAY_START_ROW;
	  m_start_row_depth = m_current_depth;
        }
 
      variable_match_operations.start_array();

      return true;
    }

    bool EndArray(rapidjson::SizeType elementCount) { 
      json_element_state.pop_back();
      m_current_depth--;
      m_current_depth_non_anonymous--;

      dec_key_path();

      if (state == row_state::ARRAY_START_ROW && (m_start_row_depth > m_current_depth)) {
	state = row_state::NA;
      }
     
      variable_match_operations.end_array();

      return true;
    }

    void set_prefix_match(std::vector<std::string>& requested_prefix_match)
    {//purpose: set the filter according to SQL statement(from clause)
      from_clause = requested_prefix_match;
      if(from_clause.size() ==0)
      {
	prefix_match = true;
	m_start_row_depth = m_current_depth;
      }
    }

    void set_statement_json_variables(std::vector<std::pair<json_variable_access*,size_t>>& statement_variables)
    {//purpose: set the json variables extracted from the SQL statement(projection columns, predicates columns)
		    variable_match_operations.init(
			  statement_variables,
			  &from_clause,
			  &key_path,
			  &m_current_depth,
			  &m_current_depth_non_anonymous,
			  &m_exact_match_cb);
    }

    void set_exact_match_callback(std::function<int(s3selectEngine::value&, int)> f)
    {//purpose: upon key is matching one of the exact filters, the callback is called.
      m_exact_match_cb = f;
    }

    void set_s3select_processing_callback(std::function<int(void)>& f)
    {//purpose: execute s3select statement on matching row (according to filters)
      m_s3select_processing = f;
    }

    void set_push_per_star_operation_callback( std::function <int(s3selectEngine::scratch_area::json_key_value_t&)> cb)
    {
      m_star_operation_cb = cb;
    }

    void set_star_operation()
    {
      m_star_operation = true;
    }

    bool is_fatal_initialization()
    {
      return m_fatal_initialization_ind;
    }

    int process_json_buffer(char* json_buffer,size_t json_buffer_sz, bool end_of_stream=false)
    {//user keeps calling with buffers, the method is not aware of the object size.


      try{
	    if(!init_buffer_stream)
	    {
		    //set the memoryStreamer
		    reader.IterativeParseInit();
		    init_buffer_stream = true;

	    }

	    //the non-processed bytes plus the next chunk are copy into main processing buffer 
	    if(!end_of_stream)
		    stream_buffer.resetBuffer(json_buffer, json_buffer_sz);

	    while (!reader.IterativeParseComplete()) {
		    reader.IterativeParseNext<rapidjson::kParseDefaultFlags>(stream_buffer, *this);

		    //once all key-values move into s3select(for further filtering and processing), it should be cleared

		    //TODO in the case the chunk is too small or some value in input is too big, the parsing will fail.
		    if (!end_of_stream && stream_buffer.next_src_==0 && stream_buffer.getBytesLeft() < 2048)
		    {//the non processed bytes will be processed on next fetched chunk
		     //TODO save remaining-bytes to internal buffer (or caller will use 2 sets of buffer)
			    stream_buffer.saveRemainingBytes();
			    return 0;
		    }
		    if(m_sql_processing_status == JSON_PROCESSING_LIMIT_REACHED)//return status(int) from callback
		    {
			    return JSON_PROCESSING_LIMIT_REACHED;
		    }

		    // error message
		    if(reader.HasParseError())  {
			    rapidjson::ParseErrorCode c = reader.GetParseErrorCode();
			    size_t ofs = reader.GetErrorOffset();
			    std::stringstream error_str;
			    error_str << "parsing error. code:" << c << " position: " << ofs << std::endl;
			    throw s3selectEngine::base_s3select_exception(error_str.str(), s3selectEngine::base_s3select_exception::s3select_exp_en_t::FATAL);
			    return -1;	  
		    }
	    }//while reader.IterativeParseComplete
	}
        catch(std::exception &e){
		std::stringstream error_str;
                error_str << "failed to process JSON : " << e.what() << std::endl;
		throw s3selectEngine::base_s3select_exception(error_str.str(), s3selectEngine::base_s3select_exception::s3select_exp_en_t::FATAL);
                return -1;
        }
	return 0;
    }
};


#endif