PartitionCompBase.h

Go to the documentation of this file.

/*****************************************************************************
 *                                                                           *
 *  Copyright 2018 Rice University                                           *
 *                                                                           *
 *  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 PDB_PARTITIONCOMPBASE_H
#define PDB_PARTITIONCOMPBASE_H

#include "Computation.h"
#include "VectorSink.h"
#include "ScanUserSet.h"
#include "TypeName.h"
#include "AbstractPartitionComp.h"
#include "HashPartitionSink.h"

namespace pdb {
template<class KeyClass, class ValueClass>
class PartitionCompBase : public AbstractPartitionComp<KeyClass, ValueClass> {

 public:


  virtual Lambda<KeyClass> getProjection(Handle<ValueClass> checkMe) = 0;


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



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

  ComputationTypeID getComputationTypeID() override {
    return PartitionCompTypeID;
  }

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


  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) override {

    PDB_COUT << "ABOUT TO GET TCAP STRING FOR PARTITION" << std::endl;
    Handle<ValueClass> checkMe = nullptr;
    std::string tupleSetName;
    std::vector<std::string> columnNames;
    std::string addedColumnName;
    int lambdaLabel = 0;


    std::string tcapString;

    PDB_COUT << "TO GET TCAP STRING FOR PROJECTION LAMBDA\n";
    Lambda<KeyClass> projectionLambda = getProjection(checkMe);

    tcapString += "\n/* Apply projection */\n";
    tcapString += projectionLambda.toTCAPString(inputTupleSetName,
                                                inputColumnNames,
                                                inputColumnsToApply,
                                                childrenLambdaNames,
                                                lambdaLabel,
                                                getComputationType(),
                                                computationLabel,
                                                tupleSetName,
                                                columnNames,
                                                addedColumnName,
                                                myLambdaName,
                                                false);

    PDB_COUT << "TO REMOVE THE KEY COLUMN\n";

    mustache::data partitionCompTCAP;
    partitionCompTCAP.set("computationType", getComputationType());
    partitionCompTCAP.set("computationLabel", std::to_string(computationLabel));
    partitionCompTCAP.set("tupleSetName", tupleSetName);
    partitionCompTCAP.set("addedColumnName", addedColumnName);

    //set the output tuple set name
    mustache::mustache outputTupleSetNameTemplate{"partitionOutFor{{computationType}}{{computationLabel}}"};
    outputTupleSetName = outputTupleSetNameTemplate.render(partitionCompTCAP);

    partitionCompTCAP.set("outputTupleSetName", outputTupleSetName);

    // set the added output column
    addedOutputColumnName = inputColumnsToApply[0];

    partitionCompTCAP.set("addedOutputColumnName", addedOutputColumnName);

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

    mustache::mustache partitionTCAPTemplate{"{{outputTupleSetName}} ({{addedOutputColumnName}})"
                                         " <= PARTITION ({{tupleSetName}}({{addedOutputColumnName}}, {{addedColumnName}}),"
                                         "'{{computationType}}_{{computationLabel}}')\n"};

    tcapString += partitionTCAPTemplate.render(partitionCompTCAP);

    // update the state of the computation
    outputColumnNames.clear();
    outputColumnNames.push_back(addedOutputColumnName);


    // update the state of the computation
    this->setTraversed(true);
    this->setOutputTupleSetName(outputTupleSetName);
    this->setOutputColumnToApply(addedOutputColumnName);

    // return the TCAP string
    return tcapString;
  }


  void setOutput(std::string dbName, std::string setName) override {
    this->materializeSelectionOut = true;
    this->outputSetScanner = makeObject<ScanUserSet<ValueClass>>();
    this->outputSetScanner->setDatabaseName(dbName);
    this->outputSetScanner->setSetName(setName);
  }

  void setBatchSize(int batchSize) override {
    if (this->outputSetScanner != nullptr) {
      this->outputSetScanner->setBatchSize(batchSize);
    }
  }

  std::string getDatabaseName() override {
    if (this->outputSetScanner != nullptr) {
      return this->outputSetScanner->getDatabaseName();
    } else {
      return "";
    }
  }

  std::string getSetName() override {
    if (this->outputSetScanner != nullptr) {
      return this->outputSetScanner->getSetName();
    } else {
      return "";
    }
  }

  ComputeSourcePtr getComputeSource(TupleSpec &outputScheme, ComputePlan &plan) override {

    if (this->outputSetScanner != nullptr) {
        std::cout << "database name is " << this->outputSetScanner->getDatabaseName() << std::endl;
        std::cout << "set name is " << this->outputSetScanner->getSetName() << std::endl;
        return outputSetScanner->getComputeSource(outputScheme, plan);
    }
    std::cout << "ERROR: get compute source for " << outputScheme << " returns nullptr" << std::endl;
    return nullptr;
  }




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

      return std::make_shared<HashPartitionSink<KeyClass, ValueClass>>(this->numPartitions, this->numNodes, consumeMe, projection);
  }


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


private:

  Handle<ScanUserSet<ValueClass>> outputSetScanner = nullptr;

};

}

#endif //PDB_PARTITIONCOMPBASE_H