@Override
  protected Pair<Option<Dataset<Row>>, String> fetchNextBatch(
      Option<String> lastCkptStr, long sourceLimit) {
    Dataset<Row> rows = null;
    final FileSystem fs = HadoopFSUtils.getFs(sourceSqlFile, sparkContext.hadoopConfiguration(), true);
    try {
      final Scanner scanner = new Scanner(fs.open(new Path(sourceSqlFile)));
      scanner.useDelimiter(";");
      while (scanner.hasNext()) {
        String sqlStr = scanner.next().trim();
        if (!sqlStr.isEmpty()) {
          LOG.info(sqlStr);
          // overwrite the same dataset object until the last statement then return.
          rows = sparkSession.sql(sqlStr);
        }
      }
      return Pair.of(Option.of(rows), shouldEmitCheckPoint ? String.valueOf(System.currentTimeMillis()) : null);
    } catch (IOException ioe) {
      throw new HoodieIOException("Error reading source SQL file.", ioe);
    }
  }
@Test
  public void shouldSetCheckpointForSqlFileBasedSourceWithEpochCheckpoint() throws IOException {
    UtilitiesTestBase.Helpers.copyToDFS(
        "streamer-config/sql-file-based-source.sql", storage,
        UtilitiesTestBase.basePath + "/sql-file-based-source.sql");

    props.setProperty(sqlFileSourceConfig, UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
    props.setProperty(sqlFileSourceConfigEmitChkPointConf, "true");

    sqlFileSource = new SqlFileBasedSource(props, jsc, sparkSession, schemaProvider);
    Pair<Option<Dataset<Row>>, String> nextBatch = sqlFileSource.fetchNextBatch(Option.empty(), Long.MAX_VALUE);

    assertEquals(10000, nextBatch.getLeft().get().count());
    long currentTimeInMillis = System.currentTimeMillis();
    long checkpointToBeUsed = Long.parseLong(nextBatch.getRight());
    assertTrue((currentTimeInMillis - checkpointToBeUsed) / 1000 < 60);
    assertTrue(currentTimeInMillis > checkpointToBeUsed);
  }