Adding upstream version 18.2.2.upstream/18.2.2

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 3153 insertions, 0 deletions

diff --git a/src/s3select/include/s3select.h b/src/s3select/include/s3select.h
new file mode 100644
index 000000000..3ac111351
--- /dev/null
+++ b/src/s3select/include/s3select.h
@@ -0,0 +1,3153 @@
+#ifndef __S3SELECT__
+#define __S3SELECT__
+#define BOOST_SPIRIT_THREADSAFE
+#define CSV_IO_NO_THREAD
+
+#pragma once
+#define BOOST_BIND_GLOBAL_PLACEHOLDERS
+#include <boost/spirit/include/classic_core.hpp>
+#include <boost/algorithm/string.hpp>
+#include <iostream>
+#include <string>
+#include <list>
+#include <deque>
+#include "s3select_oper.h"
+#include "s3select_functions.h"
+#include "s3select_csv_parser.h"
+#include "s3select_json_parser.h"
+#include <boost/function.hpp>
+#include <boost/bind.hpp>
+#include <functional>
+
+#define _DEBUG_TERM {string  token(a,b);std::cout << __FUNCTION__ << token << std::endl;}
+
+namespace s3selectEngine
+{
+
+/// AST builder
+
+class s3select_projections
+{
+
+private:
+  std::vector<base_statement*> m_projections;
+
+public:
+
+  std::vector<base_statement*>* get()
+  {
+    return &m_projections;
+  }
+
+};
+
+static s3select_reserved_word g_s3select_reserve_word;//read-only
+
+struct actionQ
+{
+// upon parser is accepting a token (lets say some number),
+// it push it into dedicated queue, later those tokens are poped out to build some "higher" contruct (lets say 1 + 2)
+// those containers are used only for parsing phase and not for runtime.
+
+  std::vector<mulldiv_operation::muldiv_t> muldivQ;
+  std::vector<addsub_operation::addsub_op_t> addsubQ;
+  std::vector<arithmetic_operand::cmp_t> arithmetic_compareQ;
+  std::vector<logical_operand::oplog_t> logical_compareQ;
+  std::vector<base_statement*> exprQ;
+  std::vector<base_statement*> funcQ;
+  std::vector<base_statement*> whenThenQ;
+  std::vector<base_statement*> inPredicateQ;
+  base_statement* inMainArg;
+  std::vector<std::string> dataTypeQ;
+  std::vector<std::string> trimTypeQ;
+  std::vector<std::string> datePartQ;
+  projection_alias alias_map;
+  std::string from_clause;
+  std::vector<std::string> json_from_clause;
+  bool limit_op;
+  unsigned long limit;
+  std::string column_prefix;
+  std::string table_alias;
+  s3select_projections  projections;
+
+  bool projection_or_predicate_state; //true->projection false->predicate(where-clause statement)
+  std::vector<base_statement*> predicate_columns;
+  std::vector<base_statement*> projections_columns; 
+  base_statement* first_when_then_expr;
+
+  std::string json_array_name; // _1.a[  ]    json_array_name = "a";  upon parser is scanning a correct json-path; json_array_name will contain the array name. 
+  std::string json_object_name; // _1.b json_object_name = "b" ; upon parser is scanning a correct json-path; json_object_name will contain the object name.
+  std::deque<size_t> json_array_index_number; //  _1.a.c[ some integer number >=0 ]; upon parser is scanning a correct json-path; json_array_index_number will contain the array index.
+					       //  or in the case of multidimensional contain seiries of index number
+			     
+  json_variable_access json_var_md;
+
+  std::vector<std::pair<json_variable_access*,size_t>> json_statement_variables_match_expression;//contains all statement variables and their search-expression for locating the correct values in input document
+
+  actionQ(): inMainArg(0),from_clause("##"),limit_op(false),column_prefix("##"),table_alias("##"),projection_or_predicate_state(true),first_when_then_expr(nullptr){}
+
+  std::map<const void*,std::vector<const char*> *> x_map;
+ 
+  ~actionQ()
+  {
+    for(auto m : x_map)
+      delete m.second;
+  }
+  
+  bool is_already_scanned(const void *th,const char *a)
+  {
+    //purpose: caller get indication in the case a specific builder is scan more than once the same text(pointer)
+    auto t = x_map.find(th);
+
+    if(t == x_map.end())
+    {
+      auto v = new std::vector<const char*>;
+      x_map.insert(std::pair<const void*,std::vector<const char*> *>(th,v));
+      v->push_back(a);
+    }
+    else
+    {
+      for(auto& c : *(t->second))
+      {
+        if( strcmp(c,a) == 0)
+          return true;
+      }
+      t->second->push_back(a);
+    }
+    return false;
+  }
+
+};
+
+class s3select;
+
+struct base_ast_builder
+{
+  void operator()(s3select* self, const char* a, const char* b) const;
+
+  virtual void builder(s3select* self, const char* a, const char* b) const = 0;
+  
+  virtual ~base_ast_builder() = default;
+};
+
+struct push_from_clause : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_from_clause g_push_from_clause;
+
+struct push_json_from_clause : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_json_from_clause g_push_json_from_clause;
+
+struct push_limit_clause : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_limit_clause g_push_limit_clause;
+
+struct push_number : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_number g_push_number;
+
+struct push_float_number : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_float_number g_push_float_number;
+
+struct push_string : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_string g_push_string;
+
+struct push_variable : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_variable g_push_variable;
+
+struct push_json_variable : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_json_variable g_push_json_variable;
+
+/////////////////////////arithmetic unit  /////////////////
+struct push_addsub : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_addsub g_push_addsub;
+
+struct push_mulop : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_mulop g_push_mulop;
+
+struct push_addsub_binop : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_addsub_binop g_push_addsub_binop;
+
+struct push_mulldiv_binop : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_mulldiv_binop g_push_mulldiv_binop;
+
+struct push_function_arg : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_function_arg g_push_function_arg;
+
+struct push_function_name : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_function_name g_push_function_name;
+
+struct push_function_expr : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_function_expr g_push_function_expr;
+
+struct push_cast_expr : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_cast_expr g_push_cast_expr;
+
+struct push_cast_decimal_expr : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_cast_decimal_expr g_push_cast_decimal_expr;
+
+struct push_decimal_operator : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_decimal_operator g_push_decimal_operator;
+
+struct push_data_type : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_data_type g_push_data_type;
+
+////////////////////// logical unit ////////////////////////
+
+struct push_compare_operator : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+
+};
+static push_compare_operator g_push_compare_operator;
+
+struct push_logical_operator : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+
+};
+static push_logical_operator g_push_logical_operator;
+
+struct push_arithmetic_predicate : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+
+};
+static push_arithmetic_predicate g_push_arithmetic_predicate;
+
+struct push_logical_predicate : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_logical_predicate g_push_logical_predicate;
+
+struct push_negation : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_negation g_push_negation;
+
+struct push_column_pos : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static  push_column_pos g_push_column_pos;
+
+struct push_projection : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_projection g_push_projection;
+
+struct push_alias_projection : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_alias_projection g_push_alias_projection;
+
+struct push_between_filter : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_between_filter g_push_between_filter;
+
+struct push_not_between_filter : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_not_between_filter g_push_not_between_filter;
+
+struct push_in_predicate : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_in_predicate g_push_in_predicate;
+
+struct push_in_predicate_arguments : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_in_predicate_arguments g_push_in_predicate_arguments;
+
+struct push_in_predicate_first_arg : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_in_predicate_first_arg g_push_in_predicate_first_arg;
+
+struct push_like_predicate_escape : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_like_predicate_escape g_push_like_predicate_escape;
+
+struct push_like_predicate_no_escape : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_like_predicate_no_escape g_push_like_predicate_no_escape;
+
+struct push_is_null_predicate : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_is_null_predicate g_push_is_null_predicate;
+
+struct push_case_when_else : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_case_when_else g_push_case_when_else;
+
+struct push_when_condition_then : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_when_condition_then g_push_when_condition_then;
+
+struct push_when_value_then : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_when_value_then g_push_when_value_then;
+
+struct push_case_value_when_value_else : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_case_value_when_value_else g_push_case_value_when_value_else;
+
+struct push_substr_from : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_substr_from g_push_substr_from;
+
+struct push_substr_from_for : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_substr_from_for g_push_substr_from_for;
+
+struct push_trim_type : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_trim_type g_push_trim_type; 
+
+struct push_trim_whitespace_both : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_trim_whitespace_both g_push_trim_whitespace_both;
+
+struct push_trim_expr_one_side_whitespace : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_trim_expr_one_side_whitespace g_push_trim_expr_one_side_whitespace;
+
+struct push_trim_expr_anychar_anyside : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_trim_expr_anychar_anyside g_push_trim_expr_anychar_anyside;
+
+struct push_datediff : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_datediff g_push_datediff;
+
+struct push_dateadd : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_dateadd g_push_dateadd;
+
+struct push_extract : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_extract g_push_extract;
+
+struct push_date_part : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_date_part g_push_date_part;
+
+struct push_time_to_string_constant : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_time_to_string_constant g_push_time_to_string_constant;
+
+struct push_time_to_string_dynamic : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_time_to_string_dynamic g_push_time_to_string_dynamic;
+
+struct push_string_to_time_constant : public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_string_to_time_constant g_push_string_to_time_constant;
+
+struct push_array_number :  public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_array_number g_push_array_number;
+
+struct push_json_array_name :  public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_json_array_name g_push_json_array_name;
+
+struct push_json_object :  public base_ast_builder
+{
+  void builder(s3select* self, const char* a, const char* b) const;
+};
+static push_json_object g_push_json_object;
+
+struct s3select : public bsc::grammar<s3select>
+{
+private:
+
+  actionQ m_actionQ;
+  scratch_area m_sca;
+  s3select_functions m_s3select_functions;
+  std::string error_description;
+  s3select_allocator m_s3select_allocator;
+  bool aggr_flow = false;
+  bool m_json_query = false;
+  std::set<base_statement*> m_ast_nodes_to_delete;
+  base_function* m_to_timestamp_for_clean = nullptr;
+
+#define BOOST_BIND_ACTION( push_name ) boost::bind( &push_name::operator(), g_ ## push_name, const_cast<s3select*>(&self), _1, _2)
+
+public:
+
+  std::set<base_statement*>& get_ast_nodes_to_delete()
+  {
+    return m_ast_nodes_to_delete;
+  }
+
+  base_function* & get_to_timestamp_for_clean()
+  {
+    return m_to_timestamp_for_clean;
+  }
+
+  actionQ* getAction()
+  {
+    return &m_actionQ;
+  }
+
+  s3select_allocator* getAllocator()
+  {
+    return &m_s3select_allocator;
+  }
+
+  s3select_functions* getS3F()
+  {
+    return &m_s3select_functions;
+  }
+
+  int semantic()
+  {
+    for (const auto &e : get_projections_list())
+    {
+      e->resolve_node();
+      //upon validate there is no aggregation-function nested calls, it validates legit aggregation call. 
+      if (e->is_nested_aggregate(aggr_flow))
+      {
+        error_description = "nested aggregation function is illegal i.e. sum(...sum ...)";
+        throw base_s3select_exception(error_description, base_s3select_exception::s3select_exp_en_t::FATAL);
+      }
+
+      e->push_for_cleanup(m_ast_nodes_to_delete);
+    }
+
+    if(get_filter())
+	    get_filter()->push_for_cleanup(m_ast_nodes_to_delete);
+
+    if (aggr_flow == true)
+    {// atleast one projection column contain aggregation function
+      for (const auto &e : get_projections_list())
+      {
+        auto aggregate_expr = e->get_aggregate();
+
+        if (aggregate_expr)
+        {
+          //per each column, subtree is mark to skip except for the aggregation function subtree. 
+          //for an example: substring( ... , sum() , count() ) :: the substring is mark to skip execution, while sum and count not.
+          e->set_skip_non_aggregate(true);
+          e->mark_aggreagtion_subtree_to_execute();
+        }
+        else
+        {
+          //in case projection column is not aggregate, the projection column must *not* contain reference to columns.
+          if(e->is_column_reference())
+          {
+            error_description = "illegal query; projection contains aggregation function is not allowed with projection contains column reference";
+            throw base_s3select_exception(error_description, base_s3select_exception::s3select_exp_en_t::FATAL);
+          }
+        }
+        
+      }
+    }
+    
+    m_json_query = (m_actionQ.json_from_clause.size() != 0);
+    
+    return 0;
+  }
+
+  int parse_query(const char* input_query)
+  {
+    if(get_projections_list().empty() == false)
+    {
+      return 0;  //already parsed
+    }
+
+
+    error_description.clear();
+    aggr_flow = false;
+
+    try
+    {
+      bsc::parse_info<> info = bsc::parse(input_query, *this, bsc::space_p);
+      auto query_parse_position = info.stop;
+
+      if (!info.full)
+      {
+        error_description = std::string("failure -->") + query_parse_position + std::string("<---");
+        return -1;
+      }
+
+      semantic();
+    }
+    catch (base_s3select_exception& e)
+    {
+      error_description.assign(e.what());
+      if (e.severity() == base_s3select_exception::s3select_exp_en_t::FATAL) //abort query execution
+      {
+        return -1;
+      }
+    }
+
+    return 0;
+  }
+
+  std::string get_error_description()
+  {
+    return error_description;
+  }
+
+  s3select()
+  {
+    m_s3select_functions.setAllocator(&m_s3select_allocator);
+    m_s3select_functions.set_AST_nodes_for_cleanup(&m_ast_nodes_to_delete);
+  }
+
+  bool is_semantic()//TBD traverse and validate semantics per all nodes
+  {
+    base_statement* cond = m_actionQ.exprQ.back();
+
+    return  cond->semantic();
+  }
+
+  std::string get_from_clause() const
+  {
+    return m_actionQ.from_clause;
+  }
+
+  bool is_limit()
+  {
+    return m_actionQ.limit_op;
+  }
+
+  unsigned long get_limit()
+  {
+    return m_actionQ.limit;
+  }
+
+  void load_schema(std::vector< std::string>& scm)
+  {
+    int i = 0;
+    for (auto& c : scm)
+    {
+      m_sca.set_column_pos(c.c_str(), i++);
+    }
+  }
+
+  base_statement* get_filter()
+  {
+    if(m_actionQ.exprQ.empty())
+    {
+      return nullptr;
+    }
+
+    return m_actionQ.exprQ.back();
+  }
+
+  std::vector<base_statement*>  get_projections_list()
+  {
+    return *m_actionQ.projections.get(); //TODO return COPY(?) or to return evalaution results (list of class value{}) / return reference(?)
+  }
+
+  scratch_area* get_scratch_area()
+  {
+    return &m_sca;
+  }
+
+  projection_alias* get_aliases()
+  {
+    return &m_actionQ.alias_map;
+  }
+
+  std::vector<std::pair<json_variable_access*,size_t>>& get_json_variables_access()
+  {
+    return m_actionQ.json_statement_variables_match_expression;
+  }
+
+  bool is_aggregate_query() const
+  {
+    return aggr_flow == true;
+  }
+
+  bool is_json_query()
+  {
+    return m_json_query;
+  }
+
+  ~s3select()
+  {
+	for(auto it : m_ast_nodes_to_delete)
+	{
+		if (it->is_function())
+      		{//upon its a function, call to the implementation destructor
+        		if(dynamic_cast<__function*>(it)->impl())
+				dynamic_cast<__function*>(it)->impl()->dtor();
+      		}
+		//calling to destrcutor of class-function itself, or non-function destructor
+		it->dtor();
+	}
+
+	for(auto x: m_actionQ.json_statement_variables_match_expression)
+	{//the json_variable_access object is allocated by S3SELECT_NEW. this object contains stl-vector that should be free 
+		x.first->~json_variable_access();
+	}
+  if(m_to_timestamp_for_clean)
+  { 
+    m_to_timestamp_for_clean->dtor();
+  }
+  }
+
+#define JSON_ROOT_OBJECT "s3object[*]"
+
+//the input is converted to lower case
+#define S3SELECT_KW( reserve_word ) bsc::as_lower_d[ reserve_word ]
+
+  template <typename ScannerT>
+  struct definition
+  {
+    explicit definition(s3select const& self)
+    {
+      ///// s3select syntax rules and actions for building AST
+
+      select_expr =  select_expr_base_ >> bsc::lexeme_d[ *(bsc::str_p(" ")|bsc::str_p(";")) ];
+
+      select_expr_base_ = select_expr_base >> S3SELECT_KW("limit") >> (limit_number)[BOOST_BIND_ACTION(push_limit_clause)] | select_expr_base;
+
+      limit_number = (+bsc::digit_p);
+
+      select_expr_base =  S3SELECT_KW("select") >> projections >> S3SELECT_KW("from") >> (from_expression)[BOOST_BIND_ACTION(push_from_clause)] >> !where_clause ;
+
+      projections = projection_expression >> *( ',' >> projection_expression) ;
+
+      projection_expression = (arithmetic_expression >> S3SELECT_KW("as") >> alias_name)[BOOST_BIND_ACTION(push_alias_projection)] | 
+                              (arithmetic_expression)[BOOST_BIND_ACTION(push_projection)] | 
+			      (arithmetic_predicate >> S3SELECT_KW("as") >> alias_name)[BOOST_BIND_ACTION(push_alias_projection)] |
+                              (arithmetic_predicate)[BOOST_BIND_ACTION(push_projection)] ;
+
+      alias_name = bsc::lexeme_d[(+bsc::alpha_p >> *bsc::digit_p)] ;
+
+      when_case_else_projection = (S3SELECT_KW("case")  >> (+when_stmt) >> S3SELECT_KW("else") >> arithmetic_expression >> S3SELECT_KW("end")) [BOOST_BIND_ACTION(push_case_when_else)];
+
+      when_stmt = (S3SELECT_KW("when") >> condition_expression >> S3SELECT_KW("then") >> arithmetic_expression)[BOOST_BIND_ACTION(push_when_condition_then)];
+
+      when_case_value_when = (S3SELECT_KW("case") >> arithmetic_expression >> 
+                              (+when_value_then) >> S3SELECT_KW("else") >> arithmetic_expression >> S3SELECT_KW("end")) [BOOST_BIND_ACTION(push_case_value_when_value_else)];
+
+      when_value_then = (S3SELECT_KW("when") >> arithmetic_expression >> S3SELECT_KW("then") >> arithmetic_expression)[BOOST_BIND_ACTION(push_when_value_then)];
+
+      from_expression = (s3_object >> variable_name ) | s3_object;
+
+      //the stdin and object_path are for debug purposes(not part of the specs)
+      s3_object = json_s3_object | S3SELECT_KW("stdin") | S3SELECT_KW("s3object") | object_path;
+
+      json_s3_object = ((S3SELECT_KW(JSON_ROOT_OBJECT)) >> *(bsc::str_p(".") >> json_path_element))[BOOST_BIND_ACTION(push_json_from_clause)];
+
+      json_path_element = bsc::lexeme_d[+( bsc::alnum_p | bsc::str_p("_")) ];
+
+      object_path = "/" >> *( fs_type >> "/") >> fs_type;
+
+      fs_type = bsc::lexeme_d[+( bsc::alnum_p | bsc::str_p(".")  | bsc::str_p("_")) ];
+
+      where_clause = S3SELECT_KW("where") >> condition_expression;
+
+      condition_expression = arithmetic_predicate;
+
+      arithmetic_predicate = (S3SELECT_KW("not") >> logical_predicate)[BOOST_BIND_ACTION(push_negation)] | logical_predicate;
+
+      logical_predicate =  (logical_and) >> *(or_op[BOOST_BIND_ACTION(push_logical_operator)] >> (logical_and)[BOOST_BIND_ACTION(push_logical_predicate)]);
+
+      logical_and =  (cmp_operand) >> *(and_op[BOOST_BIND_ACTION(push_logical_operator)] >> (cmp_operand)[BOOST_BIND_ACTION(push_logical_predicate)]);
+
+      cmp_operand = special_predicates | (factor) >> *(arith_cmp[BOOST_BIND_ACTION(push_compare_operator)] >> (factor)[BOOST_BIND_ACTION(push_arithmetic_predicate)]);
+
+      special_predicates = (is_null) | (is_not_null) | (between_predicate) | (not_between) | (in_predicate) | (like_predicate);
+
+      is_null = ((factor) >> S3SELECT_KW("is") >> S3SELECT_KW("null"))[BOOST_BIND_ACTION(push_is_null_predicate)];
+
+      is_not_null = ((factor) >> S3SELECT_KW("is") >> S3SELECT_KW("not") >> S3SELECT_KW("null"))[BOOST_BIND_ACTION(push_is_null_predicate)];
+
+      between_predicate = (arithmetic_expression >> S3SELECT_KW("between") >> arithmetic_expression >> S3SELECT_KW("and") >> arithmetic_expression)[BOOST_BIND_ACTION(push_between_filter)];
+
+      not_between = (arithmetic_expression >> S3SELECT_KW("not") >> S3SELECT_KW("between") >> arithmetic_expression >> S3SELECT_KW("and") >> arithmetic_expression)[BOOST_BIND_ACTION(push_not_between_filter)];
+
+      in_predicate = (arithmetic_expression >> S3SELECT_KW("in") >> '(' >> arithmetic_expression[BOOST_BIND_ACTION(push_in_predicate_first_arg)] >> *(',' >> arithmetic_expression[BOOST_BIND_ACTION(push_in_predicate_arguments)]) >> ')')[BOOST_BIND_ACTION(push_in_predicate)];
+      
+      like_predicate = (like_predicate_escape) |(like_predicate_no_escape);
+
+      like_predicate_no_escape = (arithmetic_expression >> S3SELECT_KW("like") >> arithmetic_expression)[BOOST_BIND_ACTION(push_like_predicate_no_escape)];
+
+      like_predicate_escape = (arithmetic_expression >> S3SELECT_KW("like") >> arithmetic_expression >> S3SELECT_KW("escape") >> arithmetic_expression)[BOOST_BIND_ACTION(push_like_predicate_escape)];
+
+      factor = arithmetic_expression  | ( '(' >> arithmetic_predicate >> ')' ) ; 
+
+      arithmetic_expression = (addsub_operand >> *(addsubop_operator[BOOST_BIND_ACTION(push_addsub)] >> addsub_operand[BOOST_BIND_ACTION(push_addsub_binop)] ));
+
+      addsub_operand = (mulldiv_operand >> *(muldiv_operator[BOOST_BIND_ACTION(push_mulop)]  >> mulldiv_operand[BOOST_BIND_ACTION(push_mulldiv_binop)] ));// this non-terminal gives precedense to  mull/div
+
+      mulldiv_operand = arithmetic_argument | ('(' >> (arithmetic_expression) >> ')') ;
+
+      list_of_function_arguments = (arithmetic_expression)[BOOST_BIND_ACTION(push_function_arg)] >> *(',' >> (arithmetic_expression)[BOOST_BIND_ACTION(push_function_arg)]);
+
+      reserved_function_names = (S3SELECT_KW("when")|S3SELECT_KW("case")|S3SELECT_KW("then")|S3SELECT_KW("not")|S3SELECT_KW("limit")|S3SELECT_KW("where")|S3SELECT_KW("in")|S3SELECT_KW("between") |
+				S3SELECT_KW("like")|S3SELECT_KW("is") );
+     
+      function = ( ((variable_name)  >> '(' )[BOOST_BIND_ACTION(push_function_name)] >> !list_of_function_arguments >> ')')[BOOST_BIND_ACTION(push_function_expr)];
+
+      arithmetic_argument = (float_number)[BOOST_BIND_ACTION(push_float_number)] |  (number)[BOOST_BIND_ACTION(push_number)] | (json_variable_name)[BOOST_BIND_ACTION(push_json_variable)] |
+			    (column_pos)[BOOST_BIND_ACTION(push_column_pos)] |
+                            (string)[BOOST_BIND_ACTION(push_string)] | (backtick_string) | (datediff) | (dateadd) | (extract) | (time_to_string_constant) | (time_to_string_dynamic) |
+                            (cast) | (substr) | (trim) | (when_case_value_when) | (when_case_else_projection) |
+                            (function) | (variable)[BOOST_BIND_ACTION(push_variable)]; //function is pushed by right-term
+
+      cast = cast_as_data_type | cast_as_decimal_expr ;
+
+      cast_as_data_type = (S3SELECT_KW("cast") >> '(' >> factor >> S3SELECT_KW("as") >> (data_type) >> ')') [BOOST_BIND_ACTION(push_cast_expr)];
+
+      cast_as_decimal_expr = (S3SELECT_KW("cast") >> '(' >> factor >> S3SELECT_KW("as") >> decimal_operator >> ')') [BOOST_BIND_ACTION(push_cast_decimal_expr)];
+
+      decimal_operator = (S3SELECT_KW("decimal") >> '(' >> (number)[BOOST_BIND_ACTION(push_number)] >> ',' >> (number)[BOOST_BIND_ACTION(push_number)] >> ')')
+					[BOOST_BIND_ACTION(push_decimal_operator)];
+
+      data_type = (S3SELECT_KW("int") | S3SELECT_KW("float") | S3SELECT_KW("string") |  S3SELECT_KW("timestamp") | S3SELECT_KW("bool"))[BOOST_BIND_ACTION(push_data_type)];
+     
+      substr = (substr_from) | (substr_from_for);
+      
+      substr_from = (S3SELECT_KW("substring") >> '(' >> (arithmetic_expression >> S3SELECT_KW("from") >> arithmetic_expression) >> ')') [BOOST_BIND_ACTION(push_substr_from)];
+
+      substr_from_for = (S3SELECT_KW("substring") >> '(' >> (arithmetic_expression >> S3SELECT_KW("from") >> arithmetic_expression >> S3SELECT_KW("for") >> arithmetic_expression) >> ')') [BOOST_BIND_ACTION(push_substr_from_for)];
+      
+      trim = (trim_whitespace_both) | (trim_one_side_whitespace) | (trim_anychar_anyside);
+
+      trim_one_side_whitespace = (S3SELECT_KW("trim") >> '(' >> (trim_type)[BOOST_BIND_ACTION(push_trim_type)] >> arithmetic_expression >> ')') [BOOST_BIND_ACTION(push_trim_expr_one_side_whitespace)];
+
+      trim_whitespace_both = (S3SELECT_KW("trim") >> '(' >> arithmetic_expression >> ')') [BOOST_BIND_ACTION(push_trim_whitespace_both)];
+
+      trim_anychar_anyside = (S3SELECT_KW("trim") >> '(' >> ((trim_remove_type)[BOOST_BIND_ACTION(push_trim_type)] >> arithmetic_expression >> S3SELECT_KW("from") >> arithmetic_expression)  >> ')') [BOOST_BIND_ACTION(push_trim_expr_anychar_anyside)];
+      
+      trim_type = ((S3SELECT_KW("leading") >> S3SELECT_KW("from")) | ( S3SELECT_KW("trailing") >> S3SELECT_KW("from")) | (S3SELECT_KW("both") >> S3SELECT_KW("from")) | S3SELECT_KW("from") ); 
+
+      trim_remove_type = (S3SELECT_KW("leading") | S3SELECT_KW("trailing") | S3SELECT_KW("both") );
+
+      datediff = (S3SELECT_KW("date_diff") >> '(' >> date_part >> ',' >> arithmetic_expression >> ',' >> arithmetic_expression >> ')') [BOOST_BIND_ACTION(push_datediff)];
+
+      dateadd = (S3SELECT_KW("date_add") >> '(' >> date_part >> ',' >> arithmetic_expression >> ',' >> arithmetic_expression >> ')') [BOOST_BIND_ACTION(push_dateadd)];
+
+      extract = (S3SELECT_KW("extract") >> '(' >> (date_part_extract)[BOOST_BIND_ACTION(push_date_part)] >> S3SELECT_KW("from") >> arithmetic_expression >> ')') [BOOST_BIND_ACTION(push_extract)];
+
+      date_part = (S3SELECT_KW("year") | S3SELECT_KW("month") | S3SELECT_KW("day") | S3SELECT_KW("hour")  | S3SELECT_KW("minute") | S3SELECT_KW("second")) [BOOST_BIND_ACTION(push_date_part)];
+
+      date_part_extract = ((date_part) |  S3SELECT_KW("week") | S3SELECT_KW("timezone_hour") | S3SELECT_KW("timezone_minute"));
+
+      time_to_string_constant = (S3SELECT_KW("to_string") >> '(' >> arithmetic_expression >> ',' >> (string)[BOOST_BIND_ACTION(push_string)] >> ')') [BOOST_BIND_ACTION(push_time_to_string_constant)];
+
+      time_to_string_dynamic = (S3SELECT_KW("to_string") >> '(' >> arithmetic_expression >> ',' >> arithmetic_expression >> ')') [BOOST_BIND_ACTION(push_time_to_string_dynamic)];
+
+      number = bsc::int_p;
+
+      float_number = bsc::real_p;
+
+      string = (bsc::str_p("\"") >> *( bsc::anychar_p - bsc::str_p("\"") ) >> bsc::str_p("\"")) | (bsc::str_p("\'") >> *( bsc::anychar_p - bsc::str_p("\'") ) >> bsc::str_p("\'"));
+
+      backtick_string = (bsc::str_p("`") >> *( bsc::anychar_p - bsc::str_p("`") ) >> bsc::str_p("`")) [BOOST_BIND_ACTION(push_string_to_time_constant)];
+
+      column_pos = (variable_name >> "." >> column_pos_name) | column_pos_name; //TODO what about space
+
+      column_pos_name = ('_'>>+(bsc::digit_p) ) | '*' ;
+
+      muldiv_operator = bsc::str_p("*") | bsc::str_p("/") | bsc::str_p("^") | bsc::str_p("%");// got precedense
+
+      addsubop_operator = bsc::str_p("+") | bsc::str_p("-");
+
+      arith_cmp = bsc::str_p("<>") | bsc::str_p(">=") | bsc::str_p("<=") | bsc::str_p("=") | bsc::str_p("<") | bsc::str_p(">") | bsc::str_p("!=");
+
+      and_op =  S3SELECT_KW("and");
+
+      or_op =  S3SELECT_KW("or");
+
+      variable_name =  bsc::lexeme_d[(+bsc::alpha_p >> *( bsc::alpha_p | bsc::digit_p | '_') ) - reserved_function_names];
+
+      variable = (variable_name >> "." >> variable_name) | variable_name;
+
+      json_variable_name = bsc::str_p("_1") >> +("." >> (json_array | json_object) );
+
+      json_object = (variable_name)[BOOST_BIND_ACTION(push_json_object)]; 
+
+      json_array = (variable_name >> +(bsc::str_p("[") >> number[BOOST_BIND_ACTION(push_array_number)] >> bsc::str_p("]")) )[BOOST_BIND_ACTION(push_json_array_name)];
+    }
+
+
+    bsc::rule<ScannerT> cast, data_type, variable, json_variable_name, variable_name, select_expr, select_expr_base, select_expr_base_, s3_object, where_clause, limit_number;
+    bsc::rule<ScannerT> number, float_number, string, backtick_string, from_expression, cast_as_data_type, cast_as_decimal_expr, decimal_operator;
+    bsc::rule<ScannerT> cmp_operand, arith_cmp, condition_expression, arithmetic_predicate, logical_predicate, factor; 
+    bsc::rule<ScannerT> trim, trim_whitespace_both, trim_one_side_whitespace, trim_anychar_anyside, trim_type, trim_remove_type, substr, substr_from, substr_from_for;
+    bsc::rule<ScannerT> datediff, dateadd, extract, date_part, date_part_extract, time_to_string_constant, time_to_string_dynamic;
+    bsc::rule<ScannerT> special_predicates, between_predicate, not_between, in_predicate, like_predicate, like_predicate_escape, like_predicate_no_escape, is_null, is_not_null;
+    bsc::rule<ScannerT> muldiv_operator, addsubop_operator, function, arithmetic_expression, addsub_operand, list_of_function_arguments, arithmetic_argument, mulldiv_operand, reserved_function_names;
+    bsc::rule<ScannerT> fs_type, object_path,json_s3_object,json_path_element,json_object,json_array;
+    bsc::rule<ScannerT> projections, projection_expression, alias_name, column_pos,column_pos_name;
+    bsc::rule<ScannerT> when_case_else_projection, when_case_value_when, when_stmt, when_value_then;
+    bsc::rule<ScannerT> logical_and,and_op,or_op;
+    bsc::rule<ScannerT> const& start() const
+    {
+      return  select_expr ;
+    }
+  };
+};
+
+void base_ast_builder::operator()(s3select *self, const char *a, const char *b) const
+{
+  //the purpose of the following procedure is to bypass boost::spirit rescan (calling to bind-action more than once per the same text)
+  //which cause wrong AST creation (and later false execution).
+  if (self->getAction()->is_already_scanned((void *)(this), const_cast<char *>(a)))
+    return;
+
+  builder(self, a, b);
+}
+
+void push_from_clause::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b),table_name,alias_name;
+
+  //should search for generic space
+  if(token.find(' ') != std::string::npos)
+  {
+    size_t pos = token.find(' '); 
+    table_name = token.substr(0,pos);
+    
+    pos = token.rfind(' ');
+    alias_name = token.substr(pos+1,token.size());
+
+    self->getAction()->table_alias = alias_name;
+
+    if(self->getAction()->column_prefix != "##" && self->getAction()->table_alias != self->getAction()->column_prefix)
+    {
+      throw base_s3select_exception(std::string("query can not contain more then a single table-alias"), base_s3select_exception::s3select_exp_en_t::FATAL);
+    }
+
+    token = table_name;
+  }
+
+  self->getAction()->from_clause = token;
+
+  self->getAction()->exprQ.clear();
+}
+
+void push_json_from_clause::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b),table_name,alias_name;
+
+  //TODO handle the star-operation ('*') in from-clause. build the parameters for json-reader search-api's.
+  std::vector<std::string> variable_key_path;
+  const char* delimiter = ".";
+  auto pos = token.find(delimiter);
+
+  if(pos != std::string::npos)
+  {
+    token = token.substr(strlen(JSON_ROOT_OBJECT)+1,token.size());
+    pos = token.find(delimiter);
+    do
+    {
+      variable_key_path.push_back(token.substr(0,pos));
+      if(pos != std::string::npos)
+	token = token.substr(pos+1,token.size());
+      else 
+	token = "";
+      pos = token.find(delimiter);
+    }while(token.size());
+  }
+  else
+  {
+    variable_key_path.push_back(JSON_ROOT_OBJECT);
+  }
+
+  self->getAction()->json_from_clause = variable_key_path;
+}
+
+void push_limit_clause::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  self->getAction()->limit_op = true;
+  try
+  {
+    self->getAction()->limit = std::stoul(token);
+  }
+  catch(std::invalid_argument& e)
+  {
+    throw base_s3select_exception(std::string("Invalid argument "), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+  catch(std::out_of_range& e)
+  {
+    throw base_s3select_exception(std::string("Out of range "), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+}
+
+void push_number::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  variable* v = S3SELECT_NEW(self, variable, atoi(token.c_str()));
+
+  self->getAction()->exprQ.push_back(v);
+}
+
+void push_float_number::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  //the parser for float(real_p) is accepting also integers, thus "blocking" integer acceptence and all are float.
+  bsc::parse_info<> info = bsc::parse(token.c_str(), bsc::int_p, bsc::space_p);
+
+  if (!info.full)
+  {
+    char* perr;
+    double d = strtod(token.c_str(), &perr);
+    variable* v = S3SELECT_NEW(self, variable, d);
+
+    self->getAction()->exprQ.push_back(v);
+  }
+  else
+  {
+    variable* v = S3SELECT_NEW(self, variable, atoi(token.c_str()));
+
+    self->getAction()->exprQ.push_back(v);
+  }
+}
+
+void push_string::builder(s3select* self, const char* a, const char* b) const
+{
+  a++;
+  b--; // remove double quotes
+  std::string token(a, b);
+
+  variable* v = S3SELECT_NEW(self, variable, token, variable::var_t::COLUMN_VALUE);
+
+  self->getAction()->exprQ.push_back(v);
+}
+
+void push_variable::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  variable* v = nullptr;
+
+  if (g_s3select_reserve_word.is_reserved_word(token))
+  {
+    if (g_s3select_reserve_word.get_reserved_word(token) == s3select_reserved_word::reserve_word_en_t::S3S_NULL)
+    {
+      v = S3SELECT_NEW(self, variable, s3select_reserved_word::reserve_word_en_t::S3S_NULL);
+    }
+    else if (g_s3select_reserve_word.get_reserved_word(token) == s3select_reserved_word::reserve_word_en_t::S3S_NAN)
+    {
+      v = S3SELECT_NEW(self, variable, s3select_reserved_word::reserve_word_en_t::S3S_NAN);
+    }
+    else if (g_s3select_reserve_word.get_reserved_word(token) == s3select_reserved_word::reserve_word_en_t::S3S_FALSE)
+    {
+      v = S3SELECT_NEW(self, variable, s3select_reserved_word::reserve_word_en_t::S3S_FALSE);
+    }
+    else if (g_s3select_reserve_word.get_reserved_word(token) == s3select_reserved_word::reserve_word_en_t::S3S_TRUE)
+    {
+      v = S3SELECT_NEW(self, variable, s3select_reserved_word::reserve_word_en_t::S3S_TRUE);
+    }
+    else
+    {
+      v = S3SELECT_NEW(self, variable, s3select_reserved_word::reserve_word_en_t::NA);
+    }
+    
+  }
+  else
+  {
+    size_t pos = token.find('.');
+    std::string alias_name;
+    if(pos != std::string::npos)
+    {
+      alias_name = token.substr(0,pos);
+      pos ++;
+      token = token.substr(pos,token.size());
+
+      if(self->getAction()->column_prefix != "##" && alias_name != self->getAction()->column_prefix)
+      {
+        throw base_s3select_exception(std::string("query can not contain more then a single table-alias"), base_s3select_exception::s3select_exp_en_t::FATAL);
+      }
+
+      self->getAction()->column_prefix = alias_name;
+    }
+    v = S3SELECT_NEW(self, variable, token);
+  }
+  
+  self->getAction()->exprQ.push_back(v);
+}
+
+void push_json_variable::builder(s3select* self, const char* a, const char* b) const
+{//purpose: handle the use case of json-variable structure (_1.a.b.c)
+
+  std::string token(a, b);
+  std::vector<std::string> variable_key_path;
+
+  //the following flow determine the index per json variable reside on statement.
+  //per each discovered json_variable, it search the json-variables-vector whether it already exists.
+  //in case it is exist, it uses its index (position in vector)
+  //in case it's not exist its pushes the variable into vector.
+  //the json-index is used upon updating the scratch area or searching for a specific json-variable value.
+
+  size_t json_index=self->getAction()->json_statement_variables_match_expression.size();
+  variable* v = nullptr;
+  json_variable_access* ja = S3SELECT_NEW(self, json_variable_access);
+  *ja = self->getAction()->json_var_md;
+  self->getAction()->json_statement_variables_match_expression.push_back(std::pair<json_variable_access*,size_t>(ja,json_index));
+  
+  v = S3SELECT_NEW(self, variable, token, variable::var_t::JSON_VARIABLE, json_index);
+  self->getAction()->exprQ.push_back(v);
+
+  self->getAction()->json_var_md.clear();
+}
+
+void push_array_number::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  //DEBUG - TEMP std::cout << "push_array_number " << token << std::endl;
+
+  self->getAction()->json_array_index_number.push_back(std::stoll(token.c_str()));
+}
+
+void push_json_array_name::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  size_t found = token.find("[");
+  std::string array_name = token.substr(0,found);
+
+  //DEBUG - TEMP std::cout << "push_json_array_name " << array_name << std::endl;
+
+  //remove white-space
+  array_name.erase(std::remove_if(array_name.begin(),
+  		array_name.end(),
+  		[](unsigned char x){return std::isspace(x);}),
+  		array_name.end());
+
+  std::vector<std::string> json_path;
+  std::vector<std::string> empty = {};
+  json_path.push_back(array_name);
+
+  self->getAction()->json_var_md.push_variable_state(json_path, -1);//pushing the array-name, {-1} means, search for object-name
+
+  while(self->getAction()->json_array_index_number.size())
+  {
+  	self->getAction()->json_var_md.push_variable_state(empty, self->getAction()->json_array_index_number.front());//pushing empty and number>=0, means array-access
+	self->getAction()->json_array_index_number.pop_front();
+  }	
+}
+
+void push_json_object::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  //DEBUG - TEMP std::cout << "push_json_object " << token << std::endl;
+
+  self->getAction()->json_object_name = token;
+  std::vector<std::string> json_path;
+  json_path.push_back(token);
+
+  self->getAction()->json_var_md.push_variable_state(json_path, -1);
+}
+
+void push_addsub::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  if (token == "+")
+  {
+    self->getAction()->addsubQ.push_back(addsub_operation::addsub_op_t::ADD);
+  }
+  else
+  {
+    self->getAction()->addsubQ.push_back(addsub_operation::addsub_op_t::SUB);
+  }
+}
+
+void push_mulop::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  if (token == "*")
+  {
+    self->getAction()->muldivQ.push_back(mulldiv_operation::muldiv_t::MULL);
+  }
+  else if (token == "/")
+  {
+    self->getAction()->muldivQ.push_back(mulldiv_operation::muldiv_t::DIV);
+  }
+  else if(token == "^")
+  {
+    self->getAction()->muldivQ.push_back(mulldiv_operation::muldiv_t::POW);
+  }
+  else
+  {
+    self->getAction()->muldivQ.push_back(mulldiv_operation::muldiv_t::MOD);
+  }
+}
+
+void push_addsub_binop::builder(s3select* self, [[maybe_unused]] const char* a,[[maybe_unused]] const char* b) const
+{
+  base_statement* l = nullptr, *r = nullptr;
+
+  r = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  l = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  addsub_operation::addsub_op_t o = self->getAction()->addsubQ.back();
+  self->getAction()->addsubQ.pop_back();
+  addsub_operation* as = S3SELECT_NEW(self, addsub_operation, l, o, r);
+  self->getAction()->exprQ.push_back(as);
+}
+
+void push_mulldiv_binop::builder(s3select* self, [[maybe_unused]] const char* a, [[maybe_unused]] const char* b) const
+{
+  base_statement* vl = nullptr, *vr = nullptr;
+
+  vr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  vl = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  mulldiv_operation::muldiv_t o = self->getAction()->muldivQ.back();
+  self->getAction()->muldivQ.pop_back();
+  mulldiv_operation* f = S3SELECT_NEW(self, mulldiv_operation, vl, o, vr);
+  self->getAction()->exprQ.push_back(f);
+}
+
+void push_function_arg::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  base_statement* be = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  base_statement* f = self->getAction()->funcQ.back();
+
+  if (dynamic_cast<__function*>(f))
+  {
+    dynamic_cast<__function*>(f)->push_argument(be);
+  }
+}
+
+void push_function_name::builder(s3select* self, const char* a, const char* b) const
+{
+  b--;
+  while (*b == '(' || *b == ' ')
+  {
+    b--; //point to function-name
+  }
+
+  std::string fn;
+  fn.assign(a, b - a + 1);
+
+  __function* func = S3SELECT_NEW(self, __function, fn.c_str(), self->getS3F());
+  self->getAction()->funcQ.push_back(func);
+}
+
+void push_function_expr::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  base_statement* func = self->getAction()->funcQ.back();
+  self->getAction()->funcQ.pop_back();
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_compare_operator::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  arithmetic_operand::cmp_t c = arithmetic_operand::cmp_t::NA;
+
+  if (token == "=")
+  {
+    c = arithmetic_operand::cmp_t::EQ;
+  }
+  else if (token == "!=" || token == "<>")
+  {
+    c = arithmetic_operand::cmp_t::NE;
+  }
+  else if (token == ">=")
+  {
+    c = arithmetic_operand::cmp_t::GE;
+  }
+  else if (token == "<=")
+  {
+    c = arithmetic_operand::cmp_t::LE;
+  }
+  else if (token == ">")
+  {
+    c = arithmetic_operand::cmp_t::GT;
+  }
+  else if (token == "<")
+  {
+    c = arithmetic_operand::cmp_t::LT;
+  }
+
+  self->getAction()->arithmetic_compareQ.push_back(c);
+}
+
+void push_logical_operator::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  logical_operand::oplog_t l = logical_operand::oplog_t::NA;
+
+  if (token == "and")
+  {
+    l = logical_operand::oplog_t::AND;
+  }
+  else if (token == "or")
+  {
+    l = logical_operand::oplog_t::OR;
+  }
+
+  self->getAction()->logical_compareQ.push_back(l);
+}
+
+void push_arithmetic_predicate::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  base_statement* vr, *vl;
+  arithmetic_operand::cmp_t c = self->getAction()->arithmetic_compareQ.back();
+  self->getAction()->arithmetic_compareQ.pop_back();
+
+  if (!self->getAction()->exprQ.empty())
+  {
+    vr = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }
+  else
+  {
+    throw base_s3select_exception(std::string("missing right operand for arithmetic-comparision expression"), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+  
+  if (!self->getAction()->exprQ.empty())
+  {
+    vl = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }
+  else
+  {
+    throw base_s3select_exception(std::string("missing left operand for arithmetic-comparision expression"), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+  
+  arithmetic_operand* t = S3SELECT_NEW(self, arithmetic_operand, vl, c, vr);
+
+  self->getAction()->exprQ.push_back(t);
+}
+
+void push_logical_predicate::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  base_statement* tl = nullptr, *tr = nullptr;
+  logical_operand::oplog_t oplog = self->getAction()->logical_compareQ.back();
+  self->getAction()->logical_compareQ.pop_back();
+
+  if (self->getAction()->exprQ.empty() == false)
+  {
+    tr = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }  
+  else 
+  {//should reject by syntax parser
+    throw base_s3select_exception(std::string("missing right operand for logical expression"), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+
+  if (self->getAction()->exprQ.empty() == false)
+  {
+    tl = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  } 
+  else 
+  {//should reject by syntax parser
+    throw base_s3select_exception(std::string("missing left operand for logical expression"), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+
+  logical_operand* f = S3SELECT_NEW(self, logical_operand, tl, oplog, tr);
+
+  self->getAction()->exprQ.push_back(f);
+}
+
+void push_negation::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  base_statement* pred = nullptr;
+
+  if (self->getAction()->exprQ.empty() == false)
+  {
+    pred = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }
+  else
+  {
+    throw base_s3select_exception(std::string("failed to create AST for NOT operator"), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+  
+  //upon NOT operator, the logical and arithmetical operators are "tagged" to negate result.
+  if (dynamic_cast<logical_operand*>(pred))
+  {
+    logical_operand* f = S3SELECT_NEW(self, logical_operand, pred);
+    self->getAction()->exprQ.push_back(f);
+  }
+  else if (dynamic_cast<__function*>(pred) || dynamic_cast<negate_function_operation*>(pred) || dynamic_cast<variable*>(pred))
+  {
+    negate_function_operation* nf = S3SELECT_NEW(self, negate_function_operation, pred);
+    self->getAction()->exprQ.push_back(nf);
+  }
+  else if(dynamic_cast<arithmetic_operand*>(pred))
+  {
+    arithmetic_operand* f = S3SELECT_NEW(self, arithmetic_operand, pred);
+    self->getAction()->exprQ.push_back(f);
+  }
+  else
+  {
+    throw base_s3select_exception(std::string("failed to create AST for NOT operator"), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+}
+
+void push_column_pos::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  std::string alias_name;
+  variable* v;
+
+  if (token == "*" || token == "* ") //TODO space should skip in boost::spirit
+  {
+    v = S3SELECT_NEW(self, variable, token, variable::var_t::STAR_OPERATION);
+
+  }
+  else
+  {
+    size_t pos = token.find('.');
+    if(pos != std::string::npos)
+    {
+      alias_name = token.substr(0,pos);
+
+      pos ++;
+      token = token.substr(pos,token.size());
+
+      if(self->getAction()->column_prefix != "##" && self->getAction()->column_prefix != alias_name)
+      {
+        throw base_s3select_exception(std::string("query can not contain more then a single table-alias"), base_s3select_exception::s3select_exp_en_t::FATAL);
+      }
+    
+      self->getAction()->column_prefix = alias_name;
+    }
+    v = S3SELECT_NEW(self, variable, token, variable::var_t::POS);
+  }
+
+  self->getAction()->exprQ.push_back(v);
+}
+
+void push_projection::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  self->getAction()->projections.get()->push_back(self->getAction()->exprQ.back());
+  self->getAction()->exprQ.pop_back();
+}
+
+void push_alias_projection::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  //extract alias name
+  const char* p = b;
+  while (*(--p) != ' ')
+    ;
+  std::string alias_name(p + 1, b);
+  base_statement* bs = self->getAction()->exprQ.back();
+
+  //mapping alias name to base-statement
+  bool res = self->getAction()->alias_map.insert_new_entry(alias_name, bs);
+  if (res == false)
+  {
+    throw base_s3select_exception(std::string("alias <") + alias_name + std::string("> is already been used in query"), base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+
+  self->getAction()->projections.get()->push_back(bs);
+  self->getAction()->exprQ.pop_back();
+}
+
+void push_between_filter::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  std::string between_function("#between#");
+
+  __function* func = S3SELECT_NEW(self, __function, between_function.c_str(), self->getS3F());
+
+  base_statement* second_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(second_expr);
+
+  base_statement* first_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(first_expr);
+
+  base_statement* main_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(main_expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_not_between_filter::builder(s3select* self, const char* a, const char* b) const
+{
+
+  static constexpr const std::string_view not_between_function("#not_between#");
+
+  __function* func = S3SELECT_NEW(self, __function, not_between_function.data(), self->getS3F());
+
+  base_statement* second_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(second_expr);
+
+  base_statement* first_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(first_expr);
+
+  base_statement* main_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(main_expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_in_predicate_first_arg::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  if(self->getAction()->exprQ.empty())
+  {
+    throw base_s3select_exception("failed to create AST for in predicate", base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+
+  self->getAction()->inPredicateQ.push_back( self->getAction()->exprQ.back() );
+  self->getAction()->exprQ.pop_back();
+
+  if(self->getAction()->exprQ.empty())
+  {
+    throw base_s3select_exception("failed to create AST for in predicate", base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+
+  self->getAction()->inMainArg = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+
+}
+
+void push_in_predicate_arguments::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  if(self->getAction()->exprQ.empty())
+  {
+    throw base_s3select_exception("failed to create AST for in predicate", base_s3select_exception::s3select_exp_en_t::FATAL);
+  }
+
+  self->getAction()->inPredicateQ.push_back( self->getAction()->exprQ.back() );
+
+  self->getAction()->exprQ.pop_back();
+
+}
+
+void push_in_predicate::builder(s3select* self, const char* a, const char* b) const
+{
+  // expr in (e1,e2,e3 ...)
+  std::string token(a, b);
+
+  std::string in_function("#in_predicate#");
+
+  __function* func = S3SELECT_NEW(self, __function, in_function.c_str(), self->getS3F());
+
+  while(!self->getAction()->inPredicateQ.empty())
+  {
+    base_statement* ei = self->getAction()->inPredicateQ.back();
+
+    self->getAction()->inPredicateQ.pop_back();
+
+    func->push_argument(ei);
+
+  }
+
+  func->push_argument( self->getAction()->inMainArg );
+
+  self->getAction()->exprQ.push_back(func);
+
+  self->getAction()->inPredicateQ.clear();
+
+  self->getAction()->inMainArg = 0;
+}
+
+void push_like_predicate_no_escape::builder(s3select* self, const char* a, const char* b) const
+{
+
+  std::string token(a, b);
+  std::string in_function("#like_predicate#");
+
+  __function* func = S3SELECT_NEW(self, __function, in_function.c_str(), self->getS3F());
+  
+  variable* v = S3SELECT_NEW(self, variable, "\\",variable::var_t::COLUMN_VALUE);
+  func->push_argument(v);
+  
+  // experimenting valgrind-issue happens only on teuthology
+  //self->getS3F()->push_for_cleanup(v);
+  
+  base_statement* like_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(like_expr);  
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_like_predicate_escape::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+  std::string in_function("#like_predicate#");
+
+  __function* func = S3SELECT_NEW(self, __function, in_function.c_str(), self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(expr);
+
+  base_statement* main_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(main_expr);
+
+  base_statement* escape_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(escape_expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_is_null_predicate::builder(s3select* self, const char* a, const char* b) const
+{
+    //expression is null, is not null 
+  std::string token(a, b);
+  bool is_null = true;
+
+  for(size_t i=0;i<token.size();i++)
+  {//TODO use other scan rules
+    bsc::parse_info<> info = bsc::parse(token.c_str()+i, (bsc::str_p("is") >> bsc::str_p("not") >> bsc::str_p("null")) , bsc::space_p);
+    if (info.full)
+      is_null = false;
+  }
+
+  std::string in_function("#is_null#");
+
+  if (is_null == false)
+  {
+    in_function = "#is_not_null#";
+  }
+
+  __function* func = S3SELECT_NEW(self, __function, in_function.c_str(), self->getS3F());
+
+  if (!self->getAction()->exprQ.empty())
+  {
+    base_statement* expr = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+    func->push_argument(expr);
+  }
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_when_condition_then::builder(s3select* self, const char* a, const char* b) const
+{
+//purpose: each new function node, provide execution for (if {condition} then {expresion} )
+  std::string token(a, b);
+
+  // _fn_when_then
+  __function* func = S3SELECT_NEW(self, __function, "#when-then#", self->getS3F());
+
+ base_statement* then_expr = self->getAction()->exprQ.back();
+ self->getAction()->exprQ.pop_back();
+
+ base_statement* when_expr = self->getAction()->exprQ.back();
+ self->getAction()->exprQ.pop_back();
+
+ func->push_argument(then_expr);
+ func->push_argument(when_expr);
+
+ self->getAction()->exprQ.push_back(func);
+
+  // the first_when_then_expr mark the first when-then expression, it is been used later upon complete the full statement (case when ... then ... else ... end)
+ if(self->getAction()->first_when_then_expr == nullptr)
+ {	
+  self->getAction()->first_when_then_expr = func;
+ }
+}
+
+void push_case_when_else::builder(s3select* self, const char* a, const char* b) const
+{
+//purpose: provide the execution for complete statement, i.e. (case when {expression} then {expression} else {expression} end)
+  std::string token(a, b);
+
+  base_statement* else_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  // _fn_case_when_else
+  __function* func = S3SELECT_NEW(self, __function, "#case-when-else#", self->getS3F());
+
+  func->push_argument(else_expr);
+
+  base_statement* when_then_func = nullptr;
+
+  // the loop ended upon reaching the first when-then
+  while(when_then_func != self->getAction()->first_when_then_expr)
+  {
+    // poping from whenThen-queue and pushing to function arguments list
+    when_then_func = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+    func->push_argument(when_then_func);
+  }
+  
+  self->getAction()->first_when_then_expr = nullptr;
+  //func is the complete statement,  implemented by _fn_case_when_else
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_case_value_when_value_else::builder(s3select* self, const char* a, const char* b) const
+{
+//purpose: provide execution for the complete statement. i.e. case-value-when-value-else-value-end
+  std::string token(a, b);
+
+  base_statement* else_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  // _fn_case_when_else
+  __function* func = S3SELECT_NEW(self, __function, "#case-when-else#", self->getS3F());
+
+  // push the else expression 
+  func->push_argument(else_expr);
+
+  // poping the case-value  
+  base_statement* case_value = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* when_then_func = nullptr;
+  
+  //poping all when-value-then expression(_fn_when_value_then) and add the case-value per each
+  while(self->getAction()->whenThenQ.empty() == false)
+  {
+    when_then_func = self->getAction()->whenThenQ.back();
+    if (dynamic_cast<__function*>(when_then_func))
+    {
+      // adding the case-value as argument
+      dynamic_cast<__function*>(when_then_func)->push_argument(case_value);
+    }
+    else 
+      throw base_s3select_exception("failed to create AST for case-value-when construct", base_s3select_exception::s3select_exp_en_t::FATAL);
+
+    self->getAction()->whenThenQ.pop_back(); 
+ 
+    func->push_argument(when_then_func);
+  }
+  //pushing the execution function for the complete statement
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_when_value_then::builder(s3select* self, const char* a, const char* b) const
+{
+  //provide execution of when-value-then-value :: _fn_when_value_then
+  std::string token(a, b);
+
+  __function* func = S3SELECT_NEW(self, __function, "#when-value-then#", self->getS3F());
+
+ base_statement* then_expr = self->getAction()->exprQ.back();
+ self->getAction()->exprQ.pop_back();
+
+ base_statement* when_expr = self->getAction()->exprQ.back();
+ self->getAction()->exprQ.pop_back();
+
+ func->push_argument(then_expr);
+ func->push_argument(when_expr);
+  //each when-value-then-value pushed to dedicated queue
+ self->getAction()->whenThenQ.push_back(func);
+}
+
+void push_decimal_operator::builder(s3select* self, const char* a, const char* b) const
+{//decimal(integer,integer)
+  std::string token(a, b);
+
+  base_statement* lhs = nullptr;
+  base_statement* rhs = nullptr;
+
+  //right side (decimal operator)
+  if (self->getAction()->exprQ.empty() == false)
+  {
+    rhs = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }
+
+  //left side (decimal operator)
+  if (self->getAction()->exprQ.empty() == false)
+  {
+    lhs = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }
+
+  __function* func = S3SELECT_NEW(self, __function, "#decimal_operator#", self->getS3F());
+
+  func->push_argument(rhs);
+  func->push_argument(lhs);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_cast_decimal_expr::builder(s3select* self, const char* a, const char* b) const
+{
+  //cast(expression as decimal(x,y))
+  std::string token(a, b);
+
+  base_statement* lhs = nullptr;
+  base_statement* rhs = nullptr;
+
+  //right side (decimal operator)
+  if (self->getAction()->exprQ.empty() == false)
+  {
+    rhs = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }
+
+  //left side - expression
+  if (self->getAction()->exprQ.empty() == false)
+  {
+    lhs = self->getAction()->exprQ.back();
+    self->getAction()->exprQ.pop_back();
+  }
+
+  __function* func = S3SELECT_NEW(self, __function, "#cast_as_decimal#", self->getS3F());
+
+  func->push_argument(rhs);
+  func->push_argument(lhs);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_cast_expr::builder(s3select* self, const char* a, const char* b) const
+{
+  //cast(expression as int/float/string/timestamp) --> new function "int/float/string/timestamp" ( args = expression )
+  std::string token(a, b);
+  
+  std::string cast_function;
+
+  cast_function = self->getAction()->dataTypeQ.back();
+  self->getAction()->dataTypeQ.pop_back();
+
+  __function* func = S3SELECT_NEW(self, __function, cast_function.c_str(), self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_data_type::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  auto cast_operator = [&](const char *s){return strncasecmp(a,s,strlen(s))==0;};
+
+  if(cast_operator("int"))
+  {
+    self->getAction()->dataTypeQ.push_back("int");
+  }else if(cast_operator("float"))
+  {
+    self->getAction()->dataTypeQ.push_back("float");
+  }else if(cast_operator("string"))
+  {
+    self->getAction()->dataTypeQ.push_back("string");
+  }else if(cast_operator("timestamp"))
+  {
+    self->getAction()->dataTypeQ.push_back("to_timestamp");
+  }else if(cast_operator("bool"))
+  {
+    self->getAction()->dataTypeQ.push_back("to_bool");
+  }
+}
+
+void push_trim_whitespace_both::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  __function* func = S3SELECT_NEW(self, __function, "#trim#", self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}  
+
+void push_trim_expr_one_side_whitespace::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  std::string trim_function;
+
+  trim_function = self->getAction()->trimTypeQ.back();
+  self->getAction()->trimTypeQ.pop_back(); 
+
+  __function* func = S3SELECT_NEW(self, __function, trim_function.c_str(), self->getS3F());
+
+  base_statement* inp_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(inp_expr);
+
+  self->getAction()->exprQ.push_back(func);
+} 
+
+void push_trim_expr_anychar_anyside::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  std::string trim_function;
+
+  trim_function = self->getAction()->trimTypeQ.back();
+  self->getAction()->trimTypeQ.pop_back(); 
+
+  __function* func = S3SELECT_NEW(self, __function, trim_function.c_str(), self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(expr);
+
+  base_statement* inp_expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(inp_expr);
+
+  self->getAction()->exprQ.push_back(func);
+} 
+
+void push_trim_type::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  auto trim_option = [&](const char *s){return strncmp(a,s,strlen(s))==0;};
+
+  if(trim_option("leading"))
+  {
+    self->getAction()->trimTypeQ.push_back("#leading#");
+  }else if(trim_option("trailing"))
+  {
+    self->getAction()->trimTypeQ.push_back("#trailing#");
+  }else 
+  {
+    self->getAction()->trimTypeQ.push_back("#trim#");
+  }
+} 
+
+void push_substr_from::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  __function* func = S3SELECT_NEW(self, __function, "substring", self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* start_position = self->getAction()->exprQ.back();
+
+  self->getAction()->exprQ.pop_back();
+  func->push_argument(start_position);
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}  
+
+void push_substr_from_for::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  __function* func = S3SELECT_NEW(self, __function, "substring", self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* start_position = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* end_position = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(end_position);
+  func->push_argument(start_position);
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_datediff::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  std::string date_op;
+
+  date_op = self->getAction()->datePartQ.back();
+  self->getAction()->datePartQ.pop_back();
+
+  std::string date_function =  "#datediff_" + date_op + "#";
+
+  __function* func = S3SELECT_NEW(self, __function, date_function.c_str(), self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* start_position = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(start_position);
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_dateadd::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  std::string date_op;
+
+  date_op = self->getAction()->datePartQ.back();
+  self->getAction()->datePartQ.pop_back();
+
+  std::string date_function =  "#dateadd_" + date_op + "#";
+
+  __function* func = S3SELECT_NEW(self, __function, date_function.c_str(), self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* start_position = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(start_position);
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_extract::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  std::string date_op;
+
+  date_op = self->getAction()->datePartQ.back();
+  self->getAction()->datePartQ.pop_back();
+
+  std::string date_function =  "#extract_" + date_op + "#";
+
+  __function* func = S3SELECT_NEW(self, __function, date_function.c_str(), self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_date_part::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  self->getAction()->datePartQ.push_back(token);
+}
+
+void push_time_to_string_constant::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  __function* func = S3SELECT_NEW(self, __function, "#to_string_constant#", self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* frmt = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(frmt);
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+
+}
+
+void push_time_to_string_dynamic::builder(s3select* self, const char* a, const char* b) const
+{
+  std::string token(a, b);
+
+  __function* func = S3SELECT_NEW(self, __function, "#to_string_dynamic#", self->getS3F());
+
+  base_statement* expr = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  base_statement* frmt = self->getAction()->exprQ.back();
+  self->getAction()->exprQ.pop_back();
+
+  func->push_argument(frmt);
+  func->push_argument(expr);
+
+  self->getAction()->exprQ.push_back(func);
+}
+
+void push_string_to_time_constant::builder(s3select* self, const char* a, const char* b) const
+{
+  //token could be a string or a timestamp, we need to check it
+  //upon it is a timestamp format, we need to push the variable as timestamp or else, it as a string
+  //the purpose is to use backticks to convert the string to timestamp in parsing time instead of processing time(Trino uses this approach)
+  
+  a++; //remove the first quote
+  b--;
+  std::string token(a, b);
+
+  _fn_to_timestamp* to_timestamp = S3SELECT_NEW(self, _fn_to_timestamp);//TODO the _fn_to_timestamp should release the memory (cleanup)
+  bs_stmt_vec_t args;
+
+  variable* var_string = S3SELECT_NEW(self, variable, token, variable::var_t::COLUMN_VALUE);
+  variable* timestamp = S3SELECT_NEW(self, variable, token, variable::var_t::COLUMN_VALUE);
+
+  (self->get_to_timestamp_for_clean()) = to_timestamp;
+  var_string->push_for_cleanup(self->get_ast_nodes_to_delete());
+  timestamp->push_for_cleanup(self->get_ast_nodes_to_delete());
+  
+  args.push_back(var_string);
+
+  try {
+    (*to_timestamp)(&args, timestamp);
+  }
+  catch(std::exception& e)
+  {
+    //it is not a timestamp, it is a string
+    self->getAction()->exprQ.push_back(var_string);
+    return;
+  }
+
+  self->getAction()->exprQ.push_back(timestamp);
+}
+
+struct s3select_csv_definitions //TODO 
+{
+    char row_delimiter;
+    char column_delimiter;
+    char output_row_delimiter;
+    char output_column_delimiter;
+    char escape_char;
+    char output_escape_char;
+    char output_quot_char;
+    char quot_char;
+    bool use_header_info;
+    bool ignore_header_info;//skip first line
+    bool quote_fields_always;
+    bool quote_fields_asneeded;
+    bool redundant_column;
+    bool comment_empty_lines;
+    std::vector<char> comment_chars;
+    std::vector<char> trim_chars;
+
+    s3select_csv_definitions():row_delimiter('\n'), column_delimiter(','), output_row_delimiter('\n'), output_column_delimiter(','), escape_char('\\'), output_escape_char('\\'), output_quot_char('"'), quot_char('"'), use_header_info(false), ignore_header_info(false), quote_fields_always(false), quote_fields_asneeded(false), redundant_column(false), comment_empty_lines(false) {}
+
+};
+ 
+
+/////// handling different object types
+class base_s3object
+{
+
+protected:
+  scratch_area* m_sa;
+  std::string m_obj_name;
+  bool m_aggr_flow = false; //TODO once per query
+  bool m_is_to_aggregate;
+  std::vector<base_statement*> m_projections;
+  base_statement* m_where_clause;
+  s3select* m_s3_select;
+  size_t m_error_count;
+  bool m_is_limit_on;
+  unsigned long m_limit;
+  unsigned long m_processed_rows;
+  size_t m_returned_bytes_size;
+  std::function<void(const char*)> fp_ext_debug_mesg;//dispache debug message into external system
+
+public:
+  s3select_csv_definitions m_csv_defintion;//TODO add method for modify
+
+  enum class Status {
+    END_OF_STREAM,
+    INITIAL_STAT,
+    NORMAL_EXIT,
+    LIMIT_REACHED,
+    SQL_ERROR
+  };
+
+  Status m_sql_processing_status;
+
+  Status get_sql_processing_status()
+  {
+	  return m_sql_processing_status;
+  }
+
+  bool is_sql_limit_reached()
+  {
+	  return m_sql_processing_status == Status::LIMIT_REACHED;
+  }
+
+  void set_base_defintions(s3select* m)
+  {
+    if(m_s3_select || !m)
+    {//not to define twice
+     //not to define with null
+  	return;
+    }
+
+    m_s3_select=m;
+    m_sa=m_s3_select->get_scratch_area();
+    m_error_count=0;
+    m_projections = m_s3_select->get_projections_list();
+    m_where_clause = m_s3_select->get_filter();
+
+    if (m_where_clause)
+    {
+      m_where_clause->traverse_and_apply(m_sa, m_s3_select->get_aliases(), m_s3_select->is_json_query());
+    }
+
+    for (auto& p : m_projections)
+    {
+      p->traverse_and_apply(m_sa, m_s3_select->get_aliases(), m_s3_select->is_json_query());
+    }
+    m_is_to_aggregate = true;//TODO not correct. should be set upon end-of-stream
+    m_aggr_flow = m_s3_select->is_aggregate_query();
+
+    m_is_limit_on = m_s3_select->is_limit();
+    if(m_is_limit_on)
+    {
+        m_limit = m_s3_select->get_limit();
+    }
+
+    m_processed_rows = 0;
+  }
+
+  base_s3object():m_sa(nullptr),m_is_to_aggregate(false),m_where_clause(nullptr),m_s3_select(nullptr),m_error_count(0),m_returned_bytes_size(0),m_sql_processing_status(Status::INITIAL_STAT){}
+
+  explicit base_s3object(s3select* m):base_s3object()
+  {
+    if(m)
+    {
+        set_base_defintions(m);
+    }
+  }
+
+  virtual bool is_end_of_stream() {return false;}
+  virtual void row_fetch_data() {}
+  virtual void row_update_data() {}
+  virtual void columnar_fetch_where_clause_columns(){}
+  virtual void columnar_fetch_projection(){}
+  // for the case were the rows are not fetched, but "pushed" by the data-source parser (JSON)
+  virtual bool multiple_row_processing(){return true;}
+
+  void set_external_debug_system(std::function<void(const char*)> fp_external)
+  {
+	fp_ext_debug_mesg = fp_external; 
+  }
+
+  size_t get_return_result_size()
+  {
+	return m_returned_bytes_size;
+  }
+
+  void result_values_to_string(multi_values& projections_resuls, std::string& result)
+  {
+    size_t i = 0;
+    std::string output_delimiter(1,m_csv_defintion.output_column_delimiter);
+    std::string output_row_delimiter(1,m_csv_defintion.output_row_delimiter);
+
+    for(auto& res : projections_resuls.values)
+    {
+	    if(fp_ext_debug_mesg)
+		      fp_ext_debug_mesg( res->to_string() );
+
+            if (m_csv_defintion.quote_fields_always) {
+              std::ostringstream quoted_result;
+              quoted_result << std::quoted(res->to_string(),m_csv_defintion.output_quot_char, m_csv_defintion.escape_char);
+              result.append(quoted_result.str());
+	      m_returned_bytes_size += quoted_result.str().size();
+            }//TODO to add asneeded
+	    else
+	    {
+            	result.append(res->to_string());
+		m_returned_bytes_size += strlen(res->to_string());
+	    }
+
+            if(!m_csv_defintion.redundant_column) {
+              if(++i < projections_resuls.values.size()) {
+                result.append(output_delimiter);
+		m_returned_bytes_size += output_delimiter.size();
+              }
+            }
+            else {
+              result.append(output_delimiter);
+	      m_returned_bytes_size += output_delimiter.size();
+            }    
+    }
+    if(!m_aggr_flow){
+	      result.append(output_row_delimiter);
+	      m_returned_bytes_size += output_delimiter.size();
+    }
+  }
+
+  Status getMatchRow( std::string& result)
+  {
+    multi_values projections_resuls;
+
+    if (m_is_limit_on && m_processed_rows == m_limit)
+    {
+      return m_sql_processing_status = Status::LIMIT_REACHED;
+    }
+    
+    if (m_aggr_flow == true)
+    {
+      do
+      {
+        row_fetch_data();
+        columnar_fetch_where_clause_columns();
+        if (is_end_of_stream())
+        {
+          if (m_is_to_aggregate)
+            for (auto& i : m_projections)
+            {
+              i->set_last_call();
+              i->set_skip_non_aggregate(false);//projection column is set to be runnable
+
+              projections_resuls.push_value( &(i->eval()) );
+            }
+
+          result_values_to_string(projections_resuls,result);
+          return m_sql_processing_status = Status::END_OF_STREAM;
+        }
+
+        m_processed_rows++;
+        if ((*m_projections.begin())->is_set_last_call())
+        {
+          //should validate while query execution , no update upon nodes are marked with set_last_call
+          throw base_s3select_exception("on aggregation query , can not stream row data post do-aggregate call", base_s3select_exception::s3select_exp_en_t::FATAL);
+        }
+
+        for (auto& a : *m_s3_select->get_aliases()->get())
+        {
+          a.second->invalidate_cache_result();
+        }
+
+	row_update_data();
+        if (!m_where_clause || m_where_clause->eval().is_true())
+	{
+	  columnar_fetch_projection();
+          for (auto i : m_projections)
+          {	
+	    i->eval();
+          }
+	}
+
+        if(m_is_limit_on && m_processed_rows == m_limit)
+        {
+	  for (auto& i : m_projections)
+	  {
+	    i->set_last_call();
+	    i->set_skip_non_aggregate(false);//projection column is set to be runnable
+	    projections_resuls.push_value( &(i->eval()) );
+	  }
+	  result_values_to_string(projections_resuls,result);
+	  return m_sql_processing_status = Status::LIMIT_REACHED;
+        }
+      }
+      while (multiple_row_processing());
+    }
+    else
+    {
+      //save the where-clause evaluation result (performance perspective)
+      bool where_clause_result = false;
+      do
+      {
+	row_fetch_data();
+	columnar_fetch_where_clause_columns();
+        if(is_end_of_stream())
+        {
+          return m_sql_processing_status = Status::END_OF_STREAM;
+        }
+
+        m_processed_rows++;
+        row_update_data();
+        for (auto& a : *m_s3_select->get_aliases()->get())
+        {
+          a.second->invalidate_cache_result();
+        }
+      }
+      while (multiple_row_processing() && m_where_clause && !(where_clause_result = m_where_clause->eval().is_true()) && !(m_is_limit_on && m_processed_rows == m_limit));
+
+ 	// in the of JSON it needs to evaluate the where-clause(for the first time)
+      if(!multiple_row_processing() && m_where_clause){
+	where_clause_result = m_where_clause->eval().is_true();
+      }
+
+      if(m_where_clause && ! where_clause_result && m_is_limit_on && m_processed_rows == m_limit)
+      {
+          return m_sql_processing_status = Status::LIMIT_REACHED;
+      }
+
+      bool found = multiple_row_processing();
+
+      if(!multiple_row_processing())
+      {
+		found = !m_where_clause || where_clause_result;	
+      }
+  
+      if(found)
+      {
+	columnar_fetch_projection();
+	projections_resuls.clear();
+	for (auto& i : m_projections)
+	{
+	  projections_resuls.push_value( &(i->eval()) );
+	}
+	result_values_to_string(projections_resuls,result);
+      }
+
+    }
+    return is_end_of_stream() ? (m_sql_processing_status = Status::END_OF_STREAM) : (m_sql_processing_status = Status::NORMAL_EXIT);
+    
+  }//getMatchRow
+
+  virtual ~base_s3object() = default;
+
+}; //base_s3object
+
+//TODO config / default-value
+#define CSV_INPUT_TYPE_RESPONSE_SIZE_LIMIT (64 * 1024)
+class csv_object : public base_s3object
+{
+
+public:
+
+  class csv_defintions : public s3select_csv_definitions
+  {};
+
+  explicit csv_object(s3select* s3_query) :
+    base_s3object(s3_query),
+    m_skip_last_line(false),
+    m_extract_csv_header_info(false),
+    m_previous_line(false),
+    m_skip_first_line(false),
+    m_processed_bytes(0) {}
+
+  csv_object(s3select* s3_query, csv_defintions csv) :
+    base_s3object(s3_query),
+    m_skip_last_line(false),
+    m_extract_csv_header_info(false),
+    m_previous_line(false),
+    m_skip_first_line(false),
+    m_processed_bytes(0)
+  {
+    m_csv_defintion = csv;
+  }
+
+  csv_object():
+    base_s3object(nullptr),
+    m_skip_last_line(false),
+    m_extract_csv_header_info(false),
+    m_previous_line(false),
+    m_skip_first_line(false),
+    m_processed_bytes(0) {}
+
+  void set_csv_query(s3select* s3_query,csv_defintions csv)
+  {
+    if(m_s3_select != nullptr) 
+    {
+      //return;
+    }
+
+    set_base_defintions(s3_query);
+    m_csv_defintion = csv;
+  }
+
+private:
+  bool m_skip_last_line;
+  std::string m_error_description;
+  char* m_stream;
+  char* m_end_stream;
+  std::vector<char*> m_row_tokens;
+  CSVParser* csv_parser;
+  bool m_extract_csv_header_info;
+  std::vector<std::string> m_csv_schema{128};
+
+  //handling arbitrary chunks (rows cut in the middle)
+  bool m_previous_line;
+  bool m_skip_first_line;
+  std::string merge_line;
+  std::string m_last_line;
+  size_t m_processed_bytes;
+  int64_t m_number_of_tokens;
+  size_t m_skip_x_first_bytes=0;
+
+  std::function<int(std::string&)> fp_s3select_result_format=nullptr;
+  std::function<int(std::string&)> fp_s3select_header_format=nullptr;
+public:
+  void set_result_formatters(	std::function<int(std::string&)>& result_format, 
+				std::function<int(std::string&)>& header_format)
+  {
+	fp_s3select_result_format = result_format;
+	fp_s3select_header_format = header_format;
+  }
+private:
+  int getNextRow()
+  {
+    size_t num_of_tokens=0;
+    m_row_tokens.clear();
+
+    if (csv_parser->read_row(m_row_tokens))
+    {
+      num_of_tokens = m_row_tokens.size();
+    }
+    else
+    {
+      return -1;
+    }
+
+    return num_of_tokens;
+  }
+
+public:
+
+  std::string get_error_description()
+  {
+    return m_error_description;
+  }
+
+  virtual ~csv_object() = default;
+
+public:
+  virtual bool is_end_of_stream()
+  {
+      return m_number_of_tokens < 0;
+  }
+
+  virtual void row_fetch_data()
+  {
+        m_number_of_tokens = getNextRow();
+  }
+  
+  virtual void row_update_data()
+  {
+        m_sa->update(m_row_tokens, m_number_of_tokens);
+  }
+
+
+  int extract_csv_header_info()
+  {
+
+    if (m_csv_defintion.ignore_header_info == true)
+    {
+      csv_parser->next_line();
+    }
+    else if(m_csv_defintion.use_header_info == true)
+    {
+      size_t num_of_tokens = getNextRow();//TODO validate number of tokens
+
+      for(size_t i=0; i<num_of_tokens; i++)
+      {
+        m_csv_schema[i].assign(m_row_tokens[i]);
+      }
+
+      m_s3_select->load_schema(m_csv_schema);
+    }
+
+    m_extract_csv_header_info = true;
+
+    return 0;
+  }
+
+
+  int run_s3select_on_stream(std::string& result, const char* csv_stream, size_t stream_length, size_t obj_size)
+  {
+    int status=0;
+    try{
+        status = run_s3select_on_stream_internal(result,csv_stream,stream_length,obj_size);
+    }
+    catch(base_s3select_exception& e)
+    {
+        m_error_description = e.what();
+        m_error_count ++;
+        if (e.severity() == base_s3select_exception::s3select_exp_en_t::FATAL || m_error_count>100)//abort query execution
+        {
+          return -1;
+        }
+    }
+    catch(chunkalloc_out_of_mem)
+    {
+      m_error_description = "out of memory";
+      return -1;
+    }
+    catch(io::error::escaped_char_missing& err)
+    {
+      m_error_description = "escaped_char_missing failure while csv parsing";
+      return -1;
+    }
+	catch(io::error::escaped_string_not_closed& err)
+    {
+      m_error_description = "escaped_string_not_closed failure while csv parsing";
+      return -1;
+    }
+	catch(io::error::line_length_limit_exceeded& err)
+    {
+      m_error_description = "line_length_limit_exceeded failure while csv parsing";
+      return -1;
+    }
+	catch(io::error::with_file_name& err)
+    {
+      m_error_description = "with_file_name failure while csv parsing";
+      return -1;
+    }
+	catch(io::error::with_file_line& err)
+    {
+      m_error_description = "with_file_line failure while csv parsing";
+      return -1;
+    }
+
+    return status;
+  }
+
+private:
+  int run_s3select_on_stream_internal(std::string& result, const char* csv_stream, size_t stream_length, size_t obj_size)
+  {
+    //purpose: the CSV data is "streaming", it may "cut" rows in the middle, in that case the "broken-line" is stores
+    //for later, upon next chunk of data is streaming, the stored-line is merge with current broken-line, and processed.
+    std::string tmp_buff;
+	
+    m_processed_bytes += stream_length;
+
+    m_skip_first_line = false;
+
+    if (m_previous_line)
+    {
+      //if previous broken line exist , merge it to current chunk
+      char* p_obj_chunk = (char*)csv_stream;
+      while (*p_obj_chunk != m_csv_defintion.row_delimiter && p_obj_chunk<(csv_stream+stream_length))
+      {
+        p_obj_chunk++;
+      }
+
+      tmp_buff.assign((char*)csv_stream, (char*)csv_stream + (p_obj_chunk - csv_stream));
+      merge_line = m_last_line + tmp_buff + m_csv_defintion.row_delimiter;
+      m_previous_line = false;
+      m_skip_first_line = true;
+      m_skip_x_first_bytes = tmp_buff.size()+1;
+
+      //processing the merged row (previous broken row)
+      run_s3select_on_object(result, merge_line.c_str(), merge_line.length(), false, false, false);
+    }
+
+    if (stream_length && csv_stream[stream_length - 1] != m_csv_defintion.row_delimiter)
+    {
+      //in case of "broken" last line
+      char* p_obj_chunk = (char*)&(csv_stream[stream_length - 1]);
+      while (*p_obj_chunk != m_csv_defintion.row_delimiter && p_obj_chunk>csv_stream)
+      {
+        p_obj_chunk--;  //scan until end-of previous line in chunk
+      }
+
+      u_int32_t skip_last_bytes = (&(csv_stream[stream_length - 1]) - p_obj_chunk);
+      m_last_line.assign(p_obj_chunk + 1, p_obj_chunk + 1 + skip_last_bytes); //save it for next chunk
+
+      m_previous_line = true;//it means to skip last line
+
+      //cut out the broken line
+      stream_length -= (m_last_line.length());
+    }
+
+    return run_s3select_on_object(result, csv_stream, stream_length, m_skip_first_line, m_previous_line, (m_processed_bytes >= obj_size));
+  }
+
+public:
+  int run_s3select_on_object(std::string& result, const char* csv_stream, size_t stream_length, bool skip_first_line, bool skip_last_line, bool do_aggregate)
+  {
+    m_stream = (char*)csv_stream;
+    m_end_stream = (char*)csv_stream + stream_length;
+    m_is_to_aggregate = do_aggregate;
+    m_skip_last_line = skip_last_line;
+
+    if(skip_first_line)
+    {
+      //the stream may start in the middle of a row (maybe in the middle of a quote).
+      //at this point the stream should skip the first row(broken row).
+      //the csv_parser should be init with the fixed stream position. 
+      m_stream += m_skip_x_first_bytes;
+      m_skip_x_first_bytes=0;
+    }
+
+    CSVParser _csv_parser("csv", m_stream, m_end_stream);
+    csv_parser = &_csv_parser;
+    csv_parser->set_csv_def(	m_csv_defintion.row_delimiter, 
+		    		m_csv_defintion.column_delimiter, 
+				m_csv_defintion.quot_char, 
+				m_csv_defintion.escape_char, 
+				m_csv_defintion.comment_empty_lines, 
+				m_csv_defintion.comment_chars, 
+				m_csv_defintion.trim_chars);
+
+
+    if(m_extract_csv_header_info == false)
+    {
+      extract_csv_header_info();
+    }
+    do
+    {
+      m_sql_processing_status = Status::INITIAL_STAT;
+      try
+      {
+        getMatchRow(result);
+      }
+      catch (base_s3select_exception& e)
+      {
+        m_error_description = e.what();
+        m_error_count ++;
+        if (e.severity() == base_s3select_exception::s3select_exp_en_t::FATAL || m_error_count>100 || (m_stream>=m_end_stream))//abort query execution
+        {
+          return -1;
+        }
+      }
+
+      if(fp_s3select_result_format && fp_s3select_header_format)
+      {
+      	if (result.size() > CSV_INPUT_TYPE_RESPONSE_SIZE_LIMIT)
+      	{//there are systems that might resject the response due to its size.
+	  fp_s3select_result_format(result);
+	  fp_s3select_header_format(result);
+      	}
+      }
+
+      if (m_sql_processing_status == Status::END_OF_STREAM)
+      {
+        break;
+      }
+      else if (m_sql_processing_status == Status::LIMIT_REACHED) // limit reached
+      {
+        break;//user should request for sql_processing_status
+      }
+
+    } while (true);
+
+    if(fp_s3select_result_format && fp_s3select_header_format)
+    {	//note: it may produce empty response(more the once)
+	//upon empty result, it should return *only* upon last call.
+	fp_s3select_result_format(result);
+	fp_s3select_header_format(result);
+    }
+
+    return 0;
+  }
+};
+
+#ifdef _ARROW_EXIST
+class parquet_object : public base_s3object
+{
+
+private:
+  std::string m_error_description;
+  parquet_file_parser* object_reader;
+  parquet_file_parser::column_pos_t m_where_clause_columns;
+  parquet_file_parser::column_pos_t m_projections_columns;
+  std::vector<parquet_file_parser::parquet_value_t> m_predicate_values;
+  std::vector<parquet_file_parser::parquet_value_t> m_projections_values;
+  bool not_to_increase_first_time;
+
+public:
+
+  parquet_object(std::string parquet_file_name, s3select *s3_query,s3selectEngine::rgw_s3select_api* rgw) : base_s3object(s3_query),object_reader(nullptr)
+  {
+    try{
+    
+      object_reader = new parquet_file_parser(parquet_file_name,rgw); //TODO uniq ptr
+    } catch(std::exception &e)
+    { 
+      throw base_s3select_exception(std::string("failure while processing parquet meta-data ") + std::string(e.what()) ,base_s3select_exception::s3select_exp_en_t::FATAL);
+    }
+
+    parquet_query_setting(nullptr);
+  }
+
+  parquet_object() : base_s3object(nullptr),object_reader(nullptr)
+  {}
+
+  void parquet_query_setting(s3select *s3_query)
+  {
+    if(s3_query)
+    {
+      set_base_defintions(s3_query);
+    }
+    load_meta_data_into_scratch_area();
+    for(auto x : m_s3_select->get_projections_list())
+    {//traverse the AST and extract all columns reside in projection statement.
+        x->extract_columns(m_projections_columns,object_reader->get_num_of_columns());
+    }
+    //traverse the AST and extract all columns reside in where clause. 
+    if(m_s3_select->get_filter())
+        m_s3_select->get_filter()->extract_columns(m_where_clause_columns,object_reader->get_num_of_columns());
+
+     not_to_increase_first_time = true;
+  }
+
+  ~parquet_object()
+  {
+    if(object_reader != nullptr)
+    {
+      delete object_reader;
+    }
+
+  }
+
+  std::string get_error_description()
+  {
+    return m_error_description;
+  }
+
+  bool is_set()
+  {
+    return m_s3_select != nullptr; 
+  }
+
+  void set_parquet_object(std::string parquet_file_name, s3select *s3_query,s3selectEngine::rgw_s3select_api* rgw) //TODO duplicate code
+  {
+    try{
+    
+      object_reader = new parquet_file_parser(parquet_file_name,rgw); //TODO uniq ptr
+    } catch(std::exception &e)
+    { 
+      throw base_s3select_exception(std::string("failure while processing parquet meta-data ") + std::string(e.what()) ,base_s3select_exception::s3select_exp_en_t::FATAL);
+    }
+
+    parquet_query_setting(s3_query);
+  }
+  
+
+  int run_s3select_on_object(std::string &result,
+        std::function<int(std::string&)> fp_s3select_result_format,
+        std::function<int(std::string&)> fp_s3select_header_format)
+  {
+	m_sql_processing_status = Status::INITIAL_STAT;
+    do
+    {
+      try
+      {
+        getMatchRow(result);
+      }
+      catch (base_s3select_exception &e)
+      {
+        m_error_description = e.what();
+        m_error_count++;
+        if (e.severity() == base_s3select_exception::s3select_exp_en_t::FATAL || m_error_count > 100) //abort query execution
+        {
+          return -1;
+        }
+      }
+      catch (std::exception &e)
+      {
+        m_error_description = e.what();
+        m_error_count++;
+        if (m_error_count > 100) //abort query execution
+        {
+          return -1;
+        }
+      }
+
+#define S3SELECT_RESPONSE_SIZE_LIMIT (4 * 1024 * 1024)
+      if (result.size() > S3SELECT_RESPONSE_SIZE_LIMIT)
+      {//AWS-cli limits response size the following callbacks send response upon some threshold
+        fp_s3select_result_format(result);
+
+        if (!is_end_of_stream() && (get_sql_processing_status() != Status::LIMIT_REACHED))
+        {
+          fp_s3select_header_format(result);
+        }
+      }
+      else
+      {
+        if (is_end_of_stream() || (get_sql_processing_status() == Status::LIMIT_REACHED))
+        {
+          fp_s3select_result_format(result);
+        }
+      }
+
+      //TODO is_end_of_stream() required?
+      if (get_sql_processing_status() == Status::END_OF_STREAM || is_end_of_stream() || get_sql_processing_status() == Status::LIMIT_REACHED)
+      {
+        break;
+      }
+
+    } while (1);
+
+    return 0;
+  }
+
+  void load_meta_data_into_scratch_area()
+  {
+    int i=0;
+    for(auto x : object_reader->get_schema())
+    {
+      m_s3_select->get_scratch_area()->set_column_pos(x.first.c_str(),i++); 
+    }
+  }
+
+  virtual bool is_end_of_stream()
+  {
+    return object_reader->end_of_stream();
+  }
+
+  virtual void columnar_fetch_where_clause_columns()
+  {
+    if(!not_to_increase_first_time)//for rownum=0 
+      object_reader->increase_rownum();
+    else
+      not_to_increase_first_time = false;
+
+    auto status = object_reader->get_column_values_by_positions(m_where_clause_columns, m_predicate_values);
+    if(status<0)//TODO exception?
+      return;
+    m_sa->update(m_predicate_values, m_where_clause_columns);
+  }
+
+  virtual void columnar_fetch_projection()
+  {
+    auto status = object_reader->get_column_values_by_positions(m_projections_columns, m_projections_values);
+    if(status<0)//TODO exception?
+      return;
+    m_sa->update(m_projections_values, m_projections_columns);
+  }
+
+};
+#endif //_ARROW_EXIST
+
+class json_object : public base_s3object
+{
+private:
+
+  JsonParserHandler JsonHandler;
+  size_t m_processed_bytes;
+  bool m_end_of_stream;
+  std::string* m_s3select_result = nullptr;
+  size_t m_row_count;
+  bool star_operation_ind;
+  std::string m_error_description;
+  bool m_init_json_processor_ind;
+
+public:
+
+  void init_json_processor(s3select* query)
+  {
+    if(m_init_json_processor_ind)
+	    return;
+
+    m_init_json_processor_ind = true;
+    std::function<int(void)> f_sql = [this](void){auto res = sql_execution_on_row_cb();return res;};
+    std::function<int(s3selectEngine::value&, int)> 
+      f_push_to_scratch = [this](s3selectEngine::value& value,int json_var_idx){return push_into_scratch_area_cb(value,json_var_idx);};
+    std::function <int(s3selectEngine::scratch_area::json_key_value_t&)>
+      f_push_key_value_into_scratch_area_per_star_operation = [this](s3selectEngine::scratch_area::json_key_value_t& key_value)
+                {return push_key_value_into_scratch_area_per_star_operation(key_value);};
+
+    //setting the container for all json-variables, to be extracted by the json reader    
+    JsonHandler.set_statement_json_variables(query->get_json_variables_access());
+
+
+    //calling to getMatchRow. processing a single row per each call.
+    JsonHandler.set_s3select_processing_callback(f_sql);
+    //upon excat match between input-json-key-path and sql-statement-variable-path the callback pushes to scratch area 
+    JsonHandler.set_exact_match_callback(f_push_to_scratch);
+    //upon star-operation(in statemenet) the callback pushes the key-path and value into scratch-area
+    JsonHandler.set_push_per_star_operation_callback(f_push_key_value_into_scratch_area_per_star_operation);
+
+    //the json-from-clause is unique and should exist. otherwise it's a failure. 
+    if(query->getAction()->json_from_clause.empty())
+    {
+	JsonHandler.m_fatal_initialization_ind = true;
+	JsonHandler.m_fatal_initialization_description = "the SQL statement is not align with the correct syntax of JSON statement. from-clause is missing.";
+	return;
+    }
+
+    //setting the from clause path 
+    if(query->getAction()->json_from_clause[0] == JSON_ROOT_OBJECT)
+    {
+      query->getAction()->json_from_clause.pop_back();
+    }
+    JsonHandler.set_prefix_match(query->getAction()->json_from_clause);
+
+    for (auto& p : m_projections)
+    {
+      if(p->is_statement_contain_star_operation())
+      {
+        star_operation_ind=true;
+        break;
+      }
+    }
+
+    if(star_operation_ind)
+    {
+      JsonHandler.set_star_operation();
+      //upon star-operation the key-path is extracted with the value, each key-value displayed in a seperate row.
+      //the return results end with a line contains the row-number.
+      m_csv_defintion.output_column_delimiter = m_csv_defintion.output_row_delimiter;
+    }
+
+    m_sa->set_parquet_type();//TODO json type
+  }
+    
+  json_object(s3select* query):base_s3object(query),m_processed_bytes(0),m_end_of_stream(false),m_row_count(0),star_operation_ind(false),m_init_json_processor_ind(false)
+  {
+    init_json_processor(query);
+  }
+
+  void set_sql_result(std::string& sql_result)
+  {
+	m_s3select_result = &sql_result; 
+  }
+
+  json_object(): base_s3object(nullptr), m_processed_bytes(0),m_end_of_stream(false),m_row_count(0),star_operation_ind(false),m_init_json_processor_ind(false) {}
+
+private:
+
+  virtual bool is_end_of_stream()
+  {
+      return m_end_of_stream == true;
+  }
+
+  virtual bool multiple_row_processing()
+  {
+    return false;
+  }
+
+  int sql_execution_on_row_cb()
+  {
+      //execute statement on row 
+      //create response (TODO callback)
+
+      size_t result_len = m_s3select_result->size();
+      int status=0;
+      try{
+	getMatchRow(*m_s3select_result);
+      }
+      catch(s3selectEngine::base_s3select_exception& e)
+      {
+	sql_error_handling(e,*m_s3select_result);
+	status = -1;
+      }
+
+      if(is_sql_limit_reached()) 
+      {
+	      status = JSON_PROCESSING_LIMIT_REACHED;//returning number since sql_execution_on_row_cb is a callback; the caller can not access the object
+      }
+
+      m_sa->clear_data(); 
+      if(star_operation_ind && (m_s3select_result->size() != result_len))
+      {//as explained above the star-operation is displayed differently
+	std::string end_of_row;
+	end_of_row = "#=== " + std::to_string(m_row_count++) + " ===#\n";
+	m_s3select_result->append(end_of_row);
+      }
+      return status;
+  }
+
+  int push_into_scratch_area_cb(s3selectEngine::value& key_value, int json_var_idx)
+  {
+    //upon exact-filter match push value to scratch area with json-idx ,  it should match variable
+    //push (key path , json-var-idx , value) json-var-idx should be attached per each exact filter
+    m_sa->update_json_varible(key_value,json_var_idx);
+    return 0;
+  }
+
+  int push_key_value_into_scratch_area_per_star_operation(s3selectEngine::scratch_area::json_key_value_t& key_value)
+  {
+    m_sa->get_star_operation_cont()->push_back( key_value );
+    return 0;
+  }
+
+  void sql_error_handling(s3selectEngine::base_s3select_exception& e,std::string& result)
+  {
+    //the JsonHandler makes the call to SQL processing, upon a failure to procees the SQL statement, 
+    //the error-handling takes care of the error flow.
+    m_error_description = e.what();
+    m_error_count++;
+    m_s3select_result->append(std::to_string(m_error_count));
+    *m_s3select_result += " : ";
+    m_s3select_result->append(m_error_description);
+    *m_s3select_result += m_csv_defintion.output_row_delimiter;
+  }
+
+public:
+
+  int run_s3select_on_stream(std::string& result, const char* json_stream, size_t stream_length, size_t obj_size)
+  {
+    int status=0;
+    m_processed_bytes += stream_length;
+    set_sql_result(result);
+
+    if(JsonHandler.is_fatal_initialization())
+    {
+      throw base_s3select_exception(JsonHandler.m_fatal_initialization_description, base_s3select_exception::s3select_exp_en_t::FATAL);
+    }
+
+    if(!stream_length || !json_stream)//TODO m_processed_bytes(?)
+    {//last processing cycle
+      JsonHandler.process_json_buffer(0, 0, true);//TODO end-of-stream = end-of-row
+      m_end_of_stream = true;
+      sql_execution_on_row_cb();
+      return 0;
+    }
+
+    try{
+    //the handler is processing any buffer size and return results per each buffer
+      status = JsonHandler.process_json_buffer((char*)json_stream, stream_length);
+    }
+    catch(std::exception &e)
+    {
+	std::string error_description = std::string("exception while processing :") + e.what();
+	throw base_s3select_exception(error_description,base_s3select_exception::s3select_exp_en_t::FATAL);
+    }
+
+    if(status<0)
+    {
+    	std::string error_description = std::string("failure upon JSON processing");
+    	throw base_s3select_exception(error_description,base_s3select_exception::s3select_exp_en_t::FATAL);
+	return -1;
+    }
+ 
+    return status; 
+  }
+
+  void set_json_query(s3select* s3_query)
+  {
+    set_base_defintions(s3_query);
+    init_json_processor(s3_query);
+  }
+
+  std::string get_error_description()
+  {
+    return m_error_description;
+  }
+
+  ~json_object() = default;
+};
+
+}; // namespace s3selectEngine
+
+#endif