docs: VoltTable migrateRowsCommon(SystemProcedureExecutionContext ctx,
              String tableName,
              String columnName,
              String compStr,
              VoltTable paramTable,
              long chunksize,
              boolean replicated) {
          // Some basic checks
          if (chunksize <= 0) {
              throw new VoltAbortException(
                      "Chunk size must be positive, current value is" + chunksize);
          }
          Table catTable = ctx.getDatabase().getTables().getIgnoreCase(tableName);
          if (catTable == null) {
              throw new VoltAbortException(
                      String.format("Table %s doesn't present in catalog", tableName));
          }
  
          if (replicated != catTable.getIsreplicated()) {
              throw new VoltAbortException(
                      String.format("%s incompatible with %s table %s.",
                              replicated ? "@MigrateRowsMP" : "@MigrateRowsSP",
                              catTable.getIsreplicated() ? "replicated" : "partitioned", tableName));
          }
          Column column = catTable.getColumns().get(columnName);
          if (column == null) {
              throw new VoltAbortException(
                      String.format("Column %s does not exist in table %s", columnName, tableName));
          }
  
          // verify all required indices
          verifyRequiredIndices(catTable, column);
          // so far should only be single column, single row table
          int columnCount = paramTable.getColumnCount();
          if (columnCount > 1) {
              throw new VoltAbortException("More than one input parameter is not supported right now.");
          }
          assert(columnCount == 1);
          paramTable.resetRowPosition();
          paramTable.advanceRow();
          Object[] params = new Object[columnCount];
          for (int i = 0; i < columnCount; i++) {
              params[i] = paramTable.get(i, paramTable.getColumnType(i));
          }
          assert (params.length == 1);
  
          VoltType expectedType = VoltType.values()[column.getType()];
          VoltType actualType = VoltType.typeFromObject(params[0]);
          if (actualType != expectedType) {
              throw new VoltAbortException(String.format("Parameter type %s doesn't match column type %s",
                              actualType.toString(), expectedType.toString()));
          }
          ComparisonOperation op = ComparisonOperation.fromString(compStr);
  
          ProcedureRunner pr = ctx.getSiteProcedureConnection().getMigrateProcRunner(
                  tableName + ".autogenMigrate"+ op.toString(), catTable, column, op);
  
          Procedure newCatProc = pr.getCatalogProcedure();
          Statement countStmt = newCatProc.getStatements().get(VoltDB.ANON_STMT_NAME + "0");
          Statement valueAtStmt = newCatProc.getStatements().get(VoltDB.ANON_STMT_NAME + "1");
          Statement migrateStmt = newCatProc.getStatements().get(VoltDB.ANON_STMT_NAME + "2");
  
          if (countStmt == null || migrateStmt == null || valueAtStmt == null) {
              throw new VoltAbortException(
                      String.format("Unable to find SQL statement for migrate on table %s",
                              tableName));
          }
  
          Object cutoffValue = null;
          VoltTable result = null;
          result = executePrecompiledSQL(countStmt, params, replicated);
          long rowCount = result.asScalarLong();
          if (exportLog.isDebugEnabled()) {
              exportLog.debug("Migrate on table " + tableName +
                      (replicated ? "" : " on partition " + ctx.getPartitionId()) +
                      " reported " + rowCount + " matching rows");
          }
  
          // If number of rows meet the criteria is more than chunk size, pick the column value
          // which offset equals to chunk size as new predicate.
          // Please be noted that it means rows be deleted can be more than chunk size, normally
          // data has higher cardinality won't exceed the limit a lot,  but low cardinality data
          // might cause this procedure to delete a large number of rows than the limit chunk size.
          if (op != ComparisonOperation.EQ) {
              if (rowCount > chunksize) {
                  result = executePrecompiledSQL(valueAtStmt, new Object[] { chunksize }, replicated);
                  cutoffValue = result.fetchRow(0).get(0, actualType);
  
                  if (exportLog.isDebugEnabled()) {
                      exportLog.debug("Migrate on table " + tableName +
                              (replicated ? "" : " on partition " + ctx.getPartitionId()) +
                              " reported " + cutoffValue + " target ttl");
                  }
              }
          }
          result = executePrecompiledSQL(migrateStmt,
                                         cutoffValue == null ? params : new Object[] {cutoffValue},
                                         replicated);
          long migratedRows = result.asScalarLong();
  
          // Return rows be deleted in this run and rows left for next run
          VoltTable retTable = new VoltTable(schema);
          retTable.addRow(migratedRows, rowCount - migratedRows);
          return retTable;
      }