@Override
    public RowData nextRecord(RowData reuse) {
        // return the next row
        row.setRowId(this.nextRow++);
        return row;
    }
@Test
    void testReadFileWithSelectFields() throws IOException {
        FileInputSplit[] splits = createSplits(testFileFlat, 4);

        long cnt = 0;
        long totalF0 = 0;

        Map<String, Object> partSpec = new HashMap<>();
        partSpec.put("f1", 1);
        partSpec.put("f3", 3L);
        partSpec.put("f5", "f5");
        partSpec.put("f8", BigDecimal.valueOf(5.333));
        partSpec.put("f13", "f13");

        // read all splits
        for (FileInputSplit split : splits) {
            try (OrcColumnarRowSplitReader reader =
                    createReader(
                            new int[] {8, 1, 3, 0, 5, 2},
                            new DataType[] {
                                /* 0 */ DataTypes.INT(),
                                /* 1 */ DataTypes.INT(), // part-1
                                /* 2 */ DataTypes.STRING(),
                                /* 3 */ DataTypes.BIGINT(), // part-2
                                /* 4 */ DataTypes.STRING(),
                                /* 5 */ DataTypes.STRING(), // part-3
                                /* 6 */ DataTypes.STRING(),
                                /* 7 */ DataTypes.INT(),
                                /* 8 */ DataTypes.DECIMAL(10, 5), // part-4
                                /* 9 */ DataTypes.STRING(),
                                /* 11*/ DataTypes.INT(),
                                /* 12*/ DataTypes.INT(),
                                /* 13*/ DataTypes.STRING(), // part-5
                                /* 14*/ DataTypes.INT()
                            },
                            partSpec,
                            split)) {
                // read and count all rows
                while (!reader.reachedEnd()) {
                    RowData row = reader.nextRecord(null);

                    // data values
                    assertThat(row.isNullAt(3)).isFalse();
                    assertThat(row.isNullAt(5)).isFalse();
                    totalF0 += row.getInt(3);
                    assertThat(row.getString(5).toString()).isNotNull();

                    // part values
                    assertThat(row.isNullAt(0)).isFalse();
                    assertThat(row.isNullAt(1)).isFalse();
                    assertThat(row.isNullAt(2)).isFalse();
                    assertThat(row.isNullAt(4)).isFalse();
                    assertThat(row.getDecimal(0, 10, 5))
                            .isEqualTo(DecimalDataUtils.castFrom(5.333, 10, 5));
                    assertThat(row.getInt(1)).isEqualTo(1);
                    assertThat(row.getLong(2)).isEqualTo(3);
                    assertThat(row.getString(4).toString()).isEqualTo("f5");
                    cnt++;
                }
            }
        }
        // check that all rows have been read
        assertThat(cnt).isEqualTo(1920800);
        assertThat(totalF0).isEqualTo(1844737280400L);
    }