AggregateCompBase.h

Go to the documentation of this file.

//
// Created by dimitrije on 3/16/18.
//

#ifndef PDB_BASEAGGREGATECOMP_H
#define PDB_BASEAGGREGATECOMP_H

#include "AbstractAggregateComp.h"

namespace pdb {
template<class OutputClass, class InputClass, class KeyClass, class ValueClass>
class AggregateCompBase : public AbstractAggregateComp {
 public:

  virtual Lambda<KeyClass> getKeyProjection(Handle<InputClass> aggMe) = 0;

  virtual Lambda<ValueClass> getValueProjection(Handle<InputClass> aggMe) = 0;

  void setOutput(std::string dbName, std::string setName) override {
    this->materializeAggOut = true;
    this->outputSetScanner = makeObject<ScanUserSet<OutputClass>>();
    this->outputSetScanner->setBatchSize(batchSize);
    this->outputSetScanner->setDatabaseName(dbName);
    this->outputSetScanner->setSetName(setName);
    this->whereHashTableSitsForThePartition = nullptr;
  }

  void setHashTablePointer(void *hashTablePointer) {
    this->whereHashTableSitsForThePartition = hashTablePointer;
  }

  void extractLambdas(std::map<std::string, GenericLambdaObjectPtr> &returnVal) override {
    int suffix = 0;
    Handle<InputClass> checkMe = nullptr;
    Lambda<KeyClass> keyLambda = getKeyProjection(checkMe);
    Lambda<ValueClass> valueLambda = getValueProjection(checkMe);
    keyLambda.toMap(returnVal, suffix);
    valueLambda.toMap(returnVal, suffix);
  }

  ComputeSinkPtr getComputeSink(TupleSpec &consumeMe,
                                TupleSpec &projection,
                                ComputePlan &plan) override {

    if (this->isUsingCombiner()) {
      return std::make_shared<ShuffleSink<KeyClass, ValueClass>>(numPartitions, consumeMe, projection);
    } else {
      if (numNodes == 0) {
        std::cout << "ERROR: cluster has 0 node" << std::endl;
        return nullptr;
      }
      if (numPartitions < numNodes) {
        std::cout << "ERROR: each node must have at least one partition" << std::endl;
        return nullptr;
      }
      return std::make_shared<CombinedShuffleSink<KeyClass, ValueClass>>(
          numPartitions / numNodes, numNodes, consumeMe, projection);
    }
  }

  ComputeSourcePtr getComputeSource(TupleSpec &outputScheme, ComputePlan &plan) override {
    // materialize aggregation result to user set
    if (this->materializeAggOut) {
      if (outputSetScanner != nullptr) {
        return outputSetScanner->getComputeSource(outputScheme, plan);
      }
      return nullptr;

      // not materialize aggregation result, keep them in hash table
    } else {
      if (whereHashTableSitsForThePartition != nullptr) {
        Handle<Object> myHashTable =
            ((Record<Object> *) whereHashTableSitsForThePartition)->getRootObject();
        std::cout << "ClusterAggregate: getComputeSource: BATCHSIZE=" << batchSize
                  << std::endl;
        return std::make_shared<MapTupleSetIterator<KeyClass, ValueClass, OutputClass>>(
            myHashTable, batchSize);
      }
      return nullptr;
    }
  }

  SimpleSingleTableQueryProcessorPtr getCombinerProcessor(
      std::vector<HashPartitionID> partitions) override {
    return make_shared<CombinerProcessor<KeyClass, ValueClass>>(partitions);
  }

  SimpleSingleTableQueryProcessorPtr getAggregationProcessor(HashPartitionID id) override {
    return make_shared<AggregationProcessor<KeyClass, ValueClass>>(id);
  }

  SimpleSingleTableQueryProcessorPtr getAggOutProcessor() override {
    return make_shared<AggOutProcessor<OutputClass, KeyClass, ValueClass>>();
  }

  void setIterator(PageCircularBufferIteratorPtr iterator) override {
    this->outputSetScanner->setIterator(iterator);
  }

  void setProxy(DataProxyPtr proxy) override {
    this->outputSetScanner->setProxy(proxy);
  }

  void setDatabaseName(std::string dbName) override {
    this->outputSetScanner->setDatabaseName(dbName);
  }

  void setSetName(std::string setName) override {
    this->outputSetScanner->setSetName(setName);
  }

  std::string getDatabaseName() override {
    return this->outputSetScanner->getDatabaseName();
  }

  std::string getSetName() override {
    return this->outputSetScanner->getSetName();
  }

  std::string getComputationType() override {
    return std::string("ClusterAggregationComp");
  }

  ComputationTypeID getComputationTypeID() override {
    return ClusterAggregationCompTypeID;
  }

  std::string getOutputType() override {
    return getTypeName<OutputClass>();
  }

  int getNumInputs() override {
    return 1;
  }

  std::string getIthInputType(int i) override {
    if (i == 0) {
      return getTypeName<InputClass>();
    } else {
      return "";
    }
  }

  void setCollectAsMap(bool collectAsMapOrNot) override {
    this->collectAsMapOrNot = collectAsMapOrNot;
  }

  bool isCollectAsMap() override {
    return this->collectAsMapOrNot;
  }

  int getNumNodesToCollect() override {
    return this->numNodesToCollect;
  }

  void setNumNodesToCollect(int numNodesToCollect) override {
    this->numNodesToCollect = numNodesToCollect;
  }

