@Override
    public CloseableIterator<ColumnarBatch> readParquetFiles(
            CloseableIterator<FileStatus> fileIter,
            StructType physicalSchema,
            Optional<Predicate> predicate) {
        return new CloseableIterator<>() {
            private int currentReadColumnarBatchIndex = -1;
            private List<ColumnarBatch> currentColumnarBatchList = Lists.newArrayList();
            private String currentFile;

            @Override
            public void close() {
                Utils.closeCloseables(fileIter);
                currentReadColumnarBatchIndex = -1;
                currentColumnarBatchList = null;
            }

            @Override
            public boolean hasNext() {
                if (hasNextToConsume()) {
                    return true;
                } else {
                    currentReadColumnarBatchIndex = -1;
                    currentColumnarBatchList = Lists.newArrayList();
                    // There is no file in reading or the current file being read has no more data.
                    // Initialize the next file reader or return false if there are no more files to
                    // read.
                    try {
                        tryGetNextFileColumnarBatch();
                    } catch (Exception ex) {
                        throw new KernelEngineException(
                                format("Error reading Parquet file: %s", currentFile), ex);
                    }

                    if (hasNextToConsume()) {
                        return true;
                    } else if (fileIter.hasNext()) {
                        // recurse since it's possible the loaded file is empty
                        return hasNext();
                    } else {
                        return false;
                    }
                }
            }

            private boolean hasNextToConsume() {
                return currentReadColumnarBatchIndex != -1 && !currentColumnarBatchList.isEmpty() &&
                        currentReadColumnarBatchIndex < currentColumnarBatchList.size();
            }

            @Override
            public ColumnarBatch next() {
                return currentColumnarBatchList.get(currentReadColumnarBatchIndex++);
            }

            private void tryGetNextFileColumnarBatch() throws ExecutionException {
                if (fileIter.hasNext()) {
                    currentFile = fileIter.next().getPath();
                    if (LogReplay.containsAddOrRemoveFileActions(physicalSchema)) {
                        Pair<String, StructType> key = Pair.create(currentFile, physicalSchema);
                        if (checkpointCache.getIfPresent(key) != null || predicate.isEmpty()) {
                            currentColumnarBatchList = checkpointCache.get(key);
                        } else {
                            currentColumnarBatchList = readParquetFile(currentFile, physicalSchema, hadoopConf);
                        }
                    } else {
                        currentColumnarBatchList = readParquetFile(currentFile, physicalSchema, hadoopConf);
                    }
                    currentReadColumnarBatchIndex = 0;
                }
            }
        };
    }
@Test
    public void testParquetMetadata() {
        String path = deltaLakePath + "/00000000000000000030.checkpoint.parquet";
        DeltaLakeParquetHandler deltaLakeParquetHandler = new DeltaLakeParquetHandler(hdfsConfiguration, checkpointCache);
        StructType readSchema = LogReplay.getAddRemoveReadSchema(true);
        FileStatus fileStatus = FileStatus.of(path, 0, 0);

        List<Row> addRows = Lists.newArrayList();
        try (CloseableIterator<ColumnarBatch> parquetIter = deltaLakeParquetHandler.readParquetFiles(
                Utils.singletonCloseableIterator(fileStatus), readSchema, Optional.empty())) {
            while (parquetIter.hasNext()) {
                ColumnarBatch columnarBatch = parquetIter.next();
                ColumnVector addsVector = columnarBatch.getColumnVector(ADD_FILE_ORDINAL);

                for (int rowId = 0; rowId < addsVector.getSize(); rowId++) {
                    if (addsVector.isNullAt(rowId)) {
                        continue;
                    }
                    getAddFilePath(addsVector, rowId);
                }

                try (CloseableIterator<Row> rows = columnarBatch.getRows()) {
                    while (rows.hasNext()) {
                        Row row = rows.next();
                        addRows.add(row);
                    }
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        Assert.assertEquals(32, addRows.size());
        List<String> pathList = Lists.newArrayList();
        Set<String> partitionValues = Sets.newHashSet();
        for (Row scanRow : addRows) {
            if (scanRow.isNullAt(InternalScanFileUtils.ADD_FILE_ORDINAL)) {
                continue;
            }
            Row addFile = getAddFileEntry(scanRow);
            pathList.add(addFile.getString(ADD_FILE_PATH_ORDINAL));
            partitionValues.addAll(InternalScanFileUtils.getPartitionValues(scanRow).values());
        }

        Assert.assertEquals(30, pathList.size());
        Assert.assertEquals(18, partitionValues.size());
        Assert.assertFalse(checkpointCache.asMap().isEmpty());
        Assert.assertTrue(checkpointCache.asMap().containsKey(Pair.create(path, readSchema)));
    }