@Override
  public long extractWatermark(IcebergSourceSplit split) {
    return split.task().files().stream()
        .map(
            scanTask -> {
              Preconditions.checkArgument(
                  scanTask.file().lowerBounds() != null
                      && scanTask.file().lowerBounds().get(eventTimeFieldId) != null,
                  "Missing statistics for column name = %s in file = %s",
                  eventTimeFieldName,
                  eventTimeFieldId,
                  scanTask.file());
              return timeUnit.toMillis(
                  Conversions.fromByteBuffer(
                      Types.LongType.get(), scanTask.file().lowerBounds().get(eventTimeFieldId)));
            })
        .min(Comparator.comparingLong(l -> l))
        .get();
  }
@TestTemplate
  public void testTimeUnit() throws IOException {
    assumeThat(columnName).isEqualTo("long_column");
    ColumnStatsWatermarkExtractor extractor =
        new ColumnStatsWatermarkExtractor(SCHEMA, columnName, TimeUnit.MICROSECONDS);

    assertThat(extractor.extractWatermark(split(0)))
        .isEqualTo(MIN_VALUES.get(0).get(columnName) / 1000L);
  }