CompletableFuture < Integer > updateBuckets ( DirectSegmentAccess segment , Collection < BucketUpdate > bucketUpdates , long firstIndexedOffset , long lastIndexedOffset , int processedCount , Duration timeout ) { UpdateInstructions update = new UpdateInstructions ( ) ; // Process each Key in the given Map. // Locate the Key's Bucket, then generate necessary Attribute Updates to integrate new Keys into it. for ( BucketUpdate bucketUpdate : bucketUpdates ) { generateAttributeUpdates ( bucketUpdate , update ) ; } if ( lastIndexedOffset > firstIndexedOffset ) { // Atomically update the Table-related attributes in the Segment's metadata, once we apply these changes. generateTableAttributeUpdates ( firstIndexedOffset , lastIndexedOffset , processedCount , update ) ; } if ( update . getAttributes ( ) . isEmpty ( ) ) { // We haven't made any updates. log . debug ( "IndexWriter[{}]: FirstIdxOffset={}, LastIdxOffset={}, No Changes." , segment . getSegmentId ( ) , firstIndexedOffset , lastIndexedOffset ) ; return CompletableFuture . completedFuture ( 0 ) ; } else { log . debug ( "IndexWriter[{}]: FirstIdxOffset={}, LastIdxOffset={}, AttrUpdates={}, Processed={}, Entries+={}, Buckets+={}." , segment . getSegmentId ( ) , firstIndexedOffset , lastIndexedOffset , update . getAttributes ( ) . size ( ) , processedCount , update . getEntryCountDelta ( ) , update . getBucketCountDelta ( ) ) ; return segment . updateAttributes ( update . getAttributes ( ) , timeout ) . thenApply ( v -> update . getAttributes ( ) . size ( ) ) ; } }
Determines what Segment Attribute Updates are necessary to apply the given bucket updates and executes them onto the given Segment .