src/selfLearning/source/RuleBasedDataPlacementOptimizerForLoadJob.cc
Handle<LambdaIdentifier> RuleBasedDataPlacementOptimizerForLoadJob :: getBestLambda() {
    if (consumers.size() == 0) {
       return nullptr;
    }
    Handle<LambdaIdentifier> partitionLambda = nullptr;
    std::cout << " there are " << consumers.size() << " pairs" << std::endl;
    //first to get the consumer job stages
    for (auto it = consumers.begin(); it != consumers.end(); ++it) {
        std::cout << "get into the loop" << std::endl;
        //for the 1st consumer job stage, apply following rules:
        std::shared_ptr<JobStageSelfLearningInfo> consumerInfo = it->second;
        //Rule 1: if the consumer job stage has a Repartition sink, we use the lambda associated with this job stage;
        //Rule 2: if the consumer job stage has a Shuffle sink, we use the lambda associated with this job stage;
        std::cout << "SinkType is " << consumerInfo->getSinkType() << std::endl;
        if ((consumerInfo->getSinkType() == "Shuffle")||(consumerInfo->getSinkType() == "Repartition")) {
             std::cout << "to get lambda for jobId = " << consumerInfo->getJobId() << ", targetComputation = " 
                      << consumerInfo->getTargetComputationName() << ", dataType = " << dataType << std::endl;
             partitionLambda = server.getLambda(consumerInfo->getJobId(), consumerInfo->getTargetComputationName(), 
                      dataType, consumerInfo->getIndexInInputs());
        } else if (consumerInfo->getSinkType() == "UserSet") {
        //Rule 3: if the direct consumer job stage has a UserSet sink, and the consumer job stage of the UserSet generated by the direct consumer job stage has a Repartition or Shuffle sink, we use the lambda associated with the indirect consumer job stage;
             //1. to get the identifiter to the output set of the direct consumer job stage
             long dataId = server.getSinkDataId(consumerInfo->getJobStageId());             
             std::string indirectDataType = server.getDataType(dataId);
             std::cout << "indirectDataType is " << indirectDataType << std::endl;
             //2. to get the consumers that consume the output set
             std::unordered_map<std::pair<long, long>, std::shared_ptr<JobStageSelfLearningInfo>, PairKeyHash, PairKeyEqual> indirectConsumers;
             std::cout << "to get consumers for data" << std::endl;
             server.getConsumerInfoForData(dataId, indirectConsumers);
             //3. follow rule-1 and rule-2
             for (auto it1 = indirectConsumers.begin(); it1 != indirectConsumers.end(); ++it1) {
                 std::shared_ptr<JobStageSelfLearningInfo> indirectConsumerInfo = it1->second;
                 std::string sinkType = indirectConsumerInfo->getSinkType();
                 std::cout << "SinkType is " << sinkType << std::endl;
                 if ((sinkType == "Shuffle") || (sinkType == "Repartition")) {
                     std::cout << "to get lambda for jobId = " << indirectConsumerInfo->getJobId()
                        << ", targetComputation = " << indirectConsumerInfo->getTargetComputationName()
                        << ", dataType = " << indirectDataType << std::endl;
                     partitionLambda = server.getLambda(indirectConsumerInfo->getJobId(), 
                        indirectConsumerInfo->getTargetComputationName(), indirectDataType,
                        indirectConsumerInfo->getIndexInInputs());
                 } 
                 if (partitionLambda->getLambdaName().find("native") != std::string::npos) {
                     std::cout << "INFO: continue, because we can't support partitioning using native lambda: " 
                       << partitionLambda->getLambdaName() << std::endl;
                    continue;
                 }
                 if (partitionLambda != nullptr) {
                     break;
                 }
             }

        }
        if (partitionLambda != nullptr) {
            break;
        }
        
    }
    return partitionLambda;
}
//Now we implement a simple rule-based optimizer, assuming there is only 1 consumer job stage
Now we implement a simple rule-based optimizer, assuming there is only 1 consumer job stage
Now we implement a simple rule - based optimizer assuming there is only 1 consumer job stage
Handle<LambdaIdentifier> RuleBasedDataPlacementOptimizerForLoadJob :: getBestLambda() {
    if (consumers.size() == 0) {
       return nullptr;
    }
    Handle<LambdaIdentifier> partitionLambda = nullptr;
    std::cout << " there are " << consumers.size() << " pairs" << std::endl;
    for (auto it = consumers.begin(); it != consumers.end(); ++it) {
        std::cout << "get into the loop" << std::endl;
        std::shared_ptr<JobStageSelfLearningInfo> consumerInfo = it->second;
        std::cout << "SinkType is " << consumerInfo->getSinkType() << std::endl;
        if ((consumerInfo->getSinkType() == "Shuffle")||(consumerInfo->getSinkType() == "Repartition")) {
             std::cout << "to get lambda for jobId = " << consumerInfo->getJobId() << ", targetComputation = " 
                      << consumerInfo->getTargetComputationName() << ", dataType = " << dataType << std::endl;
             partitionLambda = server.getLambda(consumerInfo->getJobId(), consumerInfo->getTargetComputationName(), 
                      dataType, consumerInfo->getIndexInInputs());
        } else if (consumerInfo->getSinkType() == "UserSet") {
             long dataId = server.getSinkDataId(consumerInfo->getJobStageId());             
             std::string indirectDataType = server.getDataType(dataId);
             std::cout << "indirectDataType is " << indirectDataType << std::endl;
             std::unordered_map<std::pair<long, long>, std::shared_ptr<JobStageSelfLearningInfo>, PairKeyHash, PairKeyEqual> indirectConsumers;
             std::cout << "to get consumers for data" << std::endl;
             server.getConsumerInfoForData(dataId, indirectConsumers);
             for (auto it1 = indirectConsumers.begin(); it1 != indirectConsumers.end(); ++it1) {
                 std::shared_ptr<JobStageSelfLearningInfo> indirectConsumerInfo = it1->second;
                 std::string sinkType = indirectConsumerInfo->getSinkType();
                 std::cout << "SinkType is " << sinkType << std::endl;
                 if ((sinkType == "Shuffle") || (sinkType == "Repartition")) {
                     std::cout << "to get lambda for jobId = " << indirectConsumerInfo->getJobId()
                        << ", targetComputation = " << indirectConsumerInfo->getTargetComputationName()
                        << ", dataType = " << indirectDataType << std::endl;
                     partitionLambda = server.getLambda(indirectConsumerInfo->getJobId(), 
                        indirectConsumerInfo->getTargetComputationName(), indirectDataType,
                        indirectConsumerInfo->getIndexInInputs());
                 } 
                 if (partitionLambda->getLambdaName().find("native") != std::string::npos) {
                     std::cout << "INFO: continue, because we can't support partitioning using native lambda: " 
                       << partitionLambda->getLambdaName() << std::endl;
                    continue;
                 }
                 if (partitionLambda != nullptr) {
                     break;
                 }
             }
        }
        if (partitionLambda != nullptr) {
            break;
        }
    }
    return partitionLambda;
}
Handle < LambdaIdentifier > RuleBasedDataPlacementOptimizerForLoadJob :: getBestLambda ( ) { if ( consumers . size ( ) == 0 ) { return nullptr ; } Handle < LambdaIdentifier > partitionLambda = nullptr ; std :: cout << " " << consumers . size ( ) << " " << std :: endl ; for ( auto it = consumers . begin ( ) ; it != consumers . end ( ) ; ++ it ) { std :: cout << " " << std :: endl ; std :: shared_ptr < JobStageSelfLearningInfo > consumerInfo = it -> second ; std :: cout << " " << consumerInfo -> getSinkType ( ) << std :: endl ; if ( ( consumerInfo -> getSinkType ( ) == " " ) || ( consumerInfo -> getSinkType ( ) == " " ) ) { std :: cout << " " << consumerInfo -> getJobId ( ) << " " << consumerInfo -> getTargetComputationName ( ) << " " << dataType << std :: endl ; partitionLambda = server . getLambda ( consumerInfo -> getJobId ( ) , consumerInfo -> getTargetComputationName ( ) , dataType , consumerInfo -> getIndexInInputs ( ) ) ; } else if ( consumerInfo -> getSinkType ( ) == " " ) { long dataId = server . getSinkDataId ( consumerInfo -> getJobStageId ( ) ) ; std :: string indirectDataType = server . getDataType ( dataId ) ; std :: cout << " " << indirectDataType << std :: endl ; std :: unordered_map < std :: pair < long , long > , std :: shared_ptr < JobStageSelfLearningInfo > , PairKeyHash , PairKeyEqual > indirectConsumers ; std :: cout << " " << std :: endl ; server . getConsumerInfoForData ( dataId , indirectConsumers ) ; for ( auto it1 = indirectConsumers . begin ( ) ; it1 != indirectConsumers . end ( ) ; ++ it1 ) { std :: shared_ptr < JobStageSelfLearningInfo > indirectConsumerInfo = it1 -> second ; std :: string sinkType = indirectConsumerInfo -> getSinkType ( ) ; std :: cout << " " << sinkType << std :: endl ; if ( ( sinkType == " " ) || ( sinkType == " " ) ) { std :: cout << " " << indirectConsumerInfo -> getJobId ( ) << " " << indirectConsumerInfo -> getTargetComputationName ( ) << " " << indirectDataType << std :: endl ; partitionLambda = server . getLambda ( indirectConsumerInfo -> getJobId ( ) , indirectConsumerInfo -> getTargetComputationName ( ) , indirectDataType , indirectConsumerInfo -> getIndexInInputs ( ) ) ; } if ( partitionLambda -> getLambdaName ( ) . find ( " " ) != std :: string :: npos ) { std :: cout << " " << partitionLambda -> getLambdaName ( ) << std :: endl ; continue ; } if ( partitionLambda != nullptr ) { break ; } } } if ( partitionLambda != nullptr ) { break ; } } return partitionLambda ; }
Now we implement a simple rule-based optimizer, assuming there is only 1 consumer job stage
Now we implement a simple rule - based optimizer assuming there is only 1 consumer job stage
//first to get the consumer job stages //for the 1st consumer job stage, apply following rules: //Rule 1: if the consumer job stage has a Repartition sink, we use the lambda associated with this job stage; //Rule 2: if the consumer job stage has a Shuffle sink, we use the lambda associated with this job stage; //Rule 3: if the direct consumer job stage has a UserSet sink, and the consumer job stage of the UserSet generated by the direct consumer job stage has a Repartition or Shuffle sink, we use the lambda associated with the indirect consumer job stage; //1. to get the identifiter to the output set of the direct consumer job stage //2. to get the consumers that consume the output set //3. follow rule-1 and rule-2