@Override
  public Dataset<Row> apply(
      final JavaSparkContext jsc,
      final SparkSession sparkSession,
      final Dataset<Row> rowDataset,
      final TypedProperties props) {

    final String sqlFile = getStringWithAltKeys(props, SqlTransformerConfig.TRANSFORMER_SQL_FILE);

    final FileSystem fs = HadoopFSUtils.getFs(sqlFile, jsc.hadoopConfiguration(), true);
    // tmp table name doesn't like dashes
    final String tmpTable = TMP_TABLE.concat(UUID.randomUUID().toString().replace("-", "_"));
    LOG.info("Registering tmp table: {}", tmpTable);
    rowDataset.createOrReplaceTempView(tmpTable);

    try (final Scanner scanner = new Scanner(fs.open(new Path(sqlFile)), "UTF-8")) {
      Dataset<Row> rows = null;
      // each sql statement is separated with semicolon hence set that as delimiter.
      scanner.useDelimiter(";");
      LOG.info("SQL Query for transformation:");
      while (scanner.hasNext()) {
        String sqlStr = scanner.next();
        sqlStr = sqlStr.replaceAll(SRC_PATTERN, tmpTable).trim();
        if (!sqlStr.isEmpty()) {
          LOG.info(sqlStr);
          // overwrite the same dataset object until the last statement then return.
          rows = sparkSession.sql(sqlStr);
        }
      }
      return rows;
    } catch (final IOException ioe) {
      throw new HoodieTransformExecutionException("Error reading transformer SQL file.", ioe);
    } finally {
      sparkSession.catalog().dropTempView(tmpTable);
    }
  }
@Test
  public void testSqlFileBasedTransformerInvalidSQL() throws IOException {
    UtilitiesTestBase.Helpers.copyToDFS(
        "streamer-config/sql-file-transformer-invalid.sql",
        UtilitiesTestBase.storage,
        UtilitiesTestBase.basePath + "/sql-file-transformer-invalid.sql");

    // Test if the SQL file based transformer works as expected for the invalid SQL statements.
    props.setProperty(
        "hoodie.streamer.transformer.sql.file",
        UtilitiesTestBase.basePath + "/sql-file-transformer-invalid.sql");
    assertThrows(
        ParseException.class,
        () -> sqlFileTransformer.apply(jsc, sparkSession, inputDatasetRows, props));
  }