#-------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.
#
#-------------------------------------------------------------

# In The bandit function the objective is to find an arm that optimizes
# a known functional of the unknown arm-reward distributions.
#
# INPUT:
# -----------------------------------------------------------------------------------
# X_train         ---
# Y_train         ---
# X_test          ---
# Y_test          ---
# metaList        ---
# evaluationFunc  ---
# evalFunHp       ---
# lp              ---
# primitives      ---
# params          ---
# K               ---
# R               ---
# baseLineScore   ---
# cv              ---
# cvk             ---
# verbose         ---
# output          ---
# -----------------------------------------------------------------------------------
#
# OUTPUT:
# --------------------------------------
# perf    ---
# --------------------------------------

m_bandit = function(Matrix[Double] X_train, Matrix[Double] Y_train, Matrix[Double] X_test, Matrix[Double] Y_test, List[Unknown] metaList,
  String evaluationFunc, Matrix[Double] evalFunHp, Frame[Unknown] lp, Matrix[Double] lpHp, Frame[Unknown] primitives, Frame[Unknown] param, Integer k = 3,
  Integer R=50, Double baseLineScore, Boolean cv,  Integer cvk = 2, Double ref = 0, Integer seed = -1, Boolean enablePruning = FALSE, Boolean verbose = TRUE)
  return (Frame[Unknown] bestPipeline, Matrix[Double] bestHyperparams, Matrix[Double] bestAccuracy, Frame[String] applyFunc) 
{
  print("Starting optimizer")
  totalPruneCount = 0
  FLAG_VARIABLE = 5
  pipelines_executed = 0
  HYPERPARAM_LENGTH = ((ncol(lp) + 2) * FLAG_VARIABLE * 3) + 1 ## num of col in logical * 5 meat flag vars * max hyperparam per op + 1 accuracy col
  bestPipeline = frame("", rows=1, cols=1)
  bestHyperparams = as.matrix(0)
  bestAccuracy = as.matrix(0)
  # initialize bandit variables
  # variable names follow publication where algorithm is introduced
  eta = 2  # the halving ratio is fixed to 2
  s_max = floor(log(R,eta));
  # # compute weights for R and then increase/decrease R with respect to importance of configurations

  weight = matrix(1/s_max , rows=s_max, cols=1)
  weight = cumsum(weight)
  # weight = matrix(1,  rows=s_max, cols=1)
  # print("weight matrix: "+toString(weight))
  # initialize output variables
  hparam = matrix(0, rows=k*(s_max), cols=HYPERPARAM_LENGTH)
  pipeline = matrix(0, rows=k*(s_max), cols=4)
  endIdx = matrix(k, rows=(s_max), cols=1)
  endIdx = cumsum(endIdx)
  startIdx = (endIdx - k) + 1
  logicalBest = lp
  n = ifelse(s_max >= nrow(lp), nrow(lp), n = ceil(nrow(lp)/s_max);)
  pipelineId = as.frame(seq(1, nrow(lp)))
  lp = cbind(pipelineId, lp)
  mainLookup = lp
  B = (s_max + 1) * R;
  s_max = s_max - 1
  idx = 1
  for(s in s_max:0, check = 0) { # TODO convert to parfor
    
   # result variables
    bracket_hp = matrix(0, rows=k*(s+1)+k, cols=HYPERPARAM_LENGTH)
    bracket_pipel = matrix(0, rows=k*(s+1)+k, cols=4)
    start=1; end=0;
    # # compute the number of initial pipelines n
    r = max(R * as.scalar(weight[((s_max - s) + 1)]) * eta^(-s), 1);
    configurations = lp[1:(min(n, nrow(lp)))]
    # append configuration keys for extracting the pipeline later on
    id = seq(1, nrow(configurations))
    configurations = cbind(as.frame(id), configurations)

    for(i in 0:s) {
      # successive halving
      n_i = min(max(as.integer(floor(n * eta^(-i))), 1), nrow(configurations));
      r_i = as.integer(floor(r * eta^i));
      
      if(verbose) {
        print("no of configurations ---------"+n_i)
        print("no of resources --------------"+r_i)
        print("iteration  ---------------------"+i+" out of "+s)
      }
      configurations = configurations[1:n_i, ]
      pipelines_executed = pipelines_executed + (n_i * r_i)
      [outPip,outHp, pruneCount] = run_with_hyperparam(ph_pip=configurations, r_i=r_i, X=X_train, Y=Y_train, Xtest=X_test, Ytest=Y_test, metaList=metaList,
        evaluationFunc=evaluationFunc, evalFunHp=evalFunHp, param=param, cv=cv, cvk=cvk, ref=ref, seed = seed, enablePruning=enablePruning)
      totalPruneCount = totalPruneCount + pruneCount
      # sort the pipelines by order of accuracy decreasing
      IX = order(target = outPip, by = 1, decreasing=TRUE, index.return=TRUE)
      P = table(seq(1,nrow(IX)), IX, nrow(IX), nrow(outPip));
      a = P %*% outPip
      b = P %*% outHp
      rowIndex = min(k, nrow(a))

      # maintain the brackets results
      end = end + rowIndex
      bracket_pipel[start:end, 1:ncol(a)] =  a[1:rowIndex,]
      bracket_hp[start:end, 1:ncol(b)] =  b[1:rowIndex,]
      start = end + 1

      # sort the configurations for successive halving
      avergae_perf =  getMaxPerConf(outPip, nrow(configurations)) 
      sortMask = matrix(1, rows=1, cols=ncol(configurations) + 1)
      sortMask[1,1] = 0
      configurations = frameSort(cbind(avergae_perf, configurations), sortMask, TRUE)
      configurations = configurations[, 2:ncol(configurations)]
    }
    if(n < nrow(lp))
      lp = lp[n+1:nrow(lp),]
    bracket_pipel = removeEmpty(target=bracket_pipel, margin="rows")
    bracket_hp = removeEmpty(target=bracket_hp, margin="rows")
    # keep the best k results for each bracket
    [bracket_bestPipeline, bracket_bestHyperparams] = extractBracketWinners(bracket_pipel, bracket_hp, k)
    # optimize by the features
    startOut = as.scalar(startIdx[idx])
    endOut = min(as.scalar(endIdx[idx]), (startOut + nrow(bracket_bestPipeline) - 1))
    pipeline[startOut:endOut, ] = bracket_bestPipeline
    hparam[startOut:endOut, 1:ncol(bracket_bestHyperparams)] = bracket_bestHyperparams
    idx = idx + 1
  }
  [bestPipeline, bestHyperparams, bestAccuracy] = extractTopK(pipeline, hparam, baseLineScore, k, mainLookup, lp, lpHp)

  imp = as.double(as.scalar(bestAccuracy[1, 1])) - as.double(baseLineScore)
  perf = imp > 0
  applyFunc = bestPipeline
  for(k in 1:nrow(bestPipeline))
  {
    bestPip = removeEmpty(target=bestPipeline[k], margin="cols")
    applyOp = getParamMeta(bestPip, param)
    applyFunc[k, 1:ncol(applyOp)] = applyOp
  }
  if(verbose) {
    print("dirty accuracy "+toString(baseLineScore))  
    print("topk pipelines \n"+toString(bestPipeline))
    print("topk hyper params \n"+toString(bestHyperparams))
    print("topk  scores: \n"+toString(bestAccuracy))
    print("evalHp: \n"+toString(evalFunHp))
    print("performance improvement "+ imp)
    print("total physical pipelines to be executed: "+pipelines_executed)
    print("prune count: "+totalPruneCount)
    print("actual executed pipelines: "+(pipelines_executed - totalPruneCount))
  }
}

