GenericGenerator.h

Go to the documentation of this file.

 
#pragma once
#include <memory>
#include <queue>
#include <string>
#include <unordered_map>
 
#include "GenericQueryBuilder.h"
#include "GroupByManager.h"
#include "interfaces/ConditionQuery.h"
#include "interfaces/DataGeneratorInterface.h"
#include "interfaces/Validators.h"
 
namespace GenericQuery {
using ColumnStore::DataRecord;
using ColumnStore::DataSource;
using ColumnStore::DataType;
using ColumnStore::DataValue;
using ColumnStore::RecordValidator;
 
class GenericDataGenerator : public ColumnStore::DataGeneratorInterface {
    GenericQueryBuilder builder;
    RecordValidator recordValidator;
    DataSource baseSource;
 
    std::vector<int> baseIndices;
 
    struct DataGeneraterJoin {
        std::vector<int> indices;
        int FKIndex;
        JoinValueManager manager;
        std::unordered_map<JoinValue, DataRecord, JoinHash> joinMap;
 
        DataGeneraterJoin(std::string name, Metadata metadata)
            : manager(name, metadata) {}
    };
 
    bool _hasNext = false;
 
    DataRecord nextRecord;
 
    std::vector<DataGeneraterJoin> joins;
 
    DataRecord _getNext();
 
    void _setNext();
 
    DataRecord _getCandidateNext();
 
   public:
    GenericDataGenerator(GenericQueryBuilder builder);
    bool hasNext();
 
    DataRecord next();
};
 
class GenericDataAggregator : public ColumnStore::DataGeneratorInterface {
    bool _hasNext = false;
 
    DataRecord record;
 
   public:
    GenericDataAggregator(GenericQueryBuilder builder) {
        GenericDataGenerator generator(builder);
 
        auto generatorMetadata = generator.getMetadata();
 
        auto &aggregations = builder.aggregations;
 
        std::vector<ColumnStore::Column> columns;
        for (auto &aggregate : aggregations)
            columns.emplace_back(aggregate->getColumnName(),
                                 ColumnStore::DataType::FLOAT);
 
        metadata = Metadata(new DataRecordMetadata(columns));
 
        if (!generator.hasNext()) return;
 
        for (auto &aggregate : aggregations)
            aggregate->initialize(generatorMetadata);
 
        while (generator.hasNext()) {
            auto record = generator.next();
            for (auto &aggregate : aggregations) aggregate->addRecord(record);
        }
 
        std::vector<ColumnStore::DataValue> values;
 
        for (auto &aggregate : aggregations)
            values.emplace_back(aggregate->getValue());
 
        record = ColumnStore::DataRecord(values);
        _hasNext = true;
    }
    bool hasNext() { return _hasNext; }
 
    DataRecord next() {
        _hasNext = false;
        return record;
    }
};
 
class GenericGroupByAggregator : public ColumnStore::DataGeneratorInterface {
    std::queue<ColumnStore::DataRecord> records;
 
   public:
    GenericGroupByAggregator(GenericQueryBuilder builder) {
        GenericDataGenerator generator(builder);
 
        auto generatorMetadata = generator.getMetadata();
 
        GroupByManager manager(builder.groupBys, generatorMetadata);
 
        std::vector<std::string> &groupBys = builder.groupBys;
        auto &aggregations = builder.aggregations;
 
        std::vector<ColumnStore::Column> columns;
 
        for (auto &grp : groupBys)
            columns.push_back(generatorMetadata->getColumn(grp));
 
        for (auto &aggregate : aggregations)
            columns.emplace_back(aggregate->getColumnName(),
                                 ColumnStore::DataType::FLOAT);
 
        metadata = Metadata(new DataRecordMetadata(columns));
 
        std::unordered_map<GroupByValue,
                           std::vector<ColumnStore::AggregatorQuery>, GroupHash>
            map;
 
        for (auto &aggregate : aggregations)
            aggregate->initialize(generatorMetadata);
 
        auto duplicateAggregators = [&]() {
            std::vector<ColumnStore::AggregatorQuery> temp;
            for (auto &agg : aggregations) temp.push_back(agg->clone());
            return temp;
        };
 
        while (generator.hasNext()) {
            auto record = generator.next();
            auto groupValue = manager.processRecord(record);
            if (!map.count(groupValue))
                map[groupValue] = duplicateAggregators();
 
            auto &aggregations1 = map[groupValue];
 
            for (auto &aggregate : aggregations1) aggregate->addRecord(record);
        }
 
        for (auto &[groupValue, aggregations1] : map) {
            std::vector<ColumnStore::DataValue> values = groupValue.values;
            for (auto &aggregate : aggregations1)
                values.emplace_back(aggregate->getValue());
            records.emplace(values);
        }
    }
    bool hasNext() { return !records.empty(); }
 
    DataRecord next() {
        auto temp = records.front();
        records.pop();
        return temp;
    }
};
 
}  // namespace GenericQuery