public static void insertRecordsIntoDB(WarehouseInstance primary, String DbName, String primaryDbNameExtra,
                                          String tableName, String tableNameOp, boolean isMMTable,
                             OperationType opType) throws Throwable {
    String[] resultArray = new String[]{"1", "2", "3", "4", "5"};
    String tableProperty;
    String tableStorage;

    if (primary.isAcidEnabled()) {
      tableProperty = "'transactional'='true'";
      if (isMMTable) {
        tableProperty = setMMtableProperty(tableProperty);
      }
      tableStorage = "STORED AS ORC";
    } else {
      // create non-acid table, which will be converted to acid at target cluster.
      tableProperty = "'transactional'='false'";
      if (isMMTable) {
        // for migration to MM table, storage type should be non-orc
        tableStorage = "";
      } else {
        // for migration to full acid table, storage type should be ORC
        tableStorage = "STORED AS ORC";
      }
    }

    primary.run("use " + DbName);

    switch (opType) {
      case REPL_TEST_ACID_INSERT:
        insertIntoDB(primary, DbName, tableName, tableProperty, tableStorage, resultArray);
        if (primaryDbNameExtra != null) {
          insertIntoDB(primary, primaryDbNameExtra, tableName, tableProperty, tableStorage, resultArray);
        }
        return;
      case REPL_TEST_ACID_INSERT_OVERWRITE:
        primary.run("CREATE TABLE " + tableNameOp + " (key int, value int) PARTITIONED BY (load_date date) " +
              "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( "+ tableProperty + " )")
        .run("INSERT INTO " + tableNameOp + " partition (load_date='2016-03-01') VALUES (2, 2)")
        .run("INSERT INTO " + tableNameOp + " partition (load_date='2016-03-01') VALUES (10, 12)")
        .run("INSERT INTO " + tableNameOp + " partition (load_date='2016-03-02') VALUES (11, 1)")
        .run("select key from " + tableNameOp + " order by key")
        .verifyResults(new String[]{"2", "10", "11"})
        .run("insert overwrite table " + tableNameOp + " select * from " + tableName)
        .run("CREATE TABLE " + tableNameOp + "_nopart (key int, value int) " +
                "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( "+ tableProperty + " )")
        .run("INSERT INTO " + tableNameOp + "_nopart VALUES (2, 2)")
        .run("INSERT INTO " + tableNameOp + "_nopart VALUES (10, 12)")
        .run("INSERT INTO " + tableNameOp + "_nopart VALUES (11, 1)")
        .run("select key from " + tableNameOp + "_nopart" + " order by key")
        .verifyResults(new String[]{"2", "10", "11"})
        .run("insert overwrite table " + tableNameOp + "_nopart select * from " + tableName + "_nopart")
        .run("select key from " + tableNameOp + "_nopart" + " order by key");
        break;
      case REPL_TEST_ACID_INSERT_SELECT:
        primary.run("CREATE TABLE " + tableNameOp + " (key int, value int) PARTITIONED BY (load_date date) " +
            "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( " + tableProperty + " )")
        .run("insert into " + tableNameOp + " partition (load_date) select * from " + tableName)
        .run("CREATE TABLE " + tableNameOp + "_nopart (key int, value int) " +
                "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( " + tableProperty + " )")
        .run("insert into " + tableNameOp + "_nopart select * from " + tableName + "_nopart");
        break;
      case REPL_TEST_ACID_INSERT_IMPORT:
        String path = "hdfs:///tmp/" + DbName + "/";
        String exportPath = "'" + path + tableName + "/'";
        String exportPathNoPart = "'" + path + tableName + "_nopart/'";
        primary.run("export table " + tableName + " to " + exportPath)
        .run("import table " + tableNameOp + " from " + exportPath)
        .run("export table " + tableName + "_nopart to " + exportPathNoPart)
        .run("import table " + tableNameOp + "_nopart from " + exportPathNoPart);
        break;
      case REPL_TEST_ACID_CTAS:
        primary.run("create table " + tableNameOp + " partitioned by (load_date) " + tableStorage
                            + " tblproperties (" + tableProperty + ") as select * from " + tableName)
                .run("create table " + tableNameOp + "_nopart " + tableStorage
                            + " tblproperties (" + tableProperty + ") as select * from " + tableName + "_nopart");
        break;
      case REPL_TEST_ACID_INSERT_LOADLOCAL:
        // For simplicity setting key and value as same value
        StringBuilder buf = new StringBuilder();
        boolean nextVal = false;
        for (String key : resultArray) {
          if (nextVal) {
            buf.append(',');
          }
          buf.append('(');
          buf.append(key);
          buf.append(',');
          buf.append(key);
          buf.append(')');
          nextVal = true;
        }

        primary.run("CREATE TABLE " + tableNameOp + "_temp (key int, value int) " + tableStorage + "")
        .run("INSERT INTO TABLE " + tableNameOp + "_temp VALUES " + buf.toString())
        .run("SELECT key FROM " + tableNameOp + "_temp")
        .verifyResults(resultArray)
        .run("CREATE TABLE " + tableNameOp + " (key int, value int) PARTITIONED BY (load_date date) " +
              "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( " + tableProperty + ")")
        .run("SHOW TABLES LIKE '" + tableNameOp + "'")
        .verifyResult(tableNameOp)
        .run("INSERT OVERWRITE LOCAL DIRECTORY './test.dat' " + tableStorage + " SELECT * FROM " + tableNameOp + "_temp")
        .run("LOAD DATA LOCAL INPATH './test.dat/000000_0' OVERWRITE INTO TABLE " + tableNameOp +
                " PARTITION (load_date='2008-08-15')")
        .run("CREATE TABLE " + tableNameOp + "_nopart (key int, value int) " +
                      "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( " + tableProperty + ")")
        .run("SHOW TABLES LIKE '" + tableNameOp + "_nopart'")
        .verifyResult(tableNameOp + "_nopart")
        .run("LOAD DATA LOCAL INPATH './test.dat/000000_0' OVERWRITE INTO TABLE " + tableNameOp + "_nopart");
        break;
      case REPL_TEST_ACID_INSERT_UNION:
        primary.run("CREATE TABLE " + tableNameOp + " (key int, value int) PARTITIONED BY (load_date date) " +
                "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( " + tableProperty + ")")
                .run("SHOW TABLES LIKE '" + tableNameOp + "'")
                .verifyResult(tableNameOp)
                .run("insert overwrite table " + tableNameOp + " partition (load_date) select * from " + tableName +
                    " union all select * from " + tableName)
                .run("CREATE TABLE " + tableNameOp + "_nopart (key int, value int) " +
                "CLUSTERED BY(key) INTO 3 BUCKETS " + tableStorage + " TBLPROPERTIES ( " + tableProperty + ")")
                .run("insert overwrite table " + tableNameOp + "_nopart select * from " + tableName +
                        "_nopart union all select * from " + tableName + "_nopart");
        resultArray = new String[]{"1", "2", "3", "4", "5", "1", "2", "3", "4", "5"};
        break;
      default:
        return;
    }
    primary.run("select key from " + tableNameOp + " order by key").verifyResults(resultArray);
    primary.run("select key from " + tableNameOp + "_nopart" + " order by key").verifyResults(resultArray);
  }
ReplicationTestUtils - static helper functions for replication test
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/ReplicationTestUtils.java