/*
 * 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.
 */
///////////////////////////////////////////////////////////////////////////////
// Copyright 2016, Oushu Inc.
// All rights reserved.
//
// Author:
///////////////////////////////////////////////////////////////////////////////
#include "postgres.h"

#include "cdb/cdbgang.h"
#include "cdb/cdbutil.h"
#include "cdb/cdbvars.h"
#include "cdb/scheduler.h"
#include "commands/variable.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"

#include "scheduler/cwrapper/scheduler-c.h"

#define CDB_MOTION_LOST_CONTACT_STRING \
  "Interconnect error master lost contact with segment."

typedef struct {
  int sliceIndex;
  int children;
  Slice *slice;
} sliceVec;

static int compare_slice_order(const void *aa, const void *bb) {
  sliceVec *a = (sliceVec *) aa;
  sliceVec *b = (sliceVec *) bb;

  if (a->slice == NULL)
    return 1;
  if (b->slice == NULL)
    return -1;

  /* sort the writer gang slice first, because he sets the shared snapshot */
  if (a->slice->is_writer && !b->slice->is_writer)
    return -1;
  else if (b->slice->is_writer && !a->slice->is_writer)
    return 1;

  if (a->children == b->children)
    return 0;
  else if (a->children > b->children)
    return 1;
  else
    return -1;
}
static void mark_bit(char *bits, int nth) {
  int nthbyte = nth >> 3;
  char nthbit = 1 << (nth & 7);
  bits[nthbyte] |= nthbit;
}
static void or_bits(char *dest, char *src, int n) {
  int i;

  for (i = 0; i < n; i++)
    dest[i] |= src[i];
}

static int count_bits(char *bits, int nbyte) {
  int i;
  int nbit = 0;
  int bitcount[] = { 0, 1, 1, 2, 1, 2, 2, 3, 1, 2, 2, 3, 2, 3, 3, 4 };

  for (i = 0; i < nbyte; i++) {
    nbit += bitcount[bits[i] & 0x0F];
    nbit += bitcount[(bits[i] >> 4) & 0x0F];
  }

  return nbit;
}

static int markbit_dep_children(SliceTable *sliceTable, int sliceIdx,
                                sliceVec *sliceVec, int bitmasklen, char *bits) {
  ListCell *sublist;
  Slice *slice = (Slice *) list_nth(sliceTable->slices, sliceIdx);

  foreach (sublist, slice->children)
  {
    int childIndex = lfirst_int(sublist);
    char *newbits = palloc0(bitmasklen);

    markbit_dep_children(sliceTable, childIndex, sliceVec, bitmasklen, newbits);
    or_bits(bits, newbits, bitmasklen);
    mark_bit(bits, childIndex);
    pfree(newbits);
  }

  sliceVec[sliceIdx].sliceIndex = sliceIdx;
  sliceVec[sliceIdx].children = count_bits(bits, bitmasklen);
  sliceVec[sliceIdx].slice = slice;

  return sliceVec[sliceIdx].children;
}
static int count_dependent_children(SliceTable *sliceTable, int sliceIndex,
                                    sliceVec *sliceVector, int len) {
  int ret = 0;
  int bitmasklen = (len + 7) >> 3;
  char *bitmask = palloc0(bitmasklen);

  ret = markbit_dep_children(sliceTable, sliceIndex, sliceVector, bitmasklen,
                             bitmask);
  pfree(bitmask);

  return ret;
}

static int fillSliceVector(SliceTable *sliceTbl, int rootIdx,
                           sliceVec *sliceVector, int sliceLim) {
  int top_count;

  /* count doesn't include top slice add 1 */
  top_count = 1
      + count_dependent_children(sliceTbl, rootIdx, sliceVector, sliceLim);

  qsort(sliceVector, sliceLim, sizeof(sliceVec), compare_slice_order);

  return top_count;
}

static void scheduler_serialize_common_plan(SchedulerData *data,
                                            CommonPlanContext *ctx);
static void scheduler_init_instrumentation(MyInstrumentation **myInstrument,
                                           MyInstrumentation *instr,
                                           struct PlanState *planstate,
                                           int stageNo, int segmentNum);
static CdbVisitOpt convertMyInstruemntToSliceStat(MyInstrumentation *instr,
                                                  void *context);