# this method will extract the physical pipelines for a given logical pipelines
get_physical_configurations = function(Frame[String] logical, Scalar[int] numConfigs = 10, 
  Frame[Unknown] primitives)
  return(Frame[String] physical)
{
  # load the primitives
  physical = as.frame("NaN")
  ed = primitives[, 1]
  mvi = primitives[, 2]
  outliers = primitives[,3]
  ec = primitives[, 4]
  scale = primitives[, 5]
  ci = primitives[, 6]
  dummy = primitives[,7]
  dim = primitives[, 8]
 
  operator = frame(0, rows=nrow(primitives), cols=ncol(logical))  # combine all logical primitives
  parfor(j in 1:ncol(logical), check = 0)
  {
    # extract the physical primitives
    if(as.scalar(logical[1,j]) == "ED")
      operator[, j] = ed;
    else if(as.scalar(logical[1,j]) == "EC")
      operator[, j] = ec;  
    else if(as.scalar(logical[1,j]) == "OTLR")
      operator[, j] = outliers;
    else if(as.scalar(logical[1,j]) == "MVI")
      operator[, j] = mvi;
    else if(as.scalar(logical[1,j]) == "CI")
      operator[, j] = ci;
    else if(as.scalar(logical[1,j]) == "DIM")
      operator[, j] =  dim;
    else if(as.scalar(logical[1,j]) == "DUMMY") 
      operator[, j] =  dummy; 
    else if(as.scalar(logical[1,j]) == "SCALE")
      operator[, j] = scale;
    else print("invalid operation "+as.scalar(logical[1,j]))
  }
  physical = operator 
}

