The underlying memory management is done by the BufferedBlockMgr. More...

#include <buffered-tuple-stream.h>

Collaboration diagram for impala::BufferedTupleStream:

Classes
struct	RowIdx

Public Member Functions
	BufferedTupleStream (RuntimeState state, const RowDescriptor &row_desc, BufferedBlockMgr block_mgr, BufferedBlockMgr::Client *client, bool use_initial_small_buffers=true, bool delete_on_read=false, bool read_write=false)

Status	Init (RuntimeProfile *profile=NULL, bool pinned=true)

Status	SwitchToIoBuffers (bool *got_buffer)

bool	AddRow (TupleRow row, uint8_t *dst=NULL)

uint8_t *	AllocateRow (int size)

void	GetTupleRow (const RowIdx &idx, TupleRow *row) const

Status	PrepareForRead (bool *got_buffer=NULL)

Status	PinStream (bool already_reserved, bool *pinned)

Status	UnpinStream (bool all=false)

Status	GetNext (RowBatch batch, bool eos, std::vector< RowIdx > *indices=NULL)

Status	GetRows (boost::scoped_ptr< RowBatch > batch, bool got_rows)

void	Close ()
	Must be called once at the end to cleanup all resources. Idempotent. More...

Status	status () const

int64_t	num_rows () const
	Number of rows in the stream. More...

int64_t	rows_returned () const
	Number of rows returned via GetNext(). More...

int64_t	byte_size () const
	Returns the byte size necessary to store the entire stream in memory. More...

int64_t	bytes_in_mem (bool ignore_current) const

bool	is_pinned () const

int	blocks_pinned () const

int	blocks_unpinned () const

bool	has_read_block () const

bool	has_write_block () const

bool	using_small_buffers () const

std::string	DebugString () const

template<bool HasNullableTuple>
Status	GetNextInternal (RowBatch batch, bool eos, vector< RowIdx > *indices)

Private Member Functions
template<bool HasNullableTuple>
bool	DeepCopyInternal (TupleRow row, uint8_t *dst)

bool	DeepCopy (TupleRow row, uint8_t *dst)
	Wrapper of the templated DeepCopyInternal() function. More...

Status	NewBlockForWrite (int min_size, bool *got_block)

Status	NextBlockForRead ()

int	ComputeRowSize (TupleRow *row) const
	Returns the byte size of this row when encoded in a block. More...

Status	UnpinBlock (BufferedBlockMgr::Block *block)
	Unpins block if it is an io sized block and updates tracking stats. More...

template<bool HasNullableTuple>
Status	GetNextInternal (RowBatch batch, bool eos, std::vector< RowIdx > *indices)
	Templated GetNext implementation. More...

int	ComputeNumNullIndicatorBytes (int block_size) const
	Computes the number of bytes needed for null indicators for a block of 'block_size'. More...

Private Attributes
bool	use_small_buffers_
	If true, this stream is still using small buffers. More...

const bool	delete_on_read_
	If true, blocks are deleted after they are read. More...

const bool	read_write_

RuntimeState *const	state_
	Runtime state instance used to check for cancellation. Not owned. More...

const RowDescriptor &	desc_
	Description of rows stored in the stream. More...

const bool	nullable_tuple_
	Whether any tuple in the rows is nullable. More...

int	fixed_tuple_row_size_
	Sum of the fixed length portion of all the tuples in desc_. More...

uint32_t	null_indicators_read_block_

uint32_t	null_indicators_write_block_

std::vector< std::pair< int, std::vector< SlotDescriptor * > > >	string_slots_
	Vector of all the strings slots grouped by tuple_idx. More...

BufferedBlockMgr *	block_mgr_
	Block manager and client used to allocate, pin and release blocks. Not owned. More...

BufferedBlockMgr::Client *	block_mgr_client_

std::list < BufferedBlockMgr::Block * >	blocks_
	List of blocks in the stream. More...

int64_t	total_byte_size_
	Total size of blocks_, including small blocks. More...

std::list < BufferedBlockMgr::Block * > ::iterator	read_block_

std::vector< uint8_t * >	block_start_idx_

uint8_t *	read_ptr_
	Current ptr offset in read_block_'s buffer. More...

uint32_t	read_tuple_idx_
	Current idx of the tuple read from the read_block_ buffer. More...

uint32_t	write_tuple_idx_
	Current idx of the tuple written at the write_block_ buffer. More...

int64_t	read_bytes_
	Bytes read in read_block_. More...

int64_t	rows_returned_
	Number of rows returned to the caller from GetNext(). More...

int	read_block_idx_
	The block index of the current read block. More...

BufferedBlockMgr::Block *	write_block_
	The current block for writing. NULL if there is no available block to write to. More...

