Struct TopNCache

pub struct TopNCache<const WITH_TIES: bool> {
    pub low: Option<EstimatedBTreeMap<(Vec<u8>, Vec<u8>), CompactedRow>>,
    pub middle: EstimatedBTreeMap<(Vec<u8>, Vec<u8>), CompactedRow>,
    pub high: EstimatedBTreeMap<(Vec<u8>, Vec<u8>), CompactedRow>,
    pub high_cache_capacity: usize,
    pub offset: usize,
    pub limit: usize,
    table_row_count: Option<usize>,
    data_types: Vec<DataType>,
}

Cache for ManagedTopNState.

The key in the maps CacheKey is [ order_by + remaining columns of pk ]. group_key is not included.

WITH_TIES

WITH_TIES supports the semantic of FETCH FIRST n ROWS WITH TIES and RANK() <= n.

OFFSET m FETCH FIRST n ROWS WITH TIES and m <= RANK() <= n are not supported now, since they have different semantics.

Fields

low: Option<EstimatedBTreeMap<(Vec<u8>, Vec<u8>), CompactedRow>>

Rows in the range [0, offset). Should always be synced with state table.

middle: EstimatedBTreeMap<(Vec<u8>, Vec<u8>), CompactedRow>

Rows in the range [offset, offset+limit). Should always be synced with state table.

When WITH_TIES is true, it also stores ties for the last element, and thus the size can be larger than limit.

high: EstimatedBTreeMap<(Vec<u8>, Vec<u8>), CompactedRow>

Cache of the beginning rows in the range [offset+limit, ...).

This is very similar to TopNStateCache, which only caches the top-N rows in the table and only accepts new records that are less than the largest in the cache.

When WITH_TIES is true, it guarantees that the ties of the last element are in the cache, and thus the size can be larger than rest_cache_capacity.

When the cache becomes empty, if the table_row_count is not matched, we need to view the cache as unsynced and refill it from the state table.

TODO(rc): later we should reuse TopNStateCache here.

high_cache_capacity: usize

offset: usize

limit: usize

Assumption: limit != 0

table_row_count: Option<usize>

Number of rows corresponding to the current group. This is a nice-to-have information. None means we don’t know the row count, but it doesn’t prevent us from working correctly.

data_types: Vec<DataType>

Data types for the full row.

For debug formatting only.

Implementations

impl<const WITH_TIES: bool> TopNCache<WITH_TIES>

pub fn new(offset: usize, limit: usize, data_types: Vec<DataType>) -> Self

data_types – Data types for the full row.

pub fn clear(&mut self)

Clear the cache. After this, the cache must be init again before use.

pub fn len(&self) -> usize

Get total count of entries in the cache.

pub(super) fn update_table_row_count(&mut self, table_row_count: usize)

pub fn low_is_full(&self) -> bool

pub fn middle_is_full(&self) -> bool

pub fn high_is_full(&self) -> bool

fn high_is_synced(&self) -> bool

fn last_cache_key_before_high(&self) -> Option<&(Vec<u8>, Vec<u8>)>

Trait Implementations

impl AppendOnlyTopNCacheTrait for TopNCache<false>

fn insert<S: StateStore>( &mut self, cache_key: (Vec<u8>, Vec<u8>), row_ref: RowRef<'_>, staging: &mut TopNStaging, managed_state: &mut ManagedTopNState<S>, row_deserializer: &RowDeserializer, ) -> StreamExecutorResult<()>

Insert input row to corresponding cache range according to its order key.

impl AppendOnlyTopNCacheTrait for TopNCache<true>

fn insert<S: StateStore>( &mut self, cache_key: (Vec<u8>, Vec<u8>), row_ref: RowRef<'_>, staging: &mut TopNStaging, managed_state: &mut ManagedTopNState<S>, row_deserializer: &RowDeserializer, ) -> StreamExecutorResult<()>

Insert input row to corresponding cache range according to its order key.

impl<const WITH_TIES: bool> Debug for TopNCache<WITH_TIES>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<const WITH_TIES: bool> EstimateSize for TopNCache<WITH_TIES>

fn estimated_heap_size(&self) -> usize

The estimated heap size of the current struct in bytes.

fn estimated_size(&self) -> usize

where Self: Sized,

The estimated total size of the current struct in bytes, including the estimated_heap_size and the size of Self.

impl TopNCacheTrait for TopNCache<false>

fn insert( &mut self, cache_key: (Vec<u8>, Vec<u8>), row: impl Row + Send, staging: &mut TopNStaging, )

Insert input row to corresponding cache range according to its order key.

async fn delete<S: StateStore>( &mut self, group_key: Option<impl GroupKey>, managed_state: &mut ManagedTopNState<S>, cache_key: (Vec<u8>, Vec<u8>), row: impl Row + Send, staging: &mut TopNStaging, ) -> StreamExecutorResult<()>

Delete input row from the cache.

impl TopNCacheTrait for TopNCache<true>

fn insert( &mut self, cache_key: (Vec<u8>, Vec<u8>), row: impl Row + Send, staging: &mut TopNStaging, )

Insert input row to corresponding cache range according to its order key.

async fn delete<S: StateStore>( &mut self, group_key: Option<impl GroupKey>, managed_state: &mut ManagedTopNState<S>, cache_key: (Vec<u8>, Vec<u8>), row: impl Row + Send, staging: &mut TopNStaging, ) -> StreamExecutorResult<()>

Delete input row from the cache.