  std::string toTCAPString(std::vector<InputTupleSetSpecifier> &inputTupleSets,
                           int computationLabel,
                           std::string &outputTupleSetName,
                           std::vector<std::string> &outputColumnNames,
                           std::string &addedOutputColumnName) override {

    if (inputTupleSets.empty()) {
      return "";
    }

    InputTupleSetSpecifier inputTupleSet = inputTupleSets[0];
    std::vector<std::string> childrenLambdaNames;
    std::string myLambdaName;

    return toTCAPString(inputTupleSet.getTupleSetName(),
                        inputTupleSet.getColumnNamesToKeep(),
                        inputTupleSet.getColumnNamesToApply(),
                        childrenLambdaNames,
                        computationLabel,
                        outputTupleSetName,
                        outputColumnNames,
                        addedOutputColumnName,
                        myLambdaName);
  }

  std::string toTCAPString(std::string inputTupleSetName,
                           std::vector<std::string> &inputColumnNames,
                           std::vector<std::string> &inputColumnsToApply,
                           std::vector<std::string> &childrenLambdaNames,
                           int computationLabel,
                           std::string &outputTupleSetName,
                           std::vector<std::string> &outputColumnNames,
                           std::string &addedOutputColumnName,
                           std::string &myLambdaName) {

    PDB_COUT << "To GET TCAP STRING FOR CLUSTER AGGREGATE COMP" << std::endl;

    PDB_COUT << "To GET TCAP STRING FOR AGGREGATE KEY" << std::endl;
    Handle<InputClass> checkMe = nullptr;
    Lambda<KeyClass> keyLambda = getKeyProjection(checkMe);
    std::string tupleSetName;
    std::vector<std::string> columnNames;
    std::string addedColumnName;
    int lambdaLabel = 0;

    std::vector<std::string> columnsToApply;
    for (const auto &i : inputColumnsToApply) {
      columnsToApply.push_back(i);
    }

    std::string tcapString;
    tcapString += "\n/* Extract key for aggregation */\n";
    tcapString += keyLambda.toTCAPString(inputTupleSetName,
                                         inputColumnNames,
                                         inputColumnsToApply,
                                         childrenLambdaNames,
                                         lambdaLabel,
                                         getComputationType(),
                                         computationLabel,
                                         tupleSetName,
                                         columnNames,
                                         addedColumnName,
                                         myLambdaName,
                                         false);

    PDB_COUT << "To GET TCAP STRING FOR AGGREGATE VALUE" << std::endl;

    Lambda<ValueClass> valueLambda = getValueProjection(checkMe);
    std::vector<std::string> columnsToKeep;
    columnsToKeep.push_back(addedColumnName);

    tcapString += "\n/* Extract value for aggregation */\n";
    tcapString += valueLambda.toTCAPString(tupleSetName,
                                           columnsToKeep,
                                           columnsToApply,
                                           childrenLambdaNames,
                                           lambdaLabel,
                                           getComputationType(),
                                           computationLabel,
                                           outputTupleSetName,
                                           outputColumnNames,
                                           addedOutputColumnName,
                                           myLambdaName,
                                           false);


    // create the data for the filter
    mustache::data clusterAggCompData;
    clusterAggCompData.set("computationType", getComputationType());
    clusterAggCompData.set("computationLabel", std::to_string(computationLabel));
    clusterAggCompData.set("outputTupleSetName", outputTupleSetName);
    clusterAggCompData.set("addedColumnName", addedColumnName);
    clusterAggCompData.set("addedOutputColumnName", addedOutputColumnName);

    // set the new tuple set name
    mustache::mustache newTupleSetNameTemplate{"aggOutFor{{computationType}}{{computationLabel}}"};
    std::string newTupleSetName = newTupleSetNameTemplate.render(clusterAggCompData);

    // set new added output columnName 1
    mustache::mustache newAddedOutputColumnName1Template{"aggOutFor{{computationLabel}}"};
    std::string addedOutputColumnName1 = newAddedOutputColumnName1Template.render(clusterAggCompData);

    clusterAggCompData.set("addedOutputColumnName1", addedOutputColumnName1);

    tcapString += "\n/* Apply aggregation */\n";

    mustache::mustache aggregateTemplate{"aggOutFor{{computationType}}{{computationLabel}} ({{addedOutputColumnName1}})"
                                         "<= AGGREGATE ({{outputTupleSetName}}({{addedColumnName}}, {{addedOutputColumnName}}),"
                                         "'{{computationType}}_{{computationLabel}}')\n"};

    tcapString += aggregateTemplate.render(clusterAggCompData);

    // update the state of the computation
    outputTupleSetName = newTupleSetName;
    outputColumnNames.clear();
    outputColumnNames.push_back(addedOutputColumnName1);

    this->setTraversed(true);
    this->setOutputTupleSetName(outputTupleSetName);
    this->setOutputColumnToApply(addedOutputColumnName1);
    addedOutputColumnName = addedOutputColumnName1;

    return tcapString;
  }

  Handle<ScanUserSet<OutputClass>> &getOutputSetScanner() {
    return outputSetScanner;
  }

 protected:

  Handle<ScanUserSet<OutputClass>> outputSetScanner = nullptr;

  bool collectAsMapOrNot = false;

  int numNodesToCollect = 1;
};

}

#endif //PDB_BASEAGGREGATECOMP_H