@Override
    public CloseableIterator<ColumnarBatch> readJsonFiles(
            CloseableIterator<FileStatus> scanFileIter,
            StructType physicalSchema,
            Optional<Predicate> predicate) {

        return new CloseableIterator<>() {
            private String currentFile;
            // index of the current line being read from the current read json list, -1 means no line is read yet
            private int currentReadLine = -1;
            private List<JsonNode> currentReadJsonList = Lists.newArrayList();

            @Override
            public void close() {
                Utils.closeCloseables(scanFileIter);
                currentReadLine = -1;
                currentReadJsonList = null;
            }

            @Override
            public boolean hasNext() {
                if (hasNextToConsume()) {
                    return true; // we have un-consumed last read line
                }

                // 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 {
                    tryGetNextFileJson();
                } catch (Exception ex) {
                    throw new KernelEngineException(
                            format("Error reading JSON file: %s", currentFile), ex);
                }

                return hasNextToConsume();
            }

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

            @Override
            public ColumnarBatch next() {
                try (Timer ignored = Tracers.watchScope(Tracers.get(), EXTERNAL, "DeltaLakeJsonHandler.JsonToColumnarBatch")) {
                    if (!hasNextToConsume()) {
                        throw new NoSuchElementException();
                    }

                    List<Row> rows = new ArrayList<>();
                    int currentBatchSize = 0;
                    do {
                        // hasNext already reads the next file and keeps it in member variable `cachedJsonList`
                        JsonNode jsonNode = currentReadJsonList.get(currentReadLine);
                        Row row = new io.delta.kernel.defaults.internal.data.DefaultJsonRow(
                                (ObjectNode) jsonNode, physicalSchema);
                        rows.add(row);
                        currentBatchSize++;
                        currentReadLine++;
                    } while (currentBatchSize < maxBatchSize && hasNext());

                    return new io.delta.kernel.defaults.internal.data.DefaultRowBasedColumnarBatch(
                            physicalSchema, rows);
                }
            }

            private void tryGetNextFileJson() throws ExecutionException, IOException {
                if (scanFileIter.hasNext()) {
                    currentFile = scanFileIter.next().getPath();
                    Path filePath = new Path(currentFile);
                    if (filePath.getName().equals(LAST_CHECKPOINT_FILE_NAME)) {
                        // can not read last_checkpoint file from cache
                        currentReadJsonList = readJsonFile(currentFile, hadoopConf);
                    } else {
                        currentReadJsonList = jsonCache.get(currentFile);
                    }
                    currentReadLine = 0;
                }
            }
        };
    }
@Test
    public void testReadJsonMetadata() {
        String path = deltaLakePath + "/00000000000000000031.json";
        DeltaLakeJsonHandler deltaLakeJsonHandler = new DeltaLakeJsonHandler(hdfsConfiguration, jsonCache);

        StructType readSchema = LogReplay.getAddRemoveReadSchema(true);
        FileStatus fileStatus = FileStatus.of(path, 0, 0);

        List<Row> addRows = Lists.newArrayList();
        try (CloseableIterator<ColumnarBatch> jsonIter = deltaLakeJsonHandler.readJsonFiles(
                Utils.singletonCloseableIterator(fileStatus), readSchema, Optional.empty())) {
            while (jsonIter.hasNext()) {
                ColumnarBatch columnarBatch = jsonIter.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) {
            Assert.fail();
        }

        Assert.assertEquals(2, addRows.size());
        Row scanRow = addRows.get(1);
        Row addFile = getAddFileEntry(scanRow);
        Assert.assertEquals("col_date=2024-01-06/part-00000-3c9a556a-d185-4963-869d-b059d4c9b482.c000.snappy.parquet",
                addFile.getString(ADD_FILE_PATH_ORDINAL));
        Assert.assertEquals(724, addFile.getLong(ADD_FILE_SIZE_ORDINAL));
        Assert.assertEquals(1721830614469L, addFile.getLong(ADD_FILE_MOD_TIME_ORDINAL));

        Map<String, String> partitionValues = InternalScanFileUtils.getPartitionValues(scanRow);
        Assert.assertTrue(partitionValues.containsKey("col_date"));
        Assert.assertEquals("2024-01-06", partitionValues.get("col_date"));

        Assert.assertFalse(jsonCache.asMap().isEmpty());
        Assert.assertTrue(jsonCache.asMap().containsKey(path));
    }