GenericLambdaObject.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 LAMBDA_HELPER_H
#define LAMBDA_HELPER_H

#include <memory>
#include <vector>
#include <functional>
#include <mustache.h>
#include <mustache_helper.h>
#include "Object.h"
#include "Handle.h"
#include "Ptr.h"
#include "TupleSpec.h"
#include "ComputeExecutor.h"
#include "SimpleComputeExecutor.h"
#include "ComputeInfo.h"
#include "MultiInputsBase.h"
#include "TupleSetMachine.h"
#include "LambdaTree.h"

namespace pdb {

class GenericLambdaObject;
typedef std::shared_ptr<GenericLambdaObject> GenericLambdaObjectPtr;

class GenericLambdaObject {

 private:

  std::vector<unsigned int> inputIndexes;

 public:

  virtual ~GenericLambdaObject() = default;

  void setInputIndex(int i, unsigned int index) {
    size_t numInputs = this->getNumInputs();
    if (numInputs == 0) {
      numInputs = 1;
    }
    if (inputIndexes.size() != numInputs) {
      inputIndexes.resize(numInputs);
    }
    if (i < numInputs) {
      this->inputIndexes[i] = index;
    }
  }

  virtual unsigned int getInputIndex(int i) {
    if (i >= this->getNumInputs()) {
      return (unsigned int) (-1);
    }
    return inputIndexes[i];
  }

  virtual unsigned int getNumInputs() = 0;

  virtual ComputeExecutorPtr getExecutor(TupleSpec &inputSchema,
                                         TupleSpec &attsToOperateOn,
                                         TupleSpec &attsToIncludeInOutput) = 0;

  virtual ComputeExecutorPtr getExecutor(TupleSpec &inputSchema,
                                         TupleSpec &attsToOperateOn,
                                         TupleSpec &attsToIncludeInOutput,
                                         ComputeInfoPtr) {
    return getExecutor(inputSchema, attsToOperateOn, attsToIncludeInOutput);
  }

  virtual ComputeExecutorPtr getLeftHasher(TupleSpec &inputSchema,
                                           TupleSpec &attsToOperateOn,
                                           TupleSpec &attsToIncludeInOutput) {
    std::cout << "getLeftHasher not implemented for this type!!\n";
    exit(1);
  }

  virtual ComputeExecutorPtr getLeftHasher(TupleSpec &inputSchema,
                                           TupleSpec &attsToOperateOn,
                                           TupleSpec &attsToIncludeInOutput,
                                           ComputeInfoPtr) {
    return getLeftHasher(inputSchema, attsToOperateOn, attsToIncludeInOutput);
  }

  virtual ComputeExecutorPtr getRightHasher(TupleSpec &inputSchema,
                                            TupleSpec &attsToOperateOn,
                                            TupleSpec &attsToIncludeInOutput) {
    std::cout << "getRightHasher not implemented for this type!!\n";
    exit(1);
  }

  virtual ComputeExecutorPtr getRightHasher(TupleSpec &inputSchema,
                                            TupleSpec &attsToOperateOn,
                                            TupleSpec &attsToIncludeInOutput,
                                            ComputeInfoPtr) {
    return getRightHasher(inputSchema, attsToOperateOn, attsToIncludeInOutput);
  }

  virtual std::string getTypeOfLambda() = 0;

  virtual int getNumChildren() = 0;

  virtual GenericLambdaObjectPtr getChild(int which) = 0;

  std::string getTCAPString(const std::string &inputTupleSetName,
                            const std::vector<std::string> &inputColumnNames,
                            const std::vector<std::string> &inputColumnsToApply,
                            const std::string &outputTupleSetName,
                            const std::vector<std::string> &outputColumns,
                            const std::string &outputColumnName,
                            const std::string &tcapOperation,
                            const std::string &computationNameAndLabel,
                            const std::string &lambdaNameAndLabel,
                            const std::map<std::string, std::string> &info) {

    mustache::mustache outputTupleSetNameTemplate{"{{outputTupleSetName}}({{#outputColumns}}{{value}}{{^isLast}},{{/isLast}}{{/outputColumns}}) <= "
                                                  "{{tcapOperation}} ({{inputTupleSetName}}({{#inputColumnsToApply}}{{value}}{{^isLast}},{{/isLast}}{{/inputColumnsToApply}}), "
                                                  "{{inputTupleSetName}}({{#hasColumnNames}}{{#inputColumnNames}}{{value}}{{^isLast}},{{/isLast}}{{/inputColumnNames}}{{/hasColumnNames}}), "
                                                  "'{{computationNameAndLabel}}', "
                                                  "{{#hasLambdaNameAndLabel}}'{{lambdaNameAndLabel}}', {{/hasLambdaNameAndLabel}}"
                                                  "[{{#info}}('{{key}}', '{{value}}'){{^isLast}}, {{/isLast}}{{/info}}])\n"};

    // create the data for the output columns
    mustache::data outputColumnData = mustache::from_vector<std::string>(outputColumns);

    // create the data for the input columns to apply
    mustache::data inputColumnsToApplyData = mustache::from_vector<std::string>(inputColumnsToApply);

    // create the data for the input columns to apply
    mustache::data inputColumnNamesData = mustache::from_vector<std::string>(inputColumnNames);

    // create the info data
    mustache::data infoData = mustache::from_map(info);

    // create the data for the lambda
    mustache::data lambdaData;

    lambdaData.set("outputTupleSetName", outputTupleSetName);
    lambdaData.set("outputColumns", outputColumnData);
    lambdaData.set("tcapOperation", tcapOperation);
    lambdaData.set("inputTupleSetName", inputTupleSetName);
    lambdaData.set("inputColumnsToApply", inputColumnsToApplyData);
    lambdaData.set("hasColumnNames", !inputColumnNames.empty());
    lambdaData.set("inputColumnNames", inputColumnNamesData);
    lambdaData.set("inputTupleSetName", inputTupleSetName);
    lambdaData.set("computationNameAndLabel", computationNameAndLabel);
    lambdaData.set("hasLambdaNameAndLabel", !lambdaNameAndLabel.empty());
    lambdaData.set("lambdaNameAndLabel", lambdaNameAndLabel);
    lambdaData.set("info", infoData);

    return outputTupleSetNameTemplate.render(lambdaData);
  }

