public Deferred < Long > purgeUIDMeta ( ) { // a list to store all pending deletes so we don't exit before they've // completed final ArrayList < Deferred < Object > > delete_calls = new ArrayList < Deferred < Object > > ( ) ; final Deferred < Long > result = new Deferred < Long > ( ) ; /** * Scanner callback that will recursively call itself and loop through the * rows of the UID table, issuing delete requests for all of the columns in * a row that match a meta qualifier. */ final class MetaScanner implements Callback < Deferred < Long > , ArrayList < ArrayList < KeyValue > > > { final Scanner scanner ; public MetaScanner ( ) { scanner = getScanner ( tsdb . uidTable ( ) ) ; } /** * Fetches the next group of rows from the scanner and sets this class as * a callback * @return The total number of columns deleted after completion */ public Deferred < Long > scan ( ) { return scanner . nextRows ( ) . addCallbackDeferring ( this ) ; } @ Override public Deferred < Long > call ( ArrayList < ArrayList < KeyValue > > rows ) throws Exception { if ( rows == null ) { result . callback ( columns ) ; return null ; } for ( final ArrayList < KeyValue > row : rows ) { // one delete request per row. We'll almost always delete the whole // row, so preallocate some ram. ArrayList < byte [ ] > qualifiers = new ArrayList < byte [ ] > ( row . size ( ) ) ; for ( KeyValue column : row ) { if ( Bytes . equals ( TSMeta . META_QUALIFIER ( ) , column . qualifier ( ) ) ) { qualifiers . add ( column . qualifier ( ) ) ; } else if ( Bytes . equals ( "metric_meta" . getBytes ( CHARSET ) , column . qualifier ( ) ) ) { qualifiers . add ( column . qualifier ( ) ) ; } else if ( Bytes . equals ( "tagk_meta" . getBytes ( CHARSET ) , column . qualifier ( ) ) ) { qualifiers . add ( column . qualifier ( ) ) ; } else if ( Bytes . equals ( "tagv_meta" . getBytes ( CHARSET ) , column . qualifier ( ) ) ) { qualifiers . add ( column . qualifier ( ) ) ; } } if ( qualifiers . size ( ) > 0 ) { columns += qualifiers . size ( ) ; final DeleteRequest delete = new DeleteRequest ( tsdb . uidTable ( ) , row . get ( 0 ) . key ( ) , NAME_FAMILY , qualifiers . toArray ( new byte [ qualifiers . size ( ) ] [ ] ) ) ; delete_calls . add ( tsdb . getClient ( ) . delete ( delete ) ) ; } } /** * Buffer callback used to wait on all of the delete calls for the * last set of rows returned from the scanner so we don't fill up the * deferreds array and OOM out. */ final class ContinueCB implements Callback < Deferred < Long > , ArrayList < Object > > { @ Override public Deferred < Long > call ( ArrayList < Object > deletes ) throws Exception { LOG . debug ( "[" + thread_id + "] Processed [" + deletes . size ( ) + "] delete calls" ) ; delete_calls . clear ( ) ; return scan ( ) ; } } // fetch the next set of rows after waiting for current set of delete // requests to complete Deferred . group ( delete_calls ) . addCallbackDeferring ( new ContinueCB ( ) ) ; return null ; } } // start the scan new MetaScanner ( ) . scan ( ) ; return result ; }
Scans the entire UID table and removes any UIDMeta objects found .