bool scheduler_plan_support_check(QueryDesc *queryDesc) {
  SliceTable *sliceTbl = queryDesc->estate->es_sliceTable;

  if(sliceTbl->doInstrument){
    return false;
  }

  if(sliceTbl->nInitPlans){
    return false;
  }

  ListCell *cell = NULL;
  foreach(cell, sliceTbl->slices)
  {
    Slice *slice = (Slice *) lfirst(cell);

    if (slice->directDispatch.isDirectDispatch
        || slice->gangSize == 1 && slice->gangType != GANGTYPE_UNALLOCATED) {
      return false;
    }
  }

  return true;
}

void scheduler_prepare_for_new_query(QueryDesc *queryDesc, const char *queryId,
                                     int planId) {
  // Initialize scheduler when first query
  if (MyScheduler == NULL) {
    // MyScheduler = SchedulerNew(magma_port_segment + 4);
    SchedulerSetupRpcServer(MyScheduler, GetMasterSegment()->hostip);
  }

  queryDesc->estate->scheduler_data = palloc0(sizeof(SchedulerData));
  SchedulerData *scheduler_data = queryDesc->estate->scheduler_data;
  scheduler_data->state = SS_INIT;
  scheduler_data->segmentNum = list_length(queryDesc->resource->segments);
  scheduler_data->resource = queryDesc->resource;
  scheduler_data->queryDesc = queryDesc;

  SliceTable *sliceTbl = queryDesc->estate->es_sliceTable;
  scheduler_data->localSliceId = sliceTbl->localSlice;

  sliceVec *sliceVector = NULL;
  int i;
  int slice_num;

  sliceVector = palloc0(list_length(sliceTbl->slices) * sizeof(*sliceVector));
  slice_num = fillSliceVector(sliceTbl, RootSliceIndex(queryDesc->estate),
                              sliceVector, list_length(sliceTbl->slices));

  scheduler_data->job.used_slices_num = slice_num;
  scheduler_data->job.all_slices_num = list_length(sliceTbl->slices);
  scheduler_data->job.slices = palloc0(
      scheduler_data->job.used_slices_num * sizeof(ScheduleSlice));

  for (i = 0; i < scheduler_data->job.used_slices_num; i++) {
    scheduler_data->job.slices[i].workers_num = list_length(
        scheduler_data->resource->segments);
    scheduler_data->job.slices[i].workers = palloc0(
        scheduler_data->job.slices[i].workers_num * sizeof(ScheduleWorker));
  }

  for (i = 0; i < slice_num; i++) {
    Slice *slice = sliceVector[i].slice;
    bool is_writer = false;

    is_writer = (i == 0 && !queryDesc->extended_query);
    // Not all of slices need to dispatch, such as root slice!
    // root slice is at end of sliceVector
    if (slice->gangType == GANGTYPE_UNALLOCATED) {
      scheduler_data->job.used_slices_num--;
      continue;
    }

    int j;

    // Assign the id from 0.
    for (j = 0; j < scheduler_data->job.slices[i].workers_num; j++) {
      ScheduleWorker *worker = &scheduler_data->job.slices[i].workers[j];

      worker->id.id_in_slice = j;
      worker->id.slice_id = slice->sliceIndex;
      worker->id.gang_member_num = list_length(
          scheduler_data->resource->segments);
      worker->id.command_count = gp_command_count;
      worker->id.is_writer = is_writer;
      worker->segment = list_nth(scheduler_data->resource->segments,
                                 worker->id.id_in_slice);
      worker->id.init = true;
    }
  }

  pfree(sliceVector);

  // get all stageNo from plan tree
  struct Plan *planTree = queryDesc->plannedstmt->planTree;
  scheduler_data->stageNo = palloc0(
      sizeof(int) * scheduler_data->job.used_slices_num);
  scheduler_data->stageNum = scheduler_data->job.used_slices_num;
  scheduler_data->totalStageNum = scheduler_data->job.all_slices_num;

  for (int i = 0; i < scheduler_data->job.used_slices_num; i++) {
    scheduler_data->stageNo[i] = scheduler_data->job.slices[i].workers[0].id
        .slice_id;
  }

//  bool *isInitPlan =
//      palloc0(sizeof(bool) * list_length(queryDesc->plannedstmt->subplans));
//  if (planId == 0) {
//    get_all_stageno_from_plantree(planTree, scheduler_data->stageNo,
//                                  &scheduler_data->stageNum, isInitPlan);
//    ListCell *lc;
//    int i = 0;
//    foreach (lc, queryDesc->plannedstmt->subplans) {
//      Plan *subplan = (Plan *)lfirst(lc);
//      if (!isInitPlan[i])
//        get_all_stageno_from_plantree(subplan, scheduler_data->stageNo,
//                                      &scheduler_data->stageNum, isInitPlan);
//      i++;
//    }
//  } else {
//    Plan *subplan =
//        (Plan *)list_nth(queryDesc->plannedstmt->subplans, planId - 1);
//    get_all_stageno_from_plantree(subplan, scheduler_data->stageNo,
//                                  &scheduler_data->stageNum, isInitPlan);
//  }
//  pfree(isInitPlan);

  // get all hosts from resource
  ListCell *lc;
  i = 0;
  scheduler_data->hosts = palloc0(sizeof(char *) * scheduler_data->segmentNum);
  int *lens = palloc0(sizeof(int) * scheduler_data->segmentNum);
  foreach (lc, scheduler_data->resource->segments)
  {
    Segment *seg = (Segment *) lfirst(lc);
    scheduler_data->hosts[i] = seg->hostname;
    lens[i] = strlen(seg->hostname);
    i++;
  }

  SchedulerPrepareForNewQuery(MyScheduler, queryId, scheduler_data->hosts, lens,
                              scheduler_data->segmentNum,
                              scheduler_data->stageNo,
                              scheduler_data->stageNum);
  pfree(lens);
  SchedulerCatchedError *err = SchedulerGetLastError(MyScheduler);
  if (err->errCode != ERRCODE_SUCCESSFUL_COMPLETION) {
    scheduler_data->state = SS_ERROR;
    int errCode = err->errCode;
    ereport(
        ERROR,
        (errcode(errCode), errmsg("failed to prepare scheduler for new query. %s (%d)", err->errMessage, errCode)));
  }

  scheduler_data->totalStageNum = queryDesc->estate->es_sliceTable->nInitPlans
      + queryDesc->estate->es_sliceTable->nMotions + 1;
  scheduler_data->ports = palloc0(
      sizeof(int *) * scheduler_data->totalStageNum);
  for (i = 0;
      i
          < queryDesc->estate->es_sliceTable->nInitPlans
              + queryDesc->estate->es_sliceTable->nMotions + 1; i++)
    scheduler_data->ports[i] = palloc0(
        sizeof(int) * scheduler_data->segmentNum);
  for (i = 0; i < scheduler_data->segmentNum; i++)
    for (int j = 0; j < scheduler_data->stageNum; j++)
      scheduler_data->ports[scheduler_data->stageNo[j]][i] =
          SchedulerGetListenPort(MyScheduler, i, scheduler_data->stageNo[j]);
  if (err->errCode != ERRCODE_SUCCESSFUL_COMPLETION) {
    scheduler_data->state = SS_ERROR;
    int errCode = err->errCode;
    ereport(
        ERROR,
        (errcode(errCode), errmsg("failed to get scheduler listen port. %s (%d)", err->errMessage, errCode)));
  }
}