# # this method will call the execute pipelines with their hyper-parameters
run_with_hyperparam = function(Frame[Unknown] ph_pip, Integer r_i = 1, Matrix[Double] X, Matrix[Double] Y,
  Matrix[Double] Xtest, Matrix[Double] Ytest, List[Unknown] metaList, String evaluationFunc, Matrix[Double] evalFunHp,
  Frame[Unknown] param, Boolean cv = FALSE,  Integer cvk = 2, Double ref = 0, Integer seed = -1, Boolean enablePruning = FALSE, Boolean default = FALSE)
  return (Matrix[Double] output_operator, Matrix[Double] output_hyperparam, Integer pruneCount, Matrix[Double] changesByPipMatrix)
{
  # # # TODO there is a partial overlap but it is negligible so we will not rewrite the scripts but lineage based reuse will get rid of it
  changesByPipMatrix = matrix(0, rows=nrow(ph_pip) * r_i, cols=1)
  pruneCount = 0
  output_hp = matrix(0, nrow(ph_pip)*r_i, (ncol(ph_pip)) * 5 * 3)
  output_accuracy = matrix(0, nrow(ph_pip)*r_i, 1)
  output_pipelines = matrix(0, nrow(ph_pip)*r_i, 3)
  # rows in validation set
  ids = as.matrix(ph_pip[, 1:2])
  ph_pip = ph_pip[, 3:ncol(ph_pip)]
  inputHpMatrix = matrix(0, nrow(ph_pip)*r_i, (ncol(ph_pip)) * 5 * 3 + 1)
  # prepare the pipelines and resources
  allPipelines = frame(0, rows = nrow(ph_pip) * r_i, cols=ncol(ph_pip))
  allApplyFunctions = frame(0, rows = nrow(ph_pip) * r_i, cols=ncol(ph_pip))
  allIds = matrix(0, rows = nrow(ph_pip) * r_i, cols=2)
  start = 1
  end = 0
  i = 1
  while(i <= nrow(ph_pip))
  {
    op = removeEmpty(target=ph_pip[i], margin="cols")
    [hp, applyFunctions, no_of_res, no_of_flag_vars] = getHyperparam(op, param, r_i, default, seed, enablePruning)
    tmpFrameApply = frame(0, rows=no_of_res, cols=ncol(applyFunctions))
    tmpFrameApply = freplicate(tmpFrameApply, applyFunctions)
    tmpFramePip = frame(0, rows=no_of_res, cols=ncol(op))
    tmpFramePip = freplicate(tmpFramePip, op)
    tmpIds = matrix(1, rows=no_of_res, cols=2) * ids[i]
    end = start + (no_of_res - 1)
    allPipelines[start:end, 1:ncol(op)] = tmpFramePip
    allApplyFunctions[start:end, 1:ncol(op)] = tmpFrameApply
    allIds[start:end,] = tmpIds
    if(no_of_res == 1) {
      hp = matrix(hp, rows=no_of_res, cols = ncol(hp) * ncol(op))
      inputHpMatrix[i, 1:ncol(hp)+1] = cbind(as.matrix(ncol(hp)), hp) 
      i = i + 1
    }
    else
    {
      for(r in 1:no_of_res)
      {
        indexes = matrix(no_of_res, rows=ncol(op), cols=1)
        indexes[1, 1] = r
        indexes = cumsum(indexes)
        indexes = table(indexes, 1, 1, nrow(hp), 1)
        hp_matrix = removeEmpty(target = hp, margin="rows", select = indexes)
        tmp = matrix(hp_matrix, rows=1, cols=nrow(hp_matrix) * ncol(hp_matrix))
        tmp = cbind(as.matrix(ncol(tmp)), tmp)
        inputHpMatrix[i, 1:ncol(tmp)] = tmp
        i = i + 1
      }
    }
    start = end + 1
  }
  allPipelines = removeEmpty(target = allPipelines, margin="rows")
  inputHpMatrix = removeEmpty(target = inputHpMatrix, margin="rows")
  opPre = allIds[1]
  accuracy = 0
  changesByPip = 0

  parfor(i in 1:nrow(allPipelines), check = 0) # opt=CONSTRAINED, mode=REMOTE_SPARK
  {
    evalFunOutput = as.matrix(0)
    # execute configurations with r resources
    op = removeEmpty(target=allPipelines[i], margin="cols")
    opNew = allIds[i]
    hp = inputHpMatrix[i]
    totalVals = as.scalar(hp[1, 1]) + 1
    hp = hp[, 2:totalVals]
    applyFunctions = allApplyFunctions[i]
    no_of_res = nrow(hp)
    print("PIPELINE EXECUTION START ... "+toString(op))
    hpForPruning = matrix(0, rows=1, cols=ncol(op))
    changesByOp = matrix(0, rows=1, cols=ncol(op))
    metaList2 = metaList; #ensure metaList is no result var
    metaList2["applyFunc"] = applyFunctions
    hp_matrix = matrix(hp, rows=ncol(op), cols=ncol(hp)/ncol(op)) #removeEmpty(target = hp, margin="rows", select = indexes)
    # # check if the pruning could be applied to avoid unnecessary executions
    pruneSignal = pruningSignal(opPre, opNew, hp_matrix, hpForPruning, changesByOp)
    opPre = opNew
    executionSingnal = TRUE #ifelse(enablePruning, pruneSignal, TRUE)
    ref = ifelse(enablePruning, ref, 0)
    if(executionSingnal)
    {
      t1 = time()
      if(cv)
      {
        pipList = list(ph = op, hp = hp_matrix, flags = no_of_flag_vars)
        [accuracy, evalHp, hpForPruning, changesByOp, changesByPip] = crossV(X=X, y=Y, cvk=cvk, evalFunHp=evalFunHp,
          pipList=pipList, metaList=metaList2, hpForPruning=hpForPruning, 
        changesByOp=changesByOp, evalFunc=evaluationFunc, ref=ref)
      }
      else 
      {
        [eXtrain, eYtrain, eXtest, eYtest, Tr, hpForPruning, changesByOp, changesByPip] = executePipeline(pipeline=op, 
          Xtrain=X, Ytrain=Y, Xtest=Xtest, Ytest=Ytest, metaList=metaList2,  hyperParameters=hp_matrix, hpForPruning=hpForPruning,
          changesByOp=changesByOp, flagsCount=no_of_flag_vars, test=TRUE, verbose=FALSE)
        if(max(eYtrain) == min(eYtrain)) 
          print("Y contains only one class")
        else if(changesByPip < ref)
          print("prunningAlert 2: not training the model due to minimum changes")
        else 
          evalFunOutput = eval(evaluationFunc, list(X=eXtrain, Y=eYtrain, Xtest=eXtest, Ytest=eYtest, Xorig=as.matrix(0), evalFunHp=evalFunHp))
          accuracy = as.scalar(evalFunOutput[1, 1])
      }

      # evalFunOutput = eval(evaluationFunc, argList)  
      matrix_width = as.matrix(nrow(hp_matrix) * ncol(hp_matrix))
      hp_vec = inputHpMatrix[i]
      output_accuracy[i, 1] = accuracy
      output_hp[i, 1:ncol(hp_vec)] = hp_vec
      output_pipelines[i, ] = cbind(as.matrix(i), allIds[i,1:2])
    }
    else
    {
      pruneCount = pruneCount + 1
      print("prunningAlert 1: not executing instance : "+i+" pruneCount"+pruneCount)
    }
    changesByPipMatrix[i] = changesByPip
  }
  sel = rowSums(output_hp) > 0
  output_hyperparam = removeEmpty(target=cbind(output_accuracy, output_hp), margin="rows", select = sel)
  output_operator = removeEmpty(target=cbind(output_accuracy, output_pipelines), margin="rows", select = sel)
  changesByPipMatrix = removeEmpty(target=changesByPipMatrix, margin="rows", select = sel)
}




