text: /* -------------------------------------------------------------------------
   *
   * redistrib.c
   *	  Routines related to online data redistribution
   *
   * Copyright (c) 2010-2012 Postgres-XC Development Group
   *
   *
   * IDENTIFICATION
   *	  src/backend/pgxc/locator/redistrib.c
   *
   * -------------------------------------------------------------------------
   */
  
  #include "postgres.h"
  #include "knl/knl_variable.h"
  #include "miscadmin.h"
  
  #include "access/cstore_am.h"
  #include "access/hash.h"
  #include "access/heapam.h"
  #include "access/htup.h"
  #include "access/reloptions.h"
  #include "access/tableam.h"
  #include "access/xact.h"
  #include "catalog/namespace.h"
  #include "catalog/pg_partition_fn.h"
  #include "catalog/pg_type.h"
  #include "catalog/pgxc_node.h"
  #include "commands/tablecmds.h"
  #include "cstore.h"
  #include "nodes/makefuncs.h"
  #include "nodes/parsenodes.h"
  #include "nodes/value.h"
  #include "pgxc/copyops.h"
  #include "pgxc/execRemote.h"
  #include "pgxc/pgxc.h"
  #include "pgxc/redistrib.h"
  #include "pgxc/remotecopy.h"
  #include "storage/lock/lwlock.h"
  #include "storage/off.h"
  #include "storage/proc.h"
  #include "storage/shmem.h"
  #include "utils/lsyscache.h"
  #include "utils/rel.h"
  #include "utils/rel_gs.h"
  #include "utils/snapmgr.h"
  #include "utils/syscache.h"
  #include "utils/elog.h"
  
  #define IsCommandTypePreUpdate(x) ((x) == CATALOG_UPDATE_BEFORE || (x) == CATALOG_UPDATE_BOTH)
  #define IsCommandTypePostUpdate(x) ((x) == CATALOG_UPDATE_AFTER || (x) == CATALOG_UPDATE_BOTH)
  
  #define DatumGetItemPointer(X) ((ItemPointer)DatumGetPointer(X))
  #define ItemPointerGetDatum(X) PointerGetDatum(X)
  #define PG_GETARG_ITEMPOINTER(n) DatumGetItemPointer(PG_GETARG_DATUM(n))
  #define PG_RETURN_ITEMPOINTER(x) return ItemPointerGetDatum(x)
  #define DST_MAX_SIZE 64
  
  #define RedisRelCtidFormat "(%u.%u)"
  
  /* Functions used for the execution of redistribution commands */
  static void distrib_execute_query(const char* sql, bool is_temp, ExecNodes* exec_nodes);
  static void distrib_execute_command(RedistribState* distribState, RedistribCommand* command);
  static void distrib_copy_to(RedistribState* distribState);
  static void distrib_copy_from(RedistribState* distribState, ExecNodes* exec_nodes);
  static void distrib_truncate(RedistribState* distribState, ExecNodes* exec_nodes);
  static void distrib_reindex(RedistribState* distribState, ExecNodes* exec_nodes);
  static void distrib_delete_hash(RedistribState* distribState, ExecNodes* exec_nodes);
  
  /* Functions used to build the command list */
  #ifdef ENABLE_MULTIPLE_NODES
  static void pgxc_redist_build_entry(
      RedistribState* distribState, RelationLocInfo* oldLocInfo, RelationLocInfo* newLocInfo);
  static void pgxc_redist_build_replicate(
      RedistribState* distribState, RelationLocInfo* oldLocInfo, RelationLocInfo* newLocInfo);
  static void pgxc_redist_build_replicate_to_distrib(
      RedistribState* distribState, RelationLocInfo* oldLocInfo, RelationLocInfo* newLocInfo);
  
  static void pgxc_redist_build_default(RedistribState* distribState);
  static void pgxc_redist_add_reindex(RedistribState* distribState);
  #endif
  /*
   * PGXCRedistribTable
   * Execute redistribution operations after catalog update
   */
  void PGXCRedistribTable(RedistribState* distribState, RedistribCatalog type)
  {
      ListCell* item = NULL;
  
      /* Nothing to do if no redistribution operation */
      if (distribState == NULL)
          return;
  
      /* Nothing to do if on remote node */
      if (IS_PGXC_DATANODE || IsConnFromCoord())
          return;
  
      /* Execute each command if necessary */
      foreach (item, distribState->commands) {
          RedistribCommand* command = (RedistribCommand*)lfirst(item);
  
          /* Check if command can be run */
          if (!IsCommandTypePostUpdate(type) && IsCommandTypePostUpdate(command->updateState))
              continue;
          if (!IsCommandTypePreUpdate(type) && IsCommandTypePreUpdate(command->updateState))
              continue;
  
          /* Now enter in execution list */
          distrib_execute_command(distribState, command);
      }
  }
  
  /*
   * PGXCRedistribCreateCommandList
   * Look for the list of necessary commands to perform table redistribution.
   */
  void PGXCRedistribCreateCommandList(RedistribState* distribState, RelationLocInfo* newLocInfo)
  {
  #ifndef ENABLE_MULTIPLE_NODES
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return;
  #else
      Relation rel;
      RelationLocInfo* oldLocInfo;
  
      rel = relation_open(distribState->relid, NoLock);
      oldLocInfo = RelationGetLocInfo(rel);
  
      /* Build redistribution command list */
      pgxc_redist_build_entry(distribState, oldLocInfo, newLocInfo);
  
      relation_close(rel, NoLock);
  #endif
  }
  
  #ifdef ENABLE_MULTIPLE_NODES
  /*
   * pgxc_redist_build_entry
   * Entry point for command list building
   */
  static void pgxc_redist_build_entry(
      RedistribState* distribState, RelationLocInfo* oldLocInfo, RelationLocInfo* newLocInfo)
  {
      /* If distribution has not changed at all, nothing to do */
      if (IsLocatorInfoEqual(oldLocInfo, newLocInfo))
          return;
  
      /* Evaluate cases for replicated tables */
      pgxc_redist_build_replicate(distribState, oldLocInfo, newLocInfo);
  
      /* Evaluate cases for replicated to distributed tables */
      pgxc_redist_build_replicate_to_distrib(distribState, oldLocInfo, newLocInfo);
  
      /* PGXC:perform more complex builds of command list */
  
      /* Fallback to default */
      pgxc_redist_build_default(distribState);
  }
  
  /*
   * pgxc_redist_build_replicate_to_distrib
   * Build redistribution command list from replicated to distributed
   * table.
   */
  static void pgxc_redist_build_replicate_to_distrib(
      RedistribState* distribState, RelationLocInfo* oldLocInfo, RelationLocInfo* newLocInfo)
  {
      List* removedNodes = NULL;
      List* newNodes = NULL;
  
      /* If a command list has already been built, nothing to do */
      if (list_length(distribState->commands) != 0)
          return;
  
      /* Redistribution is done from replication to distributed (with value) */
      if (!IsRelationReplicated(oldLocInfo) || !IsRelationDistributedByValue(newLocInfo))
          return;
  
      /* Get the list of nodes that are added to the relation */
      removedNodes = list_difference_int(oldLocInfo->nodeList, newLocInfo->nodeList);
  
      /* Get the list of nodes that are removed from relation */
      newNodes = list_difference_int(newLocInfo->nodeList, oldLocInfo->nodeList);
      /*
       * If some nodes are added, turn back to default, we need to fetch data
       * and then redistribute it properly.
       */
      if (newNodes != NIL)
          return;
  
      /* Nodes removed have to be truncated, so add a TRUNCATE commands to removed nodes */
      if (removedNodes != NIL) {
          ExecNodes* execNodes = makeNode(ExecNodes);
          execNodes->nodeList = removedNodes;
          /* Add TRUNCATE command */
          distribState->commands =
              lappend(distribState->commands, makeRedistribCommand(DISTRIB_TRUNCATE, CATALOG_UPDATE_BEFORE, execNodes));
      }
  
      /*
       * If the table is redistributed to a single node, a TRUNCATE on removed nodes
       * is sufficient so leave here.
       */
      if (list_length(newLocInfo->nodeList) == 1) {
          /* Add REINDEX command if necessary */
          pgxc_redist_add_reindex(distribState);
          return;
      }
  
      /*
       * If we are here we are sure that redistribution only requires to delete data on remote
       * nodes on the new subset of nodes. So launch to remote nodes a DELETE command that only
       * eliminates the data not verifying the new hashing condition.
       */
      if (newLocInfo->locatorType == LOCATOR_TYPE_HASH) {
          ExecNodes* execNodes = makeNode(ExecNodes);
          execNodes->nodeList = newLocInfo->nodeList;
          distribState->commands =
              lappend(distribState->commands, makeRedistribCommand(DISTRIB_DELETE_HASH, CATALOG_UPDATE_AFTER, execNodes));
      } else if (newLocInfo->locatorType == LOCATOR_TYPE_MODULO) {
          ExecNodes* execNodes = makeNode(ExecNodes);
          execNodes->nodeList = newLocInfo->nodeList;
          distribState->commands = lappend(
              distribState->commands, makeRedistribCommand(DISTRIB_DELETE_MODULO, CATALOG_UPDATE_AFTER, execNodes));
      } else
          ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("Incorrect redistribution operation")));
  
      /* Add REINDEX command if necessary */
      pgxc_redist_add_reindex(distribState);
  }
  
  /*
   * pgxc_redist_build_replicate
   * Build redistribution command list for replicated tables
   */
  static void pgxc_redist_build_replicate(
      RedistribState* distribState, RelationLocInfo* oldLocInfo, RelationLocInfo* newLocInfo)
  {
      List* removedNodes = NULL;
      List* newNodes = NULL;
  
      /* If a command list has already been built, nothing to do */
      if (list_length(distribState->commands) != 0)
          return;
  
      /* Case of a replicated table whose set of nodes is changed */
      if (!IsRelationReplicated(newLocInfo) || !IsRelationReplicated(oldLocInfo))
          return;
  
      /* Get the list of nodes that are added to the relation */
      removedNodes = list_difference_int(oldLocInfo->nodeList, newLocInfo->nodeList);
  
      /* Get the list of nodes that are removed from relation */
      newNodes = list_difference_int(newLocInfo->nodeList, oldLocInfo->nodeList);
      /*
       * If nodes have to be added, we need to fetch data for redistribution first.
       * So add a COPY TO command to fetch data.
       */
      if (newNodes != NIL) {
          /* Add COPY TO command */
          distribState->commands =
              lappend(distribState->commands, makeRedistribCommand(DISTRIB_COPY_TO, CATALOG_UPDATE_BEFORE, NULL));
      }
  
      /* Nodes removed have to be truncated, so add a TRUNCATE commands to removed nodes */
      if (removedNodes != NIL) {
          ExecNodes* execNodes = makeNode(ExecNodes);
          execNodes->nodeList = removedNodes;
          /* Add TRUNCATE command */
          distribState->commands =
              lappend(distribState->commands, makeRedistribCommand(DISTRIB_TRUNCATE, CATALOG_UPDATE_BEFORE, execNodes));
      }
  
      /* If necessary, COPY the data obtained at first step to the new nodes. */
      if (newNodes != NIL) {
          ExecNodes* execNodes = makeNode(ExecNodes);
          execNodes->nodeList = newNodes;
          /* Add COPY FROM command */
          distribState->commands =
              lappend(distribState->commands, makeRedistribCommand(DISTRIB_COPY_FROM, CATALOG_UPDATE_AFTER, execNodes));
      }
  
      /* Add REINDEX command if necessary */
      pgxc_redist_add_reindex(distribState);
  }
  
  /*
   * pgxc_redist_build_default
   * Build a default list consisting of
   * COPY TO -> TRUNCATE -> COPY FROM ( -> REINDEX )
   */
  static void pgxc_redist_build_default(RedistribState* distribState)
  {
      /* If a command list has already been built, nothing to do */
      if (list_length(distribState->commands) != 0)
          return;
  
      /* COPY TO command */
      distribState->commands =
          lappend(distribState->commands, makeRedistribCommand(DISTRIB_COPY_TO, CATALOG_UPDATE_BEFORE, NULL));
      /* TRUNCATE command */
      distribState->commands =
          lappend(distribState->commands, makeRedistribCommand(DISTRIB_TRUNCATE, CATALOG_UPDATE_BEFORE, NULL));
      /* COPY FROM command */
      distribState->commands =
          lappend(distribState->commands, makeRedistribCommand(DISTRIB_COPY_FROM, CATALOG_UPDATE_AFTER, NULL));
  
      /* REINDEX command */
      pgxc_redist_add_reindex(distribState);
  }
  
  /*
   * pgxc_redist_build_reindex
   * Add a reindex command if necessary
   */
  static void pgxc_redist_add_reindex(RedistribState* distribState)
  {
      Relation rel;
  
      rel = relation_open(distribState->relid, NoLock);
      /* Build REINDEX command if necessary */
      if (RelationGetIndexList(rel) != NIL) {
          distribState->commands =
              lappend(distribState->commands, makeRedistribCommand(DISTRIB_REINDEX, CATALOG_UPDATE_AFTER, NULL));
      }
  
      relation_close(rel, NoLock);
  }
  #endif
  
  /*
   * distrib_execute_command
   * Execute a redistribution operation
   */
  static void distrib_execute_command(RedistribState* distribState, RedistribCommand* command)
  {
      /* Execute redistribution command */
      switch (command->type) {
          case DISTRIB_COPY_TO:
              distrib_copy_to(distribState);
              break;
          case DISTRIB_COPY_FROM:
              distrib_copy_from(distribState, command->execNodes);
              break;
          case DISTRIB_TRUNCATE:
              distrib_truncate(distribState, command->execNodes);
              break;
          case DISTRIB_REINDEX:
              distrib_reindex(distribState, command->execNodes);
              break;
          case DISTRIB_DELETE_HASH:
          case DISTRIB_DELETE_MODULO:
              distrib_delete_hash(distribState, command->execNodes);
              break;
          case DISTRIB_NONE:
          default:
              Assert(0); /* Should not happen */
              break;
      }
  }
  
  /*
   * distrib_copy_to
   * Copy all the data of table to be distributed.
   * This data is saved in a tuplestore saved in distribution state.
   * a COPY FROM operation is always done on nodes determined by the locator data
   * in catalogs, explaining why this cannot be done on a subset of nodes. It also
   * insures that no read operations are done on nodes where data is not yet located.
   */
  static void distrib_copy_to(RedistribState* distribState)
  {
      Oid relOid = distribState->relid;
      Relation rel;
      RemoteCopyOptions* options = NULL;
      RemoteCopyData* copyState = NULL;
      Tuplestorestate* store = NULL; /* Storage of redistributed data */
  
      /* Fetch necessary data to prepare for the table data acquisition */
      options = makeRemoteCopyOptions();
  
      /* All the fields are separated by tabs in redistribution */
      options->rco_delim = (char*)palloc(2);
      options->rco_delim[0] = COPYOPS_DELIMITER;
      options->rco_delim[1] = '\0';
  
      copyState = (RemoteCopyData*)palloc0(sizeof(RemoteCopyData));
      copyState->is_from = false;
  
      /* A sufficient lock level needs to be taken at a higher level */
      rel = relation_open(relOid, NoLock);
      RemoteCopy_GetRelationLoc(copyState, rel, NIL);
      RemoteCopy_BuildStatement(copyState, rel, options, NIL, NIL);
  
      /* Inform client of operation being done */
      ereport(DEBUG1,
          (errmsg("Copying data for relation \"%s.%s\"",
              get_namespace_name(RelationGetNamespace(rel)),
              RelationGetRelationName(rel))));
  
      /* Begin the COPY process */
      copyState->connections =
          DataNodeCopyBegin(copyState->query_buf.data, copyState->exec_nodes->nodeList, GetActiveSnapshot());
  
      /* Create tuplestore storage */
      store = tuplestore_begin_heap(true, false, u_sess->attr.attr_memory.work_mem);
  
      /* Then get rows and copy them to the tuplestore used for redistribution */
      DataNodeCopyOut(copyState->exec_nodes,
          copyState->connections,
          RelationGetDescr(rel), /* Need also to set up the tuple descriptor */
          NULL,
          store, /* Tuplestore used for redistribution */
          REMOTE_COPY_TUPLESTORE);
  
      /* Do necessary clean-up */
      FreeRemoteCopyOptions(options);
  
      /* Lock is maintained until transaction commits */
      relation_close(rel, NoLock);
  
      /* Save results */
      distribState->store = store;
  }
  
  /*
   * PGXCDistribTableCopyFrom
   * Execute commands related to COPY FROM
   * Redistribute all the data of table with a COPY FROM from given tuplestore.
   */
  static void distrib_copy_from(RedistribState* distribState, ExecNodes* exec_nodes)
  {
      Oid relOid = distribState->relid;
      Tuplestorestate* store = distribState->store;
      Relation rel;
      RemoteCopyOptions* options = NULL;
      RemoteCopyData* copyState = NULL;
      bool replicated = false, contains_tuple = true;
      TupleDesc tupdesc;
  
      /* Nothing to do if on remote node */
      if (IS_PGXC_DATANODE || IsConnFromCoord())
          return;
  
      /* Fetch necessary data to prepare for the table data acquisition */
      options = makeRemoteCopyOptions();
      /* All the fields are separated by tabs in redistribution */
      options->rco_delim = (char*)palloc(2);
      options->rco_delim[0] = COPYOPS_DELIMITER;
      options->rco_delim[1] = '\0';
  
      copyState = (RemoteCopyData*)palloc0(sizeof(RemoteCopyData));
      copyState->is_from = true;
  
      /* A sufficient lock level needs to be taken at a higher level */
      rel = relation_open(relOid, NoLock);
      RemoteCopy_GetRelationLoc(copyState, rel, NIL);
      RemoteCopy_BuildStatement(copyState, rel, options, NIL, NIL);
  
      /*
       * When building COPY FROM command in redistribution list,
       * use the list of nodes that has been calculated there.
       * It might be possible that this COPY is done only on a portion of nodes.
       */
      if (exec_nodes != NULL && exec_nodes->nodeList != NIL) {
          copyState->exec_nodes->nodeList = exec_nodes->nodeList;
          copyState->rel_loc->nodeList = exec_nodes->nodeList;
      }
  
      tupdesc = RelationGetDescr(rel);
  
      /* Inform client of operation being done */
      ereport(DEBUG1,
          (errmsg("Redistributing data for relation \"%s.%s\"",
              get_namespace_name(RelationGetNamespace(rel)),
              RelationGetRelationName(rel))));
  
      /* Begin redistribution on remote nodes */
      copyState->connections =
          DataNodeCopyBegin(copyState->query_buf.data, copyState->exec_nodes->nodeList, GetActiveSnapshot());
  
      /* Transform each tuple stored into a COPY message and send it to remote nodes */
      while (contains_tuple) {
          char* data = NULL;
          int len;
          Form_pg_attribute* attr = tupdesc->attrs;
          TupleTableSlot* slot = NULL;
          ExecNodes* local_execnodes = NULL;
  
          Oid* att_type = (Oid*)palloc(tupdesc->natts * sizeof(Oid));
  
          /* Build table slot for this relation */
          slot = MakeSingleTupleTableSlot(tupdesc);
  
          /* Get tuple slot from the tuplestore */
          contains_tuple = tuplestore_gettupleslot(store, true, false, slot);
          if (!contains_tuple) {
              ExecDropSingleTupleTableSlot(slot);
              break;
          }
  
          /* Make sure the tuple is fully deconstructed */
          tableam_tslot_getallattrs(slot);
  
          /* Find value of distribution column if necessary */
          for (int i = 0; i < tupdesc->natts; i++) {
              att_type[i] = attr[i]->atttypid;
          }
  
          local_execnodes = GetRelationNodes(copyState->rel_loc,
              slot->tts_values,
              slot->tts_isnull,
              att_type,
              copyState->idx_dist_by_col,
              RELATION_ACCESS_INSERT,
              false);
  
          pfree(att_type);
  
          /* Build message to be sent to Datanodes */
          data = CopyOps_BuildOneRowTo(tupdesc, slot->tts_values, slot->tts_isnull, &len);
  
          /* Take a copy of the node lists so as not to interfere with locator info */
          local_execnodes->primarynodelist = list_copy(local_execnodes->primarynodelist);
          local_execnodes->nodeList = list_copy(local_execnodes->nodeList);
  
          /* Process data to Datanodes */
          DataNodeCopyIn(data, len, NULL, local_execnodes, copyState->connections);
  
          /* Clean up */
          pfree(data);
          FreeExecNodes(&local_execnodes);
          (void)ExecClearTuple(slot);
          ExecDropSingleTupleTableSlot(slot);
      }
  
      /* Finish the redistribution process */
      replicated = copyState->rel_loc->locatorType == LOCATOR_TYPE_REPLICATED;
      DataNodeCopyFinish(copyState->connections,
          u_sess->pgxc_cxt.NumDataNodes,
          replicated ? PGXCNodeGetNodeId(u_sess->pgxc_cxt.primary_data_node, PGXC_NODE_DATANODE) : -1,
          replicated ? COMBINE_TYPE_SAME : COMBINE_TYPE_SUM,
          rel);
  
      /* Lock is maintained until transaction commits */
      relation_close(rel, NoLock);
  }
  
  /*
   * distrib_truncate
   * Truncate all the data of specified table.
   * This is used as a second step of online data redistribution.
   */
  static void distrib_truncate(RedistribState* distribState, ExecNodes* exec_nodes)
  {
      Relation rel;
      StringInfo buf;
      Oid relOid = distribState->relid;
  
      /* Nothing to do if on remote node */
      if (IS_PGXC_DATANODE || IsConnFromCoord())
          return;
  
      /* A sufficient lock level needs to be taken at a higher level */
      rel = relation_open(relOid, NoLock);
  
      /* Inform client of operation being done */
      ereport(DEBUG1,
          (errmsg("Truncating data for relation \"%s.%s\"",
              get_namespace_name(RelationGetNamespace(rel)),
              RelationGetRelationName(rel))));
  
      /* Initialize buffer */
      buf = makeStringInfo();
  
      /* Build query to clean up table before redistribution */
      appendStringInfo(
          buf, "TRUNCATE %s.%s", get_namespace_name(RelationGetNamespace(rel)), RelationGetRelationName(rel));
  
      /*
       * Lock is maintained until transaction commits,
       * relation needs also to be closed before effectively launching the query.
       */
      relation_close(rel, NoLock);
  
      /* Execute the query */
      distrib_execute_query(buf->data, IsTempTable(relOid), exec_nodes);
  
      /* Clean buffers */
      pfree(buf->data);
      pfree(buf);
  }
  
  /*
   * distrib_reindex
   * Reindex the table that has been redistributed
   */
  static void distrib_reindex(RedistribState* distribState, ExecNodes* exec_nodes)
  {
      Relation rel;
      StringInfo buf;
      Oid relOid = distribState->relid;
  
      /* Nothing to do if on remote node */
      if (IS_PGXC_DATANODE || IsConnFromCoord())
          return;
  
      /* A sufficient lock level needs to be taken at a higher level */
      rel = relation_open(relOid, NoLock);
  
      /* Inform client of operation being done */
      ereport(DEBUG1,
          (errmsg("Reindexing relation \"%s.%s\"",
              get_namespace_name(RelationGetNamespace(rel)),
              RelationGetRelationName(rel))));
  
      /* Initialize buffer */
      buf = makeStringInfo();
  
      /* Generate the query */
      appendStringInfo(
          buf, "REINDEX TABLE %s.%s", get_namespace_name(RelationGetNamespace(rel)), RelationGetRelationName(rel));
  
      /* Execute the query */
      distrib_execute_query(buf->data, IsTempTable(relOid), exec_nodes);
  
      /* Clean buffers */
      pfree(buf->data);
      pfree(buf);
  
      /* Lock is maintained until transaction commits */
      relation_close(rel, NoLock);
  }
  
  /*
   * distrib_delete_hash
   * Perform a partial tuple deletion of remote tuples not checking the correct hash
   * condition. The new distribution condition is set up in exec_nodes when building
   * the command list.
   */
  static void distrib_delete_hash(RedistribState* distribState, ExecNodes* exec_nodes)
  {
      Relation rel;
      StringInfo buf;
      Oid relOid = distribState->relid;
      ListCell* item = NULL;
  
      /* Nothing to do if on remote node */
      if (IS_PGXC_DATANODE || IsConnFromCoord())
          return;
  
      /* A sufficient lock level needs to be taken at a higher level */
      rel = relation_open(relOid, NoLock);
  
      /* Inform client of operation being done */
      ereport(DEBUG1,
          (errmsg("Deleting necessary tuples \"%s.%s\"",
              get_namespace_name(RelationGetNamespace(rel)),
              RelationGetRelationName(rel))));
  
      /* Initialize buffer */
      buf = makeStringInfo();
  
      /* Build query to clean up table before redistribution */
      appendStringInfo(
          buf, "DELETE FROM %s.%s", get_namespace_name(RelationGetNamespace(rel)), RelationGetRelationName(rel));
  
      /*
       * Launch the DELETE query to each node as the DELETE depends on
       * local conditions for each node.
       */
      foreach (item, exec_nodes->nodeList) {
          StringInfo buf2;
          char* colname = NULL;
          RelationLocInfo* locinfo = RelationGetLocInfo(rel);
          int nodenum = lfirst_int(item);
          int nodepos = 0;
          ExecNodes* local_exec_nodes = makeNode(ExecNodes);
          ListCell* item2 = NULL;
  
          /* Here the query is launched to a unique node */
          local_exec_nodes->nodeList = lappend_int(NIL, nodenum);
  
          /* Get the hash type of relation */
  
          /* Get function hash name */
  
          /* Get distribution column name */
          if (IsRelationDistributedByValue(locinfo)) {
              colname = NULL;
          } else
              ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("Incorrect redistribution operation")));
  
          /*
           * Find the correct node position in node list of locator information.
           * So scan the node list and fetch the position of node.
           */
          foreach (item2, locinfo->nodeList) {
              int loc = lfirst_int(item2);
              if (loc == nodenum)
                  break;
              nodepos++;
          }
  
          /*
           * Then build the WHERE clause for deletion.
           * The condition that allows to keep the tuples on remote nodes
           * is of the type "RemoteNodeNumber != abs(hash_func(dis_col)) % NumDatanodes".
           * the remote Datanode has no knowledge of its position in cluster so this
           * number needs to be compiled locally on Coordinator.
           * Taking the absolute value is necessary as hash may return a negative value.
           * For hash distributions a condition with correct hash function is used.
           * For modulo distribution, well we might need a hash function call but not
           * all the time, this is determined implicitely by get_compute_hash_function.
           */
          buf2 = makeStringInfo();
          {
              /* Lets leave NULLs on the first node and delete from the rest */
              if (nodepos != 0)
                  appendStringInfo(buf2,
                      "%s WHERE %s IS NULL OR abs(%s) %% %d != %d",
                      buf->data,
                      colname,
                      colname,
                      list_length(locinfo->nodeList),
                      nodepos);
              else
                  appendStringInfo(
                      buf2, "%s WHERE abs(%s) %% %d != %d", buf->data, colname, list_length(locinfo->nodeList), nodepos);
          }
  
          /* Then launch this single query */
          distrib_execute_query(buf2->data, IsTempTable(relOid), local_exec_nodes);
  
          FreeExecNodes(&local_exec_nodes);
          pfree(buf2->data);
          pfree(buf2);
      }
  
      relation_close(rel, NoLock);
  
      /* Clean buffers */
      pfree(buf->data);
      pfree(buf);
  }
  
  /*
   * makeRedistribState
   * Build a distribution state operator
   */
  RedistribState* makeRedistribState(Oid relOid)
  {
  #ifndef ENABLE_MULTIPLE_NODES
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return NULL;
  #else
  
      RedistribState* res = (RedistribState*)palloc(sizeof(RedistribState));
      res->relid = relOid;
      res->commands = NIL;
      res->store = NULL;
      return res;
  
  #endif
  }
  
  /*
   * FreeRedistribState
   * Free given distribution state
   */
  void FreeRedistribState(RedistribState* state)
  {
  #ifndef ENABLE_MULTIPLE_NODES
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return;
  #else
      ListCell* item = NULL;
  
      /* Leave if nothing to do */
      if (!state)
          return;
  
      foreach (item, state->commands)
          FreeRedistribCommand((RedistribCommand*)lfirst(item));
      if (list_length(state->commands) > 0)
          list_free(state->commands);
      if (state->store)
          tuplestore_clear(state->store);
  #endif
  }
  
  /*
   * makeRedistribCommand
   * Build a distribution command
   */
  RedistribCommand* makeRedistribCommand(RedistribOperation type, RedistribCatalog updateState, ExecNodes* nodes)
  {
      RedistribCommand* res = (RedistribCommand*)palloc0(sizeof(RedistribCommand));
      res->type = type;
      res->updateState = updateState;
      res->execNodes = nodes;
      return res;
  }
  
  /*
   * FreeRedistribCommand
   * Free given distribution command
   */
  void FreeRedistribCommand(RedistribCommand* command)
  {
      ExecNodes* nodes = NULL;
      /* Leave if nothing to do */
      if (command == NULL)
          return;
      nodes = command->execNodes;
  
      if (nodes != NULL)
          FreeExecNodes(&nodes);
      pfree(command);
  }
  
  /*
   * distrib_execute_query
   * Execute single raw query on given list of nodes
   */
  static void distrib_execute_query(const char* sql, bool is_temp, ExecNodes* exec_nodes)
  {
      RemoteQuery* step = makeNode(RemoteQuery);
      step->combine_type = COMBINE_TYPE_SAME;
      step->exec_nodes = exec_nodes;
      step->sql_statement = pstrdup(sql);
      step->force_autocommit = false;
  
      /* Redistribution operations only concern Datanodes */
      step->exec_type = EXEC_ON_DATANODES;
      step->is_temp = is_temp;
      ExecRemoteUtility(step);
      pfree(step->sql_statement);
      pfree(step);
  
      /* Be sure to advance the command counter after the last command */
      CommandCounterIncrement();
  }
  
  /*
   * - Brief: Get relation's ctid in Reloptions
   * - Parameter:
   *      @rel_name: relation's name
   *      @partition_name: partition's name
   *      @ctid_type: start_ctid or end_ctid
   *      @result : the start or end ctid
   */
  void get_redis_rel_ctid(const char* rel_name, const char* partition_name, RedisCtidType ctid_type, ItemPointer result)
  {
      Relation rel = NULL, fake_rel = NULL;
      Partition partition = NULL;
      Oid relid = InvalidOid, part_oid = InvalidOid;
      ItemPointerData start_ctid, end_ctid;
      List* names = NIL;
  
      names = stringToQualifiedNameList(rel_name);
  
      /* if rel_name doesn't exist, RangeVarGetRelid will emit ERROR. */
      relid = RangeVarGetRelid(makeRangeVarFromNameList(names), NoLock, false);
      if (names != NIL)
          list_free(names);
  
      rel = try_relation_open(relid, AccessShareLock);
      if (rel == NULL) {
          ereport(WARNING, (errmsg("try open relation %s failed", rel_name)));
          return;
      }
  
      if (partition_name != NULL) {
          /* For partitioned table, */
          part_oid = partitionNameGetPartitionOid(
              relid, partition_name, PART_OBJ_TYPE_TABLE_PARTITION, NoLock, false, false, NULL, NULL, NoLock);
      }
  
      if (OidIsValid(part_oid)) {
          partition = partitionOpen(rel, part_oid, AccessShareLock);
          fake_rel = partitionGetRelation(rel, partition);
          RelationGetCtids(fake_rel, &start_ctid, &end_ctid);
          releaseDummyRelation(&fake_rel);
          partitionClose(rel, partition, AccessShareLock);
      } else {
          RelationGetCtids(rel, &start_ctid, &end_ctid);
      }
      relation_close(rel, AccessShareLock);
  
      /* Fetch relation's append mode status with given relid */
      if (ctid_type == REDIS_START_CTID)
          ItemPointerCopy(&start_ctid, result);
      else
          ItemPointerCopy(&end_ctid, result);
  }
  
  /*
   * - Brief: get max cuid. The caller should already have
   *          exclusive lock on relation.
   * - Parameter:
   *      @rel:  columnar store table
   * - Return:
   *      @tid:  returned max tid
   */
  void col_get_max_tid(const Relation rel, ItemPointer tid)
  {
      Oid cudescOid = rel->rd_rel->relcudescrelid;
      uint32 maxCUID = CStore::GetMaxCUID(cudescOid, rel->rd_att);
  
      Assert(tid != NULL);
      /*
       * Columnar store don't care about offsetnumber because we
       * always use append mode but to make it a valid
       * itempointer we set DefaultFullCUSize as the max offsetnumber.
       */
      ItemPointerSet(tid, maxCUID, DefaultFullCUSize);
  }
  
  /*
   * - Brief: Add start_ctid/end_ctid to DefElem List
   *			convert ctid item to string mode.
   * - Parameter:
   *      @def_list: input parameter, DefElem list to be added
   *		@item_name: start_ctid_internal/end_ctid_internal
   *		@ctid: ctid values
   * - Return:
   *      @void:
   */
  List* add_ctid_string_to_reloptions(List* def_list, const char* item_name, ItemPointer ctid)
  {
      char ctid_str[DST_MAX_SIZE];
      errno_t rc;
      int ret;
  
      /* 64 is not a friendly item. */
      rc = memset_s(ctid_str, DST_MAX_SIZE, 0, DST_MAX_SIZE);
      securec_check_c(rc, "\0", "\0");
      ret = snprintf_s(ctid_str,
          DST_MAX_SIZE,
          DST_MAX_SIZE - 1,
          RedisRelCtidFormat,
          RedisCtidGetBlockNumber(ctid),
          RedisCtidGetOffsetNumber(ctid));
      securec_check_ss_c(ret, "\0", "\0");
      def_list = lappend(def_list, makeDefElem(pstrdup(item_name), (Node*)makeString(pstrdup(ctid_str))));
      return def_list;
  }
  
  /*
   * - Brief: Remove redis reloptions from DefElem List*
   * - Parameter:
   *      @rel: pointer to reloptions list
   * - Return:
   *      @void:
   */
  void RemoveRedisRelOptionsFromList(List** reloptions)
  {
      ereport(DEBUG3, (errmsg("[%s()]: shouldn't run here", __FUNCTION__)));
      return;
  } 
  
  /*
   * - Brief: Modify relation's redistribute options
   * - Parameter:
   *      @rel: relation that needs to check
   *		@action: append/normal/refresh
   * - Return:
   *      @void:
   */
  List* AlterTableSetRedistribute(Relation rel, RedisRelAction action, char *merge_list)
  {
      List* rel_options = NIL;
      bool isCUFormat = false;
      ItemPointerData start_ctid, end_ctid;
  
      Assert(rel);
      merge_list = NULL;
      isCUFormat = RelationIsCUFormat(rel);
  
      switch (action) {
          case REDIS_REL_APPEND:
          case REDIS_REL_END_CATCHUP: {
              if (RELATION_IS_PARTITIONED(rel)) {
                  rel_options =
                      lappend(rel_options, makeDefElem(pstrdup("append_mode_internal"), (Node*)makeInteger(action)));
                  /* set start_ctid */
                  if (isCUFormat)
                      ItemPointerSet(&start_ctid, FirstCUID + 1, 0);
                  else
                      ItemPointerSet(&start_ctid, 0, 0);
  
                  ItemPointerSet(&end_ctid, 0, 0);
  
                  rel_options = add_ctid_string_to_reloptions(rel_options, "start_ctid_internal", &start_ctid);
                  rel_options = add_ctid_string_to_reloptions(rel_options, "end_ctid_internal", &end_ctid);
                  return rel_options;
              }
  
              /*
               * 1. set append_mode_internal
               * 			start_ctid = end_ctid + 1
               *			end_ctid = max_ctid
               */
              if (IS_PGXC_DATANODE) {
                  if (isCUFormat)
                      col_get_max_tid(rel, &end_ctid);
                  else
                      heap_get_max_tid(rel, &end_ctid);
              } else {
                  ItemPointerSet(&end_ctid, 0, 0);
              }
  
              /* 2. update redis rel info
               *
               * We apply the lappend for append_mode_internal for both if/else
               * case here to prevent the scenario like
               * 1. Table is first to set append_mode_internal = read_only
               * 2. Table is then to set append_mode_internal = on
               * the secondary setup will not update append_mode_internal if the
               * append is not applied to else part. Although such scenario will
               * rarely happen. transformRelOptions later will only update the value
               * of append_mode_internal.
               */
              rel_options =
                  lappend(rel_options, makeDefElem(pstrdup("append_mode_internal"), (Node*)makeInteger(action)));
              if (!RelationInRedistribute(rel)) {
                  /* set start_ctid */
                  if (isCUFormat)
                      ItemPointerSet(&start_ctid, FirstCUID + 1, 0);
                  else
                      ItemPointerSet(&start_ctid, 0, 0);
              } else {
                  if (!IS_PGXC_DATANODE) {
                      ItemPointerSet(&start_ctid, 0, 0);
                  } else {
                      /*  relation is redistributing */
                      ItemPointerData cur_start_ctid, cur_end_ctid;
  
                      RelationGetCtids(rel, &cur_start_ctid, &cur_end_ctid);
  
                      /* set start_ctid */
                      if (isCUFormat) {
                          /*
                           * Since columnar store table always use append mode, we
                           * just end_ctid.cuid + 1 to get the start_ctid
                           */
                          ItemPointerSetBlockNumber(&start_ctid, BlockIdGetBlockNumber(&(cur_end_ctid.ip_blkid)) + 1);
                          ItemPointerSetOffsetNumber(&start_ctid, FirstOffsetNumber);
                      } else {
                          if (RedisCtidGetOffsetNumber(&cur_end_ctid) == MaxOffsetNumber) {
                              ItemPointerSetBlockNumber(&start_ctid, RedisCtidGetBlockNumber(&(cur_end_ctid)) + 1);
                              ItemPointerSetOffsetNumber(&start_ctid, FirstOffsetNumber);
                          } else {
                              ItemPointerSetBlockNumber(&start_ctid, RedisCtidGetBlockNumber(&(cur_end_ctid)));
                              ItemPointerSetOffsetNumber(&start_ctid, OffsetNumberNext(cur_end_ctid.ip_posid));
                          }
                      }
                  }
              }
  
              rel_options = add_ctid_string_to_reloptions(rel_options, "start_ctid_internal", &start_ctid);
              rel_options = add_ctid_string_to_reloptions(rel_options, "end_ctid_internal", &end_ctid);
              break;
          }
          case REDIS_REL_READ_ONLY: {
              rel_options =
                  lappend(rel_options, makeDefElem(pstrdup("append_mode_internal"), (Node*)makeInteger(action)));
              break;
          }
          case REDIS_REL_REFRESH: {
              /* start_ctid = (0,0) */
              if (isCUFormat)
                  ItemPointerSet(&start_ctid, FirstCUID + 1, 0);
              else
                  ItemPointerSet(&start_ctid, 0, 0);
  
              /*  end_ctid = max_ctid; */
              if (IS_PGXC_DATANODE) {
                  if (isCUFormat)
                      col_get_max_tid(rel, &end_ctid);
                  else
                      heap_get_max_tid(rel, &end_ctid);
              } else {
                  ItemPointerSet(&end_ctid, 0, 0);
              }
  
              rel_options = add_ctid_string_to_reloptions(rel_options, "start_ctid_internal", &start_ctid);
              rel_options = add_ctid_string_to_reloptions(rel_options, "end_ctid_internal", &end_ctid);
              break;
          }
          case REDIS_REL_RESET_CTID: {
              /*
               * 1. start_ctid = (0,0)
               * 2. end_ctid = (0,0);
               */
              if (isCUFormat) {
                  ItemPointerSet(&start_ctid, FirstCUID + 1, 0);
                  ItemPointerSet(&end_ctid, 0, 0);
              } else {
                  ItemPointerSet(&start_ctid, 0, 0);
                  ItemPointerSet(&end_ctid, 0, 0);
              }
  
              rel_options = add_ctid_string_to_reloptions(rel_options, "start_ctid_internal", &start_ctid);
              rel_options = add_ctid_string_to_reloptions(rel_options, "end_ctid_internal", &end_ctid);
              break;
          }
          default:
              rel_options = NIL;
              break;
      }
  
      return rel_options;
  }
  
  /*
   * - Brief: Modify partition rel's redistribute options
   * - Parameter:
   *      @rel: relation that needs to check
   *		@defList: DefElem list
   *		@operation: Alter reloptions method
   *		@lockmode: lock method
   *		@action: append/normal/refresh
   * - Return:
   *      @void:
   */
  void AlterTableSetPartRelOptions(
      Relation rel, List* defList, AlterTableType operation, LOCKMODE lockmode, char *merge_list, RedisRelAction action)
  {
      merge_list = NULL;
      ereport(DEBUG3, (errmsg("[%s()]: shouldn't run here", __FUNCTION__)));
      return;
  }
  
  /*
   * @Description: Check and get redistribute info
   * @Param[IN] options: input user options
   * @Param[OUT] *action: 0: insert; 1: delete. *rel_cn_oid: oid from CN
   * @See also:
   */
  #ifdef ENABLE_MULTIPLE_NODES
  void CheckRedistributeOption(List* options, Oid* rel_cn_oid, RedisHtlAction* action, char *merge_list_str, Relation rel)
  #else
  void CheckRedistributeOption(List* options, Oid* rel_cn_oid, RedisHtlAction* action, Relation rel)
  #endif
  {
      ListCell* opt = NULL;
      bool append_set = false;
      bool cnoid_set = false;
  
      Assert(rel != NULL);
  
      foreach (opt, options) {
          DefElem* def = (DefElem*)lfirst(opt);
  
          if (pg_strcasecmp(def->defname, "append_mode") == 0) {
              char* mode_options = defGetString(def);
              if (pg_strcasecmp(mode_options, "on") == 0)
                  *action = REDIS_REL_APPEND;
              else if (pg_strcasecmp(mode_options, "refresh") == 0)
                  *action = REDIS_REL_REFRESH;
              else if (pg_strcasecmp(mode_options, "off") == 0)
                  *action = REDIS_REL_NORMAL;
              else if (pg_strcasecmp(mode_options, "read_only") == 0)
                  *action = REDIS_REL_READ_ONLY;
              else if (pg_strcasecmp(mode_options, "end_catchup") == 0)
                  *action = REDIS_REL_END_CATCHUP;
              else
                  ereport(ERROR,
                      (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                          errmsg("you can only take append_mode = on, off, refresh, read_only, end_catchup")));
              append_set = true;
          } else if (pg_strcasecmp(def->defname, "rel_cn_oid") == 0) {
              *rel_cn_oid = atoi(defGetString(def));
              cnoid_set = true;
  #ifdef ENABLE_MULTIPLE_NODES
          }else if (pg_strcasecmp(def->defname, "merge_list") == 0) {
              *merge_list_str = pstrdup(defGetString(def));
  #endif
          }
      }
      if (*action == REDIS_REL_APPEND && append_set && !cnoid_set) {
          ereport(ERROR,
              (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                  errmsg("Alter-Table set append mode should have 'rel_cn_oid' set together")));
      } else if (!append_set && cnoid_set) {
          ereport(ERROR,
              (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                  errmsg("Alter-Table set rel_cn_oid should have 'append_mode' set together")));
      }
  
      return;
  }
  
  /*
   * - Brief: set PGPROC as redistribution
   * - Parameter:
   *      @void:
   * - Return:
   *      @true:  succeed
   *      @false: failed
   */
  bool set_proc_redis(void)
  {
      if (!u_sess->attr.attr_sql.enable_cluster_resize) {
          /*
           * fix me here with a better ERRCODE
           */
          ereport(ERROR,
              (errcode(ERRCODE_INVALID_OPERATION),
                  (errmsg("Proc redistribution only can be set during data redistribution time"))));
  
          return false;
      }
  
      /*
       * set process as data redistribution
       */
      LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
  
      t_thrd.pgxact->vacuumFlags |= PROC_IS_REDIST;
      LWLockRelease(ProcArrayLock);
      u_sess->proc_cxt.Isredisworker = true;
  
      return true;
  }
  
  /*
   * - Brief: reset PGPROC as redistribution
   * - Parameter:
   *      @void:
   * - Return:
   *      @true:  succeed
   *      @false: failed
   */
  bool reset_proc_redis(void)
  {
      if (!u_sess->attr.attr_sql.enable_cluster_resize) {
          /*
           * fix me here with a better ERRCODE
           */
          ereport(ERROR,
              (errcode(ERRCODE_INVALID_OPERATION),
                  (errmsg("Proc redistribution only can be reset during data redistribution time"))));
          return false;
      }
  
      /*
       * reset process as data redistribution
       */
      LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
      t_thrd.pgxact->vacuumFlags &= ~PROC_IS_REDIST;
      LWLockRelease(ProcArrayLock);
      u_sess->proc_cxt.Isredisworker = false;
  
      return true;
  }
  
  /*
   * - Brief: Is t_thrd.proc a redistribution proc?
   * - Parameter:
   *      @void:
   * - Return:
   *      @true:  yes
   *      @false: no
   */
  bool IsRedistributionWorkerProcess(void)
  {
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return false;
  }
  
  /*
   * - Brief: Get Redis Ctid values
   * - Parameter:
   *      @rel: redistibuting relation, RelationData
   *		@start_ctid : start_ctid_interlal of reloptions, output value
   *		@end_ctid : end_ctid_interlal of reloptions, output value
   * - Return:
   *		@void
   */
  void RelationGetCtids(Relation rel, ItemPointer start_ctid, ItemPointer end_ctid)
  {
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return;
  }
  
  uint32 RelationGetEndBlock(Relation rel)
  {
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return 0;
  }
  
  Node* eval_redis_func_direct(Relation rel, bool is_func_get_start_ctid, int num_of_slice, int slice_index)
  {
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return NULL;
  }
  ItemPointer eval_redis_func_direct_slice(ItemPointer start_ctid, ItemPointer end_ctid, bool is_func_get_start_ctid, 
           int num_of_slices, int slice_index)
  {
      Assert(false);
      DISTRIBUTED_FEATURE_NOT_SUPPORTED();
      return NULL;
  }
  
  