void scheduler_run(SchedulerData *data, CommonPlanContext *ctx) {
  if (!data)
    return;

  scheduler_serialize_common_plan(data, ctx);
  ScheduleRun(MyScheduler, ctx->univplan);
  univPlanFreeInstance(&ctx->univplan);
  data->state = SS_RUN;
  SchedulerCatchedError *err = SchedulerGetLastError(MyScheduler);
  if (err->errCode != ERRCODE_SUCCESSFUL_COMPLETION) {
    SchedulerCancelQuery(MyScheduler);

    if (MyNewExecutor != NULL)
      MyExecutorSetCancelQuery(MyNewExecutor);

    data->state = SS_ERROR;
    int errCode = err->errCode;
    ereport(
        ERROR,
        (errcode(errCode), errmsg("failed to run scheduler. %s (%d)", err->errMessage, errCode)));
  }
}

void scheduler_getResult(struct SchedulerData *data) {
  getTasksResult(MyScheduler, &data->insertRows);
}

void scheduler_wait(SchedulerData *data) {
  if (!data || data->state == SS_OK || data->state == SS_ERROR)
    return;

  SchedulerWaitAllTasksComplete(MyScheduler);
  SchedulerCatchedError *err = SchedulerGetLastError(MyScheduler);
  data->state = SS_OK;
  if (err->errCode != ERRCODE_SUCCESSFUL_COMPLETION) {
    data->state = SS_ERROR;
    int errCode = err->errCode;
    ereport(ERROR, (errcode(errCode), errmsg("%s", err->errMessage)));
  }
}

