X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Fftacmp%2Fquery_plan.cc;h=2e2286ec989aeeb86c8683a4ee97b9abae7dd132;hb=f1754ecea2eab7bd0a302042ac82eb11667b166c;hp=c84edb97782b4b483b7ea56c01a3626fd7741461;hpb=c9783d8ea8b85d810483559e50dbf2297109e349;p=com%2Fgs-lite.git diff --git a/src/ftacmp/query_plan.cc b/src/ftacmp/query_plan.cc index c84edb9..2e2286e 100644 --- a/src/ftacmp/query_plan.cc +++ b/src/ftacmp/query_plan.cc @@ -98,16 +98,76 @@ mrg_qpn::mrg_qpn(filter_join_qpn *spx, std::string n_name, std::vector &sources, std::vector > &ifaces, ifq_t *ifdb){ + param_tbl = spx->param_tbl; + int i; + node_name = n_name; + field_entry_list *fel = new field_entry_list(); + merge_fieldpos = -1; + + disorder = 1; + + for(i=0;iselect_list.size();++i){ + data_type *dt = spx->select_list[i]->se->get_data_type()->duplicate(); + if(dt->is_temporal()){ + if(merge_fieldpos < 0){ + merge_fieldpos = i; + }else{ + fprintf(stderr,"Warning: Merge subquery %s found two temporal fields (%s, %s), using %s\n", n_name.c_str(), spx->select_list[merge_fieldpos]->name.c_str(), spx->select_list[i]->name.c_str(), spx->select_list[merge_fieldpos]->name.c_str() ); + dt->reset_temporal(); + } + } + + field_entry *fe = dt->make_field_entry(spx->select_list[i]->name); + fel->append_field(fe); + delete dt; + } + if(merge_fieldpos<0){ + fprintf(stderr,"ERROR, no temporal attribute for merge subquery %s\n",n_name.c_str()); + exit(1); + } + table_layout = new table_def( n_name.c_str(), NULL, NULL, fel, STREAM_SCHEMA); + +// NEED TO HANDLE USER_SPECIFIED SLACK + this->resolve_slack(spx->select_list[merge_fieldpos]->se, + spx->select_list[merge_fieldpos]->name, ifaces, ifdb,NULL); +// if(this->slack == NULL) +// fprintf(stderr,"Zero slack.\n"); +// else +// fprintf(stderr,"slack is %s\n",slack->to_string().c_str()); + + for(i=0;iselect_list[merge_fieldpos]->name.c_str())); + mvars[i]->set_tablevar_ref(i); + fm.push_back(new tablevar_t(sources[i].c_str())); + fm[i]->set_range_var(rvar); + } + + param_tbl = new param_table(); + std::vector param_names = spx->param_tbl->get_param_names(); + int pi; + for(pi=0;piparam_tbl->get_data_type(param_names[pi]); + param_tbl->add_param(param_names[pi],dt->duplicate(), + spx->param_tbl->handle_access(param_names[pi])); + } + definitions = spx->definitions; + +} + + + // This function translates an analyzed parse tree // into one or more query nodes (qp_node). // Currently only one node is created, but some query // fragments might create more than one query node, -// e.g. aggregation over a joim, or nested subqueries -// in the FROM clause (unless this is handles at parse tree +// e.g. aggregation over a join, or nested subqueries +// in the FROM clause (unless this is handled at parse tree // analysis time). At this stage, they will be linked // by the names in the FROM clause. -// INVARIANT : if mroe than one query node is returned, +// INVARIANT : if more than one query node is returned, // the last one represents the output of the query. vector create_query_nodes(query_summary_class *qs,table_list *Schema){ @@ -121,6 +181,16 @@ vector create_query_nodes(query_summary_class *qs,table_list *Schema) // into the qp_node constructors, // and have this code focus on building the query plan tree. +// Watchlist node + if(qs->query_type == WATCHLIST_QUERY){ + watch_tbl_qpn *watchnode = new watch_tbl_qpn(qs, Schema); + +// Done + plan_root = watchnode; + local_plan.push_back(watchnode); + } + + // MERGE node if(qs->query_type == MERGE_QUERY){ mrg_qpn *merge_node = new mrg_qpn(qs,Schema); @@ -159,7 +229,9 @@ printf("\n"); */ - } else{ + } + + if(qs->query_type == SELECT_QUERY){ // Select / Aggregation / Join if(qs->gb_tbl->size() == 0 && qs->aggr_tbl->size() == 0){ @@ -175,9 +247,15 @@ printf("\n"); plan_root = join_node; local_plan.push_back(join_node); }else{ - join_eq_hash_qpn *join_node = new join_eq_hash_qpn(qs,Schema); - plan_root = join_node; - local_plan.push_back(join_node); + if(qs->fta_tree->get_from()->get_properties() == WATCHLIST_JOIN_PROPERTY){ + watch_join_qpn *join_node = new watch_join_qpn(qs,Schema); + plan_root = join_node; + local_plan.push_back(join_node); + }else{ + join_eq_hash_qpn *join_node = new join_eq_hash_qpn(qs,Schema); + plan_root = join_node; + local_plan.push_back(join_node); + } } } }else{ @@ -2077,6 +2155,20 @@ string sgahcwcb_qpn::to_query_string(){ return(ret); } +string watch_tbl_qpn::to_query_string(){ + string ret; +// ret += "DEFINE {\n"; +// ret += "\tfilename='"+filename+";\n"; +// ret += "\trefresh_interval="+to_string(refresh_interval)+";\n}\n"; + ret += "WATCHLIST FIELDS {\n"; + std::vector fields = table_layout->get_fields(); + for(int f=0;fto_string()+"\n"; + } + ret += "}\n"; + + return ret; +} string mrg_qpn::to_query_string(){ @@ -2181,6 +2273,42 @@ string filter_join_qpn::to_query_string(){ return(ret); } +string watch_join_qpn::to_query_string(){ + + string ret = "Select "; + int s; + for(s=0;s0) ret+=", "; + ret += se_to_query_string(select_list[s]->se, NULL); + if(select_list[s]->name != "") ret += " AS "+select_list[s]->name; + } + ret += "\n"; + +// NOTE: assuming binary join. + ret += "WATCHLIST_JOIN "; + + ret += "From "; + int f; + for(f=0;f0) ret+=", "; + ret += from[f]->to_string(); + } + ret += "\n"; + + if(where.size() > 0){ + ret += "Where "; + int w; + for(w=0;w0) ret += " AND "; + ret += "(" + pred_to_query_str(where[w]->pr,NULL) + ")"; + } + ret += "\n"; + } + + return(ret); +} + + // ----------------------------------------------------------------- // Query node subclass specific processing. @@ -2390,6 +2518,18 @@ void mrg_qpn::resolve_slack(scalarexp_t *t_se, string fname, vector watch_tbl_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, table_list *Schema, int &hfta_returned, ifq_t *ifdb, int n_virtual_ifaces, int hfta_parallelism, int hfta_idx){ + // nothing to do, nothing to split, return copy of self. + + hfta_returned = 0; + + vector ret_vec; + + ret_vec.push_back(this); + return(ret_vec); + +} + vector mrg_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, table_list *Schema, int &hfta_returned, ifq_t *ifdb, int n_virtual_ifaces, int hfta_parallelism, int hfta_idx){ // nothing to do, nothing to split, return copy of self. @@ -2429,7 +2569,7 @@ vector filter_join_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, ta vector sel_names; vector > ifaces = get_ifaces(from[0], ifdb, n_virtual_ifaces, hfta_parallelism, hfta_idx); if (ifaces.empty()) { - fprintf(stderr,"INTERNAL ERROR in filter_join_qpn::split_node_for_fta - empty interface set\n"); + fprintf(stderr,"INTERNAL ERROR in filter_join_qpn::split_node_for_fta - empty interface set, node is %s\n", node_name.c_str()); exit(1); } @@ -2481,35 +2621,35 @@ vector filter_join_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, ta } for(p=0;ppr, NULL); + predicate_t *new_pr = dup_pr(shared_pred[p]->pr, NULL); cnf_elem *new_cnf = new cnf_elem(new_pr); analyze_cnf(new_cnf); fta_node->shared_pred.push_back(new_cnf); fta_node->where.push_back(new_cnf); } for(p=0;ppr, NULL); + predicate_t *new_pr = dup_pr(pred_t0[p]->pr, NULL); cnf_elem *new_cnf = new cnf_elem(new_pr); analyze_cnf(new_cnf); fta_node->pred_t0.push_back(new_cnf); fta_node->where.push_back(new_cnf); } for(p=0;ppr, NULL); + predicate_t *new_pr = dup_pr(pred_t1[p]->pr, NULL); cnf_elem *new_cnf = new cnf_elem(new_pr); analyze_cnf(new_cnf); fta_node->pred_t1.push_back(new_cnf); fta_node->where.push_back(new_cnf); } for(p=0;ppr, NULL); + predicate_t *new_pr = dup_pr(hash_eq[p]->pr, NULL); cnf_elem *new_cnf = new cnf_elem(new_pr); analyze_cnf(new_cnf); fta_node->hash_eq.push_back(new_cnf); fta_node->where.push_back(new_cnf); } for(p=0;ppr, NULL); + predicate_t *new_pr = dup_pr(postfilter[p]->pr, NULL); cnf_elem *new_cnf = new cnf_elem(new_pr); analyze_cnf(new_cnf); fta_node->postfilter.push_back(new_cnf); @@ -2543,6 +2683,155 @@ vector filter_join_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, ta } + + + + +vector watch_join_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, table_list *Schema, int &hfta_returned, ifq_t *ifdb, int n_virtual_ifaces, int hfta_parallelism, int hfta_idx){ + vector ret_vec; + +// First check if the query can be pushed to the FTA. + bool fta_ok = true; + int s; + for(s=0;sse,NULL, Ext_fcns); + } + int p; + for(p=0;ppr,NULL, Ext_fcns); + } + + if(!fta_ok){ + fprintf(stderr,"ERROR, watchlist join %s is fta-unsafe.\n",node_name.c_str()); + exit(1); + } + +// Can it be done in a single lfta? +// Get the set of interfaces it accesses. + int ierr; + int si; + vector sel_names; + vector > ifaces = get_ifaces(from[0], ifdb, n_virtual_ifaces, hfta_parallelism, hfta_idx); + if (ifaces.empty()) { + fprintf(stderr,"INTERNAL ERROR in filter_join_qpn::split_node_for_fta - empty interface set, node is %s\n", node_name.c_str()); + exit(1); + } + + if(ifaces.size() == 1){ +// Single interface, no need to merge. + hfta_returned = 0; + ret_vec.push_back(this); + +// Treat the range vars a bit differently, the 2nd is reading from a _local_ watchlist. + from[0]->set_machine(ifaces[0].first); + from[0]->set_interface(ifaces[0].second); + from[0]->set_ifq(false); + + from[1]->set_machine(ifaces[0].first); + from[1]->set_interface("_local_"); + from[1]->set_ifq(false); + + return(ret_vec); + }else{ +// Multiple interfaces, generate the interface-specific queries plus +// the merge. + hfta_returned = 1; + + vector sel_names; + for(si=0;siset_node_name( node_name ); + else{ + string new_name = "_"+node_name+"_"+ifaces[si].first+"_"+ifaces[si].second; + untaboo(new_name); + fta_node->set_node_name(new_name); + } + sel_names.push_back(fta_node->get_node_name()); + +// Assign the table + int f; + for(f=0;ffrom.push_back(from[f]->duplicate()); + fta_node->from[f]->set_machine(ifaces[si].first); + if(f==0) + fta_node->from[f]->set_interface(ifaces[si].second); + else + fta_node->from[f]->set_interface("_local_"); + fta_node->from[f]->set_ifq(false); + } + + for(s=0;sselect_list.push_back( dup_select(select_list[s], NULL) ); + } + + for(p=0;ppr, NULL); + cnf_elem *new_cnf = new cnf_elem(new_pr); + analyze_cnf(new_cnf); + fta_node->pred_t0.push_back(new_cnf); + fta_node->where.push_back(new_cnf); + } + for(p=0;ppr, NULL); + cnf_elem *new_cnf = new cnf_elem(new_pr); + analyze_cnf(new_cnf); + fta_node->pred_t1.push_back(new_cnf); + fta_node->where.push_back(new_cnf); + } + for(p=0;ppr, NULL); + cnf_elem *new_cnf = new cnf_elem(new_pr); + analyze_cnf(new_cnf); + fta_node->hash_eq[k] = new_cnf; + fta_node->where.push_back(new_cnf); + } + for(p=0;ppr, NULL); + cnf_elem *new_cnf = new cnf_elem(new_pr); + analyze_cnf(new_cnf); + fta_node->postfilter.push_back(new_cnf); + fta_node->where.push_back(new_cnf); + } + for(p=0;ppr, NULL); + cnf_elem *new_cnf = new cnf_elem(new_pr); + analyze_cnf(new_cnf); + fta_node->postfilter.push_back(new_cnf); + fta_node->where.push_back(new_cnf); + } + fta_node->key_flds = key_flds; + +// Xfer all of the parameters. +// Use existing handle annotations. + vector param_names = param_tbl->get_param_names(); + int pi; + for(pi=0;piget_data_type(param_names[pi]); + fta_node->param_tbl->add_param(param_names[pi],dt->duplicate(), + param_tbl->handle_access(param_names[pi])); + } + fta_node->definitions = definitions; + if(fta_node->resolve_if_params(ifdb, this->err_str)){ + this->error_code = 3; + return ret_vec; + } + + ret_vec.push_back(fta_node); + } + + mrg_qpn *mrg_node = new mrg_qpn((watch_join_qpn *)ret_vec[0], + node_name, sel_names,ifaces, ifdb); + ret_vec.push_back(mrg_node); + + return(ret_vec); + } + +} + // Use to search for unresolved interface param refs in an hfta. int spx_qpn::count_ifp_refs(set &ifpnames){ @@ -2606,6 +2895,10 @@ int rsgah_qpn::count_ifp_refs(set &ifpnames){ return ret; } +int watch_tbl_qpn::count_ifp_refs(set &ifpnames){ + return 0; +} + int mrg_qpn::count_ifp_refs(set &ifpnames){ return 0; } @@ -2638,6 +2931,17 @@ int filter_join_qpn::count_ifp_refs(set &ifpnames){ return ret; } +int watch_join_qpn::count_ifp_refs(set &ifpnames){ + int ret = 0; + int i; + for(i=0;ise,ifpnames); + for(i=0;ipr,ifpnames); + return ret; +} + + // Resolve interface params to string literals int filter_join_qpn::resolve_if_params( ifq_t *ifdb, string &err){ @@ -2654,6 +2958,20 @@ int filter_join_qpn::resolve_if_params( ifq_t *ifdb, string &err){ return ret; } +int watch_join_qpn::resolve_if_params( ifq_t *ifdb, string &err){ + int ret = 0; + int i; + string ifname = from[0]->get_interface(); + string ifmach = from[0]->get_machine(); + for(i=0;ise,ifmach, ifname, ifdb, err) ) + ret = 1; + for(i=0;ipr,ifmach, ifname, ifdb, err)) + ret = 1; + return ret; +} + int spx_qpn::resolve_if_params( ifq_t *ifdb, string &err){ int ret = 0; @@ -2784,7 +3102,7 @@ vector spx_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, table_list vector sel_names; vector > ifaces = get_ifaces(table_name, ifdb, n_virtual_ifaces, hfta_parallelism, hfta_idx); if (ifaces.empty()) { - fprintf(stderr,"INTERNAL ERROR in spx_qpn::split_node_for_fta - empty interface set\n"); + fprintf(stderr,"INTERNAL ERROR in spx_qpn::split_node_for_fta - empty interface set, node is %s\n", node_name.c_str()); exit(1); } @@ -3138,7 +3456,7 @@ vector rsgah_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, table_li vector sel_names; vector > ifaces = get_ifaces(table_name, ifdb, n_virtual_ifaces, hfta_parallelism, hfta_idx); if (ifaces.empty()) { - fprintf(stderr,"INTERNAL ERROR in rsgah_qpn::split_node_for_fta - empty interface set\n"); + fprintf(stderr,"INTERNAL ERROR in rsgah_qpn::split_node_for_fta - empty interface set, node is %s\n", node_name.c_str()); exit(1); } @@ -3805,7 +4123,7 @@ vector sgah_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, table_lis vector sel_names; vector > ifaces = get_ifaces(table_name, ifdb, n_virtual_ifaces, hfta_parallelism, hfta_idx); if (ifaces.empty()) { - fprintf(stderr,"INTERNAL ERROR in sgah_qpn::split_node_for_fta - empty interface set\n"); + fprintf(stderr,"INTERNAL ERROR in sgah_qpn::split_node_for_fta - empty interface set, node is %s\n", node_name.c_str()); exit(1); } @@ -4596,6 +4914,7 @@ vector join_eq_hash_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, t child_qpn->table_name = new tablevar_t( from[f]->get_interface().c_str(), from[f]->get_schema_name().c_str(), from[f]->get_ifq()); child_qpn->table_name->set_range_var(from[f]->get_var_name()); + child_qpn->table_name->set_machine(from[f]->get_machine()); child_vec.push_back(child_qpn); select_vec.push_back(&(child_qpn->select_list)); @@ -4735,7 +5054,7 @@ vector join_eq_hash_qpn::split_node_for_fta(ext_fcn_list *Ext_fcns, t spx_qpn *c_node = child_vec[f]; vector > ifaces = get_ifaces(c_node->table_name, ifdb, n_virtual_ifaces, hfta_parallelism, hfta_idx); if (ifaces.empty()) { - fprintf(stderr,"INTERNAL ERROR in join_eq_hash_qpn::split_node_for_fta - empty interface set\n"); + fprintf(stderr,"INTERNAL ERROR in join_eq_hash_qpn::split_node_for_fta - empty interface set, node is %s\n", node_name.c_str()); exit(1); } @@ -5002,6 +5321,21 @@ vector filter_join_qpn::extract_opview(table_list *Schema, vecto return(ret); } +vector watch_join_qpn::extract_opview(table_list *Schema, vector &qnodes, opview_set &opviews, string rootnm, string silo_name){ + vector ret; + int retval = process_opview(from[0],0,node_name, + Schema,qnodes,opviews,ret, rootnm, silo_name); + if(retval) exit(1); + return(ret); +} + + + +vector watch_tbl_qpn::extract_opview(table_list *Schema, vector &qnodes, opview_set &opviews, string rootnm, string silo_name){ + vector ret; + return ret; // nothing to process +} + ////////////////////////////////////////////////////////////////// @@ -5017,6 +5351,10 @@ table_def *mrg_qpn::get_fields(){ return(table_layout); } +table_def *watch_tbl_qpn::get_fields(){ + return(table_layout); +} + table_def *spx_qpn::get_fields(){ return(create_attributes(node_name, select_list)); @@ -5038,6 +5376,9 @@ table_def *filter_join_qpn::get_fields(){ return(create_attributes(node_name, select_list)); } +table_def *watch_join_qpn::get_fields(){ + return(create_attributes(node_name, select_list)); +} table_def *join_eq_hash_qpn::get_fields(){ int i, h, s, t; @@ -5139,6 +5480,57 @@ table_def *join_eq_hash_qpn::get_fields(){ } +//----------------------------------------------------------------- +// get output "keys" +// This is a guess about the set of fields which are a key +// Use as metadata output, e.g. in qtree.xml + + + +// refs to GB attribtues are keys, if a SE is not a GB colref +// but refers to a GB colref (outside of an aggregation) +// then set partial_keys to true +vector sgah_qpn::get_tbl_keys(vector &partial_keys){ + vector keys; + + set gref_set; + for(int i=0; ise->is_gb()){ + keys.push_back(select_list[s]->name); + }else{ + if(contains_gb_se(select_list[s]->se, gref_set)){ + partial_keys.push_back(select_list[s]->name); + } + } + } + return keys; +} + +vector rsgah_qpn::get_tbl_keys(vector &partial_keys){ + vector keys; + + set gref_set; + for(int i=0; ise->is_gb()){ + keys.push_back(select_list[s]->name); + }else{ + if(contains_gb_se(select_list[s]->se, gref_set)){ + partial_keys.push_back(select_list[s]->name); + } + } + } + return keys; +} + + + + //----------------------------------------------------------------- // get output tables @@ -5151,6 +5543,11 @@ table_def *join_eq_hash_qpn::get_fields(){ return(fm); } + vector watch_tbl_qpn::get_input_tbls(){ + vector ret; + return(ret); + } + vector mrg_qpn::get_input_tbls(){ return(fm); } @@ -5183,6 +5580,10 @@ table_def *join_eq_hash_qpn::get_fields(){ return(from); } + vector watch_join_qpn::get_input_tbls(){ + return(from); + } + //----------------------------------------------------------------- // get output tables @@ -5194,6 +5595,11 @@ table_def *join_eq_hash_qpn::get_fields(){ return(retval); } + vector watch_tbl_qpn::get_output_tbls(){ + vector retval(1,new tablevar_t(node_name.c_str())); + return(retval); + } + vector mrg_qpn::get_output_tbls(){ vector retval(1,new tablevar_t(node_name.c_str())); return(retval); @@ -5230,6 +5636,12 @@ table_def *join_eq_hash_qpn::get_fields(){ } + vector watch_join_qpn::get_output_tbls(){ + vector retval(1,new tablevar_t(node_name.c_str())); + return(retval); + } + + //----------------------------------------------------------------- // Bind to schema @@ -5327,6 +5739,30 @@ col_id_set filter_join_qpn::get_colrefs(bool ext_fcns_only,table_list *Schema){ return tmp_cset; } +col_id_set watch_join_qpn::get_colrefs(bool ext_fcns_only,table_list *Schema){ + col_id_set retval, tmp_cset; + int p; + for(p=0;ppr, tmp_cset, NULL); + } + int s; + for(s=0;sse, tmp_cset, NULL); + } + col_id_set::iterator cisi; + if(ext_fcns_only){ + for(cisi=tmp_cset.begin();cisi!=tmp_cset.end();++cisi){ + field_entry *fe = Schema->get_field((*cisi).schema_ref, (*cisi).field); + if(fe->get_unpack_fcns().size()>0) + retval.insert((*cisi)); + } + return retval; + } + + return tmp_cset; +} + + // Associate colrefs in SEs with this schema. @@ -5386,6 +5822,35 @@ void filter_join_qpn::bind_to_schema(table_list *Schema){ } +void watch_join_qpn::bind_to_schema(table_list *Schema){ +// Bind the tablevars in the From clause to the Schema +// (it might have changed from analysis time) + int f; + for(f=0;fget_schema_name(); + int tbl_ref = Schema->get_table_ref(snm); + if(tbl_ref >= 0) + from[f]->set_schema_ref(tbl_ref); + } + +// Bind all SEs to this schema + tablevar_list_t fm(from); + + int p; + for(p=0;ppr, &fm, Schema); + } + int s; + for(s=0;sse, &fm, Schema); + } + +// Collect set of tuples referenced in this HFTA +// input, internal, or output. + +} + + @@ -5573,6 +6038,10 @@ void sgahcwcb_qpn::bind_to_schema(table_list *Schema){ //----------------------------------------------------------------- // get_cplx_lit_tbl +cplx_lit_table *watch_tbl_qpn::get_cplx_lit_tbl(ext_fcn_list *Ext_fcns){ + return(new cplx_lit_table()); +} + cplx_lit_table *mrg_qpn::get_cplx_lit_tbl(ext_fcn_list *Ext_fcns){ return(new cplx_lit_table()); } @@ -5720,12 +6189,32 @@ cplx_lit_table *filter_join_qpn::get_cplx_lit_tbl(ext_fcn_list *Ext_fcns){ return(complex_literals); } +cplx_lit_table *watch_join_qpn::get_cplx_lit_tbl(ext_fcn_list *Ext_fcns){ + int i; + cplx_lit_table *complex_literals = new cplx_lit_table(); + + for(i=0;ise, Ext_fcns, complex_literals); + } + for(i=0;ipr,Ext_fcns, complex_literals); + } + + return(complex_literals); +} + + //----------------------------------------------------------------- // get_handle_param_tbl +vector watch_tbl_qpn::get_handle_param_tbl(ext_fcn_list *Ext_fcns){ + vector retval; + return(retval); +} + vector mrg_qpn::get_handle_param_tbl(ext_fcn_list *Ext_fcns){ vector retval; return(retval); @@ -5877,6 +6366,22 @@ vector filter_join_qpn::get_handle_param_tbl(ext_fcn_l return(retval); } +vector watch_join_qpn::get_handle_param_tbl(ext_fcn_list *Ext_fcns){ + int i; + vector retval; + + for(i=0;ise, Ext_fcns, retval); + } + for(i=0;ipr,Ext_fcns, retval); + } + + return(retval); +} + + + /////////////////////////////////////////////////////////////// /////////////////////////////////////////////////////////////// /// Functions for operator output rates estimations @@ -5909,6 +6414,12 @@ double sgahcwcb_qpn::get_rate_estimate() { return SGAHCWCB_SELECTIVITY * DEFAULT_INTERFACE_RATE; } +double watch_tbl_qpn::get_rate_estimate() { + + // dummy method for now + return DEFAULT_INTERFACE_RATE; +} + double mrg_qpn::get_rate_estimate() { // dummy method for now @@ -6522,6 +7033,29 @@ static string generate_equality_test(string &lhs_op, string &rhs_op, data_type * return(ret); } +static string generate_lt_test(string &lhs_op, string &rhs_op, data_type *dt){ + string ret; + + if(dt->complex_comparison(dt) ){ + ret.append(dt->get_hfta_comparison_fcn(dt)); + ret.append("("); + if(dt->is_buffer_type() ) + ret.append("&"); + ret.append(lhs_op); + ret.append(", "); + if(dt->is_buffer_type() ) + ret.append("&"); + ret.append(rhs_op ); + ret.append(") == 1"); + }else{ + ret.append(lhs_op ); + ret.append(" < "); + ret.append(rhs_op ); + } + + return(ret); +} + static string generate_comparison(string &lhs_op, string &rhs_op, data_type *dt){ string ret; @@ -9825,6 +10359,23 @@ string mrg_qpn::generate_operator(int i, string params){ ); } +//////////////////////////////////////////////// +/// WATCHLIST_TBL operator +/// WATCHLIST_TBL functor +//////////////////////////////////////////// + +string watch_tbl_qpn::generate_functor_name(){ + return("watch_tbl_functor_" + normalize_name(this->get_node_name())); +} + +string watch_tbl_qpn::generate_functor(table_list *schema, ext_fcn_list *Ext_fcns, vector &needs_xform){ + + return("ERROR_WATCH_TBL_FUNCTOR_NOT_YET_IMPLEMENTED"); +} + +string watch_tbl_qpn::generate_operator(int i, string params){ + return("ERROR_WATCH_TBL_FUNCTOR_NOT_YET_IMPLEMENTED"); +} ///////////////////////////////////////////////////////// ////// JOIN_EQ_HASH functor @@ -12336,13 +12887,13 @@ aggr_table_entry *ate = aggr_tbl.agr_tbl[a]; for(g=0;gis_temporal()){ - sprintf(tmpstr,"last_gb%d",g); + sprintf(tmpstr,"curr_gb%d",g); ret+="\t"+gb_tbl.get_data_type(g)->make_host_cvar(tmpstr)+";\n"; - sprintf(tmpstr,"last_flushed_gb%d",g); + sprintf(tmpstr,"last_gb%d",g); ret+="\t"+gb_tbl.get_data_type(g)->make_host_cvar(tmpstr)+";\n"; } } - ret += "\tbool needs_temporal_flush;\n"; + ret += "\tgs_int32_t needs_temporal_flush;\n"; } // The publicly exposed functions @@ -12380,6 +12931,7 @@ aggr_table_entry *ate = aggr_tbl.agr_tbl[a]; // temporal flush variables // ASSUME that structured values won't be temporal. + gs_int32_t temporal_gb = 0; if(uses_temporal_flush){ ret += "//\t\tInitialize temporal flush variables.\n"; for(g=0;gtype_indicator()); sprintf(tmpstr,"\tlast_gb%d = %s;\n",g, gl.to_hfta_C_code("").c_str()); ret.append(tmpstr); + sprintf(tmpstr,"\tcurr_gb%d = %s;\n",g, gl.to_hfta_C_code("").c_str()); + ret.append(tmpstr); + temporal_gb = g; } } - ret += "\tneeds_temporal_flush = false;\n"; + ret += "\tneeds_temporal_flush = 0;\n"; } // Init temporal attributes referenced in select list @@ -12504,35 +13059,39 @@ aggr_table_entry *ate = aggr_tbl.agr_tbl[a]; // set flush indicator and update stored GB vars if there is any change. if(uses_temporal_flush){ - ret+= "\tif( !( ("; + ret+= "\tif( ( ("; bool first_one = true; for(g=0;gis_temporal()){ - sprintf(tmpstr,"last_gb%d",g); string lhs_op = tmpstr; + sprintf(tmpstr,"curr_gb%d",g); string lhs_op = tmpstr; sprintf(tmpstr,"gbval->gb_var%d",g); string rhs_op = tmpstr; if(first_one){first_one = false;} else {ret += ") && (";} - ret += generate_equality_test(lhs_op, rhs_op, gdt); + ret += generate_lt_test(lhs_op, rhs_op, gdt); } } ret += ") ) ){\n"; for(g=0;gis_temporal()){ - if(gdt->is_buffer_type()){ + temporal_gb = g; + if(gdt->is_buffer_type()){ // TODO first, last? or delete? sprintf(tmpstr,"\t\t%s(&(gbval->gb_var%d),&last_gb%d);\n",gdt->get_hfta_buffer_replace().c_str(),g,g); }else{ - sprintf(tmpstr,"\t\tlast_flushed_gb%d = last_gb%d;\n",g,g); - ret += tmpstr; - sprintf(tmpstr,"\t\tlast_gb%d = gbval->gb_var%d;\n",g,g); + ret += "\t\tif(curr_gb"+to_string(g)+"==0){\n"; + ret += "\t\t\tlast_gb"+to_string(g)+" = gbval->gb_var"+to_string(g)+";\n"; + ret += "\t\t}else{\n"; + ret += "\t\t\tlast_gb"+to_string(g)+" = curr_gb"+to_string(g)+";\n"; + ret += "\t\t}\n"; + sprintf(tmpstr,"\t\tcurr_gb%d = gbval->gb_var%d;\n",g,g); } ret += tmpstr; } } - ret += "\t\tneeds_temporal_flush=true;\n"; + ret += "\t\tneeds_temporal_flush = curr_gb"+to_string (temporal_gb)+" - last_gb"+to_string(temporal_gb)+";\n"; ret += "\t\t}else{\n" - "\t\t\tneeds_temporal_flush=false;\n" + "\t\t\tneeds_temporal_flush=0;\n" "\t\t}\n"; } @@ -12703,13 +13262,22 @@ aggr_table_entry *ate = aggr_tbl.agr_tbl[a]; //--------------------------------------------------- // Flush test - ret += "\tbool flush_needed(){\n"; + ret += "gs_int32_t flush_needed(){\n"; if(uses_temporal_flush){ - ret += "\t\treturn needs_temporal_flush;\n"; + ret += "\treturn needs_temporal_flush;\n"; }else{ - ret += "\t\treturn false;\n"; + ret += "\treturn 0;\n"; } - ret += "\t};\n"; + ret += "};\n"; + +//------------------------------------------------ +// time bucket management + ret += "void advance_last_tb(){\n"; + ret += "\tlast_gb"+to_string(temporal_gb)+"++;\n"; + ret += "}\n\n"; + ret += "void reset_last_tb(){\n"; + ret += "\tlast_gb"+to_string(temporal_gb)+" = curr_gb"+to_string(temporal_gb)+";\n"; + ret += "}\n\n"; //--------------------------------------------------- // create output tuple @@ -12942,7 +13510,7 @@ aggr_table_entry *ate = aggr_tbl.agr_tbl[a]; if(sdt->is_temporal()){ sprintf(tmpstr,"\ttuple->tuple_var%d = ",s); ret += tmpstr; - sprintf(tmpstr,"(flush_finished) ? %s : %s ", generate_se_code(select_list[s]->se,schema).c_str(), generate_se_code_fm_aggr(select_list[s]->se,"last_flushed_gb", "", schema).c_str()); + sprintf(tmpstr,"(flush_finished) ? %s : %s ", generate_se_code(select_list[s]->se,schema).c_str(), generate_se_code_fm_aggr(select_list[s]->se,"last_gb", "", schema).c_str()); ret += tmpstr; ret += ";\n"; } @@ -13470,6 +14038,29 @@ void filter_join_qpn::create_protocol_se(vector q_sources, table_list } } +void watch_join_qpn::create_protocol_se(vector q_sources, table_list *Schema){ + int i; + vector *> src_vec; + + for(i=0;iget_protocol_se()); + else + src_vec.push_back(NULL); + } + + for(i=0;iname] = resolve_protocol_se(select_list[i]->se,src_vec,NULL,Schema); + } + + for(i=0;ipr->get_left_se(),src_vec,NULL,Schema)); + hash_src_r.push_back(resolve_protocol_se(hash_eq[kfld]->pr->get_right_se(),src_vec,NULL,Schema)); + } +} + + void sgah_qpn::create_protocol_se(vector q_sources, table_list *Schema){ int i; vector *> src_vec; @@ -13558,7 +14149,7 @@ void mrg_qpn::create_protocol_se(vector q_sources, table_list *Schema for(s=1;s q_sources, table_list *Schema protocol_map[fld_nm] = NULL; } } + +void watch_tbl_qpn::create_protocol_se(vector q_sources, table_list *Schema){ + return; +} +