@Override
    protected void processRecord(RowData row) {
        synchronized (resultLock) {
            boolean isInsertOp =
                    row.getRowKind() == RowKind.INSERT || row.getRowKind() == RowKind.UPDATE_AFTER;
            // Always set the RowKind to INSERT, so that we can compare rows correctly (RowKind will
            // be ignored),
            row.setRowKind(RowKind.INSERT);

            // insert
            if (isInsertOp) {
                processInsert(row);
            }
            // delete
            else {
                processDelete(row);
            }
        }
    }
@Test
    void testSnapshot() {
        final ResolvedSchema schema =
                ResolvedSchema.physical(
                        new String[] {"f0", "f1"},
                        new DataType[] {DataTypes.STRING(), DataTypes.INT()});

        @SuppressWarnings({"unchecked", "rawtypes"})
        final DataStructureConverter<RowData, Row> rowConverter =
                (DataStructureConverter)
                        DataStructureConverters.getConverter(schema.toPhysicalRowDataType());

        try (TestMaterializedCollectStreamResult result =
                new TestMaterializedCollectStreamResult(
                        CliClientTestUtils.createTestClient(schema),
                        Integer.MAX_VALUE,
                        createInternalBinaryRowDataConverter(schema.toPhysicalRowDataType()))) {
            result.isRetrieving = true;

            result.processRecord(Row.ofKind(RowKind.INSERT, "A", 1));
            result.processRecord(Row.ofKind(RowKind.INSERT, "B", 1));
            result.processRecord(Row.ofKind(RowKind.INSERT, "A", 1));
            result.processRecord(Row.ofKind(RowKind.INSERT, "C", 2));

            assertThat(result.snapshot(1)).isEqualTo(TypedResult.payload(4));

            assertRowEquals(
                    Collections.singletonList(Row.of("A", 1)),
                    result.retrievePage(1),
                    rowConverter);
            assertRowEquals(
                    Collections.singletonList(Row.of("B", 1)),
                    result.retrievePage(2),
                    rowConverter);
            assertRowEquals(
                    Collections.singletonList(Row.of("A", 1)),
                    result.retrievePage(3),
                    rowConverter);
            assertRowEquals(
                    Collections.singletonList(Row.of("C", 2)),
                    result.retrievePage(4),
                    rowConverter);

            result.processRecord(Row.ofKind(RowKind.UPDATE_BEFORE, "A", 1));

            assertThat(result.snapshot(1)).isEqualTo(TypedResult.payload(3));

            assertRowEquals(
                    Collections.singletonList(Row.of("A", 1)),
                    result.retrievePage(1),
                    rowConverter);
            assertRowEquals(
                    Collections.singletonList(Row.of("B", 1)),
                    result.retrievePage(2),
                    rowConverter);
            assertRowEquals(
                    Collections.singletonList(Row.of("C", 2)),
                    result.retrievePage(3),
                    rowConverter);

            result.processRecord(Row.ofKind(RowKind.UPDATE_BEFORE, "C", 2));
            result.processRecord(Row.ofKind(RowKind.UPDATE_BEFORE, "A", 1));
            result.processRecord(Row.ofKind(RowKind.UPDATE_AFTER, "D", 1));

            assertThat(result.snapshot(1)).isEqualTo(TypedResult.payload(2));

            assertRowEquals(
                    Collections.singletonList(Row.of("B", 1)),
                    result.retrievePage(1),
                    rowConverter);
            assertRowEquals(
                    Collections.singletonList(Row.of("D", 1)),
                    result.retrievePage(2),
                    rowConverter);
        }
    }