private CompletableFuture < TableWriterFlushResult > flushOnce ( DirectSegmentAccess segment , TimeoutTimer timer ) { // Index all the keys in the segment range pointed to by the aggregator. KeyUpdateCollection keyUpdates = readKeysFromSegment ( segment , this . aggregator . getFirstOffset ( ) , this . aggregator . getLastOffset ( ) , timer ) ; log . debug ( "{}: Flush.ReadFromSegment {} UpdateKeys(s)." , this . traceObjectId , keyUpdates . getUpdates ( ) . size ( ) ) ; // Group keys by their assigned TableBucket (whether existing or not), then fetch all existing keys // for each such bucket and finally (reindex) update the bucket. return this . indexWriter . groupByBucket ( segment , keyUpdates . getUpdates ( ) , timer ) . thenComposeAsync ( builders -> fetchExistingKeys ( builders , segment , timer ) . thenComposeAsync ( v -> { val bucketUpdates = builders . stream ( ) . map ( BucketUpdate . Builder :: build ) . collect ( Collectors . toList ( ) ) ; logBucketUpdates ( bucketUpdates ) ; return this . indexWriter . updateBuckets ( segment , bucketUpdates , this . aggregator . getLastIndexedOffset ( ) , keyUpdates . getLastIndexedOffset ( ) , keyUpdates . getTotalUpdateCount ( ) , timer . getRemaining ( ) ) ; } , this . executor ) , this . executor ) . thenApply ( ignored -> new TableWriterFlushResult ( keyUpdates . getLastIndexedOffset ( ) , keyUpdates . getHighestCopiedOffset ( ) ) ) ; }
Performs a single flush attempt .