int	num_pinned_

int	num_small_blocks_
	The total number of small blocks in blocks_;. More...

bool	closed_

Status	status_

int64_t	num_rows_
	Number of rows stored in the stream. More...

bool	pinned_

RuntimeProfile::Counter *	pin_timer_
	Counters added by this object to the parent runtime profile. More...

RuntimeProfile::Counter *	unpin_timer_

RuntimeProfile::Counter *	get_new_block_timer_

Detailed Description

The underlying memory management is done by the BufferedBlockMgr.

Class that provides an abstraction for a stream of tuple rows. Rows can be added to the stream and returned. Rows are returned in the order they are added.The tuple stream consists of a number of small (less than io sized blocks) before an arbitrary number of io sized blocks. The smaller blocks do not spill and are there to lower the minimum buffering requirements. For example, an operator that needs to maintain 64 streams (1 buffer per partition) would need, by default, 64 * 8MB = 512MB of buffering. A query with 5 of these operators would require 2.56 GB just to run any query, regardless of how much of that is used. This is problematic for small queries. Instead we will start with a fixed number of small buffers and only start using IO sized buffers when those fill up. The small buffers never spill. The stream will not automatically switch from using small buffers to io sized buffers. The BufferedTupleStream is not thread safe from the caller's point of view. It is expected that all the APIs are called from a single thread. Internally, the object is thread safe wrt to the underlying block mgr. Buffer management: The stream is either pinned or unpinned, set via PinStream() and UnpinStream(). Blocks are optionally deleted as they are read, set with the delete_on_read c'tor parameter. Block layout: At the header of each block, starting at position 0, there is a bitstring with null indicators for all the tuples in each row in the block. Then there are the tuple rows. We further optimize the codepaths when we know that no tuple is nullable, indicated by 'nullable_tuple_'. Tuple row layout: Tuples are stored back to back. Each tuple starts with the fixed length portion, directly followed by the var len portion. (Fixed len and var len are interleaved). If any tuple in the row is nullable, then there is a bitstring of null tuple indicators at the header of the block. The order of bits in the null indicators bitstring corresponds to the order of tuples in the block. The NULL tuples are not stored in the body of the block, only as set bits in the null indicators bitsting. The behavior of reads and writes is as follows: Read:

Delete on read (delete_on_read_): Blocks are deleted as we go through the stream. The data returned by the tuple stream is valid until the next read call so the caller does not need to copy if it is streaming.
Unpinned: Blocks remain in blocks_ and are unpinned after reading.
Pinned: Blocks remain in blocks_ and are left pinned after reading. If the next block in the stream cannot be pinned, the read call will fail and the caller needs to free memory from the underlying block mgr. Write:

Unpinned: Unpin blocks as they fill up. This means only a single (i.e. the current) block needs to be in memory regardless of the input size (if read_write is true, then two blocks need to be in memory).
Pinned: Blocks are left pinned. If we run out of blocks, the write will fail and the caller needs to free memory from the underlying block mgr. TODO: we need to be able to do read ahead in the BufferedBlockMgr. It currently only has PinAllBlocks() which is blocking. We need a non-blocking version of this or some way to indicate a block will need to be pinned soon. TODO: see if this can be merged with Sorter::Run. The key difference is that this does not need to return rows in the order they were added, which allows it to be simpler. TODO: we could compact the small buffers when we need to spill but they use very little memory so ths might not be very useful. TODO: improvements:

Think about how to layout for the var len data more, possibly filling in them from the end of the same block. Don't interleave fixed and var len data.
It would be good to allocate the null indicators at the end of each block and grow this array as new rows are inserted in the block. If we do so, then there will be fewer gaps in case of many rows with NULL tuples.
We will want to multithread this. Add a AddBlock() call so the synchronization happens at the block level. This is a natural extension.
Instead of allocating all blocks from the block_mgr, allocate some blocks that are much smaller (e.g. 16K and doubling up to the block size). This way, very small streams (a common case) will use very little memory. This small blocks are always in memory since spilling them frees up negligible memory.
Return row batches in GetNext() instead of filling one in
Should we 32-bit align the start of the tuple rows? Now it is byte-aligned.

Definition at line 109 of file buffered-tuple-stream.h.

Constructor & Destructor Documentation

BufferedTupleStream::BufferedTupleStream	(	RuntimeState *	state,
		const RowDescriptor &	row_desc,
		BufferedBlockMgr *	block_mgr,
		BufferedBlockMgr::Client *	client,
		bool	use_initial_small_buffers = `true`,
		bool	delete_on_read = `false`,
		bool	read_write = `false`
	)

