susumu.yata
null+****@clear*****
Wed Sep 24 11:22:34 JST 2014
susumu.yata 2014-09-24 11:22:34 +0900 (Wed, 24 Sep 2014) New Revision: 1ec429d7c45f853361c1aa922bf9551cc322ca0b https://github.com/groonga/grnxx/commit/1ec429d7c45f853361c1aa922bf9551cc322ca0b Message: Add XorMerger. (#66) Modified files: lib/grnxx/merger.cpp Modified: lib/grnxx/merger.cpp (+177 -3) =================================================================== --- lib/grnxx/merger.cpp 2014-09-24 11:05:50 +0900 (e24e3e4) +++ lib/grnxx/merger.cpp 2014-09-24 11:22:34 +0900 (f975111) @@ -330,7 +330,182 @@ bool OrMerger::finish(Error *error) { // -- XorMerger -- -// TODO +class XorMerger : public Merger { + public: + ~XorMerger() {} + + static unique_ptr<Merger> create(Error *error, const MergerOptions &options); + + bool reset(Error *error, + Array<Record> *input_records_1, + Array<Record> *input_records_2, + Array<Record> *output_records); + + bool finish(Error *error); + + private: + Array<Record> *input_records_1_; + Array<Record> *input_records_2_; + Array<Record> *output_records_; + MergerOperatorType operator_type_; + Int offset_; + Int limit_; + + XorMerger(MergerOperatorType operator_type, Int offset, Int limit) + : Merger(), + input_records_1_(nullptr), + input_records_2_(nullptr), + output_records_(nullptr), + operator_type_(operator_type), + offset_(offset), + limit_(limit) {} +}; + +unique_ptr<Merger> XorMerger::create(Error *error, + const MergerOptions &options) { + unique_ptr<Merger> merger( + new (nothrow) XorMerger(options.operator_type, + options.offset, + options.limit)); + if (!merger) { + GRNXX_ERROR_SET(error, NO_MEMORY, "Memory allocation failed"); + return nullptr; + } + return merger; +} + +bool XorMerger::reset(Error *, + Array<Record> *input_records_1, + Array<Record> *input_records_2, + Array<Record> *output_records) { + input_records_1_ = input_records_1; + input_records_2_ = input_records_2; + output_records_ = output_records; + return true; +} + +bool XorMerger::finish(Error *error) { + // Create a hash table from the smaller input. + Array<Record> *filter_records; + Array<Record> *stream_records; + if (input_records_1_->size() < input_records_2_->size()) { + filter_records = input_records_1_; + stream_records = input_records_2_; + } else { + filter_records = input_records_2_; + stream_records = input_records_1_; + } + std::unordered_map<Int, Float> filter; + for (Int i = 0; i < filter_records->size(); ++i) try { + filter[filter_records->get_row_id(i)] = filter_records->get_score(i); + } catch (...) { + GRNXX_ERROR_SET(error, NO_MEMORY, "Memory allocation failed"); + return false; + } + + // Filter the stream (the larger input) with the hash table. + const MergerOperatorType operator_type = operator_type_; + const bool stream_is_1 = stream_records == input_records_1_; + for (Int i = 0; i < stream_records->size(); ++i) { + auto it = filter.find(stream_records->get_row_id(i)); + if (it != filter.end()) { + filter.erase(it); + } else { + Record record; + record.row_id = stream_records->get_row_id(i); + switch (operator_type) { + case PLUS_MERGER_OPERATOR: { + record.score = stream_records->get_score(i); + break; + } + case MINUS_MERGER_OPERATOR: { + record.score = stream_records->get_score(i); + if (!stream_is_1) { + record.score = -record.score; + } + break; + } + case MULTIPLICATION_MERGER_OPERATOR: { + // TODO: I'm not sure if stream_records->get_score(i) should be used? + record.score = 0.0; + break; + } + case LHS_MERGER_OPERATOR: { + if (stream_is_1) { + record.score = stream_records->get_score(i); + } else { + record.score = 0.0; + } + break; + } + case RHS_MERGER_OPERATOR: { + if (stream_is_1) { + record.score = 0.0; + } else { + record.score = stream_records->get_score(i); + } + break; + } + case ZERO_MERGER_OPERATOR: { + record.score = 0.0; + break; + } + } + if (!output_records_->push_back(error, record)) { + return false; + } + } + } + + for (auto it : filter) { + switch (operator_type) { + case PLUS_MERGER_OPERATOR: { + break; + } + case MINUS_MERGER_OPERATOR: { + it.second = stream_is_1 ? -it.second : 0.0; + } + case MULTIPLICATION_MERGER_OPERATOR: { + // TODO: I'm not sure if it.second should be used? + it.second = 0.0; + break; + } + case LHS_MERGER_OPERATOR: { + if (stream_is_1) { + it.second = 0.0; + } + break; + } + case RHS_MERGER_OPERATOR: { + if (!stream_is_1) { + it.second = 0.0; + } + break; + } + case ZERO_MERGER_OPERATOR: { + it.second = 0.0; + break; + } + } + if (!output_records_->push_back(error, Record(it.first, it.second))) { + return false; + } + } + + // Remove out-of-range records. + if (offset_ > 0) { + for (Int i = offset_; i < output_records_->size(); ++i) { + output_records_->set(i - offset_, output_records_->get(i)); + } + output_records_->resize(nullptr, output_records_->size() - offset_); + } + if (limit_ < output_records_->size()) { + output_records_->resize(nullptr, limit_); + } + input_records_1_->clear(); + input_records_2_->clear(); + return true; +} // -- MinusMerger -- @@ -762,8 +937,7 @@ unique_ptr<Merger> Merger::create(Error *error, const MergerOptions &options) { return OrMerger::create(error, options); } case XOR_MERGER: { - GRNXX_ERROR_SET(error, NOT_SUPPORTED_YET, "Not supported yet"); - return nullptr; + return XorMerger::create(error, options); } case MINUS_MERGER: { return MinusMerger::create(error, options); -------------- next part -------------- HTML����������������������������...Download