Type Members

1.  class JoinStateManagerStoreGenerator extends Logging

   Class used to handle state store creation in SymmetricHashJoinStateManager V1 and V2

2.  case class SnapshotOptions(snapshotVersion: Long, endVersion: Long, startKeyToNumValuesStateStoreCkptId: Option[String] = None, startKeyWithIndexToValueStateStoreCkptId: Option[String] = None, endKeyToNumValuesStateStoreCkptId: Option[String] = None, endKeyWithIndexToValueStateStoreCkptId: Option[String] = None) extends Product with Serializable

   Options controlling snapshot-based state replay for state data source reader.

3.  case class StreamingSymmetricHashJoinExec(leftKeys: Seq[Expression], rightKeys: Seq[Expression], joinType: JoinType, condition: JoinConditionSplitPredicates, stateInfo: Option[StatefulOperatorStateInfo], eventTimeWatermarkForLateEvents: Option[Long], eventTimeWatermarkForEviction: Option[Long], stateWatermarkPredicates: JoinStateWatermarkPredicates, stateFormatVersion: Int, left: SparkPlan, right: SparkPlan) extends SparkPlan with BinaryExecNode with StateStoreWriter with SchemaValidationUtils with Product with Serializable

   Performs stream-stream join using symmetric hash join algorithm.

   Performs stream-stream join using symmetric hash join algorithm. It works as follows.

   /-----------------------\ left side input --------->| left side state |------\ \-----------------------/ | |--------> joined output /-----------------------\ | right side input -------->| right side state |------/ \-----------------------/

   Each join side buffers past input rows as streaming state so that the past input can be joined with future input on the other side. This buffer state is effectively a multi-map: equi-join key -> list of past input rows received with the join key

   For each input row in each side, the following operations take place. - Calculate join key from the row. - Use the join key to append the row to the buffer state of the side that the row came from. - Find past buffered values for the key from the other side. For each such value, emit the "joined row" (left-row, right-row) - Apply the optional condition to filter the joined rows as the final output.

   If a timestamp column with event time watermark is present in the join keys or in the input data, then it uses the watermark to figure out which rows in the buffer will not join with the new data, and therefore can be discarded. Depending on the provided query conditions, we can define thresholds on both state key (i.e. joining keys) and state value (i.e. input rows). There are three kinds of queries possible regarding this as explained below. Assume that watermark has been defined on both leftTime and rightTime columns used below.

   1. When timestamp/time-window + watermark is in the join keys. Example (pseudo-SQL):

   SELECT * FROM leftTable, rightTable ON leftKey = rightKey AND window(leftTime, "1 hour") = window(rightTime, "1 hour") // 1hr tumbling windows

   In this case, this operator will join rows newer than watermark which fall in the same 1 hour window. Say the event-time watermark is "12:34" (both left and right input). Then input rows can only have time > 12:34. Hence, they can only join with buffered rows where window >= 12:00 - 1:00 and all buffered rows with join window < 12:00 can be discarded. In other words, the operator will discard all state where window in state key (i.e. join key) < event time watermark. This threshold is called State Key Watermark.

   2. When timestamp range conditions are provided (no time/window + watermark in join keys). E.g.

   SELECT * FROM leftTable, rightTable ON leftKey = rightKey AND leftTime > rightTime - INTERVAL 8 MINUTES AND leftTime < rightTime + INTERVAL 1 HOUR

   In this case, the event-time watermark and the BETWEEN condition can be used to calculate a state watermark, i.e., time threshold for the state rows that can be discarded. For example, say each join side has a time column, named "leftTime" and "rightTime", and there is a join condition "leftTime > rightTime - 8 min". While processing, say the watermark on right input is "12:34". This means that from henceforth, only right inputs rows with "rightTime > 12:34" will be processed, and any older rows will be considered as "too late" and therefore dropped. Then, the left side buffer only needs to keep rows where "leftTime > rightTime - 8 min > 12:34 - 8m > 12:26". That is, the left state watermark is 12:26, and any rows older than that can be dropped from the state. In other words, the operator will discard all state where timestamp in state value (input rows) < state watermark. This threshold is called State Value Watermark (to distinguish from the state key watermark).

   Note:

   • The event watermark value of one side is used to calculate the state watermark of the other side. That is, a condition ~ "leftTime > rightTime + X" with right side event watermark is used to calculate the left side state watermark. Conversely, a condition ~ "left < rightTime + Y" with left side event watermark is used to calculate right side state watermark.
   • Depending on the conditions, the state watermark maybe different for the left and right side. In the above example, leftTime > 12:26 AND rightTime > 12:34 - 1 hour = 11:34.
   • State can be dropped from BOTH sides only when there are conditions of the above forms that define time bounds on timestamp in both directions.

   3. When both window in join key and time range conditions are present, case 1 + 2. In this case, since window equality is a stricter condition than the time range, we can use the State Key Watermark = event time watermark to discard state (similar to case 1).

   leftKeys

   Expression to generate key rows for joining from left input

   rightKeys

   Expression to generate key rows for joining from right input

   joinType

   Type of join (inner, left outer, etc.)

   condition

   Conditions to filter rows, split by left, right, and joined. See JoinConditionSplitPredicates

   stateInfo

   Version information required to read join state (buffered rows)

   eventTimeWatermarkForLateEvents

   Watermark for filtering late events, same for both sides

   eventTimeWatermarkForEviction

   Watermark for state eviction

   stateWatermarkPredicates

   Predicates for removal of state, see JoinStateWatermarkPredicates

   left

   Left child plan

   right

   Right child plan

4.  abstract class SymmetricHashJoinStateManager extends Logging

   Helper class to manage state required by a single side of org.apache.spark.sql.execution.streaming.StreamingSymmetricHashJoinExec.

   Helper class to manage state required by a single side of org.apache.spark.sql.execution.streaming.StreamingSymmetricHashJoinExec. The interface of this class is basically that of a multi-map: - Get: Returns an iterator of multiple values for given key - Append: Append a new value to the given key - Remove Data by predicate: Drop any state using a predicate condition on keys or values

5.  class SymmetricHashJoinStateManagerV1 extends SymmetricHashJoinStateManager

   Streaming join state manager that uses 4 state stores without virtual column families.

   Streaming join state manager that uses 4 state stores without virtual column families. This implementation creates a state stores based on the join side and the type of state store.

   The keyToNumValues store tracks the number of rows for each key, and the keyWithIndexToValue store contains the actual entries with an additional index column.

6.  class SymmetricHashJoinStateManagerV2 extends SymmetricHashJoinStateManager

   Streaming join state manager that uses 1 state store with virtual column families enabled.

   Streaming join state manager that uses 1 state store with virtual column families enabled. Instead of creating a new state store per join side and store type, this manager uses column families to distinguish data between the original 4 state stores.