row_desc: description of rows stored in the stream. This is the desc for rows that are added and the rows being returned. block_mgr: Underlying block mgr that owns the data blocks. delete_on_read: Blocks are deleted after they are read. use_initial_small_buffers: If true, the initial N buffers allocated for the tuple stream use smaller than io sized buffers. read_write: Stream allows interchanging read and write operations. Requires at least two blocks may be pinned.

Definition at line 43 of file buffered-tuple-stream.cc.

References blocks_, impala::TupleDescriptor::byte_size(), desc_, fixed_tuple_row_size_, null_indicators_read_block_, null_indicators_write_block_, read_block_, impala::TupleDescriptor::string_slots(), string_slots_, and impala::RowDescriptor::tuple_descriptors().

Member Function Documentation

bool impala::BufferedTupleStream::AddRow	(	TupleRow *	row,
		uint8_t **	dst = `NULL`
	)

inline

Adds a single row to the stream. Returns false if an error occurred. BufferedTupleStream will do a deep copy of the memory in the row. *dst is the ptr to the memory (in the underlying block) that this row was copied to.

Definition at line 25 of file buffered-tuple-stream.inline.h.

References closed_, ComputeRowSize(), DeepCopy(), LIKELY, NewBlockForWrite(), impala::Status::ok(), and status_.

Referenced by impala::PartitionedHashJoinNode::AppendRow(), impala::PartitionedHashJoinNode::AppendRowStreamFull(), impala::PartitionedAggregationNode::ProcessBatch(), impala::PartitionedHashJoinNode::ProcessBuildBatch(), impala::PartitionedHashJoinNode::ProcessProbeBatch(), and impala::PartitionedAggregationNode::Partition::Spill().

uint8_t * impala::BufferedTupleStream::AllocateRow ( int size )

inline

Allocates space to store a row of size 'size'. Returns NULL if there is not enough memory. The returned memory is guaranteed to fit on one block.

Definition at line 34 of file buffered-tuple-stream.inline.h.

References impala::BufferedBlockMgr::Block::AddRow(), impala::BufferedBlockMgr::Block::Allocate(), impala::BufferedBlockMgr::Block::BytesRemaining(), closed_, impala::BufferedBlockMgr::Block::is_pinned(), NewBlockForWrite(), num_rows_, impala::Status::ok(), status_, UNLIKELY, and write_block_.

Referenced by impala::PartitionedAggregationNode::ConstructIntermediateTuple().

int impala::BufferedTupleStream::blocks_pinned ( ) const

inline

Definition at line 244 of file buffered-tuple-stream.h.

References num_pinned_.

Referenced by impala::PartitionedHashJoinNode::NodeDebugString(), and impala::PartitionedHashJoinNode::ProcessBuildInput().

int impala::BufferedTupleStream::blocks_unpinned ( ) const

inline

Definition at line 245 of file buffered-tuple-stream.h.

References blocks_, num_pinned_, and num_small_blocks_.

Referenced by PinStream().

int64_t impala::BufferedTupleStream::byte_size ( ) const

inline

Returns the byte size necessary to store the entire stream in memory.

Definition at line 237 of file buffered-tuple-stream.h.

References total_byte_size_.

Referenced by impala::RowBatch::AddTupleStream().

int64_t BufferedTupleStream::bytes_in_mem ( bool ignore_current ) const

Returns the byte size of the stream that is currently pinned in memory. If ignore_current is true, the write_block_ memory is not included.

Definition at line 156 of file buffered-tuple-stream.cc.

References blocks_, and write_block_.

void BufferedTupleStream::Close ( )

Must be called once at the end to cleanup all resources. Idempotent.

Definition at line 145 of file buffered-tuple-stream.cc.

References blocks_, closed_, num_pinned_, and NumPinned().

Referenced by impala::PartitionedHashJoinNode::BuildHashTables(), impala::PartitionedHashJoinNode::Close(), impala::PartitionedHashJoinNode::ProcessBuildInput(), and impala::PartitionedAggregationNode::ProcessStream().

int BufferedTupleStream::ComputeNumNullIndicatorBytes ( int block_size ) const

private

Computes the number of bytes needed for null indicators for a block of 'block_size'.

Definition at line 415 of file buffered-tuple-stream.cc.

References desc_, fixed_tuple_row_size_, nullable_tuple_, impala::BitUtil::RoundUpNumi64(), and impala::RowDescriptor::tuple_descriptors().

Referenced by NewBlockForWrite(), NextBlockForRead(), and PrepareForRead().

int BufferedTupleStream::ComputeRowSize ( TupleRow * row ) const

private

Returns the byte size of this row when encoded in a block.

Definition at line 595 of file buffered-tuple-stream.cc.

