@Override
  public Dataset<Row> apply(JavaSparkContext jsc, SparkSession sparkSession, Dataset<Row> rowDataset,
      TypedProperties properties) {
    String transformerSQL = getStringWithAltKeys(properties, SqlTransformerConfig.TRANSFORMER_SQL);

    try {
      // tmp table name doesn't like dashes
      String tmpTable = TMP_TABLE.concat(UUID.randomUUID().toString().replace("-", "_"));
      LOG.info("Registering tmp table: {}", tmpTable);
      rowDataset.createOrReplaceTempView(tmpTable);
      String sqlStr = transformerSQL.replaceAll(SRC_PATTERN, tmpTable);
      LOG.debug("SQL Query for transformation: {}", sqlStr);
      Dataset<Row> transformed = sparkSession.sql(sqlStr);
      sparkSession.catalog().dropTempView(tmpTable);
      return transformed;
    } catch (Exception e) {
      throw new HoodieTransformExecutionException("Failed to apply sql query based transformer", e);
    }
  }
@Test
  public void testSqlQuery() {

    SparkSession spark = SparkSession
        .builder()
        .config(getSparkConfForTest(TestSqlQueryBasedTransformer.class.getName()))
        .getOrCreate();

    JavaSparkContext jsc = JavaSparkContext.fromSparkContext(spark.sparkContext());

    // prepare test data
    String testData = "{\n"
        + "  \"ts\": 1622126968000,\n"
        + "  \"uuid\": \"c978e157-72ee-4819-8f04-8e46e1bb357a\",\n"
        + "  \"rider\": \"rider-213\",\n"
        + "  \"driver\": \"driver-213\",\n"
        + "  \"begin_lat\": 0.4726905879569653,\n"
        + "  \"begin_lon\": 0.46157858450465483,\n"
        + "  \"end_lat\": 0.754803407008858,\n"
        + "  \"end_lon\": 0.9671159942018241,\n"
        + "  \"fare\": 34.158284716382845,\n"
        + "  \"partitionpath\": \"americas/brazil/sao_paulo\"\n"
        + "}";

    JavaRDD<String> testRdd = jsc.parallelize(Collections.singletonList(testData), 2);
    Dataset<Row> ds = spark.read().json(testRdd);

    // create a new column dt, whose value is transformed from ts, format is yyyyMMdd
    String transSql = "select\n"
        + "\tuuid,\n"
        + "\tbegin_lat,\n"
        + "\tbegin_lon,\n"
        + "\tdriver,\n"
        + "\tend_lat,\n"
        + "\tend_lon,\n"
        + "\tfare,\n"
        + "\tpartitionpath,\n"
        + "\trider,\n"
        + "\tts,\n"
        + "\tFROM_UNIXTIME(ts / 1000, 'yyyyMMdd') as dt\n"
        + "from\n"
        + "\t<SRC>";
    TypedProperties props = new TypedProperties();
    props.put("hoodie.streamer.transformer.sql", transSql);

    // transform
    SqlQueryBasedTransformer transformer = new SqlQueryBasedTransformer();

    // test if the class throws illegal argument exception when sql-config is missing
    assertThrows(IllegalArgumentException.class, () -> transformer.apply(jsc, spark, ds, new TypedProperties()));

    Dataset<Row> result = transformer.apply(jsc, spark, ds, props);

    // check result
    assertEquals(11, result.columns().length);
    assertNotNull(result.col("dt"));
    assertEquals("20210527", result.first().get(10).toString());

    spark.close();
  }