KeyWithIndexToValueStore — State Store (Handler) Of Join Keys With Index Of Values

KeyWithIndexToValueStore is a StateStoreHandler (of KeyWithIndexToValueType) for SymmetricHashJoinStateManager to manage a join state.

KeyToNumValuesStore KeyWithIndexToValueStore.png

Figure 1. KeyToNumValuesStore, KeyWithIndexToValueStore and Stream-Stream Join

As a StateStoreHandler, KeyWithIndexToValueStore manages a state store (that is loaded) for keys and values per the keys with index and input values schemas, respectively.

KeyWithIndexToValueStore uses a schema (for the state store) that is the key schema (of the parent SymmetricHashJoinStateManager) with an extra field index of type long.

Tip

Enable ALL logging level for org.apache.spark.sql.execution.streaming.state.SymmetricHashJoinStateManager$KeyWithIndexToValueStore to see what happens inside.

Add the following line to conf/log4j.properties:

log4j.logger.org.apache.spark.sql.execution.streaming.state.SymmetricHashJoinStateManager$KeyWithIndexToValueStore=ALL

Refer to Logging.

Looking Up State Row For Given Key and Index — `get` Method

get(
  key: UnsafeRow,
  valueIndex: Long): UnsafeRow

get simply requests the internal state store to look up the value for the given key and valueIndex.

Note	`get` is used exclusively when `SymmetricHashJoinStateManager` is requested to removeByValueCondition

Retrieving (Given Number of) Values for Key — `getAll` Method

getAll(
  key: UnsafeRow,
  numValues: Long): Iterator[KeyWithIndexAndValue]

getAll…FIXME

Note	`getAll` is used when `SymmetricHashJoinStateManager` is requested to get values for a given key and removeByKeyCondition.

Storing State Row For Given Key and Index — `put` Method

put(
  key: UnsafeRow,
  valueIndex: Long,
  value: UnsafeRow): Unit

put…FIXME

Note	`put` is used when `SymmetricHashJoinStateManager` is requested to append a new value to a given key and removeByKeyCondition.

`remove` Method

remove(
  key: UnsafeRow,
  valueIndex: Long): Unit

remove…FIXME

Note	`remove` is used when `SymmetricHashJoinStateManager` is requested to removeByKeyCondition and removeByValueCondition.

`keyWithIndexRow` Internal Method

keyWithIndexRow(
  key: UnsafeRow,
  valueIndex: Long): UnsafeRow

keyWithIndexRow uses the keyWithIndexRowGenerator to generate an UnsafeRow for the key and sets the valueIndex at the indexOrdinalInKeyWithIndexRow position.

Note	`keyWithIndexRow` is used when `KeyWithIndexToValueStore` is requested to get, getAll, put, remove and removeAllValues.

`removeAllValues` Method

removeAllValues(
  key: UnsafeRow,
  numValues: Long): Unit

removeAllValues…FIXME

Note	`removeAllValues` does not seem to be used at all.

`iterator` Method

iterator: Iterator[KeyWithIndexAndValue]

iterator…FIXME

Note	`iterator` does not seem to be used at all.

Internal Properties

Name Description

Name	Description
`indexOrdinalInKeyWithIndexRow`	Position of the index in the key row (which corresponds to the number of the key attributes) Used exclusively in the keyWithIndexRow
`keyWithIndexExprs`	keyAttributes with `Literal(1L)` expression appended Used exclusively for the keyWithIndexRowGenerator projection
`keyWithIndexRowGenerator`	`UnsafeProjection` for the keyWithIndexExprs bound to the keyAttributes Used exclusively in keyWithIndexRow

indexOrdinalInKeyWithIndexRow

Position of the index in the key row (which corresponds to the number of the key attributes)

Used exclusively in the keyWithIndexRow

keyWithIndexExprs

keyAttributes with Literal(1L) expression appended

Used exclusively for the keyWithIndexRowGenerator projection

keyWithIndexRowGenerator

UnsafeProjection for the keyWithIndexExprs bound to the keyAttributes

Used exclusively in keyWithIndexRow

KeyWithIndexToValueStore