gt:                 if (results == null) {
context: /* This file is part of VoltDB.
   * Copyright (C) 2008-2011 VoltDB Inc.
   *
   * VoltDB is free software: you can redistribute it and/or modify
   * it under the terms of the GNU General Public License as published by
   * the Free Software Foundation, either version 3 of the License, or
   * (at your option) any later version.
   *
   * VoltDB is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU General Public License for more details.
   *
   * You should have received a copy of the GNU General Public License
   * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.voltdb.sysprocs;
  
  import java.io.File;
  import java.io.FileInputStream;
  import java.io.FileNotFoundException;
  import java.io.FilenameFilter;
  import java.io.IOException;
  import java.io.PrintWriter;
  import java.io.StringWriter;
  import java.nio.ByteBuffer;
  import java.util.ArrayDeque;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.io.*;
  
  import org.voltdb.BackendTarget;
  import org.voltdb.DependencyPair;
  import org.voltdb.ExecutionSite.SystemProcedureExecutionContext;
  import org.voltdb.HsqlBackend;
  import org.voltdb.ParameterSet;
  import org.voltdb.PrivateVoltTableFactory;
  import org.voltdb.ProcInfo;
  import org.voltdb.SiteProcedureConnection;
  import org.voltdb.TheHashinator;
  import org.voltdb.VoltDB;
  import org.voltdb.VoltSystemProcedure;
  import org.voltdb.VoltTable;
  import org.voltdb.VoltTable.ColumnInfo;
  import org.voltdb.VoltType;
  import org.voltdb.VoltTypeException;
  import org.voltdb.catalog.Cluster;
  import org.voltdb.catalog.Database;
  import org.voltdb.catalog.Procedure;
  import org.voltdb.catalog.Site;
  import org.voltdb.catalog.Table;
  import org.voltdb.client.ConnectionUtil;
  import org.voltdb.dtxn.DtxnConstants;
  import org.voltdb.export.ExportManager;
  import org.voltdb.logging.VoltLogger;
  import org.voltdb.sysprocs.saverestore.ClusterSaveFileState;
  import org.voltdb.sysprocs.saverestore.SavedTableConverter;
  import org.voltdb.sysprocs.saverestore.SnapshotUtil;
  import org.voltdb.sysprocs.saverestore.TableSaveFile;
  import org.voltdb.sysprocs.saverestore.TableSaveFileState;
  import org.voltdb.utils.DBBPool.BBContainer;
  import org.voltdb.utils.CatalogUtil;
  import org.voltdb.utils.VoltFile;
  import org.json_voltpatches.*;
  import org.voltdb.utils.Pair;
  
  @ProcInfo (
      singlePartition = false
  )
  public class SnapshotRestore extends VoltSystemProcedure
  {
      private static final VoltLogger TRACE_LOG = new VoltLogger(SnapshotRestore.class.getName());
  
      private static final VoltLogger HOST_LOG = new VoltLogger("HOST");
  
      private static final int DEP_restoreScan = (int)
          SysProcFragmentId.PF_restoreScan | DtxnConstants.MULTIPARTITION_DEPENDENCY;
      private static final int DEP_restoreScanResults = (int)
          SysProcFragmentId.PF_restoreScanResults;
  
      /*
       * Plan fragments for retrieving the digests
       * for the snapshot visible at every node. Can't be combined
       * with the other scan because only one result table can be returned
       * by a plan fragment.
       */
      private static final int DEP_restoreDigestScan = (int)
          SysProcFragmentId.PF_restoreDigestScan | DtxnConstants.MULTIPARTITION_DEPENDENCY;
      private static final int DEP_restoreDigestScanResults = (int)
          SysProcFragmentId.PF_restoreDigestScanResults;
  
      /*
       * Plan fragments for distributing the full set of export sequence numbers
       * to every partition where the relevant ones can be selected
       * and forwarded to the EE. Also distributes the txnId of the snapshot
       * which is used to truncate export data on disk from after the snapshot
       */
      private static final int DEP_restoreDistributeExportSequenceNumbers = (int)
          SysProcFragmentId.PF_restoreDistributeExportSequenceNumbers | DtxnConstants.MULTIPARTITION_DEPENDENCY;
      private static final int DEP_restoreDistributeExportSequenceNumbersResults = (int)
          SysProcFragmentId.PF_restoreDistributeExportSequenceNumbersResults;
  
      private static HashSet<String>  m_initializedTableSaveFileNames = new HashSet<String>();
      private static ArrayDeque<TableSaveFile> m_saveFiles = new ArrayDeque<TableSaveFile>();
  
      public static volatile boolean m_haveDoneRestore = false;
  
      private static synchronized void initializeTableSaveFiles(
              String filePath,
              String fileNonce,
              String tableName,
              int originalHostIds[],
              int relevantPartitionIds[]) throws IOException {
          // This check ensures that only one site per host attempts to
          // distribute this table.  @SnapshotRestore sends plan fragments
          // to every site on this host with the tables and partition ID that
          // this host is going to distribute to the cluster.  The first
          // execution site to get into this synchronized method is going to
          // 'win', add the table it's doing to this set, and then do the rest
          // of the work.  Subsequent sites will just return here.
          if (!m_initializedTableSaveFileNames.add(tableName)) {
              return;
          }
  
          // To avoid pulling duplicate rows when we have multiple files
          // that contain the data for a partition, we're going to assign
          // all of the partition IDs that were passed in to one and only one
          // TableSaveFile.  We'll pull them out of this set as we find
          // files for them, and then once the set is empty we can bail out of
          // this loop.  The restore planner called in @SnapshotRestore should
          // ensure that we can, in fact, find files for all these partitions.
          HashSet<Integer> relevantPartitionSet =
              new HashSet<Integer>();
          for (int part_id : relevantPartitionIds)
          {
              relevantPartitionSet.add(part_id);
          }
  
          for (int originalHostId : originalHostIds) {
              final File f = getSaveFileForPartitionedTable(filePath, fileNonce,
                                                            tableName,
                                                            originalHostId);
              TableSaveFile savefile = getTableSaveFile(
                              f,
                              org.voltdb.VoltDB.instance().getLocalSites().size() * 4,
                              relevantPartitionSet.toArray(new Integer[relevantPartitionSet.size()]));
  
              m_saveFiles.offer(savefile);
              for (int part_id : savefile.getPartitionIds())
              {
                  relevantPartitionSet.remove(part_id);
              }
              if (relevantPartitionSet.isEmpty())
              {
                  break;
              }
              assert(m_saveFiles.peekLast().getCompleted());
          }
      }
  
      private static synchronized boolean hasMoreChunks() throws IOException {
          boolean hasMoreChunks = false;
          while (!hasMoreChunks && m_saveFiles.peek() != null) {
              TableSaveFile f = m_saveFiles.peek();
              hasMoreChunks = f.hasMoreChunks();
              if (!hasMoreChunks) {
                  try {
                      f.close();
                  } catch (IOException e) {
                  }
                  m_saveFiles.poll();
              }
          }
          return hasMoreChunks;
      }
  
      private static synchronized BBContainer getNextChunk() throws IOException {
          BBContainer c = null;
          while (c == null && m_saveFiles.peek() != null) {
              TableSaveFile f = m_saveFiles.peek();
              c = f.getNextChunk();
              if (c == null) {
                  f.close();
                  m_saveFiles.poll();
              }
          }
          return c;
      }
  
      @Override
      public void init(int numberOfPartitions, SiteProcedureConnection site,
              Procedure catProc, BackendTarget eeType, HsqlBackend hsql, Cluster cluster)
      {
          super.init(numberOfPartitions, site, catProc, eeType, hsql, cluster);
          site.registerPlanFragment(SysProcFragmentId.PF_restoreScan, this);
          site.registerPlanFragment(SysProcFragmentId.PF_restoreScanResults,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreLoadReplicatedTable,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreLoadReplicatedTableResults,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreDistributeReplicatedTable,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreDistributePartitionedTable,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreDistributePartitionedTableResults,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreSendReplicatedTable,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreSendReplicatedTableResults,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreSendPartitionedTable,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                                    PF_restoreSendPartitionedTableResults,
                                    this);
          site.registerPlanFragment(SysProcFragmentId.
                  PF_restoreDigestScan,
                  this);
          site.registerPlanFragment(SysProcFragmentId.
                  PF_restoreDigestScanResults,
                  this);
          site.registerPlanFragment(SysProcFragmentId.
                  PF_restoreDistributeExportSequenceNumbers,
                  this);
          site.registerPlanFragment(SysProcFragmentId.
                  PF_restoreDistributeExportSequenceNumbersResults,
                  this);
          m_cluster = cluster;
          m_siteId = site.getCorrespondingSiteId();
          m_hostId =
              Integer.valueOf(m_cluster.getSites().get(String.valueOf(m_siteId)).
                              getHost().getTypeName());
          // XXX HACK GIANT HACK given the current assumption that there is
          // only one database per cluster, I'm asserting this and then
          // skirting around the need to have the database name in order to get
          // to the set of tables. --izzy
          assert(m_cluster.getDatabases().size() == 1);
          m_database = m_cluster.getDatabases().get("database");
      }
  
      @Override
      public DependencyPair
      executePlanFragment(HashMap<Integer, List<VoltTable>> dependencies, long fragmentId, ParameterSet params,
                          SystemProcedureExecutionContext context)
      {
          String hostname = ConnectionUtil.getHostnameOrAddress();
          if (fragmentId == SysProcFragmentId.PF_restoreDistributeExportSequenceNumbers)
          {
              assert(params.toArray()[0] != null);
              assert(params.toArray().length == 2);
              assert(params.toArray()[0] instanceof byte[]);
              VoltTable result = new VoltTable(new VoltTable.ColumnInfo("RESULT", VoltType.STRING));
              long snapshotTxnId = ((Long)params.toArray()[1]).longValue();
  
              // Choose the lowest site ID on this host to truncate export data
              int host_id = context.getExecutionSite().getCorrespondingHostId();
              Integer lowest_site_id =
                  VoltDB.instance().getCatalogContext().siteTracker.
                  getLowestLiveExecSiteIdForHost(host_id);
              if (context.getExecutionSite().getSiteId() == lowest_site_id)
              {
              ExportManager.instance().
                  truncateExportToTxnId(snapshotTxnId);
              }
              try {
                  ByteArrayInputStream bais = new ByteArrayInputStream((byte[])params.toArray()[0]);
                  ObjectInputStream ois = new ObjectInputStream(bais);
  
                  //Sequence numbers for every table and partition
                  Map<String, Map<Integer, Long>> exportSequenceNumbers =
                      (Map<String, Map<Integer, Long>>)ois.readObject();
                  Database db = context.getDatabase();
                  Integer myPartitionId = context.getExecutionSite().getCorrespondingPartitionId();
  
                  //Iterate the export tables
                  for (Table t : db.getTables()) {
                      if (!CatalogUtil.isTableExportOnly( db, t))
                          continue;
  
                      String signature = t.getSignature();
                      String name = t.getTypeName();
  
                      //Sequence numbers for this table for every partition
                      Map<Integer, Long> sequenceNumberPerPartition = exportSequenceNumbers.get(name);
                      if (sequenceNumberPerPartition == null) {
                          HOST_LOG.warn("Could not find export sequence number for table " + name +
                                  ". This warning is safe to ignore if you are loading a pre 1.3 snapshot" +
                                  " which would not contain these sequence numbers (added in 1.3)." +
                                  " If this is a post 1.3 snapshot then the restore has failed and export sequence " +
                                  " are reset to 0");
                          continue;
                      }
  
                      Long sequenceNumber =
                          sequenceNumberPerPartition.get(myPartitionId);
                      if (sequenceNumber == null) {
                          HOST_LOG.warn("Could not find an export sequence number for table " + name +
                                  " partition " + myPartitionId +
                                  ". This warning is safe to ignore if you are loading a pre 1.3 snapshot " +
                                  " which would not contain these sequence numbers (added in 1.3)." +
                                  " If this is a post 1.3 snapshot then the restore has failed and export sequence " +
                                  " are reset to 0");
                          continue;
                      }
                      //Forward the sequence number to the EE
                      context.getExecutionEngine().exportAction(
                              false,
                              0,
                              sequenceNumber,
                              myPartitionId,
                              signature);
                  }
              } catch (Exception e) {
                  e.printStackTrace();//l4j doesn't print the stack trace
                  HOST_LOG.error(e);
                  result.addRow("FAILURE");
              }
              return new DependencyPair(DEP_restoreDistributeExportSequenceNumbers, result);
          }
          else if (fragmentId == SysProcFragmentId.PF_restoreDistributeExportSequenceNumbersResults)
          {
              TRACE_LOG.trace("Aggregating digest scan state");
              assert(dependencies.size() > 0);
              List<VoltTable> dep = dependencies.get(DEP_restoreDistributeExportSequenceNumbers);
              VoltTable result = new VoltTable(new VoltTable.ColumnInfo("RESULT", VoltType.STRING));
              for (VoltTable table : dep)
              {
                  while (table.advanceRow())
                  {
                      // the actually adds the active row... weird...
                      result.add(table);
                  }
              }
              return new DependencyPair(DEP_restoreDistributeExportSequenceNumbersResults, result);
          } else if (fragmentId == SysProcFragmentId.PF_restoreDigestScan)
          {
              VoltTable result = new VoltTable(
                      new VoltTable.ColumnInfo("DIGEST", VoltType.STRING),
                      new VoltTable.ColumnInfo("RESULT", VoltType.STRING),
                      new VoltTable.ColumnInfo("ERR_MSG", VoltType.STRING));
              // Choose the lowest site ID on this host to do the file scan
              // All other sites should just return empty results tables.
              int host_id = context.getExecutionSite().getCorrespondingHostId();
              Integer lowest_site_id =
                  VoltDB.instance().getCatalogContext().siteTracker.
                  getLowestLiveExecSiteIdForHost(host_id);
              if (context.getExecutionSite().getSiteId() == lowest_site_id)
              {
                  try {
                      // implicitly synchronized by the way restore operates.
                      // this scan must complete on every site and return results
                      // to the coordinator for aggregation before it will send out
                      // distribution fragments, so two sites on the same node
                      // can't be attempting to set and clear this HashSet simultaneously
                      TRACE_LOG.trace("Checking saved table digest state for restore of: "
                                      + m_filePath + ", " + m_fileNonce);
                      List<JSONObject> digests =
                          SnapshotUtil.retrieveDigests(m_filePath, m_fileNonce);
  
                      for (JSONObject obj : digests) {
                          result.addRow(obj.toString(), "SUCCESS", null);
                      }
                  } catch (Exception e) {
                      StringWriter sw = new StringWriter();
                      PrintWriter pw = new PrintWriter(sw);
                      e.printStackTrace(pw);
                      pw.flush();
                      e.printStackTrace();//l4j doesn't print stack traces
                      HOST_LOG.error(e);
                      result.addRow(null, "FAILURE", sw.toString());
                      return new DependencyPair(DEP_restoreDigestScan, result);
                  }
              }
              return new DependencyPair(DEP_restoreDigestScan, result);
          }
          else if (fragmentId == SysProcFragmentId.PF_restoreDigestScanResults)
          {
              TRACE_LOG.trace("Aggregating digest scan state");
              assert(dependencies.size() > 0);
              List<VoltTable> dep = dependencies.get(DEP_restoreDigestScan);
              VoltTable result = new VoltTable(
                      new VoltTable.ColumnInfo("DIGEST", VoltType.STRING),
                      new VoltTable.ColumnInfo("RESULT", VoltType.STRING),
                      new VoltTable.ColumnInfo("ERR_MSG", VoltType.STRING));
              for (VoltTable table : dep)
              {
                  while (table.advanceRow())
                  {
                      // the actually adds the active row... weird...
                      result.add(table);
                  }
              }
              return new DependencyPair(DEP_restoreDigestScanResults, result);
          } else if (fragmentId == SysProcFragmentId.PF_restoreScan)
          {
              assert(params.toArray()[0] != null);
              assert(params.toArray()[1] != null);
              VoltTable result = ClusterSaveFileState.constructEmptySaveFileStateVoltTable();
              // Choose the lowest site ID on this host to do the file scan
              // All other sites should just return empty results tables.
              int host_id = context.getExecutionSite().getCorrespondingHostId();
              Integer lowest_site_id =
                  VoltDB.instance().getCatalogContext().siteTracker.
                  getLowestLiveExecSiteIdForHost(host_id);
              if (context.getExecutionSite().getSiteId() == lowest_site_id)
              {
                  // implicitly synchronized by the way restore operates.
                  // this scan must complete on every site and return results
                  // to the coordinator for aggregation before it will send out
                  // distribution fragments, so two sites on the same node
                  // can't be attempting to set and clear this HashSet simultaneously
                  m_initializedTableSaveFileNames.clear();
                  m_saveFiles.clear();//Tests will reused a VoltDB process that fails a restore
  
                  m_filePath = (String) params.toArray()[0];
                  m_fileNonce = (String) params.toArray()[1];
                  TRACE_LOG.trace("Checking saved table state for restore of: "
                                  + m_filePath + ", " + m_fileNonce);
                  File[] savefiles = retrieveRelevantFiles(m_filePath, m_fileNonce);
                  for (File file : savefiles)
                  {
                      TableSaveFile savefile = null;
                      try
                      {
                          savefile = getTableSaveFile(file, 1, null);
                          try {
  
                              if (!savefile.getCompleted()) {
                                  continue;
                              }
  
                              String is_replicated = "FALSE";
                              if (savefile.isReplicated())
                              {
                                  is_replicated = "TRUE";
                              }
                              int partitionIds[] = savefile.getPartitionIds();
                              for (int pid : partitionIds) {
                                  result.addRow(m_hostId,
                                                hostname,
                                                savefile.getHostId(),
                                                savefile.getHostname(),
                                                savefile.getClusterName(),
                                                savefile.getDatabaseName(),
                                                savefile.getTableName(),
                                                is_replicated,
                                                pid,
                                                savefile.getTotalPartitions());
                              }
                          } finally {
                              savefile.close();
                          }
                      }
                      catch (FileNotFoundException e)
                      {
                          // retrieveRelevantFiles should always generate a list
                          // of valid present files in m_filePath, so if we end up
                          // getting here, something has gone very weird.
                          e.printStackTrace();
                      }
                      catch (IOException e)
                      {
                          // For the time being I'm content to treat this as a
                          // missing file and let the coordinator complain if
                          // it discovers that it can't build a consistent
                          // database out of the files it sees available.
                          //
                          // Maybe just a log message?  Later.
                          e.printStackTrace();
                      }
                  }
              }
              return new DependencyPair(DEP_restoreScan, result);
          }
          else if (fragmentId == SysProcFragmentId.PF_restoreScanResults)
          {
              TRACE_LOG.trace("Aggregating saved table state");
              assert(dependencies.size() > 0);
              List<VoltTable> dep = dependencies.get(DEP_restoreScan);
              VoltTable result = ClusterSaveFileState.constructEmptySaveFileStateVoltTable();
              for (VoltTable table : dep)
              {
                  while (table.advanceRow())
                  {
                      // the actually adds the active row... weird...
                      result.add(table);
                  }
              }
              return new DependencyPair(DEP_restoreScanResults, result);
          }
          else if (fragmentId == SysProcFragmentId.PF_restoreLoadReplicatedTable)
          {
              m_haveDoneRestore = true;
              assert(params.toArray()[0] != null);
              assert(params.toArray()[1] != null);
              String table_name = (String) params.toArray()[0];
              int dependency_id = (Integer) params.toArray()[1];
              TRACE_LOG.trace("Loading replicated table: " + table_name);
              String result_str = "SUCCESS";
              String error_msg = "";
              TableSaveFile savefile = null;
  
              /**
               * For replicated tables this will do the slow thing and read the file
               * once for each ExecutionSite. This could use optimization like
               * is done with the partitioned tables.
               */
              try
              {
                  savefile =
                      getTableSaveFile(getSaveFileForReplicatedTable(table_name), 3, null);
                  assert(savefile.getCompleted());
              }
              catch (IOException e)
              {
                  VoltTable result = constructResultsTable();
                  result.addRow(m_hostId, hostname, m_siteId, table_name, -1, "FAILURE",
                                "Unable to load table: " + table_name +
                                " error: " + e.getMessage());
                  return new DependencyPair(dependency_id, result);
              }
  
              try {
                  while (savefile.hasMoreChunks())
                  {
                      VoltTable table = null;
  
                          final org.voltdb.utils.DBBPool.BBContainer c = savefile.getNextChunk();
                          if (c == null) {
                              continue;//Should be equivalent to break
                          }
                          VoltTable old_table = PrivateVoltTableFactory.createVoltTableFromBuffer(c.b, true);
                          Table new_catalog_table = getCatalogTable(table_name);
                          table = SavedTableConverter.convertTable(old_table,
                                                                   new_catalog_table);
                          c.discard();
  
                      try
                      {
                          super.voltLoadTable(context.getCluster().getTypeName(),
                                              context.getDatabase().getTypeName(),
                                              table_name, table);
                      }
                      catch (VoltAbortException e)
                      {
                          result_str = "FAILURE";
                          error_msg = e.getMessage();
                          break;
                      }
                  }
  
              } catch (IOException e) {
                  VoltTable result = constructResultsTable();
                  result.addRow(m_hostId, hostname, m_siteId, table_name, -1, "FAILURE",
                                "Unable to load table: " + table_name +
                                " error: " + e.getMessage());
                  return new DependencyPair(dependency_id, result);
              } catch (VoltTypeException e) {
                  VoltTable result = constructResultsTable();
                  result.addRow(m_hostId, hostname, m_siteId, table_name, -1, "FAILURE",
                                "Unable to load table: " + table_name +
                                " error: " + e.getMessage());
                  return new DependencyPair(dependency_id, result);
              }
  
              VoltTable result = constructResultsTable();
              result.addRow(m_hostId, hostname, m_siteId, table_name, -1, result_str,
                               error_msg);
              try {
                  savefile.close();
              } catch (IOException e) {
                  // TODO Auto-generated catch block
                  e.printStackTrace();
              }
              return new DependencyPair( dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreDistributeReplicatedTable)
          {
              m_haveDoneRestore = true;
              // XXX I tested this with a hack that cannot be replicated
              // in a unit test since it requires hacks to this sysproc that
              // effectively break it
              assert(params.toArray()[0] != null);
              assert(params.toArray()[1] != null);
              assert(params.toArray()[2] != null);
              assert(params.toArray()[3] != null);
              String table_name = (String) params.toArray()[0];
              int site_id = (Integer) params.toArray()[1];
              int dependency_id = (Integer) params.toArray()[2];
              TRACE_LOG.trace("Distributing replicated table: " + table_name +
                              " to: " + site_id);
              VoltTable result = performDistributeReplicatedTable(table_name, site_id);
              return new DependencyPair(dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreSendReplicatedTable)
          {
              m_haveDoneRestore = true;
              assert(params.toArray()[0] != null);
              assert(params.toArray()[1] != null);
              assert(params.toArray()[2] != null);
              assert(params.toArray()[3] != null);
              String table_name = (String) params.toArray()[0];
              int dependency_id = (Integer) params.toArray()[1];
              VoltTable table = (VoltTable) params.toArray()[2];
              TRACE_LOG.trace("Received replicated table: " + table_name);
              String result_str = "SUCCESS";
              String error_msg = "";
              try
              {
                  super.voltLoadTable(context.getCluster().getTypeName(),
                                      context.getDatabase().getTypeName(),
                                      table_name, table);
              }
              catch (VoltAbortException e)
              {
                  result_str = "FAILURE";
                  error_msg = e.getMessage();
              }
              VoltTable result = constructResultsTable();
              result.addRow(m_hostId, hostname, m_siteId, table_name, -1,
                               result_str, error_msg);
              return new DependencyPair(dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreSendReplicatedTableResults)
          {
              m_haveDoneRestore = true;
              assert(params.toArray()[0] != null);
              int dependency_id = (Integer) params.toArray()[0];
              TRACE_LOG.trace("Received confirmmation of successful replicated table load");
              VoltTable result = constructResultsTable();
              for (int dep_id : dependencies.keySet())
              {
                  List<VoltTable> table_list = dependencies.get(dep_id);
                  assert(table_list.size() == 1);
                  VoltTable t = table_list.get(0);
                  while (t.advanceRow())
                  {
                      // this will actually add the active row of t
                      result.add(t);
                  }
              }
              return new DependencyPair(dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreLoadReplicatedTableResults)
          {
              m_haveDoneRestore = true;
              TRACE_LOG.trace("Aggregating replicated table restore results");
              assert(params.toArray()[0] != null);
              int dependency_id = (Integer) params.toArray()[0];
              assert(dependencies.size() > 0);
              VoltTable result = constructResultsTable();
              for (int dep_id : dependencies.keySet())
              {
                  List<VoltTable> table_list = dependencies.get(dep_id);
                  assert(table_list.size() == 1);
                  VoltTable t = table_list.get(0);
                  while (t.advanceRow())
                  {
                      // this will actually add the active row of t
                      result.add(t);
                  }
              }
              return new DependencyPair(dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreDistributePartitionedTable)
          {
              m_haveDoneRestore = true;
              Object paramsA[] = params.toArray();
              assert(paramsA[0] != null);
              assert(paramsA[1] != null);
              assert(paramsA[2] != null);
              assert(paramsA[3] != null);
  
              String table_name = (String) paramsA[0];
              int originalHosts[] = (int[]) paramsA[1];
              int relevantPartitions[]  = (int[]) paramsA[2];
              int dependency_id = (Integer) paramsA[3];
  
              for (int partition_id : relevantPartitions) {
                  TRACE_LOG.trace("Distributing partitioned table: " + table_name +
                                  " partition id: " + partition_id);
              }
  
              VoltTable result =
                  performDistributePartitionedTable(table_name, originalHosts,
                                                    relevantPartitions);
              return new DependencyPair( dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreDistributePartitionedTableResults)
          {
              m_haveDoneRestore = true;
              TRACE_LOG.trace("Aggregating partitioned table restore results");
              assert(params.toArray()[0] != null);
              int dependency_id = (Integer) params.toArray()[0];
              VoltTable result = constructResultsTable();
              for (int dep_id : dependencies.keySet())
              {
                  List<VoltTable> table_list = dependencies.get(dep_id);
                  assert(table_list.size() == 1);
                  VoltTable t = table_list.get(0);
                  while (t.advanceRow())
                  {
                      // this will actually add the active row of t
                      result.add(t);
                  }
              }
              return new DependencyPair(dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreSendPartitionedTable)
          {
              m_haveDoneRestore = true;
              assert(params.toArray()[0] != null);
              assert(params.toArray()[1] != null);
              assert(params.toArray()[2] != null);
              assert(params.toArray()[3] != null);
              String table_name = (String) params.toArray()[0];
              int partition_id = (Integer) params.toArray()[1];
              int dependency_id = (Integer) params.toArray()[2];
              VoltTable table = (VoltTable) params.toArray()[3];
              TRACE_LOG.trace("Received partitioned table: " + table_name);
              String result_str = "SUCCESS";
              String error_msg = "";
              try
              {
                  super.voltLoadTable(context.getCluster().getTypeName(),
                                      context.getDatabase().getTypeName(),
                                      table_name, table);
              }
              catch (VoltAbortException e)
              {
                  result_str = "FAILURE";
                  error_msg = e.getMessage();
              }
              VoltTable result = constructResultsTable();
              result.addRow(m_hostId, hostname, m_siteId, table_name, partition_id,
                            result_str, error_msg);
              return new DependencyPair(dependency_id, result);
          }
          else if (fragmentId ==
              SysProcFragmentId.PF_restoreSendPartitionedTableResults)
          {
              m_haveDoneRestore = true;
              assert(params.toArray()[0] != null);
              int dependency_id = (Integer) params.toArray()[0];
              TRACE_LOG.trace("Received confirmation of successful partitioned table load");
              VoltTable result = constructResultsTable();
              for (int dep_id : dependencies.keySet())
              {
                  List<VoltTable> table_list = dependencies.get(dep_id);
                  assert(table_list.size() == 1);
                  VoltTable t = table_list.get(0);
                  while (t.advanceRow())
                  {
                      // this will actually add the active row of t
                      result.add(t);
                  }
              }
              return new DependencyPair(dependency_id, result);
          }
  
          assert (false);
          return null;
      }
  
     // private final VoltSampler m_sampler = new VoltSampler(10, "sample" + String.valueOf(new Random().nextInt() % 10000) + ".txt");
  
      public VoltTable[] run(SystemProcedureExecutionContext ctx,
              String path, String nonce) throws VoltAbortException
      {
          if (m_haveDoneRestore) {
              throw new VoltAbortException("Cluster has already been restored or has failed a restore." +
                      " Restart the cluster before doing another restore.");
          }
  
          final long startTime = System.currentTimeMillis();
          HOST_LOG.info("Restoring from path: " + path + " with nonce: " + nonce);
  
          // Fetch all the savefile metadata from the cluster
          VoltTable[] savefile_data;
          savefile_data = performRestoreScanWork(path, nonce);
  
          List<JSONObject> digests;
          Map<String, Map<Integer, Long>> exportSequenceNumbers;
          try {
              Pair<List<JSONObject>, Map<String, Map<Integer, Long>>> digestScanResult =
                  performRestoreDigestScanWork();
              digests = digestScanResult.getFirst();
              exportSequenceNumbers = digestScanResult.getSecond();
          } catch (VoltAbortException e) {
              ColumnInfo[] result_columns = new ColumnInfo[2];
              int ii = 0;
              result_columns[ii++] = new ColumnInfo("RESULT", VoltType.STRING);
              result_columns[ii++] = new ColumnInfo("ERR_MSG", VoltType.STRING);
              VoltTable results[] = new VoltTable[] { new VoltTable(result_columns) };
              results[0].addRow("FAILURE", e.toString());
              return results;
          }
  
          ClusterSaveFileState savefile_state = null;
          try
          {
              savefile_state = new ClusterSaveFileState(savefile_data[0]);
          }
          catch (IOException e)
          {
              throw new VoltAbortException(e.getMessage());
          }
  
          HashSet<String> relevantTableNames = new HashSet<String>();
          try {
              if (digests.isEmpty()) {
                  throw new Exception("No digests found");
              }
              for (JSONObject obj : digests) {
                  JSONArray tables = obj.getJSONArray("tables");
                  for (int ii = 0; ii < tables.length(); ii++) {
                      relevantTableNames.add(tables.getString(ii));
                  }
              }
          } catch (Exception e) {
              ColumnInfo[] result_columns = new ColumnInfo[2];
              int ii = 0;
              result_columns[ii++] = new ColumnInfo("RESULT", VoltType.STRING);
              result_columns[ii++] = new ColumnInfo("ERR_MSG", VoltType.STRING);
              VoltTable results[] = new VoltTable[] { new VoltTable(result_columns) };
              results[0].addRow("FAILURE", e.toString());
              return results;
          }
          assert(relevantTableNames != null);
          assert(relevantTableNames.size() > 0);
  
          // ENG-1078: I think this giant for/if block is only good for
          // checking if there are no files for a table listed in the digest.
          // There appear to be redundant checks for that, and then the per-table
          // consistency check is preempted by the ClusterSaveFileState constructor
          // called above.
          VoltTable[] results = null;
          for (String tableName : relevantTableNames) {
              if (!savefile_state.getSavedTableNames().contains(tableName)) {
                  if (results == null) {
                      ColumnInfo[] result_columns = new ColumnInfo[2];
                      int ii = 0;
                      result_columns[ii++] = new ColumnInfo("RESULT", VoltType.STRING);
                      result_columns[ii++] = new ColumnInfo("ERR_MSG", VoltType.STRING);
                      results = new VoltTable[] { new VoltTable(result_columns) };
                  }
                  results[0].addRow("FAILURE", "Save data contains no information for table " + tableName);
                  break;
              }
  
              final TableSaveFileState saveFileState = savefile_state.getTableState(tableName);
              if (saveFileState == null)
              {
                  // Pretty sure this is unreachable
                  // See ENG-1078
  