-/* ------------------------------------------------
-Copyright 2014 AT&T Intellectual Property
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
- ------------------------------------------- */
-
-#ifndef GROUPBY_OPERATOR_H
-#define GROUPBY_OPERATOR_H
-
-#include "host_tuple.h"
-#include "base_operator.h"
-#include <list>
-#include "hash_table.h"
-
-#define _GB_FLUSH_PER_TUPLE_ 1
-
-using namespace std;
-
-template <class groupby_func, class group, class aggregate, class hasher_func, class equal_func>
-class groupby_operator : public base_operator {
-private :
- groupby_func func;
- hash_table<group*, aggregate*, hasher_func, equal_func> group_table[2];
- bool flush_finished;
- unsigned int curr_table;
- typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator flush_pos;
- int n_patterns;
-
-
-
-public:
- groupby_operator(int schema_handle, const char* name) : base_operator(name), func(schema_handle) {
- flush_finished = true;
- curr_table = 0;
- flush_pos = group_table[1-curr_table].end();
- n_patterns = func.n_groupby_patterns();
- }
-
- int accept_tuple(host_tuple& tup, list<host_tuple>& result) {
-
-// Push out completed groups
-
- if(!flush_finished) partial_flush(result);
-
- // create buffer on the stack to store key object
- char buffer[sizeof(group)];
-
- // extract the key information from the tuple and
- // copy it into buffer
- group* grp = func.create_group(tup, buffer);
- if (!grp) {
-/*
-// Ignore temp tuples until we can fix their timestamps.
-if (func.temp_status_received()) {
- tup.free_tuple();
- return 0;
-}*/
- if (func.flush_needed()){
- flush_old(result);
- }
- if (func.temp_status_received()) {
- host_tuple temp_tup;
- if (!func.create_temp_status_tuple(temp_tup, flush_finished)) {
- temp_tup.channel = output_channel;
- result.push_back(temp_tup);
- }
- }
- tup.free_tuple();
- return 0;
- }
-
- typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator iter;
- if ((iter = group_table[curr_table].find(grp)) != group_table[curr_table].end()) {
-// Temporal GBvar is part of the group so no flush is needed.
- aggregate* old_aggr = (*iter).second;
- func.update_aggregate(tup, grp, old_aggr);
- }else{
- if (func.flush_needed()) {
- flush_old(result);
- }
- if(n_patterns <= 1){
- // create a copy of the group on the heap
- group* new_grp = new group(grp); // need a copy constructor for groups
-// aggregate* aggr = (aggregate*)malloc(sizeof(aggregate));
- aggregate* aggr = new aggregate();
- // create an aggregate in preallocated buffer
- aggr = func.create_aggregate(tup, (char*)aggr);
-
- group_table[curr_table].insert(new_grp, aggr);
- }else{
- int p;
- for(p=0;p<n_patterns;++p){
- group* new_grp = new group(grp, func.get_pattern(p));
- aggregate* aggr = new aggregate();
- aggr = func.create_aggregate(tup, (char*)aggr);
- group_table[curr_table].insert(new_grp, aggr);
- }
- }
- }
- tup.free_tuple();
- return 0;
- }
-
- int partial_flush(list<host_tuple>& result) {
- host_tuple tup;
- unsigned int old_table = 1-curr_table;
- unsigned int i;
-
-// emit up to _GB_FLUSH_PER_TABLE_ output tuples.
- if (!group_table[old_table].empty()) {
- for (i=0; flush_pos != group_table[old_table].end() && i<_GB_FLUSH_PER_TUPLE_; ++flush_pos, ++i) {
- bool failed = false;
- tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);
- if (!failed) {
- tup.channel = output_channel;
- result.push_back(tup);
- }
- delete ((*flush_pos).first);
- delete ((*flush_pos).second);
-// free((*flush_pos).second);
- }
- }
-
-// Finalize processing if empty.
- if(flush_pos == group_table[old_table].end()) {
- flush_finished = true;
- group_table[old_table].clear();
- group_table[old_table].rehash();
- }
- return 0;
- }
-
- int flush(list<host_tuple>& result) {
- host_tuple tup;
- typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator iter;
- unsigned int old_table = 1-curr_table;
-
-// If the old table isn't empty, flush it now.
- if (!group_table[old_table].empty()) {
- for (; flush_pos != group_table[old_table].end(); ++flush_pos) {
- bool failed = false;
- tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);
- if (!failed) {
-
- tup.channel = output_channel;
- result.push_back(tup);
- }
- delete ((*flush_pos).first);
- delete ((*flush_pos).second);
-// free((*flush_pos).second);
- }
- group_table[old_table].clear();
- group_table[old_table].rehash();
- }
-
- flush_pos = group_table[curr_table].begin();
-// If the old table isn't empty, flush it now.
- if (!group_table[curr_table].empty()) {
- for (; flush_pos != group_table[curr_table].end(); ++flush_pos) {
- bool failed = false;
- tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);
- if (!failed) {
-
- tup.channel = output_channel;
- result.push_back(tup);
- }
- delete ((*flush_pos).first);
- delete ((*flush_pos).second);
-// free((*flush_pos).second);
- }
- group_table[curr_table].clear();
- }
-
- flush_finished = true;
-
- return 0;
- }
-
- int flush_old(list<host_tuple>& result) {
- host_tuple tup;
- typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator iter;
- unsigned int old_table = 1-curr_table;
-
-// If the old table isn't empty, flush it now.
- if (!group_table[old_table].empty()) {
- for (; flush_pos != group_table[old_table].end(); ++flush_pos) {
- bool failed = false;
- tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);
- if (!failed) {
-
- tup.channel = output_channel;
- result.push_back(tup);
- }
- delete ((*flush_pos).first);
- delete ((*flush_pos).second);
-// free((*flush_pos).second);
- }
- group_table[old_table].clear();
- group_table[old_table].rehash();
- }
-
-// swap tables, enable partial flush processing.
- flush_pos = group_table[curr_table].begin();
- curr_table = old_table;
- flush_finished = false;
-
- return 0;
- }
-
- int set_param_block(int sz, void * value) {
- func.set_param_block(sz, value);
- return 0;
- }
-
- int get_temp_status(host_tuple& result) {
- result.channel = output_channel;
- return func.create_temp_status_tuple(result, flush_finished);
- }
-
- int get_blocked_status () {
- return -1;
- }
-
- unsigned int get_mem_footprint() {
- return group_table[0].get_mem_footprint() + group_table[1].get_mem_footprint();
- }
-};
-
-#endif // GROUPBY_OPERATOR_H
+/* ------------------------------------------------\r
+Copyright 2014 AT&T Intellectual Property\r
+ Licensed under the Apache License, Version 2.0 (the "License");\r
+ you may not use this file except in compliance with the License.\r
+ You may obtain a copy of the License at\r
+\r
+ http://www.apache.org/licenses/LICENSE-2.0\r
+\r
+ Unless required by applicable law or agreed to in writing, software\r
+ distributed under the License is distributed on an "AS IS" BASIS,\r
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+ See the License for the specific language governing permissions and\r
+ limitations under the License.\r
+ ------------------------------------------- */\r
+\r
+#ifndef GROUPBY_OPERATOR_H\r
+#define GROUPBY_OPERATOR_H\r
+\r
+#include "host_tuple.h"\r
+#include "base_operator.h"\r
+#include <list>\r
+#include "hash_table.h"\r
+\r
+#define _GB_FLUSH_PER_TUPLE_ 1\r
+\r
+using namespace std;\r
+\r
+template <class groupby_func, class group, class aggregate, class hasher_func, class equal_func>\r
+class groupby_operator : public base_operator {\r
+private :\r
+ groupby_func func;\r
+ hash_table<group*, aggregate*, hasher_func, equal_func> group_table[2];\r
+ bool flush_finished;\r
+ unsigned int curr_table;\r
+ typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator flush_pos;\r
+ int n_patterns;\r
+\r
+\r
+\r
+public:\r
+ groupby_operator(int schema_handle, const char* name) : base_operator(name), func(schema_handle) {\r
+ flush_finished = true;\r
+ curr_table = 0;\r
+ flush_pos = group_table[1-curr_table].end();\r
+ n_patterns = func.n_groupby_patterns();\r
+ }\r
+\r
+ int accept_tuple(host_tuple& tup, list<host_tuple>& result) {\r
+\r
+// Push out completed groups\r
+\r
+ if(!flush_finished) partial_flush(result);\r
+\r
+ // create buffer on the stack to store key object\r
+ char buffer[sizeof(group)];\r
+\r
+ // extract the key information from the tuple and\r
+ // copy it into buffer\r
+ group* grp = func.create_group(tup, buffer);\r
+ if (!grp) {\r
+/*\r
+// Ignore temp tuples until we can fix their timestamps.\r
+if (func.temp_status_received()) {\r
+ tup.free_tuple();\r
+ return 0;\r
+}*/\r
+ if (func.flush_needed()){\r
+ flush_old(result);\r
+ }\r
+ if (func.temp_status_received()) {\r
+ host_tuple temp_tup;\r
+ if (!func.create_temp_status_tuple(temp_tup, flush_finished)) {\r
+ temp_tup.channel = output_channel;\r
+ result.push_back(temp_tup);\r
+ }\r
+ }\r
+ tup.free_tuple();\r
+ return 0;\r
+ }\r
+\r
+ typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator iter;\r
+ if ((iter = group_table[curr_table].find(grp)) != group_table[curr_table].end()) {\r
+// Temporal GBvar is part of the group so no flush is needed.\r
+ aggregate* old_aggr = (*iter).second;\r
+ func.update_aggregate(tup, grp, old_aggr);\r
+ }else{\r
+ if (func.flush_needed()) {\r
+ flush_old(result);\r
+ }\r
+ if(n_patterns <= 1){\r
+ // create a copy of the group on the heap\r
+ group* new_grp = new group(grp); // need a copy constructor for groups\r
+// aggregate* aggr = (aggregate*)malloc(sizeof(aggregate));\r
+ aggregate* aggr = new aggregate();\r
+ // create an aggregate in preallocated buffer\r
+ aggr = func.create_aggregate(tup, (char*)aggr);\r
+\r
+ group_table[curr_table].insert(new_grp, aggr);\r
+ }else{\r
+ int p;\r
+ for(p=0;p<n_patterns;++p){\r
+ group* new_grp = new group(grp, func.get_pattern(p));\r
+ aggregate* aggr = new aggregate();\r
+ aggr = func.create_aggregate(tup, (char*)aggr);\r
+ group_table[curr_table].insert(new_grp, aggr);\r
+ }\r
+ }\r
+ }\r
+ tup.free_tuple();\r
+ return 0;\r
+ }\r
+\r
+ int partial_flush(list<host_tuple>& result) {\r
+ host_tuple tup;\r
+ unsigned int old_table = 1-curr_table;\r
+ unsigned int i;\r
+\r
+// emit up to _GB_FLUSH_PER_TABLE_ output tuples.\r
+ if (!group_table[old_table].empty()) {\r
+ for (i=0; flush_pos != group_table[old_table].end() && i<_GB_FLUSH_PER_TUPLE_; ++flush_pos, ++i) {\r
+ bool failed = false;\r
+ tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);\r
+ if (!failed) {\r
+ tup.channel = output_channel;\r
+ result.push_back(tup);\r
+ }\r
+ delete ((*flush_pos).first);\r
+ delete ((*flush_pos).second);\r
+// free((*flush_pos).second);\r
+ }\r
+ }\r
+\r
+// Finalize processing if empty.\r
+ if(flush_pos == group_table[old_table].end()) {\r
+ flush_finished = true;\r
+ group_table[old_table].clear();\r
+ group_table[old_table].rehash();\r
+ }\r
+ return 0;\r
+ }\r
+\r
+ int flush(list<host_tuple>& result) {\r
+ host_tuple tup;\r
+ typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator iter;\r
+ unsigned int old_table = 1-curr_table;\r
+\r
+// If the old table isn't empty, flush it now.\r
+ if (!group_table[old_table].empty()) {\r
+ for (; flush_pos != group_table[old_table].end(); ++flush_pos) {\r
+ bool failed = false;\r
+ tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);\r
+ if (!failed) {\r
+\r
+ tup.channel = output_channel;\r
+ result.push_back(tup);\r
+ }\r
+ delete ((*flush_pos).first);\r
+ delete ((*flush_pos).second);\r
+// free((*flush_pos).second);\r
+ }\r
+ group_table[old_table].clear();\r
+ group_table[old_table].rehash();\r
+ }\r
+\r
+ flush_pos = group_table[curr_table].begin();\r
+// If the old table isn't empty, flush it now.\r
+ if (!group_table[curr_table].empty()) {\r
+ for (; flush_pos != group_table[curr_table].end(); ++flush_pos) {\r
+ bool failed = false;\r
+ tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);\r
+ if (!failed) {\r
+\r
+ tup.channel = output_channel;\r
+ result.push_back(tup);\r
+ }\r
+ delete ((*flush_pos).first);\r
+ delete ((*flush_pos).second);\r
+// free((*flush_pos).second);\r
+ }\r
+ group_table[curr_table].clear();\r
+ }\r
+\r
+ flush_finished = true;\r
+\r
+ return 0;\r
+ }\r
+\r
+ int flush_old(list<host_tuple>& result) {\r
+ host_tuple tup;\r
+ typename hash_table<group*, aggregate*, hasher_func, equal_func>::iterator iter;\r
+ unsigned int old_table = 1-curr_table;\r
+\r
+// If the old table isn't empty, flush it now.\r
+ if (!group_table[old_table].empty()) {\r
+ for (; flush_pos != group_table[old_table].end(); ++flush_pos) {\r
+ bool failed = false;\r
+ tup = func.create_output_tuple((*flush_pos).first,(*flush_pos).second, failed);\r
+ if (!failed) {\r
+\r
+ tup.channel = output_channel;\r
+ result.push_back(tup);\r
+ }\r
+ delete ((*flush_pos).first);\r
+ delete ((*flush_pos).second);\r
+// free((*flush_pos).second);\r
+ }\r
+ group_table[old_table].clear();\r
+ group_table[old_table].rehash();\r
+ }\r
+\r
+// swap tables, enable partial flush processing.\r
+ flush_pos = group_table[curr_table].begin();\r
+ curr_table = old_table;\r
+ flush_finished = false;\r
+\r
+ return 0;\r
+ }\r
+\r
+ int set_param_block(int sz, void * value) {\r
+ func.set_param_block(sz, value);\r
+ return 0;\r
+ }\r
+\r
+ int get_temp_status(host_tuple& result) {\r
+ result.channel = output_channel;\r
+ return func.create_temp_status_tuple(result, flush_finished);\r
+ }\r
+\r
+ int get_blocked_status () {\r
+ return -1;\r
+ }\r
+\r
+ unsigned int get_mem_footprint() {\r
+ return group_table[0].get_mem_footprint() + group_table[1].get_mem_footprint();\r
+ }\r
+};\r
+\r
+#endif // GROUPBY_OPERATOR_H\r
Code Review / com / gs-lite.git / blobdiff