static void scheduler_serialize_common_plan(SchedulerData *data,
                                            CommonPlanContext *ctx) {
  PlannedStmt *stmt = data->queryDesc ? data->queryDesc->plannedstmt : NULL;
  if (stmt) {
    univPlanSetDoInstrument(
        ctx->univplan, data->queryDesc->estate->es_sliceTable->doInstrument);
    univPlanSetNCrossLevelParams(ctx->univplan, stmt->nCrossLevelParams);

    // add interconnect info
    for (int i = 0;
        i
            < data->queryDesc->estate->es_sliceTable->nInitPlans
                + data->queryDesc->estate->es_sliceTable->nMotions + 1; ++i) {
      Slice *slice = list_nth(data->queryDesc->estate->es_sliceTable->slices,
                              i);
      ListCell *lc;
      uint32_t listenerNum;
      char **addr;
      int32 *port;
      if (!slice->primaryProcesses) {
        if (data->ports[i][0] == 0)
          listenerNum = 0;
        else if (slice->directDispatch.isDirectDispatch)
          listenerNum = 1;
        else if (slice->gangSize == 1)
          listenerNum = 1;
        else
          listenerNum = data->segmentNum;
        addr = palloc0(listenerNum * sizeof(char *));
        port = palloc0(listenerNum * sizeof(int32));
        for (int j = 0; j < listenerNum; j++) {
          addr[j] = data->hosts[j];
          port[j] = data->ports[i][j];
        }
      } else {
        listenerNum = slice->primaryProcesses->length;
        addr = palloc0(listenerNum * sizeof(char *));
        port = palloc0(listenerNum * sizeof(int32));
        int index = 0;
        foreach (lc, slice->primaryProcesses)
        {
          CdbProcess *proc = (CdbProcess *) lfirst(lc);
          if (proc->listenerAddr)
            addr[index] = pstrdup(proc->listenerAddr);
          else
            addr[index] = data->resource->master->hostip;
          port[index] = data->ports[i][index];
          if (port[index] == 0)
            port[index] = proc->myListenerPort;
          ++index;
        }
      }
      // scheduler no use, just set listener to -1
      univPlanReceiverAddListeners(ctx->univplan, listenerNum, -1, addr, port);
      pfree(addr);
      pfree(port);
    }

    univPlanFixVarType(ctx->univplan);

    univPlanAddGuc(ctx->univplan, "work_file_dir", rm_seg_tmp_dirs);
    univPlanAddGuc(ctx->univplan, "enable_partitioned_hashagg",
                   new_executor_enable_partitioned_hashagg_mode);
    univPlanAddGuc(ctx->univplan, "enable_partitioned_hashjoin",
                   new_executor_enable_partitioned_hashjoin_mode);
    univPlanAddGuc(ctx->univplan, "enable_external_sort",
                   new_executor_enable_external_sort_mode);
    univPlanAddGuc(ctx->univplan, "new_scheduler", new_scheduler_mode);
    char *timezone_str = show_timezone();
    univPlanAddGuc(ctx->univplan, "timezone_string", timezone_str);

    char partitioned_hash_recursive_depth_limit[4];
    sprintf(partitioned_hash_recursive_depth_limit, "%d",
            new_executor_partitioned_hash_recursive_depth_limit);
    univPlanAddGuc(ctx->univplan, "partitioned_hash_recursive_depth_limit",
                   partitioned_hash_recursive_depth_limit);

    univPlanStagize(ctx->univplan);

    // elog(DEBUG1, "common plan: %s",
    // univPlanGetJsonFormatedPlan(ctx->univplan));

    // serialize common plan for QD executes
    int len;
    const char *val = univPlanSerialize(ctx->univplan, &len, true);
    data->newPlan = (CommonPlan *) palloc0(sizeof(CommonPlan));
    data->newPlan->len = len;
    data->newPlan->str = (char *) palloc0(len);
    memcpy(data->newPlan->str, val, len);

    data->queryDesc->newPlan = data->newPlan;

    int numSlicesToDispatch = data->queryDesc->plannedstmt->planTree
        ->nMotionNodes;
    uint64 planSizeInKb = ((uint64) len * (uint64) numSlicesToDispatch)
        / (uint64) 1024;
    elog(LOG,
    "Dispatch new plan instead of old plan, size to "
    "scheduler: " UINT64_FORMAT "KB",
    planSizeInKb);
  }
}

