Operator DynamicFilter
The DynamicFilter is a version of the Filter operator that can decide at run time which input tuples are passed through, based on control input it receives.
Checkpointed data
When the DynamicFilter operator is checkpointed, the current set of valid expression to match the input tuples is saved in checkpoint. Logic state variables (if present) are also included in checkpoint.
Behavior in a consistent region
The DynamicFilter operator can be an operator within the reachability graph of a consistent region. It cannot be the start of a consistent region. When in a consistent region, the DynamicFilter checkpoints and resets the current set of valid expression to match the input tuples. In a consistent region, the Streams instance automatically checkpoints and resets logic state variables (if there is any).
Checkpointing behavior in an autonomous region
When the DynamicFilter operator is in an autonomous region and configured with config checkpoint : periodic(T) clause, a background thread in SPL Runtime checkpoints the operator every T seconds, and such periodic checkpointing activity is asynchronous to tuple processing.
When the DynamicFilter operator is in an autonomous region and configured with config checkpoint : operatorDriven clause, no checkpoint is taken at runtime. Such operator driven checkpointing behavior is subject to change in the future.
Examples
This example uses the DynamicFilter operator.
composite Main {
graph
stream<rstring key, uint32 age> Data = Beacon () {}
stream<uint32 age> AddKeys = Beacon () {}
stream<uint32 removeKey> RemoveKeys = Beacon () {}
// Only pass tuples that match the current key
stream<rstring key, uint32 age> Out = DynamicFilter(Data; AddKeys; RemoveKeys)
{
param
key : Data.age; // key expression to be used for for matches
addKey : AddKeys.age - 1u; // value to be add to valid keys
removeKey : removeKey + 1u; // value to be removed from valid keys
}
}
Summary
Properties
- Implementation
- C++
- Threading
- Always - Operator always provides a single threaded execution context.
- Ports (0)
-
The DynamicFilter operator has three input ports. The first port is the stream of tuples to be filtered. The tuples on the first port are passed through the DynamicFilter if their key matches a valid key within the operator.
- Properties
-
- Optional: false
- ControlPort: false
- TupleMutationAllowed: false
- WindowingMode: NonWindowed
- WindowPunctuationInputMode: Oblivious
- Ports (1)
-
The second port is a stream of tuples that contain expressions to be added to the valid keys in the DynamicFilter operator.
- Properties
-
- Optional: false
- ControlPort: false
- TupleMutationAllowed: false
- WindowingMode: NonWindowed
- WindowPunctuationInputMode: Oblivious
- Ports (2)
-
The third port is a stream of tuples that contain expressions to be removed from the valid keys in the DynamicFilter operator.
- Properties
-
- Optional: true
- ControlPort: false
- TupleMutationAllowed: false
- WindowingMode: NonWindowed
- WindowPunctuationInputMode: Oblivious
- Ports (0)
-
The DynamicFilter operator has one or two output ports, which produce tuples that match the filter list. Tuples from the first input stream that contain valid keys are submitted to the first output port. The schemas of the output ports must match the schema of the first input port.
- Assignments
- This port set does not allow assignments to output attributes.
- Properties
-
- Optional: false
- TupleMutationAllowed: false
- WindowPunctuationOutputMode: Preserving
- Ports (1)
-
If present, this output port contains the tuples that do not contain a valid key. The schemas of the output ports must match the schema of the first input port.
- Assignments
- This port set allows any SPL expression of the correct type to be assigned to output attributes. Attributes not assigned in the output clause will be automatically assigned from the attributes of the input ports that have the same name and type. If there is no such input attribute, an error is reported at compile-time.
- Properties
-
- Optional: true
- TupleMutationAllowed: false
- WindowPunctuationOutputMode: Preserving
Optional: removeKey
- addKey
-
Specifies a list of expressions that are used to used to add keys to the set of valid keys. The number and types of the addKey expressions must match those specified in the key parameter.
- Properties
-
- Optional: false
- ExpressionMode: Expression
- PortScope: 1
- key
-
Specifies a list of expressions that are used to determine the key on the input tuple.
- Properties
-
- Optional: false
- ExpressionMode: Expression
- PortScope: 0
- removeKey
-
Specifies a list of expressions that are used to remove keys from the set of valid keys. This parameter cannot be specified when the operator is not configured with three input ports. Otherwise, it must be specified.
The number and types of the removeKey expressions must match those specified in the key parameter.
- Properties
-
- Optional: true
- ExpressionMode: Expression
- PortScope: 2
- DynamicFilter
-
stream<${schema}> ${outputStream} = DynamicFilter(${inputStream1};${inputStream2}) { param key: ${keyExpression}; addKey: ${addKeyExpression}; }