# extract the hyper-parameters for pipelines
getHyperparam = function(Frame[Unknown] pipeline, Frame[Unknown]  hpList, Integer no_of_res, Boolean default, Integer seed = -1, Boolean enablePruning)
  return (Matrix[Double] paramMatrix, Frame[Unknown] applyFunc, Integer no_of_res, Integer NUM_META_FLAGS)
{
  allParam = 0;
  NUM_META_FLAGS = 5
  NUM_DEFAULT_VALUES = 4

  # load the hyper-parameters values
  paramList = list()
  # store the row indexes of the operator matches
  [applyFunc, indexes, paramCount] = getParamMeta(pipeline, hpList)

  hpList = hpList[, 3:ncol(hpList)]
  DEFAULT_INDEX = 7
  START_INDEX = 11 # value from where the hyper-params starts after skipping meta flags

  # if there are no hyper-parameters than change the values of resources
  # so that the pipeline is only executed once and no resource are wasted, saving looping
  no_of_res = ifelse(sum(paramCount) > 0, no_of_res, 1)
  # the below matrix stores the different combinations of hyper-parameter value for each pipeline
  # if the resource value is greater than zero this means for 1 pipeline it will store r rows where each row store set
  # of hyperparameter values for ith pipeline. If resource value rv = 10 and ncol(pip) = 3 then the output matrix will have
  # 10*3= 30 rows and 1:10 hyper-paramters for i-the pipeline 11:20 for (i+1)-th pipeline and so on
  # this matrix stores no. of hps, values of hps, and flags
  paramMatrix = matrix(0, rows=ncol(pipeline)*no_of_res, cols=max(paramCount)+NUM_META_FLAGS+1)

  parfor(i in 1:ncol(pipeline), check=0) {
    op = as.scalar(pipeline[1, i])
    index = as.scalar(indexes[i])
    no_of_param = as.integer(as.scalar(paramCount[i]))
    # extract hasY and verbose flags
    attachMask = matrix(as.scalar(hpList[index, 2]), rows=no_of_res, cols=1)
    attachFD = matrix(as.scalar(hpList[index, 3]), rows=no_of_res, cols=1)
    attachY = matrix(as.scalar(hpList[index, 4]), rows=no_of_res, cols=1)
    isVerbose = matrix(as.scalar(hpList[index, 5]), rows=no_of_res, cols=1)
    dataFlag = matrix(as.scalar(hpList[index, 6]), rows=no_of_res, cols=1)
    if(no_of_param > 0) {
      paramIdx = START_INDEX
      typeIdx = START_INDEX
      OpParam = matrix(0, rows=no_of_res, cols=max(paramCount))
      if(default | no_of_res == 1) {
        OpParam[1, 1:no_of_param] = as.matrix(hpList[index, DEFAULT_INDEX:DEFAULT_INDEX+(no_of_param - 1)])
      }
      else {
        for(j in 1:no_of_param) {
          type = as.scalar(hpList[index, typeIdx])
          paramValIndex = (no_of_param) + paramIdx
          minVal =  as.scalar(hpList[index, paramValIndex])
          maxVal = as.scalar(hpList[index, paramValIndex + 1])
          if(type == "FP") {
            val = rand(rows=no_of_res, cols=1, min=minVal, max=maxVal, pdf="uniform", seed=seed);
            OpParam[, j] = val;
          }
          else if(type == "INT") {
            if(as.integer(maxVal) > no_of_res)
              val = sample(as.integer(maxVal), no_of_res, FALSE, seed)
            else 
              val = sample(as.integer(maxVal), no_of_res, TRUE, seed)
            less_than_min = val < as.integer(minVal);
            val = (less_than_min * minVal) + val;
            OpParam[, j] = val;
          }
          else if(type == "BOOL") {
            if(maxVal == 1) {
              s = sample(2, no_of_res, TRUE, seed);
              b = s - 1;
              OpParam[, j] = b;
            } 
            else
              OpParam[, j] = matrix(0, rows=no_of_res, cols=1)
          }
          else
            print("invalid data type")  # TODO handle string set something like {,,}
          
          paramIdx = paramIdx + 2
          typeIdx = typeIdx + 1
        }
      }
      if((op == "outlierBySd" | op == "outlierByIQR" | op == "imputeByFd") & no_of_res > 1 & enablePruning) 
        OpParam = order(target=OpParam, by = 1, decreasing = FALSE, index.return = FALSE)
      # hyper-parameter vector contains no. of hp, values of hp, and flag values
      OpParam = cbind(matrix(no_of_param, rows=nrow(OpParam), cols=1),OpParam, attachMask,
        attachFD, attachY, isVerbose, dataFlag)
    }
    else {
      # no hyper-parameters, so create a dummy matrix of zeros so flags are always aligned
      dummy = matrix(0, rows=no_of_res, cols=max(paramCount)+1)
      OpParam = cbind(dummy, attachMask, attachFD, attachY)
      OpParam = cbind(OpParam, isVerbose, dataFlag)
    }
    paramMatrix[((i-1)*no_of_res)+1:i*no_of_res, 1:ncol(OpParam)] = OpParam
  }
}


