public static List<ColumnarBatch> readParquetFile(String filePath, StructType physicalSchema, Configuration hadoopConf) {
        try (Timer ignored = Tracers.watchScope(Tracers.get(), EXTERNAL,
                "DeltaLakeParquetHandler.readParquetFileAndGetColumnarBatch")) {
            io.delta.kernel.defaults.internal.parquet.ParquetFileReader batchReader =
                    new io.delta.kernel.defaults.internal.parquet.ParquetFileReader(hadoopConf);
            CloseableIterator<ColumnarBatch> currentFileReader = batchReader.read(filePath, physicalSchema, Optional.empty());

            List<ColumnarBatch> result = Lists.newArrayList();
            while (currentFileReader != null && currentFileReader.hasNext()) {
                result.add(currentFileReader.next());
            }
            Utils.closeCloseables(currentFileReader);
            return result;
        }
    }
@Test
    public void testCheckpointCache() throws ExecutionException {
        LoadingCache<Pair<String, StructType>, List<ColumnarBatch>> checkpointCache = CacheBuilder.newBuilder()
                .expireAfterWrite(3600, TimeUnit.SECONDS)
                .weigher((key, value) ->
                        Math.toIntExact(SizeEstimator.estimate(key) + SizeEstimator.estimate(value)))
                .maximumWeight(1024 * 2)
                .concurrencyLevel(1)
                .build(new CacheLoader<>() {
                    @NotNull
                    @Override
                    public List<ColumnarBatch> load(@NotNull Pair<String, StructType> pair) {
                        return DeltaLakeParquetHandler.readParquetFile(pair.first, pair.second, hdfsConfiguration);
                    }
                });
        List<ColumnarBatch> columnarBatches = Lists.newArrayList();
        new MockUp<DeltaLakeParquetHandler>() {
            @Mock
            public List<ColumnarBatch> readParquetFile(@NotNull String path, @NotNull StructType schema,
                                                       @NotNull Configuration hdfsConfiguration) {
                return columnarBatches;
            }
        };

        List<StructField> fields = ImmutableList.of(
                new StructField("col1", IntegerType.INTEGER, true),
                new StructField("col2", StringType.STRING, true)
        );
        StructType deltaType = new io.delta.kernel.types.StructType(fields);

        String location1 = "hdfs://127.0.0.1:9000/delta_lake/00000000000000000030.checkpoint.parquet.1";
        Pair<String, StructType> pair1 = Pair.create(location1, deltaType);

        checkpointCache.get(pair1);
        Assert.assertEquals(1, checkpointCache.size());

        String location2 = "hdfs://127.0.0.1:9000/delta_lake/00000000000000000030.checkpoint.parquet.2";
        Pair<String, StructType> pair2 = Pair.create(location2, deltaType);
        checkpointCache.get(pair2);
        Assert.assertEquals(1, checkpointCache.size());
        Assert.assertFalse(checkpointCache.asMap().containsKey(pair1));
        Assert.assertTrue(checkpointCache.asMap().containsKey(pair2));
    }