@Override
  public final long getEstimatedSizeBytes(PipelineOptions options) throws IOException {
    // This implementation of method getEstimatedSizeBytes is provided to simplify subclasses. Here
    // we perform the size estimation of files and file patterns using the interface provided by
    // FileSystem.
    String fileOrPattern = fileOrPatternSpec.get();

    if (mode == Mode.FILEPATTERN) {
      Long maybeNumBytes = filesSizeBytes.get();
      if (maybeNumBytes != null) {
        return maybeNumBytes;
      }

      long totalSize = 0;
      List<Metadata> allMatches = FileSystems.match(fileOrPattern, emptyMatchTreatment).metadata();
      for (Metadata metadata : allMatches) {
        totalSize += metadata.sizeBytes();
      }
      LOG.info(
          "Filepattern {} matched {} files with total size {}",
          fileOrPattern,
          allMatches.size(),
          totalSize);

      filesSizeBytes.compareAndSet(null, totalSize);
      return totalSize;
    } else {
      long start = getStartOffset();
      long end = Math.min(getEndOffset(), getMaxEndOffset(options));
      return end - start;
    }
  }
@Test
  public void testEstimatedSizeOfFilePattern() throws Exception {
    List<String> data1 = createStringDataset(3, 20);
    File file1 = createFileWithData("file1", data1);

    List<String> data2 = createStringDataset(3, 40);
    File file2 = createFileWithData("file2", data2);

    List<String> data3 = createStringDataset(3, 30);
    File file3 = createFileWithData("file3", data3);

    List<String> data4 = createStringDataset(3, 45);
    createFileWithData("otherfile", data4);

    List<String> data5 = createStringDataset(3, 53);
    createFileWithData("anotherfile", data5);

    TestFileBasedSource source =
        new TestFileBasedSource(new File(file1.getParent(), "file*").getPath(), 64, null);

    // Estimated size of the file pattern based source should be the total size of files that the
    // corresponding pattern is expanded into.
    assertEquals(
        file1.length() + file2.length() + file3.length(), source.getEstimatedSizeBytes(null));
  }