Fragment 62633 (Actor 743983,743984)
StreamMaterialize { columns: [party_id, customer_relationship_id], stream_key: [customer_relationship_id], pk_columns: [customer_relationship_id], pk_conflict: NoCheck }
├── output: [ internal_last_seen_value(customer_relationships_next.party_id), customer_relationships_next.id ]
├── stream key: [ customer_relationships_next.id ]
└── StreamProject { exprs: [internal_last_seen_value(customer_relationships_next.party_id), customer_relationships_next.id] }
├── output: [ internal_last_seen_value(customer_relationships_next.party_id), customer_relationships_next.id ]
├── stream key: [ customer_relationships_next.id ]
└── StreamHashAgg { group_key: [customer_relationships_next.id], aggs: [internal_last_seen_value(customer_relationships_next.party_id), count] }
├── output: [ customer_relationships_next.id, internal_last_seen_value(customer_relationships_next.party_id), count ]
├── stream key: [ customer_relationships_next.id ]
└── StreamLocalityProvider { locality_columns: [customer_relationships_next.id] }
├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, parties.id ]
├── stream key: [ customer_relationships_next.id, customer_relationships_next.party_id ]
└── MergeExecutor
├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, parties.id ]
└── stream key: [ customer_relationships_next.party_id, customer_relationships_next.id ]
Fragment 62634 (Actor 743985,743986)
StreamSyncLogStore
├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, parties.id ]
├── stream key: [ customer_relationships_next.party_id, customer_relationships_next.id ]
└── StreamHashJoin { type: Inner, predicate: customer_relationships_next.party_id = parties.id }
├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, parties.id ]
├── stream key: [ customer_relationships_next.party_id, customer_relationships_next.id ]
├── MergeExecutor
│ ├── output: [ customer_relationships_next.id, customer_relationships_next.party_id ]
│ └── stream key: [ customer_relationships_next.party_id, customer_relationships_next.id ]
└── MergeExecutor { output: [ parties.id ], stream key: [ parties.id ] }
Fragment 62635 (Actor 744016,744015)
StreamLocalityProvider { locality_columns: [customer_relationships_next.party_id] }
├── output: [ customer_relationships_next.id, customer_relationships_next.party_id ]
├── stream key: [ customer_relationships_next.party_id, customer_relationships_next.id ]
└── MergeExecutor { output: [ customer_relationships_next.id, customer_relationships_next.party_id ], stream key: [ customer_relationships_next.id ] }
Fragment 62636 (Actor 744027,744028)
StreamProject { exprs: [customer_relationships_next.id, customer_relationships_next.party_id] } { output: [ customer_relationships_next.id, customer_relationships_next.party_id ], stream key: [ customer_relationships_next.id ] }
└── StreamDynamicFilter { predicate: ($expr2 > now), output_watermarks: [[$expr2]], output: [customer_relationships_next.id, customer_relationships_next.party_id, $expr2], cleaned_by_watermark: true }
├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, $expr2 ]
├── stream key: [ customer_relationships_next.id ]
├── StreamProject { exprs: [customer_relationships_next.id, customer_relationships_next.party_id, AtTimeZone(Coalesce(customer_relationships_next.effective_to, '9999-12-31':Date)::Timestamp, 'UTC':Varchar) as $expr2] }
│ ├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, $expr2 ]
│ ├── stream key: [ customer_relationships_next.id ]
│ └── StreamDynamicFilter { predicate: ($expr1 <= now), output: [customer_relationships_next.id, customer_relationships_next.party_id, customer_relationships_next.effective_to, $expr1], cleaned_by_watermark: true }
│ ├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, customer_relationships_next.effective_to, $expr1 ]
│ ├── stream key: [ customer_relationships_next.id ]
│ ├── StreamProject { exprs: [customer_relationships_next.id, customer_relationships_next.party_id, customer_relationships_next.effective_to, AtTimeZone(customer_relationships_next.effective_from::Timestamp, 'UTC':Varchar) as $expr1] }
│ │ ├── output: [ customer_relationships_next.id, customer_relationships_next.party_id, customer_relationships_next.effective_to, $expr1 ]
│ │ ├── stream key: [ customer_relationships_next.id ]
│ │ └── StreamFilter { predicate: (customer_relationships_next.type = 'CUSTOMER':Varchar) AND (customer_relationships_next.status = 'ACTIVE':Varchar) AND IsNull(customer_relationships_next.disabled_at) }
│ │ ├── output:
│ │ │ ┌── customer_relationships_next.id
│ │ │ ├── customer_relationships_next.party_id
│ │ │ ├── customer_relationships_next.effective_from
│ │ │ ├── customer_relationships_next.effective_to
│ │ │ ├── customer_relationships_next.type
│ │ │ ├── customer_relationships_next.status
│ │ │ └── customer_relationships_next.disabled_at
│ │ ├── stream key: [ customer_relationships_next.id ]
│ │ └── StreamTableScan { table: customer_relationships_next, columns: [id, party_id, effective_from, effective_to, type, status, disabled_at] }
│ │ ├── output:
│ │ │ ┌── customer_relationships_next.id
│ │ │ ├── customer_relationships_next.party_id
│ │ │ ├── customer_relationships_next.effective_from
│ │ │ ├── customer_relationships_next.effective_to
│ │ │ ├── customer_relationships_next.type
│ │ │ ├── customer_relationships_next.status
│ │ │ └── customer_relationships_next.disabled_at
│ │ ├── stream key: [ customer_relationships_next.id ]
│ │ ├── Upstream { output: [ id, party_id, effective_from, effective_to, type, status, disabled_at ], stream key: [] }
│ │ └── BatchPlanNode { output: [ id, party_id, effective_from, effective_to, type, status, disabled_at ], stream key: [] }
│ └── MergeExecutor { output: [ now ], stream key: [] }
└── MergeExecutor { output: [ now ], stream key: [] }
Fragment 62637 (Actor 744017)
StreamNow { output: [ now ], stream key: [] }
Fragment 62638 (Actor 744018)
StreamNow { output: [ now ], stream key: [] }
Fragment 62639 (Actor 744030,744029)
StreamProject { exprs: [parties.id] } { output: [ parties.id ], stream key: [ parties.id ] }
└── StreamFilter { predicate: IsNull(parties.disabled_at) } { output: [ parties.id, parties.disabled_at ], stream key: [ parties.id ] }
└── StreamTableScan { table: parties, columns: [id, disabled_at] } { output: [ parties.id, parties.disabled_at ], stream key: [ parties.id ] }
├── Upstream { output: [ id, disabled_at ], stream key: [] }
└── BatchPlanNode { output: [ id, disabled_at ], stream key: [] }