Snapshot Load Processors
Processor inventory for the Snapshot Load flow of the
sqlserver-multidatabase connector (v0.26.0), taken from the exported flow
definition.
⚙ 5 processors, plus a nested Perform Snapshot Load group (25)
Snapshot Load performs the initial bulk copy of each configured table before incremental
change tracking takes over. The top-level group selects and registers tables; the nested
Perform Snapshot Load group creates destination objects, reads source rows in parallel, and
streams them into Snowflake. Processor types prefixed MultiDatabase* are the
multi-database connector variants.
Table Selection & Registration (5)
| Processor | Type | Purpose |
|---|---|---|
| Set Tables for Replication | GenerateFlowFile | Emits the configured table set to drive the snapshot cycle |
| Verify Configured Tables | MultiDatabaseListTableNames | Validates the Included Table Names resolve at the source |
| Pick Tables For Replication | MultiDatabasePickTablesForReplication | Selects which tables to snapshot this cycle |
| Register New Tables for Replication | MultiDatabaseUpdateTableState | Marks newly added tables in the state store |
| Remove Table From Replication | MultiDatabaseUpdateTableState | Removes de-configured tables from the state store |
Nested: Object Naming & Creation (5)
| Processor | Type | Purpose |
|---|---|---|
| Determine Destination Object Names | SetAttributesValidatingReferences | Resolves destination schema/table names (Destination Schema Pattern) |
| Apply Object Identifier Resolution Strategy | UpdateAttribute | Applies case-sensitivity rules to identifiers |
| Create Snowflake Schema | UpdateSnowflakeSchema | Creates the destination schema if missing |
| Create Snowflake Table | UpdateSnowflakeTable | Creates the destination table |
| Create Snowflake Table | UpdateSnowflakeTable | Secondary create path for the destination table |
Nested: Read & Load (6)
| Processor | Type | Purpose |
|---|---|---|
| Fetch Source Table Schema | MultiDatabaseFetchSourceTableSchema | Reads the source table column definitions |
| Fetch Table Rows | MultiDatabaseFetchTableSnapshot | Bulk-reads source rows in parallel chunks (Concurrent Select Queries For Snapshot) |
| Add Snowflake Specific Columns | UpdateRecord | Adds connector metadata columns before upload |
| Remove Snowflake Specific Columns | UpdateRecord | Strips helper columns not persisted to the destination |
| Restore Chunks Order | EnforceOrder | Re-orders parallel chunks before load |
| Upload Rows via Snowpipe Streaming | PutSnowpipeStreaming | Streams snapshot rows into the destination table |
Nested: Gating & Control Flow (5)
| Processor | Type | Purpose |
|---|---|---|
| Check If Primary Key Is Present | RouteOnAttribute | Requires a primary key before snapshotting the table |
| Check if Configured to Skip Snapshot | RouteOnAttribute | Honors the skip-snapshot setting for the table |
| Check if Configured to Skip Snapshot | RouteOnAttribute | Second skip-snapshot gate on an alternate path |
| Check if All Rows Stored | RouteOnAttribute | Confirms every chunk uploaded before completing |
| Clear Content | ReplaceText | Clears row content once uploaded |
Nested: State & Logging (4)
| Processor | Type | Purpose |
|---|---|---|
| Wait for Incremental Load to Begin | MultiDatabaseWaitForTableState | Hands off to incremental once the snapshot finishes |
| Mark Snapshot Complete | MultiDatabaseUpdateTableState | Records snapshot completion in the state store |
| Mark Table Replication Failed | MultiDatabaseUpdateTableState | Records table-level failure in the state store |
| Log Snapshot Complete Message | LogMessage | Emits a completion log line for the table |
Nested: Error Handling (5)
| Processor | Type | Purpose |
|---|---|---|
| Update Failure Reason With CLEARING_FLOWFILE_FAILED | UpdateAttribute | Sets error code when clearing FlowFile content fails |
| Update Failure Reason With MISSING_PRIMARY_KEYS | UpdateAttribute | Sets error code when the table has no primary key |
| Update Failure Reason With ORDER_ENFORCEMENT_FAILED | UpdateAttribute | Sets error code when chunk ordering fails |
| Update Failure Reason With SNOWFLAKE_OBJECT_ALREADY_EXISTS | UpdateAttribute | Sets error code when the destination object already exists |
| Update Failure Reason With SNOWPIPE_UPLOAD_FAILED | UpdateAttribute | Sets error code when Snowpipe Streaming fails |
Note: The full connector flow definition (59 processors across Snapshot Load,
Incremental Load, and Stream Staleness Prevention) is exported to
connector-flow/sqlserver-multidatabase.flow.json in the
reference implementation repository,
taken from the deployed sqlserver-multidatabase connector v0.26.0. It can be
imported into any Openflow runtime.
Incremental Load Processors
Processor inventory for the Incremental Load flow of the
sqlserver-multidatabase connector (v0.26.0), taken from the exported flow
definition.
⚙ 19 processors, plus a nested Create Journal Table group (6)
The Incremental Load flow handles change capture, destination object naming, schema evolution,
the journal merge, state coordination, and error recovery. Processor types prefixed
MultiDatabase* are the multi-database connector variants.
Change Capture & Upload (4)
| Processor | Type | Purpose |
|---|---|---|
| Read SQLServer Change Tracking tables | MultiDatabaseCaptureChangeSqlServer | Polls CT tables for the net INSERT/UPDATE/DELETE change per interval |
| Schedule Warehouse | MergeContent | Bins changes and gates the merge schedule (Merge Task Schedule CRON) |
| Remove Rows from FlowFile | ReplaceText | Clears row content after upload, before signaling the merge |
| Upload Rows via Snowpipe Streaming | PutSnowpipeStreaming | Streams changes into the journal table |
Merge to Destination (2)
| Processor | Type | Purpose |
|---|---|---|
| Merge Journal to Destination | MultiDatabaseMergeSnowflakeJournalTable | Applies UPSERT/DELETE from journal to destination |
| Drop Journal Stream | UpdateSnowflakeStream | Cleans up the journal stream after merge |
Destination Object Naming (2)
| Processor | Type | Purpose |
|---|---|---|
| Determine Destination Object Names | SetAttributesValidatingReferences | Resolves destination schema/table names (Destination Schema Pattern) |
| Apply Object Identifier Resolution Strategy | UpdateAttribute | Applies case-sensitivity rules to identifiers |
Schema Evolution (3)
| Processor | Type | Purpose |
|---|---|---|
| Process Schema Changes | MultiDatabaseEnrichCdcStream | Detects and processes DDL changes from source |
| Drop First DDL | RouteOnAttribute | Filters the initial DDL event to prevent duplicates |
| Alter Destination Table | UpdateSnowflakeTable | Applies ALTER TABLE to the Snowflake destination |
State Management (2)
| Processor | Type | Purpose |
|---|---|---|
| Wait for Snapshot Load to Finish | MultiDatabaseWaitForTableState | Blocks incremental until the snapshot completes |
| Set Source Table FQN | UpdateAttribute | Adds the fully-qualified source table name to the FlowFile |
Error Handling (6)
| Processor | Type | Purpose |
|---|---|---|
| Update Failure Reason With CONTENT_MERGE_FAILED | UpdateAttribute | Sets error code when the merge/batching step fails |
| Update Failure Reason With SNOWFLAKE_OBJECT_OPERATION_FAILED | UpdateAttribute | Sets error code when a DDL operation fails |
| Update Failure Reason With SNOWPIPE_UPLOAD_FAILED | UpdateAttribute | Sets error code when Snowpipe Streaming fails |
| Update Failure Reason With VALUE_MAPPING_ERROR | UpdateAttribute | Sets error code when a value cannot be mapped |
| Mark Replication as Failed | MultiDatabaseUpdateTableState | Records table-level failure in the state store |
| Mark Replication as Failed | MultiDatabaseUpdateTableState | Alternate failure path |
Nested: Create Journal Table (6)
| Processor | Type | Purpose |
|---|---|---|
| Create Schema If Not Exists | UpdateSnowflakeSchema | Creates the destination schema if missing |
| Create Journal Table | UpdateSnowflakeTable | Creates the journal table in Snowflake |
| Create Journal Table Stream | UpdateSnowflakeStream | Creates the append-only stream on the journal table |
| Convert To Journal Schema | MultiDatabaseConvertToJournalSchema | Derives the journal table schema from the source table |
| Mark Replication as Failed | MultiDatabaseUpdateTableState | Records failure if journal creation fails |
| Update Failure Reason With SNOWFLAKE_OBJECT_OPERATION_FAILED | UpdateAttribute | Sets error code when a journal DDL operation fails |
Stream Staleness Prevention Processors
Processor inventory for the Stream Staleness Prevention flow of the
sqlserver-multidatabase connector (v0.26.0), taken from the exported flow
definition.
⚙ 4 processors
Each journal table has a Snowflake stream, and a stream becomes stale once its offset falls
outside the table's data retention window. If the source goes quiet longer than that window,
the stream would expire and force a costly reinitialization. This small housekeeping flow runs
on a schedule to read each journal stream's position and advance its STALE_AFTER
boundary, keeping the streams alive through periods of low change volume.
Staleness Prevention (4)
| Processor | Type | Purpose |
|---|---|---|
| Fetch Journal Streams | ExecuteSQLRecord | Queries Snowflake for the journal streams to check |
| Check If Journal Streams Are Present | RouteOnAttribute | Proceeds only when one or more streams are returned |
| Get Stream Coordinates | EvaluateJsonPath | Extracts each stream's current offset from the query result |
| Advance Stream's STALE_AFTER | ExecuteSQL | Pushes the stream's STALE_AFTER boundary forward so it does not expire |