public void saveDelta() throws IOException {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS",config.fsDefaultFs);
    FileSystem fs = FileSystem.get(conf);
    Statement stmt = null;
    try {
      if (config.fromCommitTime == null) {
        config.fromCommitTime = inferCommitTime(fs);
        LOG.info("FromCommitTime inferred as " + config.fromCommitTime);
      }

      LOG.info("FromCommitTime - " + config.fromCommitTime);
      String sourceTableLocation = getTableLocation(config.sourceDb, config.sourceTable);
      String lastCommitTime = getLastCommitTimePulled(fs, sourceTableLocation);
      if (lastCommitTime == null) {
        LOG.info("Nothing to pull. However we will continue to create a empty table");
        lastCommitTime = config.fromCommitTime;
      }

      Connection conn = getConnection();
      stmt = conn.createStatement();
      // drop the temp table if exists
      String tempDbTable = config.tmpDb + "." + config.targetTable + "__" + config.sourceTable;
      String tempDbTablePath =
          config.hoodieTmpDir + "/" + config.targetTable + "__" + config.sourceTable + "/" + lastCommitTime;
      executeStatement("drop table if exists " + tempDbTable, stmt);
      deleteHDFSPath(fs, tempDbTablePath);
      if (!ensureTempPathExists(fs, lastCommitTime)) {
        throw new IllegalStateException("Could not create target path at "
            + new Path(config.hoodieTmpDir, config.targetTable + "/" + lastCommitTime));
      }

      initHiveBeelineProperties(stmt);
      executeIncrementalSQL(tempDbTable, tempDbTablePath, stmt);
      LOG.info("Finished HoodieReader execution");
    } catch (SQLException e) {
      LOG.error("Exception when executing SQL", e);
      throw new IOException("Could not scan " + config.sourceTable + " incrementally", e);
    } finally {
      try {
        if (stmt != null) {
          stmt.close();
        }
      } catch (SQLException e) {
        LOG.error("Could not close the resultSet opened ", e);
      }
    }
  }
@Test
  @EnabledIf(value = "org.apache.hudi.HoodieSparkUtils#isSpark2", disabledReason = "Disable due to hive not support avro 1.10.2.")
  public void testPuller() throws IOException, URISyntaxException {
    createTables();
    HiveIncrementalPuller.Config cfg = getHivePullerConfig("select name from testdb.test1 where `_hoodie_commit_time` > '%s'");
    HoodieHiveSyncClient hiveClient = new HoodieHiveSyncClient(new HiveSyncConfig(hiveSyncProps, HiveTestUtil.getHiveConf()));
    hiveClient.createDatabase(cfg.tmpDb);
    HiveIncrementalPuller puller = new HiveIncrementalPuller(cfg);
    puller.saveDelta();
    HoodieHiveSyncClient assertingClient = new HoodieHiveSyncClient(new HiveSyncConfig(getAssertionSyncConfig(cfg.tmpDb), HiveTestUtil.getHiveConf()));
    String tmpTable = cfg.targetTable + "__" + cfg.sourceTable;
    assertTrue(assertingClient.tableExists(tmpTable));
  }