# extract the top k pipelines as a final result after deduplication and sorting
extractTopK = function(Matrix[Double] pipeline, Matrix[Double] hyperparam, 
  Double baseLineScore, Integer k, Frame[Unknown] mainLookup, Frame[Unknown] bestLogical, Matrix[Double] lpHp)
  return (Frame[Unknown] bestPipeline, Matrix[Double] bestHyperparams, Matrix[Double] bestAccuracy)
{

  bestLogical = as.matrix(bestLogical[, 1])
  pipeline = pipeline[, ncol(pipeline)]
  blN = nrow(bestLogical)
  lpHp = lpHp[nrow(lpHp)-blN+1:nrow(lpHp),]
  IX = order(target = rbind(lpHp, hyperparam), by = 1, decreasing=TRUE, index.return=TRUE)
  P = table(seq(1,nrow(IX)), IX, nrow(IX), nrow(rbind(lpHp, hyperparam)));
  hyperparam = P %*% rbind(lpHp, hyperparam)
  pipeline = P %*% rbind(bestLogical, pipeline)

  # remove the row with accuracy less than test accuracy 
  mask = (hyperparam[, 1] < baseLineScore) == 0
  if(sum(mask) == 0)
    mask[1, 1] = 1
  hyperparam = removeEmpty(target = hyperparam, margin = "rows", select = mask)
  pipeline = removeEmpty(target = pipeline, margin = "rows", select = mask)
 
  rowIndex = min(nrow(hyperparam), k)
  # select the top k
  bestAccuracy = hyperparam[1:rowIndex, 1]
  bestHyperparams = hyperparam[1:rowIndex, 2:ncol(hyperparam)] 
  pipeline = pipeline[1:rowIndex]
  # # # lookup for the pipelines
  pipCode = pipeline[, 1]
  
  bestPipeline = frame(data="0", rows=nrow(pipeline), cols=ncol(mainLookup))
  parfor(i in 1: nrow(pipeline)) {
    index = as.scalar(pipCode[i])
    bestPipeline[i] = mainLookup[index]
  }

  bestPipeline = bestPipeline[, 2:ncol(bestPipeline)]

}

