Computation.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 COMPUTATION_H
#define COMPUTATION_H

#include "Object.h"
#include "Lambda.h"
#include "ComputeSource.h"
#include "ComputeSink.h"
#include "SinkMerger.h"
#include "SinkShuffler.h"
#include "InputTupleSetSpecifier.h"
#include "PDBString.h"
#include <map>

namespace pdb {

class ComputePlan;

enum ComputationTypeID {
    WriteUserSetTypeID,
    ScanUserSetTypeID,
    SelectionCompTypeID,
    PartitionCompTypeID,
    PartitionTransformationCompTypeID,
    AbstractPartitionCompID,
    AggregateCompTypeID,
    MultiSelectionCompTypeID,
    SetWriterTypeID,
    AbstractAggregateCompTypeID,
    ScanSetTypeID,
    JoinCompBaseTypeID,
    JoinCompTypeID,
    ClusterAggregationCompTypeID
};

class Computation : public Object {

public:

    virtual void extractLambdas(std::map<std::string, GenericLambdaObjectPtr>& returnVal) {}

    virtual ComputeSourcePtr getComputeSource(TupleSpec& produceMe, ComputePlan& plan) {
        return nullptr;
    }

    virtual ComputeSinkPtr getComputeSink(TupleSpec& consumeMe,
                                          TupleSpec& projection,
                                          ComputePlan& plan) {
        return nullptr;
    }

    virtual ComputeSinkPtr getComputeSink(TupleSpec& consumeMe,
                                          TupleSpec& whichAttsToOpOn,
                                          TupleSpec& projection,
                                          ComputePlan& plan) {
        return getComputeSink(consumeMe, projection, plan);
    }

    virtual SinkMergerPtr getSinkMerger(TupleSpec& consumeMe,
                                        TupleSpec& projection,
                                        ComputePlan& plan) {
        return nullptr;
    }


    virtual SinkMergerPtr getSinkMerger(TupleSpec& consumeMe,
                                        TupleSpec& whichAttsToOpOn,
                                        TupleSpec& projection,
                                        ComputePlan& plan) {
        return getSinkMerger(consumeMe, projection, plan);
    }


    virtual SinkShufflerPtr getSinkShuffler(TupleSpec& consumeMe,
                                            TupleSpec& projection,
                                            ComputePlan& plan) {
        return nullptr;
    }


    virtual SinkShufflerPtr getSinkShuffler(TupleSpec& consumeMe,
                                            TupleSpec& whichAttsToOpOn,
                                            TupleSpec& projection,
                                            ComputePlan& plan) {
        return getSinkShuffler(consumeMe, projection, plan);
    }

    // returns the type of this Computation
    virtual std::string getComputationType() = 0;

    // JiaNote: below function returns a TCAP string for this Computation
    virtual std::string toTCAPString(std::vector<InputTupleSetSpecifier>& inputTupleSets,
                                     int computationLabel,
                                     std::string& outputTupleSetName,
                                     std::vector<std::string>& outputColumnNames,
                                     std::string& addedOutputColumnName) = 0;

    virtual std::string getIthInputType(int i) = 0;

    bool hasInput() {
        return !inputs.isNullPtr();
    }

    Handle<Computation>& getIthInput(int i) const {
        return (*inputs)[i];
    }

    virtual int getNumInputs() = 0;

    virtual std::string getOutputType() = 0;

    virtual ComputationTypeID getComputationTypeID() = 0;

    int getNumConsumers() {
        return numConsumers;
    }

    void setNumConsumers(int numConsumers) {
        this->numConsumers = numConsumers;
    }

    bool setInput(Handle<Computation> toMe) {
        return setInput(0, toMe);
    }

    bool setInput(int whichSlot, Handle<Computation> toMe) {

        // set the array of inputs if it is a nullptr
        if (inputs == nullptr) {
            inputs = makeObject<Vector<Handle<Computation>>>(getNumInputs());
            for (int i = 0; i < getNumInputs(); i++) {
                inputs->push_back(nullptr);
            }
        }

        // if we are adding this query to a valid slot
        if (whichSlot < getNumInputs()) {


            // make sure the output type of the guy we are accepting meets the input type
            if (getIthInputType(whichSlot) != toMe->getOutputType()) {
                std::cout << "Cannot set output of query node with output of type "
                          << toMe->getOutputType() << " to be the input";
                std::cout << " of a query with input type " << getIthInputType(whichSlot) << ".\n";
                return false;
            }
            (*inputs)[whichSlot] = toMe;
            toMe->setNumConsumers(toMe->getNumConsumers() + 1);

        } else {

            return false;
        }
        return true;
    }


    bool isTraversed() {

        return traversed;
    }

    void setTraversed(bool traversed) {

        this->traversed = traversed;
    }

    std::string getOutputTupleSetName() {

        if (traversed) {
            return outputTupleSetName;
        }
        return "";
    }

    void setOutputTupleSetName(std::string outputTupleSetName) {
        this->outputTupleSetName = outputTupleSetName;
    }

    std::string getOutputColumnToApply() {

        if (traversed) {
            return outputColumnToApply;
        }
        return "";
    }

    void setOutputColumnToApply(std::string outputColumnToApply) {
        this->outputColumnToApply = outputColumnToApply;
    }

    virtual void setOutput(std::string dbName, std::string setName) {}

    virtual std::string getDatabaseName() {
        return "";
    }

    virtual std::string getSetName() {
        return "";
    }

    virtual bool needsMaterializeOutput() {
        return false;
    }

    virtual void setBatchSize(int batchSize) {}

    virtual bool isUsingCombiner() {
        std::cout << "Only aggregation needs to set flag for combiner" << std::endl;
        return false;
    }

    virtual void setUsingCombiner(bool useCombinerOrNot) {
        std::cout << "Only aggregation needs to set flag for combiner" << std::endl;
    }


    void setAllocatorPolicy(AllocatorPolicy myPolicy) {
        this->myAllocatorPolicy = myPolicy;
    }

    void setObjectPolicy(ObjectPolicy myPolicy) {
        this->myObjectPolicy = myPolicy;
    }

    AllocatorPolicy getAllocatorPolicy() {
        return this->myAllocatorPolicy;
    }

    ObjectPolicy getObjectPolicy() {
        return this->myObjectPolicy;
    }

    virtual void setCollectAsMap(bool collectAsMapOrNot){};

    virtual bool isCollectAsMap() {
        return false;
    }

    virtual int getNumNodesToCollect() {
        return 0;
    }

    virtual void setNumNodesToCollect(int numNodesToCollect) {}

private:

    Handle<Vector<Handle<Computation>>> inputs = nullptr;

    bool traversed = false;

    String outputTupleSetName = "";

    String outputColumnToApply = "";

    int numConsumers = 0;

    AllocatorPolicy myAllocatorPolicy = AllocatorPolicy::defaultAllocator;

    ObjectPolicy myObjectPolicy = ObjectPolicy::defaultObject;
};
}

#endif