References fixed_tuple_row_size_, impala::Tuple::GetStringSlot(), impala::TupleRow::GetTuple(), impala::Tuple::IsNull(), impala::StringValue::len, impala::SlotDescriptor::null_indicator_offset(), nullable_tuple_, string_slots_, and impala::SlotDescriptor::tuple_offset().

Referenced by AddRow().

string BufferedTupleStream::DebugString ( ) const

Definition at line 93 of file buffered-tuple-stream.cc.

References blocks_, closed_, delete_on_read_, num_pinned_, num_rows_, pinned_, read_block_, rows_returned_, and write_block_.

Referenced by DeepCopyInternal(), GetNextInternal(), and NextBlockForRead().

bool BufferedTupleStream::DeepCopy	(	TupleRow *	row,
		uint8_t **	dst
	)

private

Wrapper of the templated DeepCopyInternal() function.

Definition at line 22 of file buffered-tuple-stream-ir.cc.

References nullable_tuple_.

Referenced by AddRow().

template<bool HasNullableTuple>

bool BufferedTupleStream::DeepCopyInternal	(	TupleRow *	row,
		uint8_t **	dst
	)

private

Copies 'row' into write_block_. Returns false if there is not enough space in 'write_block_'. *dst is the ptr to the memory (in the underlying write block) where this row was copied to.

Definition at line 32 of file buffered-tuple-stream-ir.cc.

References impala::BufferedBlockMgr::Block::AddRow(), impala::BufferedBlockMgr::Block::Allocate(), impala::BufferedBlockMgr::Block::buffer(), impala::BufferedBlockMgr::Block::BytesRemaining(), impala::BufferedBlockMgr::Block::DebugString(), DebugString(), desc_, fixed_tuple_row_size_, impala::Tuple::GetStringSlot(), impala::TupleRow::GetTuple(), impala::BufferedBlockMgr::Block::is_pinned(), impala::Tuple::IsNull(), impala::StringValue::len, LIKELY, impala::SlotDescriptor::null_indicator_offset(), null_indicators_write_block_, num_rows_, impala::StringValue::ptr, impala::BufferedBlockMgr::Block::ReturnAllocation(), string_slots_, impala::RowDescriptor::tuple_descriptors(), impala::SlotDescriptor::tuple_offset(), UNLIKELY, write_block_, and write_tuple_idx_.

Status BufferedTupleStream::GetNext	(	RowBatch *	batch,
		bool *	eos,
		std::vector< RowIdx > *	indices = `NULL`
	)

Get the next batch of output rows. Memory is still owned by the BufferedTupleStream and must be copied out by the caller. If 'indices' is non-NULL, that is also populated for each returned row with the index for that row.

Definition at line 447 of file buffered-tuple-stream.cc.

References nullable_tuple_.

Referenced by GetRows(), impala::PartitionedHashJoinNode::NextSpilledProbeRowBatch(), impala::PartitionedHashJoinNode::OutputNullAwareNullProbe(), impala::PartitionedHashJoinNode::ProcessBuildInput(), impala::PartitionedAggregationNode::ProcessStream(), and impala::SimpleTupleStreamTest::ReadValues().

template<bool HasNullableTuple>

Status impala::BufferedTupleStream::GetNextInternal	(	RowBatch *	batch,
		bool *	eos,
		std::vector< RowIdx > *	indices
	)

private

Templated GetNext implementation.

template<bool HasNullableTuple>

Status impala::BufferedTupleStream::GetNextInternal	(	RowBatch *	batch,
		bool *	eos,
		vector< RowIdx > *	indices
	)

Definition at line 457 of file buffered-tuple-stream.cc.

References impala::RowBatch::AddRows(), blocks_, impala::RowBatch::capacity(), closed_, impala::RowBatch::CommitRows(), DebugString(), delete_on_read_, desc_, impala::RowDescriptor::Equals(), fixed_tuple_row_size_, impala::RowBatch::GetRow(), impala::Tuple::GetStringSlot(), impala::TupleRow::GetTuple(), is_pinned(), impala::Tuple::IsNull(), impala::StringValue::len, impala::RowBatch::MarkNeedToReturn(), NextBlockForRead(), impala::SlotDescriptor::null_indicator_offset(), null_indicators_read_block_, impala::RowBatch::num_rows(), num_rows_, impala::Status::OK, pinned_, impala::StringValue::ptr, read_block_, read_block_idx_, read_bytes_, read_ptr_, read_tuple_idx_, RETURN_IF_ERROR, impala::RowBatch::row_desc(), rows_returned_, impala::TupleRow::SetTuple(), string_slots_, impala::RowDescriptor::tuple_descriptors(), impala::SlotDescriptor::tuple_offset(), and UNLIKELY.

