public List<DataRecord> merge(final List<DataRecord> dataRecords) {
        Map<DataRecord.Key, DataRecord> result = new HashMap<>();
        dataRecords.forEach(each -> {
            if (PipelineSQLOperationType.INSERT == each.getType()) {
                mergeInsert(each, result);
            } else if (PipelineSQLOperationType.UPDATE == each.getType()) {
                mergeUpdate(each, result);
            } else if (PipelineSQLOperationType.DELETE == each.getType()) {
                mergeDelete(each, result);
            }
        });
        return new ArrayList<>(result.values());
    }
@Test
    void assertUpdatePrimaryKeyBeforeUpdatePrimaryKey() {
        DataRecord beforeDataRecord = mockUpdateDataRecord(1, 2, 10, 50);
        DataRecord afterDataRecord = mockUpdateDataRecord(2, 3, 10, 50);
        Collection<DataRecord> actual = groupEngine.merge(Arrays.asList(beforeDataRecord, afterDataRecord));
        assertThat(actual.size(), is(1));
        DataRecord dataRecord = actual.iterator().next();
        assertThat(dataRecord.getType(), is(PipelineSQLOperationType.UPDATE));
        assertThat(dataRecord.getTableName(), is("order"));
        assertThat(dataRecord.getActualTableName(), is("order_0"));
        assertThat(dataRecord.getCommitTime(), is(456L));
        assertColumnsMatched(dataRecord.getColumn(0), new Column("id", 1, 3, true, true));
        assertColumnsMatched(dataRecord.getColumn(1), new Column("user_id", 10, 10, false, false));
        assertColumnsMatched(dataRecord.getColumn(2), new Column("total_price", 50, 50, false, false));
    }