  virtual std::string toTCAPStringForCartesianJoin(int lambdaLabel,
                                                   std::string computationName,
                                                   int computationLabel,
                                                   std::string &outputTupleSetName,
                                                   std::vector<std::string> &outputColumns,
                                                   std::string &outputColumnName,
                                                   std::string &myLambdaName,
                                                   MultiInputsBase *multiInputsComp) {
    std::cout << "toTCAPStringForCartesianJoin() should not be implemented here!" << std::endl;
    exit(1);
  }

  virtual std::string toTCAPString(std::vector<std::string> &inputTupleSetNames,
                                   std::vector<std::string> &inputColumnNames,
                                   std::vector<std::string> &inputColumnsToApply,
                                   std::vector<std::string> &childrenLambdaNames,
                                   int lambdaLabel,
                                   std::string computationName,
                                   int computationLabel,
                                   std::string &outputTupleSetName,
                                   std::vector<std::string> &outputColumns,
                                   std::string &outputColumnName,
                                   std::string &myLambdaName,
                                   MultiInputsBase *multiInputsComp = nullptr,
                                   bool amIPartOfJoinPredicate = false,
                                   bool amILeftChildOfEqualLambda = false,
                                   bool amIRightChildOfEqualLambda = false,
                                   std::string parentLambdaName = "",
                                   bool isSelfJoin = false) {
    std::string tcapString;
    std::string lambdaType = getTypeOfLambda();
    if ((lambdaType.find("==") != std::string::npos) ||
        (lambdaType.find("&&") != std::string::npos)) {
      return "";
    }

    if ((lambdaType.find("native_lambda") != std::string::npos) && (multiInputsComp != nullptr)
        && amIPartOfJoinPredicate &&
        !amIRightChildOfEqualLambda
        && ((parentLambdaName.empty()) || (parentLambdaName.find("&&") != std::string::npos))) {

      return toTCAPStringForCartesianJoin(lambdaLabel,
                                          computationName,
                                          computationLabel,
                                          outputTupleSetName,
                                          outputColumns,
                                          outputColumnName,
                                          myLambdaName,
                                          multiInputsComp);
    }

    std::string computationNameWithLabel = computationName + "_" + std::to_string(computationLabel);
    myLambdaName = getTypeOfLambda() + "_" + std::to_string(lambdaLabel);
    std::string inputTupleSetName = inputTupleSetNames[0];
    std::string tupleSetMidTag = "OutFor";

    std::vector<std::string> originalInputColumnsToApply;

    int myIndex;
    if (multiInputsComp != nullptr) {
      if (amILeftChildOfEqualLambda || amIRightChildOfEqualLambda) {
        tupleSetMidTag = "Extracted";
      }
      myIndex = this->getInputIndex(0);
      PDB_COUT << myLambdaName + ": myIndex=" << myIndex << std::endl;
      inputTupleSetName = multiInputsComp->getTupleSetNameForIthInput(myIndex);
      PDB_COUT << "inputTupleSetName=" << inputTupleSetName << std::endl;
      inputColumnNames = multiInputsComp->getInputColumnsForIthInput(myIndex);

      inputColumnsToApply.clear();

      if (this->getNumInputs() == 1) {
        inputColumnsToApply.push_back(multiInputsComp->getNameForIthInput(myIndex));
        originalInputColumnsToApply.push_back(multiInputsComp->getNameForIthInput(myIndex));
      } else {
        for (int i = 0; i < this->getNumInputs(); i++) {
          int index = this->getInputIndex(i);
          inputColumnsToApply.push_back(multiInputsComp->getNameForIthInput(index));
          originalInputColumnsToApply.push_back(
              multiInputsComp->getNameForIthInput(myIndex));
        }
      }
      multiInputsComp->setLambdasForIthInputAndPredicate(
          myIndex, parentLambdaName, myLambdaName);
    }

    PDB_COUT << "input columns to apply: " << std::endl;
    for (const auto &i : originalInputColumnsToApply) {
      PDB_COUT << i << std::endl;
    }

    outputTupleSetName = lambdaType.substr(0, 5) + "_" + std::to_string(lambdaLabel) + tupleSetMidTag + computationName
        + std::to_string(computationLabel);
    outputColumnName =
        lambdaType.substr(0, 5) + "_" + std::to_string(lambdaLabel) + "_" + std::to_string(computationLabel)
            + tupleSetMidTag;

    outputColumns.clear();
    for (const auto &inputColumnName : inputColumnNames) {
      outputColumns.push_back(inputColumnName);
    }
    outputColumns.push_back(outputColumnName);

    // the additional info about this attribute access lambda
    std::map<std::string, std::string> info;

    // fill in the info
    info["lambdaType"] = getTypeOfLambda();

    tcapString += getTCAPString(inputTupleSetName,
                                inputColumnNames,
                                inputColumnsToApply,
                                outputTupleSetName,
                                outputColumns,
                                outputColumnName,
                                "APPLY",
                                computationNameWithLabel,
                                myLambdaName,
                                info);

    if (multiInputsComp != nullptr) {
      if (amILeftChildOfEqualLambda || amIRightChildOfEqualLambda) {
        inputTupleSetName = outputTupleSetName;
        inputColumnNames.clear();
        for (const auto &outputColumn : outputColumns) {
          // we want to remove the extracted value column from here
          if (outputColumn != outputColumnName) {
            inputColumnNames.push_back(outputColumn);
          }
        }
        inputColumnsToApply.clear();
        inputColumnsToApply.push_back(outputColumnName);

        std::string hashOperator;
        if (amILeftChildOfEqualLambda) {
          hashOperator = "HASHLEFT";
        } else {
          hashOperator = "HASHRIGHT";
        }
        outputTupleSetName = outputTupleSetName + "_hashed";
        outputColumnName = outputColumnName + "_hash";
        outputColumns.clear();

        for (const auto &inputColumnName : inputColumnNames) {
          outputColumns.push_back(inputColumnName);
        }
        outputColumns.push_back(outputColumnName);

        tcapString += getTCAPString(inputTupleSetName,
                                    inputColumnNames,
                                    inputColumnsToApply,
                                    outputTupleSetName,
                                    outputColumns,
                                    outputColumnName,
                                    hashOperator,
                                    computationNameWithLabel,
                                    parentLambdaName,
                                    std::map<std::string, std::string>());
      }
      if (!isSelfJoin) {
        for (unsigned int index = 0; index < multiInputsComp->getNumInputs(); index++) {
          std::string curInput = multiInputsComp->getNameForIthInput(index);
          PDB_COUT << "curInput is " << curInput << std::endl;
          auto iter = std::find(outputColumns.begin(), outputColumns.end(), curInput);
          if (iter != outputColumns.end()) {
            PDB_COUT << "MultiInputsBase with index=" << index << " is updated."
                     << std::endl;
            multiInputsComp->setTupleSetNameForIthInput(index, outputTupleSetName);
            multiInputsComp->setInputColumnsForIthInput(index, outputColumns);
            multiInputsComp->setInputColumnsToApplyForIthInput(index, outputColumnName);
          }
          PDB_COUT << std::endl;
          auto iter1 = std::find(originalInputColumnsToApply.begin(),
                                 originalInputColumnsToApply.end(),
                                 curInput);
          if (iter1 != originalInputColumnsToApply.end()) {
            PDB_COUT << "MultiInputsBase with index=" << index << " is updated."
                     << std::endl;
            multiInputsComp->setTupleSetNameForIthInput(index, outputTupleSetName);
            multiInputsComp->setInputColumnsForIthInput(index, outputColumns);
            multiInputsComp->setInputColumnsToApplyForIthInput(index, outputColumnName);
          }
        }
      } else {
        // only update myIndex
        multiInputsComp->setTupleSetNameForIthInput(myIndex, outputTupleSetName);
        multiInputsComp->setInputColumnsForIthInput(myIndex, outputColumns);
        multiInputsComp->setInputColumnsToApplyForIthInput(myIndex, outputColumnName);
      }
    }
    return tcapString;
  }

  virtual std::map<std::string, std::string> getInfo() = 0;

  virtual std::string getOutputType() = 0;
};

template<typename Out>
class TypedLambdaObject : public GenericLambdaObject {

 public:

  virtual ~TypedLambdaObject() = default;

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

#endif