public IcebergTableStats collectTableStats(String fqtn) {
    Table table = getTable(fqtn);

    TableStatsCollector tableStatsCollector;
    try {
      tableStatsCollector = new TableStatsCollector(fs(), spark, fqtn, table);
    } catch (IOException e) {
      log.error("Unable to initialize file system for table stats collection", e);
      return null;
    }

    IcebergTableStats tableStats = tableStatsCollector.collectTableStats();
    return tableStats;
  }
@Test
  public void testCollectTableStats() throws Exception {
    final String tableName = "db.test_collect_table_stats";
    final int numInserts = 3;
    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
      prepareTable(ops, tableName);
      IcebergTableStats stats = ops.collectTableStats(tableName);

      // Validate empty data files case
      Assertions.assertEquals(stats.getNumReferencedDataFiles(), 0);
      Assertions.assertEquals(stats.getNumExistingMetadataJsonFiles(), 1);
      long modifiedTimeStamp = System.currentTimeMillis();

      populateTable(ops, tableName, 1);
      stats = ops.collectTableStats(tableName);
      Assertions.assertEquals(stats.getNumReferencedDataFiles(), 1);
      Assertions.assertTrue(stats.getTableLastUpdatedTimestamp() >= modifiedTimeStamp);

      // Capture first snapshot timestamp
      Table table = ops.getTable(tableName);
      long oldestSnapshot = table.currentSnapshot().timestampMillis();

      // Add more records and validate other stats
      populateTable(ops, tableName, numInserts);
      table = ops.getTable(tableName);
      log.info("Loaded table {}, location {}", table.name(), table.location());
      stats = ops.collectTableStats(tableName);
      Assertions.assertEquals(stats.getCurrentSnapshotId(), table.currentSnapshot().snapshotId());
      Assertions.assertEquals(stats.getNumReferencedDataFiles(), numInserts + 1);
      Assertions.assertEquals(stats.getNumExistingMetadataJsonFiles(), numInserts + 2);
      Assertions.assertEquals(
          stats.getCurrentSnapshotTimestamp(), table.currentSnapshot().timestampMillis());
      Assertions.assertEquals(stats.getOldestSnapshotTimestamp(), oldestSnapshot);
      Assertions.assertEquals(
          stats.getNumObjectsInDirectory(),
          stats.getNumReferencedDataFiles()
              + stats.getNumExistingMetadataJsonFiles()
              + stats.getNumReferencedManifestFiles()
              + stats.getNumReferencedManifestLists());
    }
  }