signature:
public class Image { /** * Poll for new messages in a stream . If new messages are found beyond the last consumed position then they
 * will be delivered to the { @ link FragmentHandler } up to a limited number of fragments as specified .
 * Use a { @ link FragmentAssembler } to assemble messages which span multiple fragments .
 * @ param fragmentHandler to which message fragments are delivered .
 * @ param fragmentLimit for the number of fragments to be consumed during one polling operation .
 * @ return the number of fragments that have been consumed .
 * @ see FragmentAssembler
 * @ see ImageFragmentAssembler */
public int poll ( final FragmentHandler fragmentHandler , final int fragmentLimit ) { } }

implementation: if ( isClosed ) { return 0 ; } final long position = subscriberPosition . get ( ) ; return TermReader . read ( activeTermBuffer ( position ) , ( int ) position & termLengthMask , fragmentHandler , fragmentLimit , header , errorHandler , position , subscriberPosition ) ;


signature:
public class Image { /** * Poll for new messages in a stream . If new messages are found beyond the last consumed position then they
 * will be delivered to the { @ link ControlledFragmentHandler } up to a limited number of fragments as specified .
 * Use a { @ link ControlledFragmentAssembler } to assemble messages which span multiple fragments .
 * @ param handler to which message fragments are delivered .
 * @ param fragmentLimit for the number of fragments to be consumed during one polling operation .
 * @ return the number of fragments that have been consumed .
 * @ see ControlledFragmentAssembler
 * @ see ImageControlledFragmentAssembler */
public int controlledPoll ( final ControlledFragmentHandler handler , final int fragmentLimit ) { } }

implementation: if ( isClosed ) { return 0 ; } int fragmentsRead = 0 ; long initialPosition = subscriberPosition . get ( ) ; int initialOffset = ( int ) initialPosition & termLengthMask ; int resultingOffset = initialOffset ; final UnsafeBuffer termBuffer = activeTermBuffer ( initialPosition ) ; final int capacity = termBuffer . capacity ( ) ; final Header header = this . header ; header . buffer ( termBuffer ) ; try { while ( fragmentsRead < fragmentLimit && resultingOffset < capacity ) { final int length = frameLengthVolatile ( termBuffer , resultingOffset ) ; if ( length <= 0 ) { break ; } final int frameOffset = resultingOffset ; final int alignedLength = BitUtil . align ( length , FRAME_ALIGNMENT ) ; resultingOffset += alignedLength ; if ( isPaddingFrame ( termBuffer , frameOffset ) ) { continue ; } header . offset ( frameOffset ) ; final Action action = handler . onFragment ( termBuffer , frameOffset + HEADER_LENGTH , length - HEADER_LENGTH , header ) ; if ( action == ABORT ) { resultingOffset -= alignedLength ; break ; } ++ fragmentsRead ; if ( action == BREAK ) { break ; } else if ( action == COMMIT ) { initialPosition += ( resultingOffset - initialOffset ) ; initialOffset = resultingOffset ; subscriberPosition . setOrdered ( initialPosition ) ; } } } catch ( final Throwable t ) { errorHandler . onError ( t ) ; } finally { final long resultingPosition = initialPosition + ( resultingOffset - initialOffset ) ; if ( resultingPosition > initialPosition ) { subscriberPosition . setOrdered ( resultingPosition ) ; } } return fragmentsRead ;


signature:
public class Image { /** * Peek for new messages in a stream by scanning forward from an initial position . If new messages are found then
 * they will be delivered to the { @ link ControlledFragmentHandler } up to a limited position .
 * Use a { @ link ControlledFragmentAssembler } to assemble messages which span multiple fragments . Scans must also
 * start at the beginning of a message so that the assembler is reset .
 * @ param initialPosition from which to peek forward .
 * @ param handler to which message fragments are delivered .
 * @ param limitPosition up to which can be scanned .
 * @ return the resulting position after the scan terminates which is a complete message .
 * @ see ControlledFragmentAssembler
 * @ see ImageControlledFragmentAssembler */
public long controlledPeek ( final long initialPosition , final ControlledFragmentHandler handler , final long limitPosition ) { } }

implementation: if ( isClosed ) { return 0 ; } validatePosition ( initialPosition ) ; int initialOffset = ( int ) initialPosition & termLengthMask ; int offset = initialOffset ; long position = initialPosition ; final UnsafeBuffer termBuffer = activeTermBuffer ( initialPosition ) ; final int capacity = termBuffer . capacity ( ) ; final Header header = this . header ; header . buffer ( termBuffer ) ; long resultingPosition = initialPosition ; try { while ( position < limitPosition && offset < capacity ) { final int length = frameLengthVolatile ( termBuffer , offset ) ; if ( length <= 0 ) { break ; } final int frameOffset = offset ; final int alignedLength = BitUtil . align ( length , FRAME_ALIGNMENT ) ; offset += alignedLength ; if ( isPaddingFrame ( termBuffer , frameOffset ) ) { position += ( offset - initialOffset ) ; initialOffset = offset ; resultingPosition = position ; continue ; } header . offset ( frameOffset ) ; final Action action = handler . onFragment ( termBuffer , frameOffset + HEADER_LENGTH , length - HEADER_LENGTH , header ) ; if ( action == ABORT ) { break ; } position += ( offset - initialOffset ) ; initialOffset = offset ; if ( ( header . flags ( ) & END_FRAG_FLAG ) == END_FRAG_FLAG ) { resultingPosition = position ; } if ( action == BREAK ) { break ; } } } catch ( final Throwable t ) { errorHandler . onError ( t ) ; } return resultingPosition ;


signature:
public class Image { /** * Poll for new messages in a stream . If new messages are found beyond the last consumed position then they
 * will be delivered to the { @ link BlockHandler } up to a limited number of bytes .
 * A scan will terminate if a padding frame is encountered . If first frame in a scan is padding then a block
 * for the padding is notified . If the padding comes after the first frame in a scan then the scan terminates
 * at the offset the padding frame begins . Padding frames are delivered singularly in a block .
 * Padding frames may be for a greater range than the limit offset but only the header needs to be valid so
 * relevant length of the frame is { @ link io . aeron . protocol . DataHeaderFlyweight # HEADER _ LENGTH } .
 * @ param handler to which block is delivered .
 * @ param blockLengthLimit up to which a block may be in length .
 * @ return the number of bytes that have been consumed . */
public int blockPoll ( final BlockHandler handler , final int blockLengthLimit ) { } }

implementation: if ( isClosed ) { return 0 ; } final long position = subscriberPosition . get ( ) ; final int termOffset = ( int ) position & termLengthMask ; final UnsafeBuffer termBuffer = activeTermBuffer ( position ) ; final int limitOffset = Math . min ( termOffset + blockLengthLimit , termBuffer . capacity ( ) ) ; final int resultingOffset = TermBlockScanner . scan ( termBuffer , termOffset , limitOffset ) ; final int length = resultingOffset - termOffset ; if ( resultingOffset > termOffset ) { try { final int termId = termBuffer . getInt ( termOffset + TERM_ID_FIELD_OFFSET , LITTLE_ENDIAN ) ; handler . onBlock ( termBuffer , termOffset , length , sessionId , termId ) ; } catch ( final Throwable t ) { errorHandler . onError ( t ) ; } finally { subscriberPosition . setOrdered ( position + length ) ; } } return length ;


signature:
public class Image { /** * Poll for new messages in a stream . If new messages are found beyond the last consumed position then they
 * will be delivered to the { @ link RawBlockHandler } up to a limited number of bytes .
 * This method is useful for operations like bulk archiving a stream to file .
 * A scan will terminate if a padding frame is encountered . If first frame in a scan is padding then a block
 * for the padding is notified . If the padding comes after the first frame in a scan then the scan terminates
 * at the offset the padding frame begins . Padding frames are delivered singularly in a block .
 * Padding frames may be for a greater range than the limit offset but only the header needs to be valid so
 * relevant length of the frame is { @ link io . aeron . protocol . DataHeaderFlyweight # HEADER _ LENGTH } .
 * @ param handler to which block is delivered .
 * @ param blockLengthLimit up to which a block may be in length .
 * @ return the number of bytes that have been consumed . */
public int rawPoll ( final RawBlockHandler handler , final int blockLengthLimit ) { } }

implementation: if ( isClosed ) { return 0 ; } final long position = subscriberPosition . get ( ) ; final int termOffset = ( int ) position & termLengthMask ; final int activeIndex = indexByPosition ( position , positionBitsToShift ) ; final UnsafeBuffer termBuffer = termBuffers [ activeIndex ] ; final int capacity = termBuffer . capacity ( ) ; final int limitOffset = Math . min ( termOffset + blockLengthLimit , capacity ) ; final int resultingOffset = TermBlockScanner . scan ( termBuffer , termOffset , limitOffset ) ; final int length = resultingOffset - termOffset ; if ( resultingOffset > termOffset ) { try { final long fileOffset = ( ( long ) capacity * activeIndex ) + termOffset ; final int termId = termBuffer . getInt ( termOffset + TERM_ID_FIELD_OFFSET , LITTLE_ENDIAN ) ; handler . onBlock ( logBuffers . fileChannel ( ) , fileOffset , termBuffer , termOffset , length , sessionId , termId ) ; } catch ( final Throwable t ) { errorHandler . onError ( t ) ; } finally { subscriberPosition . setOrdered ( position + length ) ; } } return length ;


signature:
public class Header { /** * Get the current position to which the image has advanced on reading this message .
 * @ return the current position to which the image has advanced on reading this message . */
public final long position ( ) { } }

implementation: final int resultingOffset = BitUtil . align ( termOffset ( ) + frameLength ( ) , FRAME_ALIGNMENT ) ; return computePosition ( termId ( ) , resultingOffset , positionBitsToShift , initialTermId ) ;


signature:
public class SubscriptionMessageFlyweight { /** * Set channel field in ASCII
 * @ param channel field value
 * @ return flyweight */
public SubscriptionMessageFlyweight channel ( final String channel ) { } }

implementation: lengthOfChannel = buffer . putStringAscii ( offset + CHANNEL_OFFSET , channel ) ; return this ;


signature:
public class FileStoreLogFactory { /** * Create new { @ link RawLog } in the publications directory for the supplied triplet .
 * @ param channel address on the media to send to .
 * @ param sessionId under which transmissions are made .
 * @ param streamId within the channel address to separate message flows .
 * @ param correlationId to use to distinguish this publication
 * @ param termBufferLength length of each term
 * @ param useSparseFiles for the log buffer .
 * @ return the newly allocated { @ link RawLog } */
public RawLog newPublication ( final String channel , final int sessionId , final int streamId , final long correlationId , final int termBufferLength , final boolean useSparseFiles ) { } }

implementation: return newInstance ( publicationsDir , channel , sessionId , streamId , correlationId , termBufferLength , useSparseFiles ) ;


signature:
public class FileStoreLogFactory { /** * Create new { @ link RawLog } in the rebuilt publication images directory for the supplied triplet .
 * @ param channel address on the media to listened to .
 * @ param sessionId under which transmissions are made .
 * @ param streamId within the channel address to separate message flows .
 * @ param correlationId to use to distinguish this connection
 * @ param termBufferLength to use for the log buffer
 * @ param useSparseFiles for the log buffer .
 * @ return the newly allocated { @ link RawLog } */
public RawLog newImage ( final String channel , final int sessionId , final int streamId , final long correlationId , final int termBufferLength , final boolean useSparseFiles ) { } }

implementation: return newInstance ( imagesDir , channel , sessionId , streamId , correlationId , termBufferLength , useSparseFiles ) ;


signature:
public class ServiceHeartbeat { /** * Allocate a counter to represent the heartbeat of a clustered service .
 * @ param aeron to allocate the counter .
 * @ param tempBuffer to use for building the key and label without allocation .
 * @ param serviceId of the service heartbeat .
 * @ return the { @ link Counter } for the commit position . */
public static Counter allocate ( final Aeron aeron , final MutableDirectBuffer tempBuffer , final int serviceId ) { } }