# extract the top k pipelines for each bracket, the intermediate results
extractBracketWinners = function(Matrix[Double] pipeline, Matrix[Double] hyperparam, Integer k)
  return (Matrix[Double] bestPipeline, Matrix[Double] bestHyperparams)
{
  # bestPipeline = frameSort(bestPipeline)
  hyperparam = order(target = hyperparam, by = 1, decreasing=TRUE, index.return=FALSE)
  pipeline = order(target = pipeline, by = 1, decreasing=TRUE, index.return=FALSE)
  rowIndex = min(k, nrow(pipeline))

  pipeline = pipeline[1:rowIndex,]
  bestHyperparams = hyperparam[1:rowIndex,]
  bestPipeline = pipeline[1:rowIndex]
}

###########################################################################
# The function will return the max performance by each individual pipeline
############################################################################
getMaxPerConf = function(Matrix[Double] pipelines, Double size)
return (Frame[Unknown] maxperconf)
{
  tab = removeEmpty(target=table(pipelines[, 2], pipelines[, 3], pipelines[, 1]), margin="cols")
  maxperconf = frame(0, rows=size, cols=1)
  maxperconf[1:ncol(tab),] = as.frame(t(colMaxs(tab)))
}

crossV = function(Matrix[double] X, Matrix[double] y, Integer cvk, Matrix[Double] evalFunHp, List[Unknown] pipList, List[Unknown] metaList,
  Matrix[Double] hpForPruning = as.matrix(0), Matrix[Double] changesByOp = as.matrix(0), String evalFunc, Double ref = 0) 
return (Double accuracy, Matrix[Double] evalFunHp, Matrix[Double] hpForPruning, Matrix[Double] changesByOp, Double allChanges)
{

  # # in the below condition we compute the hp using cv method on train dataset
  if(is.na(as.scalar(evalFunHp[1,1]))) {
    forEvalHp = eval(evalFunc, list(X=X, Y=y, Xtest=X, Ytest=y, Xorig=as.matrix(0), evalFunHp=evalFunHp))
    evalFunHp = forEvalHp[1, 2:ncol(forEvalHp)]
  } 
  changesByPip = 0
  cvChanges = matrix(0, rows=cvk, cols=ncol(changesByOp))
  accuracyMatrix = matrix(0, cvk, 1)
  allChanges = matrix(0, cvk, 1)
  #create empty lists
  dataset_X = list(); #empty list
  dataset_y = list();
  fs = ceil(nrow(X)/cvk);
  off = fs - 1;
  #divide X, y into lists of k matrices
  for (i in seq(1, cvk)) {  
    dataset_X = append(dataset_X, X[i*fs-off : min(i*fs, nrow(X)),]);
    dataset_y = append(dataset_y, y[i*fs-off : min(i*fs, nrow(y)),]);
  }

  beta_list = list();
  #keep one fold for testing in each iteration
  for (i in seq(1, cvk)) {
    [tmpX, testX] = remove(dataset_X, i); 
    [tmpy, testy] = remove(dataset_y, i);
    trainX = rbind(tmpX);
    trainy = rbind(tmpy);
    testX = as.matrix(testX)
    testy = as.matrix(testy)
    if(as.scalar(pipList['flags']) != 0)  # this flag is zero when CV is called from the dirtyScore function, means only accuracy calculation but no pipeline execution
    {
      [trainX, trainy, testX, testy, Tr, hpForPruning, changesByOp, changesByPip] = executePipeline(pipeline=as.frame(pipList['ph']),
        Xtrain=trainX, Ytrain=trainy, Xtest= testX, Ytest=testy, metaList=metaList, hyperParameters=as.matrix(pipList['hp']), hpForPruning=hpForPruning,
        changesByOp=changesByOp, flagsCount=as.scalar(pipList['flags']), test=TRUE, verbose=FALSE)
      #TODO double check why this is necessary
      mincol = min(ncol(cvChanges),ncol(changesByOp))
      cvChanges[cvk,1:mincol] = changesByOp[,1:mincol];
      allChanges[i] = changesByPip
    }
    if(changesByPip < ref)
      print("prunning alert 2: no training the model due to minimum changes")
    else { 
      res = eval(evalFunc, list(X=trainX, Y=trainy, Xtest=testX, Ytest=testy, Xorig=as.matrix(0), evalFunHp=evalFunHp))
      accuracyMatrix[i] = res[1, 1]
    }
    
  }
  allChanges = min(allChanges)
  changesByOp = colMaxs(cvChanges)
  accuracy =  mean(accuracyMatrix)
  print("mean: \n"+toString(accuracyMatrix))
  print("cv accuracy: "+toString(accuracy))
}

