KeyToNumValuesStore — State Store (Handler) Of Join Keys And Counts

KeyToNumValuesStore is a StateStoreHandler (of KeyToNumValuesType) for SymmetricHashJoinStateManager to manage a join state.

KeyToNumValuesStore KeyWithIndexToValueStore.png

Figure 1. KeyToNumValuesStore, KeyWithIndexToValueStore and Stream-Stream Join

As a StateStoreHandler, KeyToNumValuesStore manages a state store (that is loaded) with the join keys (per key schema) and their count (per value schema).

KeyToNumValuesStore uses the schema for values in the state store with one field value (of type long) that is the number of value rows (count).

Tip

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

Add the following line to conf/log4j.properties:

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

Refer to Logging.

Looking Up Number Of Value Rows For Given Key (Value Count) — `get` Method

get(key: UnsafeRow): Long

get requests the StateStore for the value for the given key and returns the long value at 0th position (of the row found) or 0.

Note	`get` is used when `SymmetricHashJoinStateManager` is requested for the values for a given key and append a new value to a given key.

Storing Key Count For Given Key — `put` Method

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

put stores the numValues at the 0th position (of the internal unsafe row) and requests the StateStore to store it with the given key.

put requires that the numValues count is greater than 0 (or throws an IllegalArgumentException).

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

All State Keys and Values — `iterator` Method

iterator: Iterator[KeyAndNumValues]

iterator simply requests the StateStore for all state keys and values.

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

Removing State Key — `remove` Method

remove(key: UnsafeRow): Unit

remove simply requests the StateStore to remove the given key.

Note	`remove` is used when…FIXME

KeyToNumValuesStore