PartitionTransformationCompBase.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 PARTITION_TRANSFORMATION_COMP_BASE_H
#define PARTITION_TRANSFORMATION_COMP_BASE_H

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

namespace pdb {
template<class KeyClass, class ValueClass>
class PartitionTransformationCompBase : 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("PartitionTransformationComp");
  }

  ComputationTypeID getComputationTypeID() override {
    return PartitionTransformationCompTypeID;
  }


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


  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 SELECTION" << 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,
                                                outputTupleSetName,
                                                outputColumnNames,
                                                addedOutputColumnName,
                                                myLambdaName,
                                                true);

    // 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<KeyClass>>();
    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->materializeSelectionOut) {
      if (this->outputSetScanner != nullptr) {
        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 {

    if (this->materializeSelectionOut) {
      return std::make_shared<HashPartitionTransformationSink<KeyClass>>(this->numPartitions, this->numNodes, consumeMe, projection);
    }
    return nullptr;
  }


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

private:

  Handle<ScanUserSet<KeyClass>> outputSetScanner = nullptr;

};

}

#endif //PDB_PARTITIONCOMPBASE_H