pruningSignal = function(Matrix[Double] pipPre, Matrix[Double] pipNew, Matrix[Double] hp_matrix, Matrix[Double] hpForPruning, Matrix[Double] changesByOp)
return(Boolean execute)
{
  execute = TRUE
  prune = (hpForPruning > 0) & (changesByOp == 0)
  changeCount = 0
  # # if there exist a case where the changes done by an operation are zeros
  # check if pipelines are same
  samePip = FALSE
  samePip = sum(pipPre == pipNew) == ncol(pipPre)
  if(sum(prune) > 0 & samePip )
  {
    # get the non-zero index of hpForPruning
    idx = (hpForPruning > 0) * t(seq(1, ncol(hpForPruning)))
    idx = removeEmpty(target=idx, margin="cols")
    for(i in 1:ncol(idx)) {
      index = as.scalar(idx[1, i])
      inProcessHp = as.scalar(hp_matrix[index, 2])
      prvHp = as.scalar(hpForPruning[1, index])
      if(inProcessHp > prvHp)
        changeCount = changeCount + 1
    }
  }
  execute = !(changeCount > 0)
}



getParamMeta = function(Frame[Unknown] pipeline, Frame[Unknown] hpList)
return(Frame[Unknown] applyFunc, Matrix[Double] indexes, Matrix[Double] paramCount)
{
  indexes = matrix(0, rows= ncol(pipeline), cols=1)
  paramCount = matrix(0, rows= ncol(pipeline), cols=1)
  applyList = hpList[, 1]
  applyFunc = pipeline
  parfor(k in 1:ncol(pipeline))
  {
    op = as.scalar(pipeline[1,k])
    hasParam = map(hpList[,2], "x->x.split(\",\")[0].equals(\""+op+"\")")
    # convert the boolean vector to 0/1 matrix representation
    m_hasParam = hasParam == frame("true", rows=nrow(hasParam), cols=1)
    m_hasParam = as.matrix(m_hasParam)
    # compute the relevant index
    index = m_hasParam * seq(1, nrow(m_hasParam))
    index = as.scalar(removeEmpty(target = index, margin = "rows"))
    indexes[k] = index
    paramCount[k] = as.integer(as.scalar(hpList[index, 3]))
    applyFunc[1, k] = as.scalar(hpList[index, 1])
  }
}



