@Override
    protected void processRecord(RowData row) {
        // limit the materialized table
        if (materializedTable.size() - validRowPosition >= maxRowCount) {
            cleanUp();
        }
        materializedTable.add(row);
    }
@Test
    void testLimitedSnapshot() throws Exception {
        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 (TestMaterializedCollectBatchResult result =
                new TestMaterializedCollectBatchResult(
                        CliClientTestUtils.createTestClient(schema),
                        2, // limit the materialized table to 2 rows
                        3,
                        createInternalBinaryRowDataConverter(
                                schema.toPhysicalRowDataType()))) { // with 3 rows overcommitment
            result.isRetrieving = true;

            result.processRecord(Row.of("D", 1));
            result.processRecord(Row.of("A", 1));
            result.processRecord(Row.of("B", 1));
            result.processRecord(Row.of("A", 1));

            assertRowEquals(
                    Arrays.asList(
                            null, null, Row.of("B", 1), Row.of("A", 1)), // two over-committed rows
                    result.getMaterializedTable(),
                    rowConverter);

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

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

            result.processRecord(Row.of("C", 1));

            assertRowEquals(
                    Arrays.asList(Row.of("A", 1), Row.of("C", 1)), // limit clean up has taken place
                    result.getMaterializedTable(),
                    rowConverter);

            result.processRecord(Row.of("A", 1));

            assertRowEquals(
                    Arrays.asList(null, Row.of("C", 1), Row.of("A", 1)),
                    result.getMaterializedTable(),
                    rowConverter);
        }
    }