Status BufferedTupleStream::GetRows	(	boost::scoped_ptr< RowBatch > *	batch,
		bool *	got_rows
	)

Returns all the rows in the stream in batch. This pins the entire stream in the process. *got_rows is false if the stream could not be pinned.

Definition at line 431 of file buffered-tuple-stream.cc.

References block_mgr_, block_mgr_client_, desc_, impala::BufferedBlockMgr::get_tracker(), GetNext(), num_rows(), impala::Status::OK, PinStream(), PrepareForRead(), and RETURN_IF_ERROR.

Referenced by impala::PartitionedHashJoinNode::EvaluateNullProbe().

void impala::BufferedTupleStream::GetTupleRow	(	const RowIdx &	idx,
		TupleRow *	row
	)		const

inline

Populates 'row' with the row at 'idx'. The stream must be pinned. The row must have been allocated with the stream's row desc.

Definition at line 49 of file buffered-tuple-stream.inline.h.

References impala::BufferedTupleStream::RowIdx::block(), block_start_idx_, blocks_, closed_, delete_on_read_, desc_, impala::BufferedTupleStream::RowIdx::idx(), is_pinned(), nullable_tuple_, impala::BufferedTupleStream::RowIdx::offset(), impala::TupleRow::SetTuple(), and impala::RowDescriptor::tuple_descriptors().

Referenced by impala::HashTable::GetRow().

bool impala::BufferedTupleStream::has_read_block ( ) const

inline

Definition at line 246 of file buffered-tuple-stream.h.

References blocks_, and read_block_.

bool impala::BufferedTupleStream::has_write_block ( ) const

inline

Definition at line 247 of file buffered-tuple-stream.h.

References write_block_.

Referenced by impala::PartitionedHashJoinNode::BuildHashTables(), and impala::PartitionedAggregationNode::ProcessBatch().

Status BufferedTupleStream::Init	(	RuntimeProfile *	profile = `NULL`,
		bool	pinned = `true`
	)

Initializes the tuple stream object. Must be called once before any of the other APIs. If pinned, the tuple stream starts of pinned, otherwise it is unpinned. If profile is non-NULL, counters are created.

Definition at line 116 of file buffered-tuple-stream.cc.

References ADD_TIMER, block_mgr_, block_mgr_client_, fixed_tuple_row_size_, get_new_block_timer_, INITIAL_BLOCK_SIZES, impala::BufferedBlockMgr::max_block_size(), impala::BufferedBlockMgr::MemLimitTooLowError(), NewBlockForWrite(), impala::Status::OK, pin_timer_, PrepareForRead(), read_write_, RETURN_IF_ERROR, unpin_timer_, UnpinStream(), use_small_buffers_, and write_block_.

Referenced by impala::PartitionedHashJoinNode::Prepare().

bool impala::BufferedTupleStream::is_pinned ( ) const

inline

Definition at line 243 of file buffered-tuple-stream.h.

References pinned_.

Referenced by impala::PartitionedHashJoinNode::BuildHashTables(), GetNextInternal(), GetTupleRow(), and impala::PartitionedAggregationNode::ProcessBatch().

Status BufferedTupleStream::NewBlockForWrite	(	int	min_size,
		bool *	got_block
	)

private

Gets a new block from the block_mgr_, updating write_block_ and write_tuple_idx_, and setting *got_block. If there are no blocks available, *got_block is set to false and write_block_ is unchanged. min_size is the minimum number of bytes required for this block.

Definition at line 178 of file buffered-tuple-stream.cc.

References impala::BufferedBlockMgr::Block::Allocate(), block_mgr_, block_mgr_client_, block_start_idx_, blocks_, impala::BufferedBlockMgr::Block::buffer(), closed_, ComputeNumNullIndicatorBytes(), get_new_block_timer_, impala::BufferedBlockMgr::GetNewBlock(), INITIAL_BLOCK_SIZES, impala::BufferedBlockMgr::Block::is_max_size(), impala::BufferedBlockMgr::Block::is_pinned(), impala::BufferedBlockMgr::max_block_size(), null_indicators_write_block_, num_pinned_, impala::BufferedBlockMgr::Block::num_rows(), NUM_SMALL_BLOCKS, num_small_blocks_, NumPinned(), impala::Status::OK, pinned_, impala::PrettyPrinter::Print(), read_block_, RETURN_IF_ERROR, SCOPED_TIMER, total_byte_size_, use_small_buffers_, write_block_, and write_tuple_idx_.

Referenced by AddRow(), AllocateRow(), Init(), and SwitchToIoBuffers().

Status BufferedTupleStream::NextBlockForRead ( )

private

Reads the next block from the block_mgr_. This blocks if necessary. Updates read_block_, read_ptr_, read_tuple_idx_ and read_bytes_.