# # this method will call the execute pipelines with their hyper-parameters
run_with_hyperparamNested = function(Frame[Unknown] ph_pip, Integer r_i = 1, Matrix[Double] X, Matrix[Double] Y,
  Matrix[Double] Xtest, Matrix[Double] Ytest, List[Unknown] metaList, String evaluationFunc, Matrix[Double] evalFunHp,
  Frame[Unknown] param, Boolean cv = FALSE,  Integer cvk = 2, Double ref = 0, Integer seed = -1, Boolean enablePruning = FALSE, Boolean default = FALSE)
  return (Matrix[Double] output_operator, Matrix[Double] output_hyperparam, Integer pruneCount, Matrix[Double] changesByPipMatrix)
{
  # # # TODO there is a partial overlap but it is negligible so we will not rewrite the scripts but lineage based reuse will get rid of it
  changesByPipMatrix = matrix(0, rows=nrow(ph_pip) * r_i, cols=1)
  pruneCount = 0
  output_hp = matrix(0, nrow(ph_pip)*r_i, (ncol(ph_pip)) * 5 * 3)
  output_accuracy = matrix(0, nrow(ph_pip)*r_i, 1)
  output_pipelines = matrix(0, nrow(ph_pip)*r_i, 3)
  # rows in validation set
  clone_X = X
  clone_Y = Y
  clone_Xtest = Xtest
  clone_Ytest = Ytest
  index = 1
  ids = as.matrix(ph_pip[, 1:2])
  ph_pip = ph_pip[, 3:ncol(ph_pip)]

  parfor(i in 1:nrow(ph_pip), check = 0) # , opt=CONSTRAINED, mode=REMOTE_SPARK
  {
    # execute configurations with r resources
    op = removeEmpty(target=ph_pip[i], margin="cols")
    # print("PIPELINE EXECUTION START ... "+toString(op))
    [hp, applyFunctions, no_of_res, no_of_flag_vars] = getHyperparam(op, param, r_i, default, seed, enablePruning)
    hpForPruning = matrix(0, rows=1, cols=ncol(op))
    changesByOp = matrix(0, rows=1, cols=ncol(op))
    metaList2 = metaList; #ensure metaList is no result var
    metaList2["applyFunc"] = applyFunctions
    for(r in 1:no_of_res)
    {
      # as the matrix first block of r rows belongs to first operator and r+1 block of rows to second operator 
      # we need to extract a row from each block
      indexes = matrix(no_of_res, rows=ncol(op), cols=1)
      indexes[1, 1] = r
      indexes = cumsum(indexes)
      indexes = table(indexes, 1, 1, nrow(hp), 1)
      hp_matrix = removeEmpty(target = hp, margin="rows", select = indexes)
      # # check if the pruning could be applied to avoid unnecessary executions
      pruneSignal = pruningSignalNested(op, hp_matrix, hpForPruning, changesByOp)
      executionSingnal = ifelse(enablePruning, pruneSignal, TRUE)
      ref = ifelse(enablePruning, ref, 0)
      if(executionSingnal)
      {
        t1 = time()
        if(cv)
        {
          pipList = list(ph = op, hp = hp_matrix, flags = no_of_flag_vars)
          [accuracy, evalHp, hpForPruning, changesByOp, changesByPip] = crossV(X=X, y=Y, cvk=cvk, evalFunHp=evalFunHp,
            pipList=pipList, metaList=metaList2, hpForPruning=hpForPruning, 
          changesByOp=changesByOp, evalFunc=evaluationFunc, ref=ref)
        }
        else 
        {
          [eXtrain, eYtrain, eXtest, eYtest, Tr, hpForPruning, changesByOp, changesByPip] = executePipeline(pipeline=op, 
            Xtrain=X, Ytrain=Y, Xtest=Xtest, Ytest=Ytest, metaList=metaList2,  hyperParameters=hp_matrix, hpForPruning=hpForPruning,
            changesByOp=changesByOp, flagsCount=no_of_flag_vars, test=TRUE, verbose=FALSE)
          if(max(eYtrain) == min(eYtrain)) 
            print("Y contains only one class")
          else if(changesByPip < ref)
            print("prunning alert 2: no training the model due to minimum changes")
          else 
            evalFunOutput = eval(evaluationFunc, list(X=eXtrain, Y=eYtrain, Xtest=eXtest, Ytest=eYtest, Xorig=as.matrix(0), evalFunHp=evalFunHp))
            accuracy = as.scalar(evalFunOutput[1, 1])
        }

        # evalFunOutput = eval(evaluationFunc, argList)  
        matrix_width = as.matrix(nrow(hp_matrix) * ncol(hp_matrix))
        hp_vec = cbind(matrix_width, matrix(hp_matrix, rows=1, cols=nrow(hp_matrix)*ncol(hp_matrix), byrow=TRUE))
        index = (i - 1) * no_of_res + r
        output_accuracy[index, 1] = accuracy
        output_hp[index, 1:ncol(hp_vec)] = hp_vec
        output_pipelines[index, ] = cbind(as.matrix(index), ids[i,1:2])
      }
      else
      {
        pruneCount = pruneCount + 1
        print("prunningAlert: not executing instance : "+r+" pruneCount"+pruneCount)
      }
      changesByPipMatrix[index] = changesByPip
      index = index + 1
    }
  }
  sel = rowSums(output_hp) > 0
  output_hyperparam = removeEmpty(target=cbind(output_accuracy, output_hp), margin="rows", select = sel)
  output_operator = removeEmpty(target=cbind(output_accuracy, output_pipelines), margin="rows", select = sel)
  changesByPipMatrix = removeEmpty(target=changesByPipMatrix, margin="rows", select = sel)
}

pruningSignalNested = function(Frame[String] op, Matrix[Double] hp_matrix, Matrix[Double] hpForPruning, Matrix[Double] changesByOp)
return(Boolean execute)
{
  execute = TRUE
  prune = (hpForPruning > 0) & (changesByOp == 0)
  changeCount = 0
  # # if there exist a case where the changes done by an operation are zeros
  # check if pipelines are same
  samePip = FALSE

  if(sum(prune) > 0)
  {
    # get the non-zero index of hpForPruning
    idx = (hpForPruning > 0) * t(seq(1, ncol(hpForPruning)))
    idx = removeEmpty(target=idx, margin="cols")
    for(i in 1:ncol(idx)) {
      index = as.scalar(idx[1, i])
      inProcessHp = as.scalar(hp_matrix[index, 2])
      prvHp = as.scalar(hpForPruning[1, index])
      if(inProcessHp > prvHp)
        changeCount = changeCount + 1
    }
  }
  execute = !(changeCount > 0)
}