public static Record convertDataRecordToRecord(final String database, final String schema, final DataRecord dataRecord) {
        List<TableColumn> before = new LinkedList<>();
        List<TableColumn> after = new LinkedList<>();
        for (Column column : dataRecord.getColumns()) {
            before.add(TableColumn.newBuilder().setName(column.getName()).setValue(Any.pack(ColumnValueConvertUtils.convertToProtobufMessage(column.getOldValue()))).build());
            after.add(TableColumn.newBuilder().setName(column.getName()).setValue(Any.pack(ColumnValueConvertUtils.convertToProtobufMessage(column.getValue()))).build());
        }
        MetaData metaData = MetaData.newBuilder().setDatabase(database).setSchema(Strings.nullToEmpty(schema)).setTable(dataRecord.getTableName()).build();
        return DataRecordResult.Record.newBuilder().setMetaData(metaData).addAllBefore(before).addAllAfter(after).setTransactionCommitMillis(dataRecord.getCommitTime())
                .setDataChangeType(getDataChangeType(dataRecord.getType())).build();
    }
@Test
    void assertConvertDataRecordToRecord() throws InvalidProtocolBufferException, SQLException {
        DataRecord dataRecord = new DataRecord(PipelineSQLOperationType.INSERT, "t_order", new IntegerPrimaryKeyIngestPosition(0L, 1L), 2);
        dataRecord.addColumn(new Column("order_id", BigInteger.ONE, false, true));
        dataRecord.addColumn(new Column("price", BigDecimal.valueOf(123L), false, false));
        dataRecord.addColumn(new Column("user_id", Long.MAX_VALUE, false, false));
        dataRecord.addColumn(new Column("item_id", Integer.MAX_VALUE, false, false));
        dataRecord.addColumn(new Column("create_date", LocalDate.now(), false, false));
        dataRecord.addColumn(new Column("create_date2", Date.valueOf(LocalDate.now()), false, false));
        dataRecord.addColumn(new Column("create_time", LocalTime.now(), false, false));
        dataRecord.addColumn(new Column("create_time2", OffsetTime.now(), false, false));
        dataRecord.addColumn(new Column("create_datetime", LocalDateTime.now(), false, false));
        dataRecord.addColumn(new Column("create_datetime2", OffsetDateTime.now(), false, false));
        dataRecord.addColumn(new Column("empty", null, false, false));
        Blob mockedBlob = mock(Blob.class);
        when(mockedBlob.getBytes(anyLong(), anyInt())).thenReturn(new byte[]{-1, 0, 1});
        dataRecord.addColumn(new Column("data_blob", mockedBlob, false, false));
        Clob mockedClob = mock(Clob.class);
        when(mockedClob.getSubString(anyLong(), anyInt())).thenReturn("clob\n");
        dataRecord.addColumn(new Column("text_clob", mockedClob, false, false));
        dataRecord.addColumn(new Column("update_time", new Timestamp(System.currentTimeMillis()), false, false));
        TypeRegistry registry = TypeRegistry.newBuilder().add(EmptyProto.getDescriptor().getMessageTypes()).add(TimestampProto.getDescriptor().getMessageTypes())
                .add(WrappersProto.getDescriptor().getMessageTypes()).build();
        Record expectedRecord = DataRecordResultConvertUtils.convertDataRecordToRecord("test", null, dataRecord);
        String print = JsonFormat.printer().usingTypeRegistry(registry).print(expectedRecord);
        Builder actualRecord = Record.newBuilder();
        JsonFormat.parser().usingTypeRegistry(registry).merge(print, actualRecord);
        assertThat(actualRecord.build(), is(expectedRecord));
    }