Definition at line 248 of file buffered-tuple-stream.cc.

References block_mgr_, block_mgr_client_, blocks_, closed_, ComputeNumNullIndicatorBytes(), DebugString(), impala::BufferedBlockMgr::DebugString(), impala::BufferedBlockMgr::Block::Delete(), delete_on_read_, impala::BufferedBlockMgr::Block::is_max_size(), null_indicators_read_block_, num_pinned_, NumPinned(), impala::Status::OK, pin_timer_, pinned_, read_block_, read_block_idx_, read_bytes_, read_ptr_, read_tuple_idx_, RETURN_IF_ERROR, SCOPED_TIMER, unpin_timer_, UnpinBlock(), and write_block_.

Referenced by GetNextInternal().

int64_t impala::BufferedTupleStream::num_rows ( ) const

inline

Number of rows in the stream.

Definition at line 231 of file buffered-tuple-stream.h.

References num_rows_.

Status BufferedTupleStream::PinStream	(	bool	already_reserved,
		bool *	pinned
	)

Pins all blocks in this stream and switches to pinned mode. If there is not enough memory, *pinned is set to false and the stream is unmodified. If already_reserved is true, the caller has already made a reservation on block_mgr_client_ to pin the stream.

Definition at line 357 of file buffered-tuple-stream.cc.

References block_mgr_, block_mgr_client_, block_start_idx_, blocks_, blocks_unpinned(), closed_, impala::BufferedBlockMgr::DebugString(), delete_on_read_, num_pinned_, NumPinned(), impala::Status::OK, pin_timer_, pinned_, RETURN_IF_ERROR, SCOPED_TIMER, impala::BufferedBlockMgr::TryAcquireTmpReservation(), and VLOG_QUERY.

Referenced by GetRows().

Status BufferedTupleStream::PrepareForRead ( bool * got_buffer = NULL )

Prepares the stream for reading. If read_write_, this does not need to be called in order to begin reading, otherwise this must be called after the last AddRow() and before GetNext(). If got_buffer is NULL, this function will fail (with a bad status) if no buffer is available. If got_buffer is non-null, this function will not fail on OOM and *got_buffer is true if a buffer was pinned.

Definition at line 314 of file buffered-tuple-stream.cc.

References blocks_, closed_, ComputeNumNullIndicatorBytes(), impala::BufferedBlockMgr::Block::is_pinned(), null_indicators_read_block_, num_pinned_, NumPinned(), impala::Status::OK, pin_timer_, pinned_, read_block_, read_block_idx_, read_bytes_, read_ptr_, read_tuple_idx_, read_write_, RETURN_IF_ERROR, rows_returned_, SCOPED_TIMER, UnpinBlock(), and write_block_.

Referenced by GetRows(), Init(), impala::PartitionedHashJoinNode::PrepareNextPartition(), impala::PartitionedHashJoinNode::PrepareNullAwareNullProbe(), impala::PartitionedHashJoinNode::PrepareNullAwarePartition(), impala::PartitionedHashJoinNode::ProcessBuildInput(), and impala::PartitionedAggregationNode::ProcessStream().

int64_t impala::BufferedTupleStream::rows_returned ( ) const

inline

Number of rows returned via GetNext().

Definition at line 234 of file buffered-tuple-stream.h.

References rows_returned_.

Referenced by impala::PartitionedHashJoinNode::NextSpilledProbeRowBatch().

Status impala::BufferedTupleStream::status ( ) const

inline

Returns the status of the stream. We don't want to return a more costly Status object on AddRow() which is way that API returns a bool.

Definition at line 228 of file buffered-tuple-stream.h.

References status_.

Referenced by impala::PartitionedHashJoinNode::AppendRowStreamFull(), impala::PartitionedAggregationNode::ProcessBatch(), impala::PartitionedHashJoinNode::ProcessBuildBatch(), impala::PartitionedHashJoinNode::ProcessProbeBatch(), and impala::PartitionedAggregationNode::Partition::Spill().

Status BufferedTupleStream::SwitchToIoBuffers ( bool * got_buffer )

Must be called for streams using small buffers to switch to IO-sized buffers. TODO: this does not seem like the best mechanism.

Definition at line 136 of file buffered-tuple-stream.cc.

References block_mgr_, impala::BufferedBlockMgr::max_block_size(), NewBlockForWrite(), impala::Status::OK, use_small_buffers_, and write_block_.

Referenced by impala::PartitionedHashJoinNode::BuildHashTables().

Status BufferedTupleStream::UnpinBlock ( BufferedBlockMgr::Block * block )

private

Unpins block if it is an io sized block and updates tracking stats.