void scheduler_cleanup(SchedulerData *data) {
  if (!data)
    return;

  if (data->stageNo) {
    pfree(data->stageNo);
    data->stageNo = NULL;
  }
  if (data->hosts) {
    pfree(data->hosts);
    data->hosts = NULL;
  }
  if (data->ports) {
    for (int i = 0;
        i
            < data->queryDesc->estate->es_sliceTable->nInitPlans
                + data->queryDesc->estate->es_sliceTable->nMotions + 1; i++) {
      pfree(data->ports[i]);
      data->ports[i] = NULL;
    }
    pfree(data->ports);
    data->ports = NULL;
  }
  if (data->slices) {
    for (int i = 0; i < data->totalStageNum; i++) {
      for (int j = data->slices[i].nStatInst / data->segmentNum; j > 0; j--)
        pfree(data->slices[i].instr[j - 1]);
    }
    pfree(data->slices);
    data->slices = NULL;
  }
  if (data->myInstrument) {
    pfree(data->myInstrument);
    data->myInstrument = NULL;
  }
  pfree(data);
}

/*
 * scheduler_catch_error
 *  Cancel the workermgr work and report error based on bug on schedulerer or
 *  executors.
 */
void scheduler_catch_error(SchedulerData *data) {
  if (!data)
    return;

  int qderrcode = elog_geterrcode();
  bool useQeError = false;
  SchedulerCatchedError *qeError = SchedulerGetLastError(MyScheduler);

  /* Init state means no thread and data is not correct! */
  if (!data->state == SS_INIT) {
    /* Have to figure out exception source */
    SchedulerCollectError(MyScheduler);
    qeError = SchedulerGetLastError(MyScheduler);

    SchedulerCancelQuery(MyScheduler);
  }

  /*
   * When a QE stops executing a command due to an error, as a
   * consequence there can be a cascade of interconnect errors
   * (usually "sender closed connection prematurely") thrown in
   * downstream processes (QEs and QD).  So if we are handling
   * an interconnect error, and a QE hit a more interesting error,
   * we'll let the QE's error report take precedence.
   */
  if (qderrcode == ERRCODE_GP_INTERCONNECTION_ERROR) {
    bool qd_lost_flag = false;
    char *qderrtext = elog_message();

    if (qderrtext && strcmp(qderrtext, CDB_MOTION_LOST_CONTACT_STRING) == 0)
      qd_lost_flag = true;

    if (qeError->errCode != ERRCODE_SUCCESSFUL_COMPLETION) {
      if (qd_lost_flag && qeError->errCode == ERRCODE_GP_INTERCONNECTION_ERROR)
        useQeError = true;
      else if (qeError->errCode != ERRCODE_GP_INTERCONNECTION_ERROR)
        useQeError = true;
    }
  }

  if (useQeError) {
    /* Too bad, our gang got an error. */
    PG_TRY()
          ;
          {
            ereport(
                ERROR,
                (errcode(qeError->errCode), errmsg("%s", qeError->errMessage)));
          }PG_CATCH();
        {
        }PG_END_TRY();
  }
}

void scheduler_print_stats(struct SchedulerData *data, StringInfo buf) {
  if (!data)
    return;
  SchedulerStats *stats = SchedulerGetStats(MyScheduler);
  appendStringInfo(buf, "New Scheduler statistics:\n");
  if (stats->rpcServerSetupTime)
    appendStringInfo(buf, "  setup rpc server: %dms;",
                     stats->rpcServerSetupTime);
  appendStringInfo(buf, "  setup listener port: %dms;",
                   stats->setupListenerPortTime);
  appendStringInfo(buf, "  dispatch task: %dms;", stats->dispatchTaskTime);
  appendStringInfo(buf, "  wait task complete: %dms;\n",
                   stats->waitTaskCompleteTime);
}

