docs: public long run(SystemProcedureExecutionContext ctx,
              String tableName, byte upsertMode, VoltTable table)
              throws VoltAbortException {
  
          // if tableName is replicated, just send table everywhere.
          // otherwise, create a VoltTable for each partition and
          // split up the incoming table .. then send those partial
          // tables to the appropriate sites.
  
          Table catTable = ctx.getDatabase().getTables().getIgnoreCase(tableName);
          if (catTable == null) {
              throw new VoltAbortException("Table not present in catalog.");
          }
  
          if (!catTable.getIsreplicated()) {
              throw new VoltAbortException("LoadMultipartitionTable no longer supports loading partitioned tables" +
                      " use CRUD procs instead");
          }
  
          boolean isUpsert = (upsertMode != 0);
  
          // TODO verify table has right schema as catTable
  
          // use loadTable path for bulk insert
          if (!isUpsert && table.getRowCount() > 1) {
              VoltTable[] results = createAndExecuteSysProcPlan(SysProcFragmentId.PF_distribute,
                      SysProcFragmentId.PF_aggregate, tableName, table);
              return results[0].asScalarLong();
          }
  
          if (isUpsert) {
              boolean hasPkey = false;
              for (Constraint c : catTable.getConstraints()) {
                  if (c.getType() == ConstraintType.PRIMARY_KEY.getValue()) {
                      hasPkey = true;
                      break;
                  }
              }
              if (!hasPkey) {
                  throw new VoltAbortException(
                          String.format("The --update argument cannot be used for LoadMultipartitionTable because the table %s does not have a primary key. "
                                  + "Either remove the --update argument or add a primary key to the table.",
                                  tableName));
              }
          }
  
          // action should be either "insert" or "upsert"
          final String action = (isUpsert ? "upsert" :"insert");
  
          // fix any case problems
          tableName = catTable.getTypeName();
  
          // check that the schema of the input matches
          int columnCount = table.getColumnCount();
  
          // find the insert statement for this table
          String crudProcName = String.format("%s.%s", tableName.toUpperCase(),action);
          Procedure proc = ctx.ensureDefaultProcLoaded(crudProcName);
          if (proc == null) {
              throw new VoltAbortException(
                      String.format("Unable to locate auto-generated CRUD %s statement for table %s",
                              action, tableName));
          }
          // ensure MP fragment tasks load the plan for the table loading procedure
          m_runner.setProcNameToLoadForFragmentTasks(crudProcName);
  
          Statement catStmt = proc.getStatements().get(VoltDB.ANON_STMT_NAME + "0");
          if (catStmt == null) {
              throw new VoltAbortException(
                      String.format("Unable to find SQL statement for found table %s: BAD",
                              tableName));
          }
  
          // create a SQLStmt instance on the fly (unusual to do)
          SQLStmt stmt = new SQLStmt(catStmt.getSqltext());
          m_runner.initSQLStmt(stmt, catStmt);
  
  
          long queued = 0;
          long executed = 0;
  
          // make sure at the start of the table
          table.resetRowPosition();
          for (int i = 1; table.advanceRow(); ++i) {
              Object[] params = new Object[columnCount];
  
              // get the parameters from the volt table
              for (int col = 0; col < columnCount; ++col) {
                  params[col] = table.get(col, table.getColumnType(col));
              }
  
              // queue an insert and count it
              voltQueueSQL(stmt, params);
              ++queued;
  
              // every 100 statements, exec the batch
              // 100 is an arbitrary number
              if ((i % 100) == 0) {
                  executed += executeSQL(false);
              }
          }
          // execute any leftover batched statements
          if (queued > executed) {
              executed += executeSQL(true);
          }
  
          return executed;
      }