Definition at line 168 of file buffered-tuple-stream.cc.

References blocks_, impala::BufferedBlockMgr::Block::is_max_size(), impala::BufferedBlockMgr::Block::is_pinned(), num_pinned_, NumPinned(), impala::Status::OK, RETURN_IF_ERROR, SCOPED_TIMER, impala::BufferedBlockMgr::Block::Unpin(), and unpin_timer_.

Referenced by NextBlockForRead(), PrepareForRead(), and UnpinStream().

Status BufferedTupleStream::UnpinStream ( bool all = false )

Unpins stream. If all is true, all blocks are unpinned, otherwise all blocks except the write_block_ and read_block_ are unpinned.

Definition at line 396 of file buffered-tuple-stream.cc.

References blocks_, closed_, impala::BufferedBlockMgr::Block::is_pinned(), impala::Status::OK, pinned_, read_block_, read_write_, RETURN_IF_ERROR, SCOPED_TIMER, unpin_timer_, UnpinBlock(), and write_block_.

Referenced by impala::PartitionedHashJoinNode::CleanUpHashPartitions(), and Init().

bool impala::BufferedTupleStream::using_small_buffers ( ) const

inline

Definition at line 248 of file buffered-tuple-stream.h.

References use_small_buffers_.

Referenced by impala::PartitionedHashJoinNode::BuildHashTables().

Member Data Documentation

BufferedBlockMgr* impala::BufferedTupleStream::block_mgr_

private

Block manager and client used to allocate, pin and release blocks. Not owned.

Definition at line 288 of file buffered-tuple-stream.h.

Referenced by GetRows(), Init(), NewBlockForWrite(), NextBlockForRead(), PinStream(), and SwitchToIoBuffers().

BufferedBlockMgr::Client* impala::BufferedTupleStream::block_mgr_client_

private

Definition at line 289 of file buffered-tuple-stream.h.

Referenced by GetRows(), Init(), NewBlockForWrite(), NextBlockForRead(), and PinStream().

std::vector<uint8_t*> impala::BufferedTupleStream::block_start_idx_

private

For each block in the stream, the buffer of the start of the block. This is only valid when the stream is pinned, giving random access to data in the stream. This is not maintained for delete_on_read_.

Definition at line 304 of file buffered-tuple-stream.h.

Referenced by GetTupleRow(), NewBlockForWrite(), and PinStream().

std::list<BufferedBlockMgr::Block*> impala::BufferedTupleStream::blocks_

private

List of blocks in the stream.

Definition at line 292 of file buffered-tuple-stream.h.

Referenced by blocks_unpinned(), BufferedTupleStream(), bytes_in_mem(), Close(), DebugString(), GetNextInternal(), GetTupleRow(), has_read_block(), NewBlockForWrite(), NextBlockForRead(), PinStream(), PrepareForRead(), UnpinBlock(), and UnpinStream().

bool impala::BufferedTupleStream::closed_

private

Definition at line 335 of file buffered-tuple-stream.h.

Referenced by AddRow(), AllocateRow(), Close(), DebugString(), GetNextInternal(), GetTupleRow(), NewBlockForWrite(), NextBlockForRead(), PinStream(), PrepareForRead(), and UnpinStream().

const bool impala::BufferedTupleStream::delete_on_read_

private

If true, blocks are deleted after they are read.

Definition at line 257 of file buffered-tuple-stream.h.

Referenced by DebugString(), GetNextInternal(), GetTupleRow(), NextBlockForRead(), and PinStream().

const RowDescriptor& impala::BufferedTupleStream::desc_

private

Description of rows stored in the stream.

Definition at line 268 of file buffered-tuple-stream.h.

Referenced by BufferedTupleStream(), ComputeNumNullIndicatorBytes(), DeepCopyInternal(), GetNextInternal(), GetRows(), and GetTupleRow().

int impala::BufferedTupleStream::fixed_tuple_row_size_

private

Sum of the fixed length portion of all the tuples in desc_.

Definition at line 274 of file buffered-tuple-stream.h.

Referenced by BufferedTupleStream(), ComputeNumNullIndicatorBytes(), ComputeRowSize(), DeepCopyInternal(), GetNextInternal(), and Init().

RuntimeProfile::Counter* impala::BufferedTupleStream::get_new_block_timer_

private

Definition at line 350 of file buffered-tuple-stream.h.

Referenced by Init(), and NewBlockForWrite().

uint32_t impala::BufferedTupleStream::null_indicators_read_block_

private

Max size (in bytes) of null indicators bitstring in the current read and write blocks. If 0, it means that there is no need to store null indicators for this RowDesc. We calculate this value based on the block's size and the fixed_tuple_row_size_. When not 0, this value is also an upper bound for the number of (rows * tuples_per_row) in this block.