implementation: tempBuffer . putInt ( SERVICE_ID_OFFSET , serviceId ) ; final int labelOffset = BitUtil . align ( KEY_LENGTH , SIZE_OF_INT ) ; int labelLength = 0 ; labelLength += tempBuffer . putStringWithoutLengthAscii ( labelOffset + labelLength , NAME ) ; labelLength += tempBuffer . putIntAscii ( labelOffset + labelLength , serviceId ) ; return aeron . addCounter ( SERVICE_HEARTBEAT_TYPE_ID , tempBuffer , 0 , KEY_LENGTH , tempBuffer , labelOffset , labelLength ) ;


signature:
public class ServiceHeartbeat { /** * Find the active counter id for heartbeat of a given service id .
 * @ param counters to search within .
 * @ param serviceId to search for .
 * @ return the counter id if found otherwise { @ link CountersReader # NULL _ COUNTER _ ID } . */
public static int findCounterId ( final CountersReader counters , final int serviceId ) { } }

implementation: final DirectBuffer buffer = counters . metaDataBuffer ( ) ; for ( int i = 0 , size = counters . maxCounterId ( ) ; i < size ; i ++ ) { if ( counters . getCounterState ( i ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( i ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == SERVICE_HEARTBEAT_TYPE_ID && buffer . getInt ( recordOffset + KEY_OFFSET + SERVICE_ID_OFFSET ) == serviceId ) { return i ; } } } return NULL_COUNTER_ID ;


signature:
public class BacklogStat { /** * Take a snapshot of all the backlog information and group by stream .
 * @ return a snapshot of all the backlog information and group by stream . */
public Map < StreamCompositeKey , StreamBacklog > snapshot ( ) { } }

implementation: final Map < StreamCompositeKey , StreamBacklog > streams = new HashMap < > ( ) ; counters . forEach ( ( counterId , typeId , keyBuffer , label ) -> { if ( ( typeId >= PUBLISHER_LIMIT_TYPE_ID && typeId <= RECEIVER_POS_TYPE_ID ) || typeId == SENDER_LIMIT_TYPE_ID || typeId == PER_IMAGE_TYPE_ID || typeId == PUBLISHER_POS_TYPE_ID ) { final StreamCompositeKey key = new StreamCompositeKey ( keyBuffer . getInt ( SESSION_ID_OFFSET ) , keyBuffer . getInt ( STREAM_ID_OFFSET ) , keyBuffer . getStringAscii ( CHANNEL_OFFSET ) ) ; final StreamBacklog streamBacklog = streams . computeIfAbsent ( key , ( ignore ) -> new StreamBacklog ( ) ) ; final long registrationId = keyBuffer . getLong ( REGISTRATION_ID_OFFSET ) ; final long value = counters . getCounterValue ( counterId ) ; switch ( typeId ) { case PublisherLimit . PUBLISHER_LIMIT_TYPE_ID : streamBacklog . createPublisherIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createPublisherIfAbsent ( ) . limit ( value ) ; break ; case PublisherPos . PUBLISHER_POS_TYPE_ID : streamBacklog . createPublisherIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createPublisherIfAbsent ( ) . position ( value ) ; break ; case SenderPos . SENDER_POSITION_TYPE_ID : streamBacklog . createSenderIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createSenderIfAbsent ( ) . position ( value ) ; break ; case SenderLimit . SENDER_LIMIT_TYPE_ID : streamBacklog . createSenderIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createSenderIfAbsent ( ) . limit ( value ) ; break ; case ReceiverHwm . RECEIVER_HWM_TYPE_ID : streamBacklog . createReceiverIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createReceiverIfAbsent ( ) . highWaterMark ( value ) ; break ; case ReceiverPos . RECEIVER_POS_TYPE_ID : streamBacklog . createReceiverIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createReceiverIfAbsent ( ) . position ( value ) ; break ; case SubscriberPos . SUBSCRIBER_POSITION_TYPE_ID : streamBacklog . subscriberBacklogs ( ) . put ( registrationId , new Subscriber ( value ) ) ; break ; } } } ) ; return streams ;


signature:
public class BacklogStat { /** * Print a snapshot of the stream backlog with some explanation to a { @ link PrintStream } .
 * Each stream will be printed in its own section .
 * @ param out to which the stream backlog will be written . */
public void print ( final PrintStream out ) { } }

implementation: final StringBuilder builder = new StringBuilder ( ) ; for ( final Map . Entry < StreamCompositeKey , StreamBacklog > entry : snapshot ( ) . entrySet ( ) ) { builder . setLength ( 0 ) ; final StreamCompositeKey key = entry . getKey ( ) ; builder . append ( "sessionId=" ) . append ( key . sessionId ( ) ) . append ( " streamId=" ) . append ( key . streamId ( ) ) . append ( " channel=" ) . append ( key . channel ( ) ) . append ( " : " ) ; final StreamBacklog streamBacklog = entry . getValue ( ) ; if ( streamBacklog . publisher ( ) != null ) { builder . append ( "\n┌─for publisher " ) . append ( streamBacklog . publisher ( ) . registrationId ( ) ) . append ( " the last sampled position is " ) . append ( streamBacklog . publisher ( ) . position ( ) ) . append ( " (~" ) . append ( streamBacklog . publisher ( ) . remainingWindow ( ) ) . append ( " bytes before back-pressure)" ) ; final Sender sender = streamBacklog . sender ( ) ; if ( sender != null ) { final long senderBacklog = sender . backlog ( streamBacklog . publisher ( ) . position ( ) ) ; builder . append ( "\n└─sender " ) . append ( sender . registrationId ( ) ) ; if ( senderBacklog >= 0 ) { builder . append ( " has to send " ) . append ( senderBacklog ) . append ( " bytes" ) ; } else { builder . append ( " is at position " ) . append ( sender . position ( ) ) ; } builder . append ( " (" ) . append ( sender . window ( ) ) . append ( " bytes remaining in the sender window)" ) ; } else { builder . append ( "\n└─no sender yet..." ) ; } } if ( streamBacklog . receiver ( ) != null ) { builder . append ( "\n┌─receiver " ) . append ( streamBacklog . receiver ( ) . registrationId ( ) ) . append ( " is at position " ) . append ( streamBacklog . receiver ( ) . position ( ) ) ; final Iterator < Map . Entry < Long , Subscriber > > subscriberIterator = streamBacklog . subscriberBacklogs ( ) . entrySet ( ) . iterator ( ) ; while ( subscriberIterator . hasNext ( ) ) { final Map . Entry < Long , Subscriber > subscriber = subscriberIterator . next ( ) ; builder . append ( subscriberIterator . hasNext ( ) ? "\n├" : "\n└" ) . append ( "─subscriber " ) . append ( subscriber . getKey ( ) ) . append ( " has " ) . append ( subscriber . getValue ( ) . backlog ( streamBacklog . receiver ( ) . highWaterMark ( ) ) ) . append ( " backlog bytes" ) ; } } builder . append ( '\n' ) ; out . println ( builder ) ; }


signature:
public class LossDetector { /** * Scan for gaps and handle received data .
 * The handler keeps track from scan to scan what is a gap and what must have been repaired .
 * @ param termBuffer to scan
 * @ param rebuildPosition to start scanning from
 * @ param hwmPosition to scan up to
 * @ param nowNs time in nanoseconds
 * @ param termLengthMask used for offset calculation
 * @ param positionBitsToShift used for position calculation
 * @ param initialTermId used by the scanner
 * @ return packed outcome of the scan . */
public long scan ( final UnsafeBuffer termBuffer , final long rebuildPosition , final long hwmPosition , final long nowNs , final int termLengthMask , final int positionBitsToShift , final int initialTermId ) { } }

implementation: boolean lossFound = false ; int rebuildOffset = ( int ) rebuildPosition & termLengthMask ; if ( rebuildPosition < hwmPosition ) { final int rebuildTermCount = ( int ) ( rebuildPosition >>> positionBitsToShift ) ; final int hwmTermCount = ( int ) ( hwmPosition >>> positionBitsToShift ) ; final int rebuildTermId = initialTermId + rebuildTermCount ; final int hwmTermOffset = ( int ) hwmPosition & termLengthMask ; final int limitOffset = rebuildTermCount == hwmTermCount ? hwmTermOffset : termLengthMask + 1 ; rebuildOffset = scanForGap ( termBuffer , rebuildTermId , rebuildOffset , limitOffset , this ) ; if ( rebuildOffset < limitOffset ) { if ( scannedTermOffset != activeTermOffset || scannedTermId != activeTermId ) { activateGap ( nowNs ) ; lossFound = true ; } checkTimerExpiry ( nowNs ) ; } } return pack ( rebuildOffset , lossFound ) ;


signature:
public class ConcurrentPublication { /** * Non - blocking publish of a partial buffer containing a message .
 * @ param buffer containing message .
 * @ param offset offset in the buffer at which the encoded message begins .
 * @ param length in bytes of the encoded message .
 * @ param reservedValueSupplier { @ link ReservedValueSupplier } for the frame .
 * @ return The new stream position , otherwise a negative error value of { @ link # NOT _ CONNECTED } ,
 * { @ link # BACK _ PRESSURED } , { @ link # ADMIN _ ACTION } , { @ link # CLOSED } , or { @ link # MAX _ POSITION _ EXCEEDED } . */
public long offer ( final DirectBuffer buffer , final int offset , final int length , final ReservedValueSupplier reservedValueSupplier ) { } }

implementation: long newPosition = CLOSED ; if ( ! isClosed ) { final long limit = positionLimit . getVolatile ( ) ; final int termCount = activeTermCount ( logMetaDataBuffer ) ; final TermAppender termAppender = termAppenders [ indexByTermCount ( termCount ) ] ; final long rawTail = termAppender . rawTailVolatile ( ) ; final long termOffset = rawTail & 0xFFFF_FFFFL ; final int termId = termId ( rawTail ) ; final long position = computeTermBeginPosition ( termId , positionBitsToShift , initialTermId ) + termOffset ; if ( termCount != ( termId - initialTermId ) ) { return ADMIN_ACTION ; } if ( position < limit ) { final int resultingOffset ; if ( length <= maxPayloadLength ) { checkPositiveLength ( length ) ; resultingOffset = termAppender . appendUnfragmentedMessage ( headerWriter , buffer , offset , length , reservedValueSupplier , termId ) ; } else { checkMaxMessageLength ( length ) ; resultingOffset = termAppender . appendFragmentedMessage ( headerWriter , buffer , offset , length , maxPayloadLength , reservedValueSupplier , termId ) ; } newPosition = newPosition ( termCount , ( int ) termOffset , termId , position , resultingOffset ) ; } else { newPosition = backPressureStatus ( position , length ) ; } } return newPosition ;


signature:
public class ConcurrentPublication { /** * Non - blocking publish of a message composed of two parts , e . g . a header and encapsulated payload .
 * @ param bufferOne containing the first part of the message .
 * @ param offsetOne at which the first part of the message begins .
 * @ param lengthOne of the first part of the message .
 * @ param bufferTwo containing the second part of the message .
 * @ param offsetTwo at which the second part of the message begins .
 * @ param lengthTwo of the second part of the message .
 * @ param reservedValueSupplier { @ link ReservedValueSupplier } for the frame .
 * @ return The new stream position , otherwise a negative error value of { @ link # NOT _ CONNECTED } ,
 * { @ link # BACK _ PRESSURED } , { @ link # ADMIN _ ACTION } , { @ link # CLOSED } , or { @ link # MAX _ POSITION _ EXCEEDED } . */
public long offer ( final DirectBuffer bufferOne , final int offsetOne , final int lengthOne , final DirectBuffer bufferTwo , final int offsetTwo , final int lengthTwo , final ReservedValueSupplier reservedValueSupplier ) { } }

implementation: long newPosition = CLOSED ; if ( ! isClosed ) { final long limit = positionLimit . getVolatile ( ) ; final int termCount = activeTermCount ( logMetaDataBuffer ) ; final TermAppender termAppender = termAppenders [ indexByTermCount ( termCount ) ] ; final long rawTail = termAppender . rawTailVolatile ( ) ; final long termOffset = rawTail & 0xFFFF_FFFFL ; final int termId = termId ( rawTail ) ; final long position = computeTermBeginPosition ( termId , positionBitsToShift , initialTermId ) + termOffset ; if ( termCount != ( termId - initialTermId ) ) { return ADMIN_ACTION ; } final int length = validateAndComputeLength ( lengthOne , lengthTwo ) ; if ( position < limit ) { final int resultingOffset ; if ( length <= maxPayloadLength ) { resultingOffset = termAppender . appendUnfragmentedMessage ( headerWriter , bufferOne , offsetOne , lengthOne , bufferTwo , offsetTwo , lengthTwo , reservedValueSupplier , termId ) ; } else { checkMaxMessageLength ( length ) ; resultingOffset = termAppender . appendFragmentedMessage ( headerWriter , bufferOne , offsetOne , lengthOne , bufferTwo , offsetTwo , lengthTwo , maxPayloadLength , reservedValueSupplier , termId ) ; } newPosition = newPosition ( termCount , ( int ) termOffset , termId , position , resultingOffset ) ; } else { newPosition = backPressureStatus ( position , length ) ; } } return newPosition ;


signature:
public class ConcurrentPublication { /** * Try to claim a range in the publication log into which a message can be written with zero copy semantics .
 * Once the message has been written then { @ link BufferClaim # commit ( ) } should be called thus making it available .
 * < b > Note : < / b > This method can only be used for message lengths less than MTU length minus header .
 * If the claim is held for more than the aeron . publication . unblock . timeout system property then the driver will
 * assume the publication thread is dead and will unblock the claim thus allowing other threads to make progress or
 * to reach end - of - stream ( EOS ) .
 * < pre > { @ code
 * final BufferClaim bufferClaim = new BufferClaim ( ) ; / / Can be stored and reused to avoid allocation
 * if ( publication . tryClaim ( messageLength , bufferClaim ) > 0L )
 * try
 * final MutableDirectBuffer buffer = bufferClaim . buffer ( ) ;
 * final int offset = bufferClaim . offset ( ) ;
 * / / Work with buffer directly or wrap with a flyweight
 * finally
 * bufferClaim . commit ( ) ;
 * } < / pre >
 * @ param length of the range to claim , in bytes . .
 * @ param bufferClaim to be populated if the claim succeeds .
 * @ return The new stream position , otherwise a negative error value of { @ link # NOT _ CONNECTED } ,
 * { @ link # BACK _ PRESSURED } , { @ link # ADMIN _ ACTION } , { @ link # CLOSED } , or { @ link # MAX _ POSITION _ EXCEEDED } .
 * @ throws IllegalArgumentException if the length is greater than { @ link # maxPayloadLength ( ) } within an MTU .
 * @ see BufferClaim # commit ( )
 * @ see BufferClaim # abort ( ) */
public long tryClaim ( final int length , final BufferClaim bufferClaim ) { } }

implementation: checkPayloadLength ( length ) ; long newPosition = CLOSED ; if ( ! isClosed ) { final long limit = positionLimit . getVolatile ( ) ; final int termCount = activeTermCount ( logMetaDataBuffer ) ; final TermAppender termAppender = termAppenders [ indexByTermCount ( termCount ) ] ; final long rawTail = termAppender . rawTailVolatile ( ) ; final long termOffset = rawTail & 0xFFFF_FFFFL ; final int termId = termId ( rawTail ) ; final long position = computeTermBeginPosition ( termId , positionBitsToShift , initialTermId ) + termOffset ; if ( termCount != ( termId - initialTermId ) ) { return ADMIN_ACTION ; } if ( position < limit ) { final int resultingOffset = termAppender . claim ( headerWriter , length , bufferClaim , termId ) ; newPosition = newPosition ( termCount , ( int ) termOffset , termId , position , resultingOffset ) ; } else { newPosition = backPressureStatus ( position , length ) ; } } return newPosition ;


signature:
public class TermBlockScanner { /** * Scan a term buffer for a block of message fragments from an offset up to a limitOffset .
 * A scan will terminate if a padding frame is encountered . If first frame in a scan is padding then a block
 * for the padding is notified . If the padding comes after the first frame in a scan then the scan terminates
 * at the offset the padding frame begins . Padding frames are delivered singularly in a block .
 * Padding frames may be for a greater range than the limit offset but only the header needs to be valid so
 * relevant length of the frame is { @ link io . aeron . protocol . DataHeaderFlyweight # HEADER _ LENGTH } .
 * @ param termBuffer to scan for message fragments .
 * @ param termOffset at which the scan should begin .
 * @ param limitOffset at which the scan should stop .
 * @ return the offset at which the scan terminated . */
public static int scan ( final UnsafeBuffer termBuffer , final int termOffset , final int limitOffset ) { } }

implementation: int offset = termOffset ; while ( offset < limitOffset ) { final int frameLength = frameLengthVolatile ( termBuffer , offset ) ; if ( frameLength <= 0 ) { break ; } final int alignedFrameLength = align ( frameLength , FRAME_ALIGNMENT ) ; if ( isPaddingFrame ( termBuffer , offset ) ) { if ( termOffset == offset ) { offset += alignedFrameLength ; } break ; } if ( offset + alignedFrameLength > limitOffset ) { break ; } offset += alignedFrameLength ; } return offset ;


signature:
public class ConsensusModule { /** * Launch an { @ link ConsensusModule } with that communicates with an out of process { @ link io . aeron . archive . Archive }
 * and { @ link io . aeron . driver . MediaDriver } then awaits shutdown signal .
 * @ param args command line argument which is a list for properties files as URLs or filenames . */
public static void main ( final String [ ] args ) { } }

implementation: loadPropertiesFiles ( args ) ; try ( ConsensusModule consensusModule = launch ( ) ) { consensusModule . context ( ) . shutdownSignalBarrier ( ) . await ( ) ; System . out . println ( "Shutdown ConsensusModule..." ) ; }


signature:
public class HeartbeatStatus { /** * Allocate a counter for tracking the last heartbeat of an entity .
 * @ param tempBuffer to be used for labels and key .
 * @ param name of the counter for the label .
 * @ param typeId of the counter for classification .
 * @ param countersManager from which to allocated the underlying storage .
 * @ param registrationId to be associated with the counter .
 * @ return a new { @ link AtomicCounter } for tracking the last heartbeat . */
public static AtomicCounter allocate ( final MutableDirectBuffer tempBuffer , final String name , final int typeId , final CountersManager countersManager , final long registrationId ) { } }

implementation: return new AtomicCounter ( countersManager . valuesBuffer ( ) , allocateCounterId ( tempBuffer , name , typeId , countersManager , registrationId ) , countersManager ) ;


signature:
public class Aeron { /** * Create an Aeron instance and connect to the media driver .
 * Threads required for interacting with the media driver are created and managed within the Aeron instance .
 * If an exception occurs while trying to establish a connection then the { @ link Context # close ( ) } method
 * will be called on the passed context .
 * @ param ctx for configuration of the client .
 * @ return the new { @ link Aeron } instance connected to the Media Driver . */
public static Aeron connect ( final Context ctx ) { } }

implementation: try { final Aeron aeron = new Aeron ( ctx ) ; if ( ctx . useConductorAgentInvoker ( ) ) { aeron . conductorInvoker . start ( ) ; } else { AgentRunner . startOnThread ( aeron . conductorRunner , ctx . threadFactory ( ) ) ; } return aeron ; } catch ( final Exception ex ) { ctx . close ( ) ; throw ex ; }


signature:
public class Aeron { /** * Print out the values from { @ link # countersReader ( ) } which can be useful for debugging .
 * @ param out to where the counters get printed . */
public void printCounters ( final PrintStream out ) { } }

implementation: final CountersReader counters = countersReader ( ) ; counters . forEach ( ( value , id , label ) -> out . format ( "%3d: %,20d - %s%n" , id , value , label ) ) ;


signature:
public class Aeron { /** * Add a new { @ link Subscription } for subscribing to messages from publishers .
 * This method will override the default handlers from the { @ link Aeron . Context } , i . e .
 * { @ link Aeron . Context # availableImageHandler ( AvailableImageHandler ) } and
 * { @ link Aeron . Context # unavailableImageHandler ( UnavailableImageHandler ) } . Null values are valid and will
 * result in no action being taken .
 * @ param channel for receiving the messages known to the media layer .
 * @ param streamId within the channel scope .
 * @ param availableImageHandler called when { @ link Image } s become available for consumption . Null is valid if no
 * action is to be taken .
 * @ param unavailableImageHandler called when { @ link Image } s go unavailable for consumption . Null is valid if no
 * action is to be taken .
 * @ return the { @ link Subscription } for the channel and streamId pair . */
public Subscription addSubscription ( final String channel , final int streamId , final AvailableImageHandler availableImageHandler , final UnavailableImageHandler unavailableImageHandler ) { } }

implementation: return conductor . addSubscription ( channel , streamId , availableImageHandler , unavailableImageHandler ) ;


signature:
public class Aeron { /** * Allocate a counter on the media driver and return a { @ link Counter } for it .
 * The counter should be freed by calling { @ link Counter # close ( ) } .
 * @ param typeId for the counter .
 * @ param keyBuffer containing the optional key for the counter .
 * @ param keyOffset within the keyBuffer at which the key begins .
 * @ param keyLength of the key in the keyBuffer .
 * @ param labelBuffer containing the mandatory label for the counter . The label should not be length prefixed .
 * @ param labelOffset within the labelBuffer at which the label begins .
 * @ param labelLength of the label in the labelBuffer .
 * @ return the newly allocated counter .
 * @ see org . agrona . concurrent . status . CountersManager # allocate ( int , DirectBuffer , int , int , DirectBuffer , int , int ) */
public Counter addCounter ( final int typeId , final DirectBuffer keyBuffer , final int keyOffset , final int keyLength , final DirectBuffer labelBuffer , final int labelOffset , final int labelLength ) { } }

implementation: return conductor . addCounter ( typeId , keyBuffer , keyOffset , keyLength , labelBuffer , labelOffset , labelLength ) ;


signature:
public class Context { /** * This is called automatically by { @ link Aeron # connect ( Aeron . Context ) } and its overloads .
 * There is no need to call it from a client application . It is responsible for providing default
 * values for options that are not individually changed through field setters .
 * @ return this for a fluent API . */
public Context conclude ( ) { } }

implementation: super . conclude ( ) ; if ( null == clientLock ) { clientLock = new ReentrantLock ( ) ; } if ( null == epochClock ) { epochClock = new SystemEpochClock ( ) ; } if ( null == nanoClock ) { nanoClock = new SystemNanoClock ( ) ; } if ( null == idleStrategy ) { idleStrategy = new SleepingMillisIdleStrategy ( Configuration . IDLE_SLEEP_MS ) ; } if ( cncFile ( ) != null ) { connectToDriver ( ) ; } interServiceTimeoutNs = CncFileDescriptor . clientLivenessTimeout ( cncMetaDataBuffer ) ; if ( interServiceTimeoutNs <= keepAliveIntervalNs ) { throw new ConfigurationException ( "interServiceTimeoutNs=" + interServiceTimeoutNs + " <= keepAliveIntervalNs=" + keepAliveIntervalNs ) ; } if ( null == toDriverBuffer ) { toDriverBuffer = new ManyToOneRingBuffer ( CncFileDescriptor . createToDriverBuffer ( cncByteBuffer , cncMetaDataBuffer ) ) ; } if ( null == toClientBuffer ) { toClientBuffer = new CopyBroadcastReceiver ( new BroadcastReceiver ( CncFileDescriptor . createToClientsBuffer ( cncByteBuffer , cncMetaDataBuffer ) ) ) ; } if ( countersMetaDataBuffer ( ) == null ) { countersMetaDataBuffer ( CncFileDescriptor . createCountersMetaDataBuffer ( cncByteBuffer , cncMetaDataBuffer ) ) ; } if ( countersValuesBuffer ( ) == null ) { countersValuesBuffer ( CncFileDescriptor . createCountersValuesBuffer ( cncByteBuffer , cncMetaDataBuffer ) ) ; } if ( null == logBuffersFactory ) { logBuffersFactory = new MappedLogBuffersFactory ( ) ; } if ( null == errorHandler ) { errorHandler = Configuration . DEFAULT_ERROR_HANDLER ; } if ( null == driverProxy ) { clientId = toDriverBuffer . nextCorrelationId ( ) ; driverProxy = new DriverProxy ( toDriverBuffer , clientId ) ; } return this ;


signature:
public class Context { /** * Clean up all resources that the client uses to communicate with the Media Driver . */
public void close ( ) { } }

implementation: final MappedByteBuffer cncByteBuffer = this . cncByteBuffer ; this . cncByteBuffer = null ; IoUtil . unmap ( cncByteBuffer ) ; super . close ( ) ;


signature:
public class ControlResponseAdapter { /** * Dispatch a descriptor message to a consumer by reading the fields in the correct order .
 * @ param decoder which wraps the encoded message ready for reading .
 * @ param consumer to which the decoded fields should be passed . */
public static void dispatchDescriptor ( final RecordingDescriptorDecoder decoder , final RecordingDescriptorConsumer consumer ) { } }

implementation: consumer . onRecordingDescriptor ( decoder . controlSessionId ( ) , decoder . correlationId ( ) , decoder . recordingId ( ) , decoder . startTimestamp ( ) , decoder . stopTimestamp ( ) , decoder . startPosition ( ) , decoder . stopPosition ( ) , decoder . initialTermId ( ) , decoder . segmentFileLength ( ) , decoder . termBufferLength ( ) , decoder . mtuLength ( ) , decoder . sessionId ( ) , decoder . streamId ( ) , decoder . strippedChannel ( ) , decoder . originalChannel ( ) , decoder . sourceIdentity ( ) ) ;


signature:
public class TermRebuilder { /** * Insert a packet of frames into the log at the appropriate termOffset as indicated by the term termOffset header .
 * If the packet has already been inserted then this is a noop .
 * @ param termBuffer into which the packet should be inserted .
 * @ param termOffset in the term at which the packet should be inserted .
 * @ param packet containing a sequence of frames .
 * @ param length of the packet of frames in bytes . */
public static void insert ( final UnsafeBuffer termBuffer , final int termOffset , final UnsafeBuffer packet , final int length ) { } }

implementation: if ( 0 == termBuffer . getInt ( termOffset ) ) { termBuffer . putBytes ( termOffset + HEADER_LENGTH , packet , HEADER_LENGTH , length - HEADER_LENGTH ) ; termBuffer . putLong ( termOffset + 24 , packet . getLong ( 24 ) ) ; termBuffer . putLong ( termOffset + 16 , packet . getLong ( 16 ) ) ; termBuffer . putLong ( termOffset + 8 , packet . getLong ( 8 ) ) ; termBuffer . putLongOrdered ( termOffset , packet . getLong ( 0 ) ) ; }


signature:
public class CapacityByteArrayOutputStream { /** * Construct a CapacityByteArrayOutputStream configured such that its initial slab size is
 * determined by { @ link # initialSlabSizeHeuristic } , with targetCapacity = = maxCapacityHint
 * @ param minSlabSize a minimum slab size
 * @ param maxCapacityHint a hint for the maximum required capacity
 * @ param targetNumSlabs the target number of slabs
 * @ param allocator an allocator to use when creating byte buffers for slabs
 * @ return a capacity baos */
public static CapacityByteArrayOutputStream withTargetNumSlabs ( int minSlabSize , int maxCapacityHint , int targetNumSlabs , ByteBufferAllocator allocator ) { } }

implementation: return new CapacityByteArrayOutputStream ( initialSlabSizeHeuristic ( minSlabSize , maxCapacityHint , targetNumSlabs ) , maxCapacityHint , allocator ) ;


signature:
public class CapacityByteArrayOutputStream { /** * the new slab is guaranteed to be at least minimumSize
 * @ param minimumSize the size of the data we want to copy in the new slab */
private void addSlab ( int minimumSize ) { } }

implementation:
int nextSlabSize ; if ( bytesUsed == 0 ) { nextSlabSize = initialSlabSize ; } else if ( bytesUsed > maxCapacityHint / 5 ) { // to avoid an overhead of up to twice the needed size , we get linear when approaching target page size
nextSlabSize = maxCapacityHint / 5 ; } else { // double the size every time
nextSlabSize = bytesUsed ; } if ( nextSlabSize < minimumSize ) { LOG . debug ( "slab size {} too small for value of size {}. Bumping up slab size" , nextSlabSize , minimumSize ) ; nextSlabSize = minimumSize ; } LOG . debug ( "used {} slabs, adding new slab of size {}" , slabs . size ( ) , nextSlabSize ) ; this . currentSlab = allocator . allocate ( nextSlabSize ) ; this . slabs . add ( currentSlab ) ; this . bytesAllocated += nextSlabSize ; this . currentSlabIndex = 0 ;


signature:
public class CapacityByteArrayOutputStream { /** * Writes the complete contents of this buffer to the specified output stream argument . the output
 * stream ' s write method < code > out . write ( slab , 0 , slab . length ) < / code > ) will be called once per slab .
 * @ param out the output stream to which to write the data .
 * @ exception IOException if an I / O error occurs . */
public void writeTo ( OutputStream out ) throws IOException { } }

implementation: for ( int i = 0 ; i < slabs . size ( ) - 1 ; i ++ ) { writeToOutput ( out , slabs . get ( i ) , slabs . get ( i ) . position ( ) ) ; } writeToOutput ( out , currentSlab , currentSlabIndex ) ;


signature:
public class CapacityByteArrayOutputStream { /** * When re - using an instance with reset , it will adjust slab size based on previous data size .
 * The intent is to reuse the same instance for the same type of data ( for example , the same column ) .
 * The assumption is that the size in the buffer will be consistent . */
public void reset ( ) { } }

implementation:
// readjust slab size .
// 7 = 2 ^ 3 - 1 so that doubling the initial size 3 times will get to the same size
this . initialSlabSize = max ( bytesUsed / 7 , initialSlabSize ) ; LOG . debug ( "initial slab of size {}" , initialSlabSize ) ; for ( ByteBuffer slab : slabs ) { allocator . release ( slab ) ; } this . slabs . clear ( ) ; this . bytesAllocated = 0 ; this . bytesUsed = 0 ; this . currentSlab = EMPTY_SLAB ; this . currentSlabIndex = 0 ;


signature:
public class CapacityByteArrayOutputStream { /** * Replace the byte stored at position index in this stream with value
 * @ param index which byte to replace
 * @ param value the value to replace it with */
public void setByte ( long index , byte value ) { } }

implementation:
checkArgument ( index < bytesUsed , "Index: " + index + " is >= the current size of: " + bytesUsed ) ; long seen = 0 ; for ( int i = 0 ; i < slabs . size ( ) ; i ++ ) { ByteBuffer slab = slabs . get ( i ) ; if ( index < seen + slab . limit ( ) ) { // ok found index
slab . put ( ( int ) ( index - seen ) , value ) ; break ; } seen += slab . limit ( ) ; }


signature:
public class ColumnIndexBuilder { /** * Adds the data from the specified statistics to this builder
 * @ param stats
 * the statistics to be added */
public void add ( Statistics < ? > stats ) { } }

implementation: if ( stats . hasNonNullValue ( ) ) { nullPages . add ( false ) ; Object min = stats . genericGetMin ( ) ; Object max = stats . genericGetMax ( ) ; addMinMax ( min , max ) ; pageIndexes . add ( nextPageIndex ) ; minMaxSize += sizeOf ( min ) ; minMaxSize += sizeOf ( max ) ; } else { nullPages . add ( true ) ; } nullCounts . add ( stats . getNumNulls ( ) ) ; ++ nextPageIndex ;


signature:
public class ColumnIndexBuilder { /** * min [ i ] > = min [ i + 1 ] & & max [ i ] > = max [ i + 1] */
private boolean isDescending ( PrimitiveComparator < Binary > comparator ) { } }

implementation: for ( int i = 1 , n = pageIndexes . size ( ) ; i < n ; ++ i ) { if ( compareMinValues ( comparator , i - 1 , i ) < 0 || compareMaxValues ( comparator , i - 1 , i ) < 0 ) { return false ; } } return true ;


signature:
public class ValuesReader { /** * Called to initialize the column reader from a part of a page .
 * The underlying implementation knows how much data to read , so a length
 * is not provided .
 * Each page may contain several sections :
 * < ul >
 * < li > repetition levels column
 * < li > definition levels column
 * < li > data column
 * < / ul >
 * This function is called with ' offset ' pointing to the beginning of one of these sections ,
 * and should return the offset to the section following it .
 * @ param valueCount count of values in this page
 * @ param page the array to read from containing the page data ( repetition levels , definition levels , data )
 * @ param offset where to start reading from in the page
 * @ throws IOException
 * @ deprecated Will be removed in 2.0.0 */
@ Deprecated public void initFromPage ( int valueCount , ByteBuffer page , int offset ) throws IOException { } }

implementation: if ( offset < 0 ) { throw new IllegalArgumentException ( "Illegal offset: " + offset ) ; } actualOffset = offset ; ByteBuffer pageWithOffset = page . duplicate ( ) ; pageWithOffset . position ( offset ) ; initFromPage ( valueCount , ByteBufferInputStream . wrap ( pageWithOffset ) ) ; actualOffset = - 1 ;


signature:
public class ValuesReader { /** * Same functionality as method of the same name that takes a ByteBuffer instead of a byte [ ] .
 * This method is only provided for backward compatibility and will be removed in a future release .
 * Please update any code using it as soon as possible .
 * @ see # initFromPage ( int , ByteBuffer , int ) */
@ Deprecated public void initFromPage ( int valueCount , byte [ ] page , int offset ) throws IOException { } }

implementation: this . initFromPage ( valueCount , ByteBuffer . wrap ( page ) , offset ) ;


signature:
public class ValuesReader { /** * Called to initialize the column reader from a part of a page .
 * Implementations must consume all bytes from the input stream , leaving the
 * stream ready to read the next section of data . The underlying
 * implementation knows how much data to read , so a length is not provided .
 * Each page may contain several sections :
 * < ul >
 * < li > repetition levels column
 * < li > definition levels column
 * < li > data column
 * < / ul >
 * @ param valueCount count of values in this page
 * @ param in an input stream containing the page data at the correct offset
 * @ throws IOException if there is an exception while reading from the input stream */
public void initFromPage ( int valueCount , ByteBufferInputStream in ) throws IOException { } }

implementation: if ( actualOffset != - 1 ) { throw new UnsupportedOperationException ( "Either initFromPage(int, ByteBuffer, int) or initFromPage(int, ByteBufferInputStream) must be implemented in " + getClass ( ) . getName ( ) ) ; } initFromPage ( valueCount , in . slice ( valueCount ) , 0 ) ;


signature:
public class AvroParquetInputFormat { /** * Set the subset of columns to read ( projection pushdown ) . Specified as an Avro
 * schema , the requested projection is converted into a Parquet schema for Parquet
 * column projection .
 * This is useful if the full schema is large and you only want to read a few
 * columns , since it saves time by not reading unused columns .
 * If a requested projection is set , then the Avro schema used for reading
 * must be compatible with the projection . For instance , if a column is not included
 * in the projection then it must either not be included or be optional in the read
 * schema . Use { @ link # setAvroReadSchema ( org . apache . hadoop . mapreduce . Job ,
 * org . apache . avro . Schema ) } to set a read schema , if needed .
 * @ param job a job
 * @ param requestedProjection the requested projection schema
 * @ see # setAvroReadSchema ( org . apache . hadoop . mapreduce . Job , org . apache . avro . Schema )
 * @ see org . apache . parquet . avro . AvroParquetOutputFormat # setSchema ( org . apache . hadoop . mapreduce . Job , org . apache . avro . Schema ) */
public static void setRequestedProjection ( Job job , Schema requestedProjection ) { } }

implementation: AvroReadSupport . setRequestedProjection ( ContextUtil . getConfiguration ( job ) , requestedProjection ) ;


signature:
public class AvroParquetInputFormat { /** * Override the Avro schema to use for reading . If not set , the Avro schema used for
 * writing is used .
 * Differences between the read and write schemas are resolved using
 * < a href = " http : / / avro . apache . org / docs / current / spec . html # Schema + Resolution " > Avro ' s schema resolution rules < / a > .
 * @ param job a job
 * @ param avroReadSchema the requested schema
 * @ see # setRequestedProjection ( org . apache . hadoop . mapreduce . Job , org . apache . avro . Schema )
 * @ see org . apache . parquet . avro . AvroParquetOutputFormat # setSchema ( org . apache . hadoop . mapreduce . Job , org . apache . avro . Schema ) */
public static void setAvroReadSchema ( Job job , Schema avroReadSchema ) { } }

implementation: AvroReadSupport . setAvroReadSchema ( ContextUtil . getConfiguration ( job ) , avroReadSchema ) ;


signature:
public class AvroParquetInputFormat { /** * Uses an instance of the specified { @ link AvroDataSupplier } class to control how the
 * { @ link org . apache . avro . specific . SpecificData } instance that is used to find
 * Avro specific records is created .
 * @ param job a job
 * @ param supplierClass an avro data supplier class */
public static void setAvroDataSupplier ( Job job , Class < ? extends AvroDataSupplier > supplierClass ) { } }

implementation: AvroReadSupport . setAvroDataSupplier ( ContextUtil . getConfiguration ( job ) , supplierClass ) ;


signature:
public class FilteringGroupConverter { /** * TODO : making the assumption that getConverter ( i ) is only called once , is that valid ? */
@ Override public Converter getConverter ( int fieldIndex ) { } }

implementation:
// get the real converter from the delegate
Converter delegateConverter = checkNotNull ( delegate . getConverter ( fieldIndex ) , "delegate converter" ) ; // determine the indexFieldPath for the converter proxy we ' re about to make , which is
// this converter ' s path + the requested fieldIndex
List < Integer > newIndexFieldPath = new ArrayList < Integer > ( indexFieldPath . size ( ) + 1 ) ; newIndexFieldPath . addAll ( indexFieldPath ) ; newIndexFieldPath . add ( fieldIndex ) ; if ( delegateConverter . isPrimitive ( ) ) { PrimitiveColumnIO columnIO = getColumnIO ( newIndexFieldPath ) ; ColumnPath columnPath = ColumnPath . get ( columnIO . getColumnDescriptor ( ) . getPath ( ) ) ; ValueInspector [ ] valueInspectors = getValueInspectors ( columnPath ) ; return new FilteringPrimitiveConverter ( delegateConverter . asPrimitiveConverter ( ) , valueInspectors ) ; } else { return new FilteringGroupConverter ( delegateConverter . asGroupConverter ( ) , newIndexFieldPath , valueInspectorsByColumn , columnIOsByIndexFieldPath ) ; }


signature:
public class ColumnIndexFilter { /** * Calculates the row ranges containing the indexes of the rows might match the specified filter .
 * @ param filter
 * to be used for filtering the rows
 * @ param columnIndexStore
 * the store for providing column / offset indexes
 * @ param paths
 * the paths of the columns used in the actual projection ; a column not being part of the projection will be
 * handled as containing { @ code null } values only even if the column has values written in the file
 * @ param rowCount
 * the total number of rows in the row - group
 * @ return the ranges of the possible matching row indexes ; the returned ranges will contain all the rows if any of
 * the required offset index is missing */
public static RowRanges calculateRowRanges ( FilterCompat . Filter filter , ColumnIndexStore columnIndexStore , Set < ColumnPath > paths , long rowCount ) { } }

implementation: return filter . accept ( new FilterCompat . Visitor < RowRanges > ( ) { @ Override public RowRanges visit ( FilterPredicateCompat filterPredicateCompat ) { try { return filterPredicateCompat . getFilterPredicate ( ) . accept ( new ColumnIndexFilter ( columnIndexStore , paths , rowCount ) ) ; } catch ( MissingOffsetIndexException e ) { LOGGER . info ( e . getMessage ( ) ) ; return RowRanges . createSingle ( rowCount ) ; } } @ Override public RowRanges visit ( UnboundRecordFilterCompat unboundRecordFilterCompat ) { return RowRanges . createSingle ( rowCount ) ; } @ Override public RowRanges visit ( NoOpFilter noOpFilter ) { return RowRanges . createSingle ( rowCount ) ; } } ) ;


signature:
public class ParquetThriftInputFormat { /** * Call this method when setting up your Hadoop job if reading into a Thrift object
 * that is not encoded into the parquet - serialized thrift metadata ( for example ,
 * writing with Apache Thrift , but reading back into Twitter Scrooge version of
 * the same thrift definition , or a different but compatible Apache Thrift class ) .
 * @ param conf a mapred jobconf
 * @ param klass a thrift class
 * @ param < T > the Java type of records the configured ReadSupport will produce */
public static < T > void setThriftClass ( JobConf conf , Class < T > klass ) { } }

implementation: conf . set ( ThriftReadSupport . THRIFT_READ_CLASS_KEY , klass . getName ( ) ) ;


signature:
public class HiveSchemaConverter { /** * 1 anonymous element " array _ element " */
private static GroupType convertArrayType ( final String name , final ListTypeInfo typeInfo ) { } }

implementation: final TypeInfo subType = typeInfo . getListElementTypeInfo ( ) ; return listWrapper ( name , listType ( ) , new GroupType ( Repetition . REPEATED , ParquetHiveSerDe . ARRAY . toString ( ) , convertType ( "array_element" , subType ) ) ) ;


signature:
public class HiveSchemaConverter { /** * An optional group containing multiple elements */
private static GroupType convertStructType ( final String name , final StructTypeInfo typeInfo ) { } }

implementation: final List < String > columnNames = typeInfo . getAllStructFieldNames ( ) ; final List < TypeInfo > columnTypes = typeInfo . getAllStructFieldTypeInfos ( ) ; return new GroupType ( Repetition . OPTIONAL , name , convertTypes ( columnNames , columnTypes ) ) ;


signature:
public class HiveSchemaConverter { /** * 2 elements : " key " , " value " */
private static GroupType convertMapType ( final String name , final MapTypeInfo typeInfo ) { } }

implementation: final Type keyType = convertType ( ParquetHiveSerDe . MAP_KEY . toString ( ) , typeInfo . getMapKeyTypeInfo ( ) , Repetition . REQUIRED ) ; final Type valueType = convertType ( ParquetHiveSerDe . MAP_VALUE . toString ( ) , typeInfo . getMapValueTypeInfo ( ) ) ; return ConversionPatterns . mapType ( Repetition . OPTIONAL , name , keyType , valueType ) ;


signature:
public class GlobExpander { /** * Expands a string with braces ( " { } " ) into all of its possible permutations .
 * We call anything inside of { } braces a " one - of " group .
 * The only special characters in this glob syntax are ' } ' , ' { ' and ' , '
 * The top - level pattern must not contain any commas , but a " one - of " group separates
 * its elements with commas , and a one - of group may contain sub one - of groups .
 * For example :
 * start { a , b , c } end - & gt ; startaend , startbend , startcend
 * start { a , { b , c } , d } - & gt ; startaend , startbend , startcend , startdend
 * { a , b , c } - & gt ; a , b , c
 * start { a , b { x , y } } - & gt ; starta , startbx , startby
 * @ param globPattern a string in the format described above
 * @ return a list of all the strings that would satisfy globPattern , including duplicates */
public static List < String > expand ( String globPattern ) { } }

implementation: return GlobExpanderImpl . expand ( GlobParser . parse ( globPattern ) ) ;


signature:
public class RowRanges { /** * Creates a new RowRanges object with the single range [ 0 , rowCount - 1 ] . */
static RowRanges createSingle ( long rowCount ) { } }

implementation: RowRanges ranges = new RowRanges ( ) ; ranges . add ( new Range ( 0 , rowCount - 1 ) ) ; return ranges ;


signature:
public class RowRanges { /** * Creates a new RowRanges object with the following ranges .
 * [ firstRowIndex [ 0 ] , lastRowIndex [ 0 ] ] ,
 * [ firstRowIndex [ 1 ] , lastRowIndex [ 1 ] ] ,
 * [ firstRowIndex [ n ] , lastRowIndex [ n ] ]
 * ( See OffsetIndex . getFirstRowIndex and OffsetIndex . getLastRowIndex for details . )
 * The union of the ranges are calculated so the result ranges always contain the disjunct ranges . See union for
 * details . */
static RowRanges create ( long rowCount , PrimitiveIterator . OfInt pageIndexes , OffsetIndex offsetIndex ) { } }

implementation: RowRanges ranges = new RowRanges ( ) ; while ( pageIndexes . hasNext ( ) ) { int pageIndex = pageIndexes . nextInt ( ) ; ranges . add ( new Range ( offsetIndex . getFirstRowIndex ( pageIndex ) , offsetIndex . getLastRowIndex ( pageIndex , rowCount ) ) ) ; } return ranges ;


signature:
public class RowRanges { /** * Calculates the union of the two specified RowRanges object . The union of two range is calculated if there are no
 * elements between them . Otherwise , the two disjunct ranges are stored separately .
 * For example :
 * [ 113 , 241 ] ∪ [ 221 , 340 ] = [ 113 , 330]
 * [ 113 , 230 ] ∪ [ 231 , 340 ] = [ 113 , 340]
 * while
 * [ 113 , 230 ] ∪ [ 232 , 340 ] = [ 113 , 230 ] , [ 232 , 340]
 * The result RowRanges object will contain all the row indexes that were contained in one of the specified objects . */
static RowRanges union ( RowRanges left , RowRanges right ) { } }

implementation: RowRanges result = new RowRanges ( ) ; Iterator < Range > it1 = left . ranges . iterator ( ) ; Iterator < Range > it2 = right . ranges . iterator ( ) ; if ( it2 . hasNext ( ) ) { Range range2 = it2 . next ( ) ; while ( it1 . hasNext ( ) ) { Range range1 = it1 . next ( ) ; if ( range1 . isAfter ( range2 ) ) { result . add ( range2 ) ; range2 = range1 ; Iterator < Range > tmp = it1 ; it1 = it2 ; it2 = tmp ; } else { result . add ( range1 ) ; } } result . add ( range2 ) ; } else { it2 = it1 ; } while ( it2 . hasNext ( ) ) { result . add ( it2 . next ( ) ) ; } return result ;


signature:
public class RowRanges { /** * Calculates the intersection of the two specified RowRanges object . Two ranges intersect if they have common
 * elements otherwise the result is empty .
 * For example :
 * [ 113 , 241 ] ∩ [ 221 , 340 ] = [ 221 , 241]
 * while
 * [ 113 , 230 ] ∩ [ 231 , 340 ] = < EMPTY >
 * The result RowRanges object will contain all the row indexes there were contained in both of the specified objects */
static RowRanges intersection ( RowRanges left , RowRanges right ) { } }

implementation: RowRanges result = new RowRanges ( ) ; int rightIndex = 0 ; for ( Range l : left . ranges ) { for ( int i = rightIndex , n = right . ranges . size ( ) ; i < n ; ++ i ) { Range r = right . ranges . get ( i ) ; if ( l . isBefore ( r ) ) { break ; } else if ( l . isAfter ( r ) ) { rightIndex = i + 1 ; continue ; } result . add ( Range . intersection ( l , r ) ) ; } } return result ;


signature:
public class RowRanges { /** * Adds a range to the end of the list of ranges . It maintains the disjunct ascending order ( * ) of the ranges by
 * trying to union the specified range to the last ranges in the list . The specified range shall be larger ( * ) than
 * the last one or might be overlapped with some of the last ones .
 * ( * ) [ a , b ] < [ c , d ] if b < c */
private void add ( Range range ) { } }

implementation: Range rangeToAdd = range ; for ( int i = ranges . size ( ) - 1 ; i >= 0 ; -- i ) { Range last = ranges . get ( i ) ; assert ! last . isAfter ( range ) ; Range u = Range . union ( last , rangeToAdd ) ; if ( u == null ) { break ; } rangeToAdd = u ; ranges . remove ( i ) ; } ranges . add ( rangeToAdd ) ;


signature:
public class IntColumnChunkMetaData { /** * stores a positive long into an int ( assuming it fits )
 * @ param value
 * @ return */
private int positiveLongToInt ( long value ) { } }

implementation: if ( ! ColumnChunkMetaData . positiveLongFitsInAnInt ( value ) ) { throw new IllegalArgumentException ( "value should be positive and fit in an int: " + value ) ; } return ( int ) ( value + Integer . MIN_VALUE ) ;


signature:
public class GlobParser { /** * Parse a String into a { @ link GlobNodeSequence }
 * See { @ link org . apache . parquet . Strings # expandGlob ( String ) } */
public static GlobNodeSequence parse ( String pattern ) { } }

implementation:
/* * The parse algorithm works as follows , assuming we are parsing :
 * " apache { one , pre { x , y } post , two } parquet { a , b } "
 * 1 ) Begin scanning the string until we find the first {
 * 2 ) Now that we ' ve found the beginning of a glob group , scan forwards
 * until the end of this glob group ( by counting { and } we see until we find
 * the closing } for the group we found in step 1 ) .
 * 3 ) Once the matching closing } is found we need to do two things . First , everything
 * from the end of the last group up to start of this group is an Atom , so in the example
 * above , once we ' ve found that " { one , pre { x , y } post , two } " is the first group , we need to grab
 * " apache " and treat it as an atom and add it to our sequence .
 * Then , we parse " { one , pre { x , y } post , two } " using a similar but slightly different function ( parseOneOf )
 * and add the result from that to our sequence .
 * 4 ) Repeat until the end of the string - - so next we find { a , b } and add " parquet " as an Atom and parse
 * { a , b } using parseOneOf . */
if ( pattern . isEmpty ( ) || pattern . equals ( "{}" ) ) { return new GlobNodeSequence ( Arrays . < GlobNode > asList ( new Atom ( "" ) ) ) ; } // the outer parse method needs to parse the pattern into a
// GlobNodeSequence , though it may end up being a singleton sequence
List < GlobNode > children = new ArrayList < GlobNode > ( ) ; int unmatchedBraces = 0 ; // count of unmatched braces
int firstBrace = 0 ; // open brace of current group being processsed
int anchor = 0 ; // first un - parsed character position
for ( int i = 0 ; i < pattern . length ( ) ; i ++ ) { char c = pattern . charAt ( i ) ; switch ( c ) { case ',' : if ( unmatchedBraces == 0 ) { // commas not allowed in the top level expression
// TODO : maybe turn this check off ?
throw new GlobParseException ( "Unexpected comma outside of a {} group:\n" + annotateMessage ( pattern , i ) ) ; } break ; case '{' : if ( unmatchedBraces == 0 ) { // this is the first brace of an outermost { } group
firstBrace = i ; } unmatchedBraces ++ ; break ; case '}' : unmatchedBraces -- ; if ( unmatchedBraces < 0 ) { throw new GlobParseException ( "Unexpected closing }:\n" + annotateMessage ( pattern , i ) ) ; } if ( unmatchedBraces == 0 ) { // grab everything from the end of the last group up to here ,
// not including the close brace , it is an Atom in our sequence
// ( assuming it ' s not empty )
if ( anchor != firstBrace ) { // not empty !
// ( substring ' s end param is exclusive )
children . add ( new Atom ( pattern . substring ( anchor , firstBrace ) ) ) ; } // grab the group , parse it , add it to our sequence , and then continue
// note that we skip the braces on both sides ( substring ' s end param is exclusive )
children . add ( parseOneOf ( pattern . substring ( firstBrace + 1 , i ) ) ) ; // we have now parsed all the way up to here , the next un - parsed char is i + 1
anchor = i + 1 ; } break ; } } if ( unmatchedBraces > 0 ) { throw new GlobParseException ( "Not enough close braces in: " + pattern ) ; } if ( anchor != pattern . length ( ) ) { // either there were no { } groups , or there were some characters after the
// last } , either way whatever is left ( could be the entire input ) is an Atom
// in our sequence
children . add ( new Atom ( pattern . substring ( anchor , pattern . length ( ) ) ) ) ; } return new GlobNodeSequence ( children ) ;


signature:
public class GlobParser { /** * for pretty printing which character had the error */
private static String annotateMessage ( String message , int pos ) { } }

implementation: StringBuilder sb = new StringBuilder ( message ) ; sb . append ( '\n' ) ; for ( int i = 0 ; i < pos ; i ++ ) { sb . append ( '-' ) ; } sb . append ( '^' ) ; return sb . toString ( ) ;


signature:
public class RunLengthBitPackingHybridEncoder { /** * If we are currently writing a bit - packed - run , update the
 * bit - packed - header and consider this run to be over
 * does nothing if we ' re not currently writing a bit - packed run */
private void endPreviousBitPackedRun ( ) { } }

implementation:
if ( bitPackedRunHeaderPointer == - 1 ) { // we ' re not currently in a bit - packed - run
return ; } // create bit - packed - header , which needs to fit in 1 byte
byte bitPackHeader = ( byte ) ( ( bitPackedGroupCount << 1 ) | 1 ) ; // update this byte
baos . setByte ( bitPackedRunHeaderPointer , bitPackHeader ) ; // mark that this run is over
bitPackedRunHeaderPointer = - 1 ; // reset the number of groups
bitPackedGroupCount = 0 ;


signature:
public class ProtocolReadToWrite { /** * reads one record from in and writes it to out
 * exceptions are not recoverable as record might be halfway written
 * @ param in input protocol
 * @ param out output protocol
 * @ throws TException if there is an error while reading or writing */
@ Override public void readOne ( TProtocol in , TProtocol out ) throws TException { } }

implementation: readOneStruct ( in , out ) ;


signature:
public class FilterCompat { /** * Given a FilterPredicate , return a Filter that wraps it .
 * This method also logs the filter being used and rewrites
 * the predicate to not include the not ( ) operator .
 * @ param filterPredicate a filter predicate
 * @ return a filter for the given predicate */
public static Filter get ( FilterPredicate filterPredicate ) { } }

implementation:
checkNotNull ( filterPredicate , "filterPredicate" ) ; LOG . info ( "Filtering using predicate: {}" , filterPredicate ) ; // rewrite the predicate to not include the not ( ) operator
FilterPredicate collapsedPredicate = LogicalInverseRewriter . rewrite ( filterPredicate ) ; if ( ! filterPredicate . equals ( collapsedPredicate ) ) { LOG . info ( "Predicate has been collapsed to: {}" , collapsedPredicate ) ; } return new FilterPredicateCompat ( collapsedPredicate ) ;


signature:
public class FilterCompat { /** * Given either a FilterPredicate or the class of an UnboundRecordFilter , or neither ( but not both )
 * return a Filter that wraps whichever was provided .
 * Either filterPredicate or unboundRecordFilterClass must be null , or an exception is thrown .
 * If both are null , the no op filter will be returned .
 * @ param filterPredicate a filter predicate , or null
 * @ param unboundRecordFilter an unbound record filter , or null
 * @ return a Filter wrapping either the predicate or the unbound record filter ( from the old API ) */
public static Filter get ( FilterPredicate filterPredicate , UnboundRecordFilter unboundRecordFilter ) { } }

implementation: checkArgument ( filterPredicate == null || unboundRecordFilter == null , "Cannot provide both a FilterPredicate and an UnboundRecordFilter" ) ; if ( filterPredicate != null ) { return get ( filterPredicate ) ; } if ( unboundRecordFilter != null ) { return get ( unboundRecordFilter ) ; } return NOOP ;


signature:
public class ParquetFileReader { /** * for files provided , check if there ' s a summary file .
 * If a summary file is found it is used otherwise the file footer is used .
 * @ param configuration the hadoop conf to connect to the file system ;
 * @ param partFiles the part files to read
 * @ return the footers for those files using the summary file if possible .
 * @ throws IOException if there is an exception while reading footers
 * @ deprecated metadata files are not recommended and will be removed in 2.0.0 */
@ Deprecated public static List < Footer > readAllFootersInParallelUsingSummaryFiles ( Configuration configuration , List < FileStatus > partFiles ) throws IOException { } }

implementation: return readAllFootersInParallelUsingSummaryFiles ( configuration , partFiles , false ) ;


signature:
public class ParquetFileReader { /** * for files provided , check if there ' s a summary file .
 * If a summary file is found it is used otherwise the file footer is used .
 * @ param configuration the hadoop conf to connect to the file system ;
 * @ param partFiles the part files to read
 * @ param skipRowGroups to skipRowGroups in the footers
 * @ return the footers for those files using the summary file if possible .
 * @ throws IOException if there is an exception while reading footers
 * @ deprecated metadata files are not recommended and will be removed in 2.0.0 */
@ Deprecated public static List < Footer > readAllFootersInParallelUsingSummaryFiles ( final Configuration configuration , final Collection < FileStatus > partFiles , final boolean skipRowGroups ) throws IOException { } }

implementation:
// figure out list of all parents to part files
Set < Path > parents = new HashSet < Path > ( ) ; for ( FileStatus part : partFiles ) { parents . add ( part . getPath ( ) . getParent ( ) ) ; } // read corresponding summary files if they exist
List < Callable < Map < Path , Footer > > > summaries = new ArrayList < Callable < Map < Path , Footer > > > ( ) ; for ( final Path path : parents ) { summaries . add ( new Callable < Map < Path , Footer > > ( ) { @ Override public Map < Path , Footer > call ( ) throws Exception { ParquetMetadata mergedMetadata = readSummaryMetadata ( configuration , path , skipRowGroups ) ; if ( mergedMetadata != null ) { final List < Footer > footers ; if ( skipRowGroups ) { footers = new ArrayList < Footer > ( ) ; for ( FileStatus f : partFiles ) { footers . add ( new Footer ( f . getPath ( ) , mergedMetadata ) ) ; } } else { footers = footersFromSummaryFile ( path , mergedMetadata ) ; } Map < Path , Footer > map = new HashMap < Path , Footer > ( ) ; for ( Footer footer : footers ) { // the folder may have been moved
footer = new Footer ( new Path ( path , footer . getFile ( ) . getName ( ) ) , footer . getParquetMetadata ( ) ) ; map . put ( footer . getFile ( ) , footer ) ; } return map ; } else { return Collections . emptyMap ( ) ; } } } ) ; } Map < Path , Footer > cache = new HashMap < Path , Footer > ( ) ; try { List < Map < Path , Footer > > footersFromSummaries = runAllInParallel ( configuration . getInt ( PARQUET_READ_PARALLELISM , 5 ) , summaries ) ; for ( Map < Path , Footer > footers : footersFromSummaries ) { cache . putAll ( footers ) ; } } catch ( ExecutionException e ) { throw new IOException ( "Error reading summaries" , e ) ; } // keep only footers for files actually requested and read file footer if not found in summaries
List < Footer > result = new ArrayList < Footer > ( partFiles . size ( ) ) ; List < FileStatus > toRead = new ArrayList < FileStatus > ( ) ; for ( FileStatus part : partFiles ) { Footer f = cache . get ( part . getPath ( ) ) ; if ( f != null ) { result . add ( f ) ; } else { toRead . add ( part ) ; } } if ( toRead . size ( ) > 0 ) { // read the footers of the files that did not have a summary file
LOG . info ( "reading another {} footers" , toRead . size ( ) ) ; result . addAll ( readAllFootersInParallel ( configuration , toRead , skipRowGroups ) ) ; } return result ;


signature:
public class ParquetFileReader { /** * read all the footers of the files provided
 * ( not using summary files )
 * @ param configuration the conf to access the File System
 * @ param partFiles the files to read
 * @ param skipRowGroups to skip the rowGroup info
 * @ return the footers
 * @ throws IOException if there is an exception while reading footers
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static List < Footer > readAllFootersInParallel ( final Configuration configuration , List < FileStatus > partFiles , final boolean skipRowGroups ) throws IOException { } }

implementation: List < Callable < Footer > > footers = new ArrayList < Callable < Footer > > ( ) ; for ( final FileStatus currentFile : partFiles ) { footers . add ( new Callable < Footer > ( ) { @ Override public Footer call ( ) throws Exception { try { return new Footer ( currentFile . getPath ( ) , readFooter ( configuration , currentFile , filter ( skipRowGroups ) ) ) ; } catch ( IOException e ) { throw new IOException ( "Could not read footer for file " + currentFile , e ) ; } } } ) ; } try { return runAllInParallel ( configuration . getInt ( PARQUET_READ_PARALLELISM , 5 ) , footers ) ; } catch ( ExecutionException e ) { throw new IOException ( "Could not read footer: " + e . getMessage ( ) , e . getCause ( ) ) ; }


signature:
public class ParquetFileReader { /** * Read the footers of all the files under that path ( recursively )
 * not using summary files .
 * @ param configuration a configuration
 * @ param fileStatus a file status to recursively list
 * @ param skipRowGroups whether to skip reading row group metadata
 * @ return a list of footers
 * @ throws IOException if an exception is thrown while reading the footers
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static List < Footer > readAllFootersInParallel ( Configuration configuration , FileStatus fileStatus , boolean skipRowGroups ) throws IOException { } }

implementation: List < FileStatus > statuses = listFiles ( configuration , fileStatus ) ; return readAllFootersInParallel ( configuration , statuses , skipRowGroups ) ;


signature:
public class ParquetFileReader { /** * Read the footers of all the files under that path ( recursively )
 * not using summary files .
 * rowGroups are not skipped
 * @ param configuration the configuration to access the FS
 * @ param fileStatus the root dir
 * @ return all the footers
 * @ throws IOException if an exception is thrown while reading the footers
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static List < Footer > readAllFootersInParallel ( Configuration configuration , FileStatus fileStatus ) throws IOException { } }

implementation: return readAllFootersInParallel ( configuration , fileStatus , false ) ;


signature:
public class ParquetFileReader { /** * this always returns the row groups
 * @ param configuration a configuration
 * @ param pathStatus a file status to read footers from
 * @ return a list of footers
 * @ throws IOException if an exception is thrown while reading the footers
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static List < Footer > readFooters ( Configuration configuration , FileStatus pathStatus ) throws IOException { } }

implementation: return readFooters ( configuration , pathStatus , false ) ;


signature:
public class ParquetFileReader { /** * Read the footers of all the files under that path ( recursively )
 * using summary files if possible
 * @ param configuration the configuration to access the FS
 * @ param pathStatus the root dir
 * @ param skipRowGroups whether to skip reading row group metadata
 * @ return all the footers
 * @ throws IOException if an exception is thrown while reading the footers
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static List < Footer > readFooters ( Configuration configuration , FileStatus pathStatus , boolean skipRowGroups ) throws IOException { } }

implementation: List < FileStatus > files = listFiles ( configuration , pathStatus ) ; return readAllFootersInParallelUsingSummaryFiles ( configuration , files , skipRowGroups ) ;


signature:
public class ParquetFileReader { /** * Specifically reads a given summary file
 * @ param configuration a configuration
 * @ param summaryStatus file status for a summary file
 * @ return the metadata translated for each file
 * @ throws IOException if an exception is thrown while reading the summary file
 * @ deprecated metadata files are not recommended and will be removed in 2.0.0 */
@ Deprecated public static List < Footer > readSummaryFile ( Configuration configuration , FileStatus summaryStatus ) throws IOException { } }

implementation: final Path parent = summaryStatus . getPath ( ) . getParent ( ) ; ParquetMetadata mergedFooters = readFooter ( configuration , summaryStatus , filter ( false ) ) ; return footersFromSummaryFile ( parent , mergedFooters ) ;


signature:
public class ParquetFileReader { /** * Reads the meta data block in the footer of the file
 * @ param configuration a configuration
 * @ param file the parquet File
 * @ return the metadata blocks in the footer
 * @ throws IOException if an error occurs while reading the file
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static final ParquetMetadata readFooter ( Configuration configuration , Path file ) throws IOException { } }

implementation: return readFooter ( configuration , file , NO_FILTER ) ;


signature:
public class ParquetFileReader { /** * Reads the meta data in the footer of the file .
 * Skipping row groups ( or not ) based on the provided filter
 * @ param configuration a configuration
 * @ param file the Parquet File
 * @ param filter the filter to apply to row groups
 * @ return the metadata with row groups filtered .
 * @ throws IOException if an error occurs while reading the file
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
public static ParquetMetadata readFooter ( Configuration configuration , Path file , MetadataFilter filter ) throws IOException { } }

implementation: return readFooter ( HadoopInputFile . fromPath ( file , configuration ) , filter ) ;


signature:
public class ParquetFileReader { /** * Reads the meta data block in the footer of the file
 * @ param configuration a configuration
 * @ param file the parquet File
 * @ param filter the filter to apply to row groups
 * @ return the metadata blocks in the footer
 * @ throws IOException if an error occurs while reading the file
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static final ParquetMetadata readFooter ( Configuration configuration , FileStatus file , MetadataFilter filter ) throws IOException { } }

implementation: return readFooter ( HadoopInputFile . fromStatus ( file , configuration ) , filter ) ;


signature:
public class ParquetFileReader { /** * Reads the meta data block in the footer of the file using provided input stream
 * @ param file a { @ link InputFile } to read
 * @ param filter the filter to apply to row groups
 * @ return the metadata blocks in the footer
 * @ throws IOException if an error occurs while reading the file
 * @ deprecated will be removed in 2.0.0;
 * use { @ link ParquetFileReader # open ( InputFile , ParquetReadOptions ) } */
@ Deprecated public static final ParquetMetadata readFooter ( InputFile file , MetadataFilter filter ) throws IOException { } }

implementation: ParquetReadOptions options ; if ( file instanceof HadoopInputFile ) { options = HadoopReadOptions . builder ( ( ( HadoopInputFile ) file ) . getConfiguration ( ) ) . withMetadataFilter ( filter ) . build ( ) ; } else { options = ParquetReadOptions . builder ( ) . withMetadataFilter ( filter ) . build ( ) ; } try ( SeekableInputStream in = file . newStream ( ) ) { return readFooter ( file , options , in ) ; }


signature:
public class ParquetFileReader { /** * Open a { @ link InputFile file } .
 * @ param file an input file
 * @ return an open ParquetFileReader
 * @ throws IOException if there is an error while opening the file */
public static ParquetFileReader open ( InputFile file ) throws IOException { } }

implementation: return new ParquetFileReader ( file , ParquetReadOptions . builder ( ) . build ( ) ) ;


signature:
public class ParquetFileReader { /** * Open a { @ link InputFile file } with { @ link ParquetReadOptions options } .
 * @ param file an input file
 * @ param options parquet read options
 * @ return an open ParquetFileReader
 * @ throws IOException if there is an error while opening the file */
public static ParquetFileReader open ( InputFile file , ParquetReadOptions options ) throws IOException { } }

implementation: return new ParquetFileReader ( file , options ) ;


signature:
public class ParquetFileReader { /** * Reads all the columns requested from the row group at the current file position .
 * @ throws IOException if an error occurs while reading
 * @ return the PageReadStore which can provide PageReaders for each column . */
public PageReadStore readNextRowGroup ( ) throws IOException { } }

implementation:
if ( currentBlock == blocks . size ( ) ) { return null ; } BlockMetaData block = blocks . get ( currentBlock ) ; if ( block . getRowCount ( ) == 0 ) { throw new RuntimeException ( "Illegal row group of 0 rows" ) ; } this . currentRowGroup = new ColumnChunkPageReadStore ( block . getRowCount ( ) ) ; // prepare the list of consecutive parts to read them in one scan
List < ConsecutivePartList > allParts = new ArrayList < ConsecutivePartList > ( ) ; ConsecutivePartList currentParts = null ; for ( ColumnChunkMetaData mc : block . getColumns ( ) ) { ColumnPath pathKey = mc . getPath ( ) ; BenchmarkCounter . incrementTotalBytes ( mc . getTotalSize ( ) ) ; ColumnDescriptor columnDescriptor = paths . get ( pathKey ) ; if ( columnDescriptor != null ) { long startingPos = mc . getStartingPos ( ) ; // first part or not consecutive = > new list
if ( currentParts == null || currentParts . endPos ( ) != startingPos ) { currentParts = new ConsecutivePartList ( startingPos ) ; allParts . add ( currentParts ) ; } currentParts . addChunk ( new ChunkDescriptor ( columnDescriptor , mc , startingPos , ( int ) mc . getTotalSize ( ) ) ) ; } } // actually read all the chunks
ChunkListBuilder builder = new ChunkListBuilder ( ) ; for ( ConsecutivePartList consecutiveChunks : allParts ) { consecutiveChunks . readAll ( f , builder ) ; } for ( Chunk chunk : builder . build ( ) ) { currentRowGroup . addColumn ( chunk . descriptor . col , chunk . readAllPages ( ) ) ; } // avoid re - reading bytes the dictionary reader is used after this call
if ( nextDictionaryReader != null ) { nextDictionaryReader . setRowGroup ( currentRowGroup ) ; } advanceToNextBlock ( ) ; return currentRowGroup ;


signature:
public class ParquetFileReader { /** * Reads all the columns requested from the row group at the current file position . It may skip specific pages based
 * on the column indexes according to the actual filter . As the rows are not aligned among the pages of the different
 * columns row synchronization might be required . See the documentation of the class SynchronizingColumnReader for
 * details .
 * @ return the PageReadStore which can provide PageReaders for each column
 * @ throws IOException
 * if any I / O error occurs while reading */
public PageReadStore readNextFilteredRowGroup ( ) throws IOException { } }

implementation:
if ( currentBlock == blocks . size ( ) ) { return null ; } if ( ! options . useColumnIndexFilter ( ) ) { return readNextRowGroup ( ) ; } BlockMetaData block = blocks . get ( currentBlock ) ; if ( block . getRowCount ( ) == 0 ) { throw new RuntimeException ( "Illegal row group of 0 rows" ) ; } ColumnIndexStore ciStore = getColumnIndexStore ( currentBlock ) ; RowRanges rowRanges = getRowRanges ( currentBlock ) ; long rowCount = rowRanges . rowCount ( ) ; if ( rowCount == 0 ) { // There are no matching rows - > skipping this row - group
advanceToNextBlock ( ) ; return readNextFilteredRowGroup ( ) ; } if ( rowCount == block . getRowCount ( ) ) { // All rows are matching - > fall back to the non - filtering path
return readNextRowGroup ( ) ; } this . currentRowGroup = new ColumnChunkPageReadStore ( rowRanges ) ; // prepare the list of consecutive parts to read them in one scan
ChunkListBuilder builder = new ChunkListBuilder ( ) ; List < ConsecutivePartList > allParts = new ArrayList < ConsecutivePartList > ( ) ; ConsecutivePartList currentParts = null ; for ( ColumnChunkMetaData mc : block . getColumns ( ) ) { ColumnPath pathKey = mc . getPath ( ) ; ColumnDescriptor columnDescriptor = paths . get ( pathKey ) ; if ( columnDescriptor != null ) { OffsetIndex offsetIndex = ciStore . getOffsetIndex ( mc . getPath ( ) ) ; OffsetIndex filteredOffsetIndex = filterOffsetIndex ( offsetIndex , rowRanges , block . getRowCount ( ) ) ; for ( OffsetRange range : calculateOffsetRanges ( filteredOffsetIndex , mc , offsetIndex . getOffset ( 0 ) ) ) { BenchmarkCounter . incrementTotalBytes ( range . getLength ( ) ) ; long startingPos = range . getOffset ( ) ; // first part or not consecutive = > new list
if ( currentParts == null || currentParts . endPos ( ) != startingPos ) { currentParts = new ConsecutivePartList ( startingPos ) ; allParts . add ( currentParts ) ; } ChunkDescriptor chunkDescriptor = new ChunkDescriptor ( columnDescriptor , mc , startingPos , ( int ) range . getLength ( ) ) ; currentParts . addChunk ( chunkDescriptor ) ; builder . setOffsetIndex ( chunkDescriptor , filteredOffsetIndex ) ; } } } // actually read all the chunks
for ( ConsecutivePartList consecutiveChunks : allParts ) { consecutiveChunks . readAll ( f , builder ) ; } for ( Chunk chunk : builder . build ( ) ) { currentRowGroup . addColumn ( chunk . descriptor . col , chunk . readAllPages ( ) ) ; } // avoid re - reading bytes the dictionary reader is used after this call
if ( nextDictionaryReader != null ) { nextDictionaryReader . setRowGroup ( currentRowGroup ) ; } advanceToNextBlock ( ) ; return currentRowGroup ;


signature:
public class ParquetFileReader { /** * Returns a { @ link DictionaryPageReadStore } for the row group that would be
 * returned by calling { @ link # readNextRowGroup ( ) } or skipped by calling
 * { @ link # skipNextRowGroup ( ) } .
 * @ return a DictionaryPageReadStore for the next row group */
public DictionaryPageReadStore getNextDictionaryReader ( ) { } }

implementation: if ( nextDictionaryReader == null && currentBlock < blocks . size ( ) ) { this . nextDictionaryReader = getDictionaryReader ( blocks . get ( currentBlock ) ) ; } return nextDictionaryReader ;


signature:
public class ParquetFileReader { /** * Reads and decompresses a dictionary page for the given column chunk .
 * Returns null if the given column chunk has no dictionary page .
 * @ param meta a column ' s ColumnChunkMetaData to read the dictionary from
 * @ return an uncompressed DictionaryPage or null
 * @ throws IOException if there is an error while reading the dictionary */
DictionaryPage readDictionary ( ColumnChunkMetaData meta ) throws IOException { } }

implementation:
if ( ! meta . getEncodings ( ) . contains ( Encoding . PLAIN_DICTIONARY ) && ! meta . getEncodings ( ) . contains ( Encoding . RLE_DICTIONARY ) ) { return null ; } // TODO : this should use getDictionaryPageOffset ( ) but it isn ' t reliable .
if ( f . getPos ( ) != meta . getStartingPos ( ) ) { f . seek ( meta . getStartingPos ( ) ) ; } PageHeader pageHeader = Util . readPageHeader ( f ) ; if ( ! pageHeader . isSetDictionary_page_header ( ) ) { return null ; // TODO : should this complain ?
} DictionaryPage compressedPage = readCompressedDictionary ( pageHeader , f ) ; BytesInputDecompressor decompressor = options . getCodecFactory ( ) . getDecompressor ( meta . getCodec ( ) ) ; return new DictionaryPage ( decompressor . decompress ( compressedPage . getBytes ( ) , compressedPage . getUncompressedSize ( ) ) , compressedPage . getDictionarySize ( ) , compressedPage . getEncoding ( ) ) ;


signature:
public class MemoryManager { /** * Add a new writer and its memory allocation to the memory manager .
 * @ param writer the new created writer
 * @ param allocation the requested buffer size */
synchronized void addWriter ( InternalParquetRecordWriter writer , Long allocation ) { } }

implementation: Long oldValue = writerList . get ( writer ) ; if ( oldValue == null ) { writerList . put ( writer , allocation ) ; } else { throw new IllegalArgumentException ( "[BUG] The Parquet Memory Manager should not add an " + "instance of InternalParquetRecordWriter more than once. The Manager already contains " + "the writer: " + writer ) ; } updateAllocation ( ) ;


signature:
public class MemoryManager { /** * Remove the given writer from the memory manager .
 * @ param writer the writer that has been closed */
synchronized void removeWriter ( InternalParquetRecordWriter writer ) { } }

implementation: if ( writerList . containsKey ( writer ) ) { writerList . remove ( writer ) ; } if ( ! writerList . isEmpty ( ) ) { updateAllocation ( ) ; }


signature:
public class MemoryManager { /** * Update the allocated size of each writer based on the current allocations and pool size . */
private void updateAllocation ( ) { } }

implementation:
long totalAllocations = 0 ; for ( Long allocation : writerList . values ( ) ) { totalAllocations += allocation ; } if ( totalAllocations <= totalMemoryPool ) { scale = 1.0 ; } else { scale = ( double ) totalMemoryPool / totalAllocations ; LOG . warn ( String . format ( "Total allocation exceeds %.2f%% (%,d bytes) of heap memory\n" + "Scaling row group sizes to %.2f%% for %d writers" , 100 * memoryPoolRatio , totalMemoryPool , 100 * scale , writerList . size ( ) ) ) ; for ( Runnable callBack : callBacks . values ( ) ) { // we do not really want to start a new thread here .
callBack . run ( ) ; } } int maxColCount = 0 ; for ( InternalParquetRecordWriter w : writerList . keySet ( ) ) { maxColCount = Math . max ( w . getSchema ( ) . getColumns ( ) . size ( ) , maxColCount ) ; } for ( Map . Entry < InternalParquetRecordWriter , Long > entry : writerList . entrySet ( ) ) { long newSize = ( long ) Math . floor ( entry . getValue ( ) * scale ) ; if ( scale < 1.0 && minMemoryAllocation > 0 && newSize < minMemoryAllocation ) { throw new ParquetRuntimeException ( String . format ( "New Memory allocation %d bytes" + " is smaller than the minimum allocation size of %d bytes." , newSize , minMemoryAllocation ) ) { } ; } entry . getKey ( ) . setRowGroupSizeThreshold ( newSize ) ; LOG . debug ( String . format ( "Adjust block size from %,d to %,d for writer: %s" , entry . getValue ( ) , newSize , entry . getKey ( ) ) ) ; }


signature:
public class MemoryManager { /** * Register callback and deduplicate it if any .
 * @ param callBackName the name of callback . It should be identical .
 * @ param callBack the callback passed in from upper layer , such as Hive . */
public void registerScaleCallBack ( String callBackName , Runnable callBack ) { } }

implementation: Preconditions . checkNotNull ( callBackName , "callBackName" ) ; Preconditions . checkNotNull ( callBack , "callBack" ) ; if ( callBacks . containsKey ( callBackName ) ) { throw new IllegalArgumentException ( "The callBackName " + callBackName + " is duplicated and has been registered already." ) ; } else { callBacks . put ( callBackName , callBack ) ; }


signature:
public class ParquetFileWriter { /** * start the file
 * @ throws IOException if there is an error while writing */
public void start ( ) throws IOException { } }

implementation: state = state . start ( ) ; LOG . debug ( "{}: start" , out . getPos ( ) ) ; out . write ( MAGIC ) ;


signature:
public class ParquetFileWriter { /** * start a block
 * @ param recordCount the record count in this block
 * @ throws IOException if there is an error while writing */
public void startBlock ( long recordCount ) throws IOException { } }

implementation:
state = state . startBlock ( ) ; LOG . debug ( "{}: start block" , out . getPos ( ) ) ; // out . write ( MAGIC ) ; / / TODO : add a magic delimiter
alignment . alignForRowGroup ( out ) ; currentBlock = new BlockMetaData ( ) ; currentRecordCount = recordCount ; currentColumnIndexes = new ArrayList < > ( ) ; currentOffsetIndexes = new ArrayList < > ( ) ;


signature:
public class ParquetFileWriter { /** * start a column inside a block
 * @ param descriptor the column descriptor
 * @ param valueCount the value count in this column
 * @ param compressionCodecName a compression codec name
 * @ throws IOException if there is an error while writing */
public void startColumn ( ColumnDescriptor descriptor , long valueCount , CompressionCodecName compressionCodecName ) throws IOException { } }

implementation:
state = state . startColumn ( ) ; encodingStatsBuilder . clear ( ) ; currentEncodings = new HashSet < Encoding > ( ) ; currentChunkPath = ColumnPath . get ( descriptor . getPath ( ) ) ; currentChunkType = descriptor . getPrimitiveType ( ) ; currentChunkCodec = compressionCodecName ; currentChunkValueCount = valueCount ; currentChunkFirstDataPage = out . getPos ( ) ; compressedLength = 0 ; uncompressedLength = 0 ; // The statistics will be copied from the first one added at writeDataPage ( s ) so we have the correct typed one
currentStatistics = null ; columnIndexBuilder = ColumnIndexBuilder . getBuilder ( currentChunkType , columnIndexTruncateLength ) ; offsetIndexBuilder = OffsetIndexBuilder . getBuilder ( ) ; firstPageOffset = - 1 ;


signature:
public class ParquetFileWriter { /** * writes a dictionary page page
 * @ param dictionaryPage the dictionary page
 * @ throws IOException if there is an error while writing */
public void writeDictionaryPage ( DictionaryPage dictionaryPage ) throws IOException { } }

implementation:
state = state . write ( ) ; LOG . debug ( "{}: write dictionary page: {} values" , out . getPos ( ) , dictionaryPage . getDictionarySize ( ) ) ; currentChunkDictionaryPageOffset = out . getPos ( ) ; int uncompressedSize = dictionaryPage . getUncompressedSize ( ) ; int compressedPageSize = ( int ) dictionaryPage . getBytes ( ) . size ( ) ; // TODO : fix casts
metadataConverter . writeDictionaryPageHeader ( uncompressedSize , compressedPageSize , dictionaryPage . getDictionarySize ( ) , dictionaryPage . getEncoding ( ) , out ) ; long headerSize = out . getPos ( ) - currentChunkDictionaryPageOffset ; this . uncompressedLength += uncompressedSize + headerSize ; this . compressedLength += compressedPageSize + headerSize ; LOG . debug ( "{}: write dictionary page content {}" , out . getPos ( ) , compressedPageSize ) ; dictionaryPage . getBytes ( ) . writeAllTo ( out ) ; encodingStatsBuilder . addDictEncoding ( dictionaryPage . getEncoding ( ) ) ; currentEncodings . add ( dictionaryPage . getEncoding ( ) ) ;


signature:
public class ParquetFileWriter { /** * writes a single page
 * @ param valueCount count of values
 * @ param uncompressedPageSize the size of the data once uncompressed
 * @ param bytes the compressed data for the page without header
 * @ param rlEncoding encoding of the repetition level
 * @ param dlEncoding encoding of the definition level
 * @ param valuesEncoding encoding of values
 * @ throws IOException if there is an error while writing */
@ Deprecated public void writeDataPage ( int valueCount , int uncompressedPageSize , BytesInput bytes , Encoding rlEncoding , Encoding dlEncoding , Encoding valuesEncoding ) throws IOException { } }

implementation:
state = state . write ( ) ; // We are unable to build indexes without rowCount so skip them for this column
offsetIndexBuilder = OffsetIndexBuilder . getNoOpBuilder ( ) ; columnIndexBuilder = ColumnIndexBuilder . getNoOpBuilder ( ) ; long beforeHeader = out . getPos ( ) ; LOG . debug ( "{}: write data page: {} values" , beforeHeader , valueCount ) ; int compressedPageSize = ( int ) bytes . size ( ) ; metadataConverter . writeDataPageV1Header ( uncompressedPageSize , compressedPageSize , valueCount , rlEncoding , dlEncoding , valuesEncoding , out ) ; long headerSize = out . getPos ( ) - beforeHeader ; this . uncompressedLength += uncompressedPageSize + headerSize ; this . compressedLength += compressedPageSize + headerSize ; LOG . debug ( "{}: write data page content {}" , out . getPos ( ) , compressedPageSize ) ; bytes . writeAllTo ( out ) ; encodingStatsBuilder . addDataEncoding ( valuesEncoding ) ; currentEncodings . add ( rlEncoding ) ; currentEncodings . add ( dlEncoding ) ; currentEncodings . add ( valuesEncoding ) ;


signature:
public class ParquetFileWriter { /** * writes a single page
 * @ param valueCount count of values
 * @ param uncompressedPageSize the size of the data once uncompressed
 * @ param bytes the compressed data for the page without header
 * @ param statistics statistics for the page
 * @ param rlEncoding encoding of the repetition level
 * @ param dlEncoding encoding of the definition level
 * @ param valuesEncoding encoding of values
 * @ throws IOException if there is an error while writing
 * @ deprecated this method does not support writing column indexes ; Use
 * { @ link # writeDataPage ( int , int , BytesInput , Statistics , long , Encoding , Encoding , Encoding ) } instead */
@ Deprecated public void writeDataPage ( int valueCount , int uncompressedPageSize , BytesInput bytes , Statistics statistics , Encoding rlEncoding , Encoding dlEncoding , Encoding valuesEncoding ) throws IOException { } }

implementation:
// We are unable to build indexes without rowCount so skip them for this column
offsetIndexBuilder = OffsetIndexBuilder . getNoOpBuilder ( ) ; columnIndexBuilder = ColumnIndexBuilder . getNoOpBuilder ( ) ; innerWriteDataPage ( valueCount , uncompressedPageSize , bytes , statistics , rlEncoding , dlEncoding , valuesEncoding ) ;


signature:
public class ParquetFileWriter { /** * Writes a single page
 * @ param valueCount count of values
 * @ param uncompressedPageSize the size of the data once uncompressed
 * @ param bytes the compressed data for the page without header
 * @ param statistics the statistics of the page
 * @ param rowCount the number of rows in the page
 * @ param rlEncoding encoding of the repetition level
 * @ param dlEncoding encoding of the definition level
 * @ param valuesEncoding encoding of values
 * @ throws IOException if any I / O error occurs during writing the file */
public void writeDataPage ( int valueCount , int uncompressedPageSize , BytesInput bytes , Statistics statistics , long rowCount , Encoding rlEncoding , Encoding dlEncoding , Encoding valuesEncoding ) throws IOException { } }

implementation: long beforeHeader = out . getPos ( ) ; innerWriteDataPage ( valueCount , uncompressedPageSize , bytes , statistics , rlEncoding , dlEncoding , valuesEncoding ) ; offsetIndexBuilder . add ( ( int ) ( out . getPos ( ) - beforeHeader ) , rowCount ) ;