public static Map<Long, List<DataFile>> snapshotToDataFiles(Table table) throws IOException {
        table.refresh();

        Map<Long, List<DataFile>> result = Maps.newHashMap();
        Snapshot current = table.currentSnapshot();
        while (current != null) {
            TableScan tableScan = table.newScan();
            if (current.parentId() != null) {
                // Collect the data files that was added only in current snapshot.
                tableScan = tableScan.appendsBetween(current.parentId(), current.snapshotId());
            } else {
                // Collect the data files that was added in the oldest snapshot.
                tableScan = tableScan.useSnapshot(current.snapshotId());
            }
            try (CloseableIterable<FileScanTask> scanTasks = tableScan.planFiles()) {
                result.put(
                        current.snapshotId(),
                        ImmutableList.copyOf(Iterables.transform(scanTasks, FileScanTask::file)));
            }

            // Continue to traverse the parent snapshot if exists.
            if (current.parentId() == null) {
                break;
            }
            // Iterate to the parent snapshot.
            current = table.snapshot(current.parentId());
        }
        return result;
    }
Assert table contains the expected list of records after waiting up to {@code maxCheckCount}
with {@code checkInterval}
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/test/java/org/apache/inlong/sort/iceberg/sink/util/SimpleDataUtil.java