Definition at line 281 of file buffered-tuple-stream.h.

Referenced by BufferedTupleStream(), GetNextInternal(), NextBlockForRead(), and PrepareForRead().

uint32_t impala::BufferedTupleStream::null_indicators_write_block_

private

Definition at line 282 of file buffered-tuple-stream.h.

Referenced by BufferedTupleStream(), DeepCopyInternal(), and NewBlockForWrite().

const bool impala::BufferedTupleStream::nullable_tuple_

private

Whether any tuple in the rows is nullable.

Definition at line 271 of file buffered-tuple-stream.h.

Referenced by ComputeNumNullIndicatorBytes(), ComputeRowSize(), DeepCopy(), GetNext(), and GetTupleRow().

int impala::BufferedTupleStream::num_pinned_

private

Number of pinned blocks in blocks_, stored to avoid iterating over the list to compute bytes_in_mem and bytes_unpinned. This does not include small blocks.

Definition at line 330 of file buffered-tuple-stream.h.

Referenced by blocks_pinned(), blocks_unpinned(), Close(), DebugString(), NewBlockForWrite(), NextBlockForRead(), PinStream(), PrepareForRead(), and UnpinBlock().

int64_t impala::BufferedTupleStream::num_rows_

private

Number of rows stored in the stream.

Definition at line 339 of file buffered-tuple-stream.h.

Referenced by AllocateRow(), DebugString(), DeepCopyInternal(), GetNextInternal(), and num_rows().

int impala::BufferedTupleStream::num_small_blocks_

private

The total number of small blocks in blocks_;.

Definition at line 333 of file buffered-tuple-stream.h.

Referenced by blocks_unpinned(), and NewBlockForWrite().

RuntimeProfile::Counter* impala::BufferedTupleStream::pin_timer_

private

Counters added by this object to the parent runtime profile.

Definition at line 348 of file buffered-tuple-stream.h.

Referenced by Init(), NextBlockForRead(), PinStream(), and PrepareForRead().

bool impala::BufferedTupleStream::pinned_

private

If true, this stream has been explicitly pinned by the caller. This changes the memory management of the stream. The blocks are not unpinned until the caller calls UnpinAllBlocks(). If false, only the write_block_ and/or read_block_ are pinned (both are if read_write_ is true).

Definition at line 345 of file buffered-tuple-stream.h.

Referenced by DebugString(), GetNextInternal(), is_pinned(), NewBlockForWrite(), NextBlockForRead(), PinStream(), PrepareForRead(), and UnpinStream().

std::list<BufferedBlockMgr::Block*>::iterator impala::BufferedTupleStream::read_block_

private

Iterator pointing to the current block for read. If read_write_, this is always a valid block, otherwise equal to list.end() until PrepareForRead() is called.

Definition at line 299 of file buffered-tuple-stream.h.

Referenced by BufferedTupleStream(), DebugString(), GetNextInternal(), has_read_block(), NewBlockForWrite(), NextBlockForRead(), PrepareForRead(), and UnpinStream().

int impala::BufferedTupleStream::read_block_idx_

private

The block index of the current read block.

Definition at line 322 of file buffered-tuple-stream.h.

Referenced by GetNextInternal(), NextBlockForRead(), and PrepareForRead().

int64_t impala::BufferedTupleStream::read_bytes_

private

Bytes read in read_block_.

Definition at line 316 of file buffered-tuple-stream.h.

Referenced by GetNextInternal(), NextBlockForRead(), and PrepareForRead().

uint8_t* impala::BufferedTupleStream::read_ptr_

private

Current ptr offset in read_block_'s buffer.

Definition at line 307 of file buffered-tuple-stream.h.

Referenced by GetNextInternal(), NextBlockForRead(), and PrepareForRead().

uint32_t impala::BufferedTupleStream::read_tuple_idx_

private

Current idx of the tuple read from the read_block_ buffer.

Definition at line 310 of file buffered-tuple-stream.h.

Referenced by GetNextInternal(), NextBlockForRead(), and PrepareForRead().

const bool impala::BufferedTupleStream::read_write_

private

If true, read and write operations may be interleaved. Otherwise all calls to AddRow() must occur before calling PrepareForRead() and subsequent calls to GetNext().

Definition at line 262 of file buffered-tuple-stream.h.

Referenced by Init(), PrepareForRead(), and UnpinStream().

int64_t impala::BufferedTupleStream::rows_returned_

private

Number of rows returned to the caller from GetNext().

Definition at line 319 of file buffered-tuple-stream.h.

Referenced by DebugString(), GetNextInternal(), PrepareForRead(), and rows_returned().