public void purge ( ) throws IOException { this . datasetOwner = getOwner ( ) ; State state = new State ( this . state ) ; this . datasetOwnerFs = ProxyUtils . getOwnerFs ( state , this . datasetOwner ) ; try ( HiveProxyQueryExecutor queryExecutor = ProxyUtils . getQueryExecutor ( state , this . datasetOwner ) ) { if ( this . simulate ) { log . info ( "Simulate is set to true. Wont't run actual queries" ) ; return ; } String originalPartitionLocation = getOriginalPartitionLocation ( ) ; // Create the staging table and staging partition queryExecutor . executeQueries ( HivePurgerQueryTemplate . getCreateStagingTableQuery ( this ) , this . datasetOwner ) ; this . startTime = getLastModifiedTime ( originalPartitionLocation ) ; // Execute purge queries, that is insert filtered data into the staging partition queryExecutor . executeQueries ( this . purgeQueries , this . datasetOwner ) ; this . endTime = getLastModifiedTime ( originalPartitionLocation ) ; // Create a backup table and partition pointing to the original partition location queryExecutor . executeQueries ( HivePurgerQueryTemplate . getBackupQueries ( this ) , this . datasetOwner ) ; String commitPolicyString = this . state . getProp ( ComplianceConfigurationKeys . PURGER_COMMIT_POLICY_CLASS , ComplianceConfigurationKeys . DEFAULT_PURGER_COMMIT_POLICY_CLASS ) ; CommitPolicy < PurgeableHivePartitionDataset > commitPolicy = GobblinConstructorUtils . invokeConstructor ( CommitPolicy . class , commitPolicyString ) ; if ( ! commitPolicy . shouldCommit ( this ) ) { log . error ( "Last modified time before start of execution : " + this . startTime ) ; log . error ( "Last modified time after execution of purge queries : " + this . endTime ) ; throw new RuntimeException ( "Failed to commit. File modified during job run." ) ; } // Alter the original table partition to start pointing to the cleaned-partition-location/staging-partition-location queryExecutor . executeQueries ( HivePurgerQueryTemplate . getAlterOriginalPartitionLocationQueries ( this ) , this . datasetOwner ) ; // Drop the staging table queryExecutor . executeQueries ( HivePurgerQueryTemplate . getDropStagingTableQuery ( this ) , this . datasetOwner ) ; } catch ( SQLException e ) { throw new IOException ( e ) ; } }
This method is responsible for actual purging . - It first creates a staging table partition with the same schema as of original table partition . - Staging table partition is then populated by original table left outer joined with compliance id table .