@Override
  public Dataset<FileStat> get() {
    return spark
        .sql(String.format("SELECT file_path, file_size_in_bytes FROM %s.data_files", tableName))
        .map(new FileStatMapper(), Encoders.bean(FileStat.class));
  }
@Test
  public void testTableFileStats() throws Exception {
    final String testTable = "db.test_table_file_stats";
    try (SparkSession spark = getSparkSession()) {
      spark.sql("USE openhouse");
      spark.sql(String.format("CREATE TABLE %s (id INT, data STRING)", testTable));
      spark.sql(String.format("INSERT INTO %s VALUES (0, '')", testTable));
      spark.sql(
          String.format("INSERT INTO %s VALUES (100000000, '000000000000000000000')", testTable));
      TableFileStats tableFileStats =
          TableFileStats.builder().spark(spark).tableName(testTable).build();
      Map<String, Long> stats =
          tableFileStats.get().collectAsList().stream()
              .collect(Collectors.toMap(FileStat::getPath, FileStat::getSize));
      FileSystem fs = FileSystem.get(spark.sparkContext().hadoopConfiguration());
      Path tableDirectory =
          new Path(
                  spark
                      .sql(
                          String.format(
                              "SHOW TBLPROPERTIES %s ('openhouse.tableLocation')", testTable))
                      .collectAsList()
                      .get(0)
                      .getString(1))
              .getParent();
      Map<String, Long> expectedStats = new HashMap<>();
      for (FileStatus fileStatus : fs.listStatus(new Path(tableDirectory, "data"))) {
        expectedStats.put(
            fileStatus.getPath().toString().substring("file:".length()), fileStatus.getLen());
      }
      Assertions.assertEquals(expectedStats, stats);
    }
  }