void scheduler_receive_computenode_stats(SchedulerData *data,
                                         struct PlanState *planstate) {
  SchedulerRecveiveAllTaskStats(MyScheduler);
  SchedulerCatchedError *err = SchedulerGetLastError(MyScheduler);
  data->state = SS_OK;
  if (err->errCode != ERRCODE_SUCCESSFUL_COMPLETION) {
    data->state = SS_ERROR;
    int errCode = err->errCode;
    ereport(ERROR, (errcode(errCode), errmsg("%s", err->errMessage)));
  }
  data->myInstrument = palloc0(
      sizeof(MyInstrumentation *) * data->totalStageNum);
  int i;
  for (i = 0; i < data->totalStageNum; i++) {
    data->myInstrument[i] = palloc0(
        sizeof(MyInstrumentation) * data->segmentNum);
  }
  scheduler_init_instrumentation(data->myInstrument, data->myInstrument[0],
                                 planstate, 0, data->segmentNum);
  data->slices = palloc0(sizeof(SchedulerSliceStats) * data->totalStageNum);
  for (i = 0; i < data->stageNum * data->segmentNum; i++) {
    int32_t stageNo;
    int32_t segmentNo;
    SchedulerGetOneTaskInfo(MyScheduler, i, &stageNo, &segmentNo);
    SchedulerGetOneTaskStats(MyScheduler, i,
                             &data->myInstrument[stageNo][segmentNo],
                             &data->slices[stageNo].nStatInst);
  }
  for (i = 0; i < data->totalStageNum; i++) {
    data->slices[i].instr = palloc0(
        sizeof(MyInstrumentation *) * data->slices[i].nStatInst);
    for (int j = 0; j < data->segmentNum; j++)
      if (data->slices[i].nStatInst > 0)
        myinstrument_walk_node(&data->myInstrument[i][j],
                               convertMyInstruemntToSliceStat,
                               &data->slices[i]);
  }
}

static void scheduler_init_instrumentation(MyInstrumentation **myInstrument,
                                           MyInstrumentation *instr,
                                           struct PlanState *planstate,
                                           int stageNo, int segmentNum) {
  int segNum = 1;
  switch (nodeTag(planstate->plan)) {
    case T_Motion: {
      Motion *m = (Motion *) planstate->plan;
      stageNo = m->motionID;
      instr = myInstrument[stageNo];
      segNum = segmentNum;
      break;
    }
  }
  if (planstate->lefttree) {
    for (int i = 0; i < segNum; i++) {
      instr[i].leftTree = palloc0(sizeof(MyInstrumentation));
      scheduler_init_instrumentation(myInstrument, instr[i].leftTree,
                                     planstate->lefttree, stageNo, segmentNum);
    }
  }
  if (planstate->righttree) {
    for (int i = 0; i < segNum; i++) {
      instr[i].rightTree = palloc0(sizeof(MyInstrumentation));
      scheduler_init_instrumentation(myInstrument, instr[i].rightTree,
                                     planstate->righttree, stageNo, segmentNum);
    }
  }
  if (nodeTag(planstate->plan) == T_Append) {
    AppendState *appendstate = (AppendState *) planstate;
    ListCell *lc;
    for (int i = 0; i < segNum; i++) {
      MyInstrumentation *subInstr = &instr[i];
      int appendNum = list_length(((Append *) planstate->plan)->appendplans);
      for (int j = 0; j < appendNum; j++) {
        PlanState *appendplanstate = appendstate->appendplans[j];
        subInstr->subTree = palloc0(sizeof(MyInstrumentation));
        scheduler_init_instrumentation(myInstrument, subInstr->subTree,
                                       appendplanstate, stageNo, segmentNum);
        subInstr = subInstr->subTree;
      }
    }
  }
}

static CdbVisitOpt convertMyInstruemntToSliceStat(MyInstrumentation *instr,
                                                  void *context) {
  SchedulerSliceStats *sliceState = (SchedulerSliceStats *) context;
  sliceState->instr[sliceState->iStatInst] = instr;
  sliceState->iStatInst++;
  return CdbVisit_Walk;
}

CdbVisitOpt myinstrument_walk_node(
    MyInstrumentation *instr,
    CdbVisitOpt (*walker)(MyInstrumentation *instr, void *context),
    void *context) {
  CdbVisitOpt whatnext;

  if (instr == NULL)
    return CdbVisit_Walk;

  whatnext = walker(instr, context);
  if (whatnext == CdbVisit_Walk) {
    if (instr->leftTree && whatnext == CdbVisit_Walk)
      whatnext = myinstrument_walk_node(instr->leftTree, walker, context);
    if (instr->rightTree && whatnext == CdbVisit_Walk)
      whatnext = myinstrument_walk_node(instr->rightTree, walker, context);
    if (instr->subTree && whatnext == CdbVisit_Walk)
      whatnext = myinstrument_walk_node(instr->subTree, walker, context);
    if (instr->subplan && whatnext == CdbVisit_Walk) {
      whatnext = myinstrument_walk_node(instr->subplan, walker, context);
    }
    if (instr->subplanSibling && whatnext == CdbVisit_Walk) {
      whatnext = myinstrument_walk_node(instr->subplanSibling, walker, context);
    }
  } else if (whatnext == CdbVisit_Skip) {
    whatnext = CdbVisit_Walk;
  }

  Assert(whatnext != CdbVisit_Skip);
  return whatnext;
}