Legacy Pipeline Atlas worker-1snapshot 2026-07-15

MDClarity · task primitive

ExecuteMergeSql

Overview

ExecuteMergeSqlTask merges data from a SQL query into a target database table by performing insert, update, and optional delete operations based on merge keys. It executes a custom SQL query against the same database, an external SQL Server, or Snowflake, then intelligently merges the results into the target table using batched operations for optimal performance.

Use this task when: Synchronizing data from external databases or complex queries into destination tables, performing incremental data refreshes from source systems, or maintaining derived/aggregated tables with automatic cleanup of obsolete records.

Don't use this task when: Loading delimited files (use LoadDataFileTask), importing data without merge logic (use ImportSqlTask), or needing full control over SQL execution (write custom SQL or stored procedures).

Parameters

Parameter Type Required Default Description
MergeDataSourceType MergeDataSourceType No ThisDatabase Source database type: ThisDatabase (current customer DB), SqlServerExternal (external SQL Server), or Snowflake
ConnectionString string Conditional "Server = {Server}; Database = {Database}; Trusted_Connection = True" Connection string for external data source. Required when MergeDataSourceType is not ThisDatabase. Use connection string placeholders like {Server} and {Database} for templating.
Query string Yes None SQL query to execute against the source database. Must return all columns that exist in the target table plus all merge key columns. Cannot include reserved audit columns (CreatedOn, UpdatedOn, DeletedOn) in SELECT list.
TableName string Yes None Name of the target table in the customer database to merge data into. Table must exist before task execution.
MergeMode MergeMode No InsertUpdate Merge behavior: InsertUpdate (insert new, update existing), InsertUpdateSoftDelete (also set DeletedOn for missing records), or InsertUpdateHardDelete (physically delete missing records)
BatchSize int No 10000 Number of rows to process per batch. Controls memory usage and transaction size. Larger batches improve throughput but increase memory and lock duration.
MergeKeys List<string> Yes None Column names that uniquely identify rows for matching source to target (typically primary key columns). Used for insert/update decisions and sync operations.
ParentMergeKeys List<string> Conditional None Subset of MergeKeys used for determining deletion scope when MergeMode includes deletes and there are multiple merge keys. Required when MergeMode is InsertUpdateSoftDelete or InsertUpdateHardDelete and MergeKeys has more than one column.
MaxInvalidRows int No 0 Maximum number of validation errors to tolerate before task fails. Set to 0 for fail-fast behavior. Higher values allow processing to continue despite data quality issues.
BulkCopyTimeoutSeconds int No 300 Timeout in seconds for bulk copy operations (insert and update batches). Increase for large batches or slow networks.
BulkCopyUseTableLock bool No true Whether to use table-level locks during bulk copy operations. true improves performance with exclusive access; false allows concurrent readers.

Execution Flow

flowchart TD
    A[Start ExecuteMergeSql] --> B[Validate Settings]
    B --> C{Source Type?}
    C -->|ThisDatabase| D[Use Customer Connection]
    C -->|External| E[Use ConnectionString]
    D --> F[Create DataMerger Factory]
    E --> F
    F --> G[Setup: Ensure Audit Columns Exist]
    G --> H[Get Target Table Schema]
    H --> I[Validate Merge Keys in Source]
    I --> J{RequiresSync?}
    J -->|Yes| K[Create Temp Key Table for All IDs]
    J -->|No| L[Start Batch Processing Loop]
    K --> L
    L --> M[Read Batch from Source Query]
    M --> N[Upload Batch Keys to Temp Table]
    N --> O[Query Existing Target Rows by Keys]
    O --> P[Compare Source vs Existing]
    P --> Q{Row Exists?}
    Q -->|No| R[Add to Insert Queue]
    Q -->|Yes| S{Data Different?}
    S -->|No| T[Skip - No Changes]
    S -->|Yes| U[Add to Update Queue]
    R --> V[Bulk Insert New Rows]
    U --> W[Bulk Copy to Temp Update Table]
    T --> X{More Batches?}
    V --> X
    W --> Y[Execute UPDATE from Temp]
    Y --> X
    X -->|Yes| M
    X -->|No| Z{RequiresSync?}
    Z -->|No| END[End]
    Z -->|Yes| AA[Index Temp Key Table]
    AA --> AB{MergeMode?}
    AB -->|InsertUpdateSoftDelete| AC[UPDATE DeletedOn for Missing Rows]
    AB -->|InsertUpdateHardDelete| AD[DELETE Missing Rows]
    AC --> END
    AD --> END

    style G fill:#e1f5ff
    style P fill:#e1f5ff
    style V fill:#e1f5ff
    style W fill:#e1f5ff
    style Y fill:#e1f5ff
    style AC fill:#fff4e1
    style AD fill:#ffe1e1

Processing Algorithm

The task performs a batched merge operation with the following algorithm:

  1. Initialization: Validates settings, ensures audit columns (CreatedOn, UpdatedOn, and optionally DeletedOn) exist on target table, and identifies common columns between source and target.

  2. Batch Processing: For each batch of BatchSize rows from the source query:

    • Upload merge keys to a temporary table
    • Query existing target rows that match these keys
    • Compare source rows to existing rows column-by-column
    • Partition rows into insert queue (new) or update queue (changed)
    • Bulk insert new rows with CreatedOn and UpdatedOn set to current time
    • Bulk copy updates to temporary table, then merge to target with UpdatedOn set to current time
    • Preserve original CreatedOn for updated rows

  3. Sync Phase (if MergeMode includes deletes):

    • After all batches processed, identify target rows whose keys were NOT in the source data
    • Filter by ParentMergeKeys if specified (limits deletion scope to parent entities)
    • Execute soft delete (UPDATE DeletedOn = GETDATE()) or hard delete (DELETE)

Change Detection

The task only updates rows when data actually changes:

Usage Examples

Example 1: Basic merge from same database

var task = new ExecuteMergeSqlTask
{
    MergeDataSourceType = MergeDataSourceType.ThisDatabase,
    Query = "SELECT AccountID, FacilityID, EstimateAmount FROM AccountEstimate WHERE FacilityID = 'MAIN'",
    TableName = "AccountEstimateSnapshot",
    MergeMode = MergeMode.InsertUpdate,
    MergeKeys = new List<string> { "AccountID", "FacilityID" },
    BatchSize = 10000
    // Uses default connection to current database
};

Example 2: Sync from external SQL Server with soft delete

var task = new ExecuteMergeSqlTask
{
    MergeDataSourceType = MergeDataSourceType.SqlServerExternal,
    ConnectionString = "Server=external-sql.example.com;Database=SourceDB;User Id=integration;Password=***;",
    Query = @"
        SELECT
            p.PatientID,
            p.FirstName,
            p.LastName,
            p.DateOfBirth,
            p.PhoneNumber
        FROM Patients p
        WHERE p.IsActive = 1",
    TableName = "PatientCache",
    MergeMode = MergeMode.InsertUpdateSoftDelete,
    MergeKeys = new List<string> { "PatientID" },
    BatchSize = 5000,
    BulkCopyTimeoutSeconds = 600
    // Rows in PatientCache that don't exist in source will have DeletedOn set
};

Example 3: Composite key with parent-scoped deletion

var task = new ExecuteMergeSqlTask
{
    MergeDataSourceType = MergeDataSourceType.ThisDatabase,
    Query = @"
        SELECT
            v.VisitID,
            c.ChargeID,
            c.ChargeCode,
            c.Amount
        FROM Visit v
        INNER JOIN Charge c ON v.VisitID = c.VisitID
        WHERE v.VisitDate >= DATEADD(day, -90, GETDATE())",
    TableName = "VisitChargeCache",
    MergeMode = MergeMode.InsertUpdateHardDelete,
    MergeKeys = new List<string> { "VisitID", "ChargeID" },
    ParentMergeKeys = new List<string> { "VisitID" },
    // Only deletes charges for visits that appear in the source query
    // If a visit isn't in the query, its charges are left untouched
    BatchSize = 10000
};

Example 4: Snowflake external source

var task = new ExecuteMergeSqlTask
{
    MergeDataSourceType = MergeDataSourceType.Snowflake,
    ConnectionString = "account=myaccount;user=myuser;password=***;db=analytics;schema=public;warehouse=compute_wh",
    Query = "SELECT account_key, total_charges, payment_amount, outstanding_balance FROM account_summary",
    TableName = "AccountSummary",
    MergeMode = MergeMode.InsertUpdate,
    MergeKeys = new List<string> { "account_key" },
    BatchSize = 20000,
    MaxInvalidRows = 10 // Tolerate up to 10 data validation errors
};

Related Tasks Comparison

Task Use Case Merge Logic Data Source Delete/Sync Support
ExecuteMergeSql Query-based merge with sync Insert/Update/Delete based on keys SQL query (same DB, external SQL, Snowflake) Yes (soft or hard delete)
ImportSqlTask Simple append or replace Append or full table replace SQL query (external SQL Server) No
LoadDataFileTask File import Based on file spec settings Delimited files (CSV, pipe, etc.) No
LoadVisitsTask Visit batch preparation Custom visit loading with merge Multiple SQL queries No (batch table specific)
SyncBatchTask Promote batch to production Delete and insert (batch → prod) Batch tables No (different pattern)

Choose ExecuteMergeSql when:

Performance Considerations

Database Impact

Optimization Tips

Scenario Recommendation Rationale
Large datasets (>1M rows) Increase BatchSize to 20000-50000 Reduces overhead of batch management, improves throughput
Concurrent access required Set BulkCopyUseTableLock = false Allows readers during merge, though slower
Slow source queries Add indexes to source query tables Source query runs for each batch read; optimize with EXPLAIN PLAN
Many updates, few inserts Keep default BatchSize at 10000 Updates require temp table and merge SQL; moderate batch size balances memory and performance
External data sources Increase BulkCopyTimeoutSeconds to 600+ Network latency can slow bulk operations
Data quality issues Set MaxInvalidRows > 0 Allows processing to continue, logs validation errors for later review
Off-hours processing Set BulkCopyUseTableLock = true Maximum performance with exclusive access

Batch Size Selection

Sync Operation Performance

When MergeMode includes delete operations:

Error Handling

The task employs a fail-fast approach with configurable tolerance:

// Row validation happens during batch read from source
foreach (DataRow sourceRow in reader)
{
    try
    {
        // Validate against target table schema
        // Checks: data types, string lengths, nullable constraints
    }
    catch (Exception ex)
    {
        invalidRowCount++;
        if (invalidRowCount > MaxInvalidRows)
        {
            throw new MaxInvalidRowsException("Exceeded max invalid rows", ex);
        }
        // Otherwise, log and continue
    }
}

Key behaviors:

Important: Validation errors log the merge key values to help identify problematic source rows.

Common Pitfalls

Issue Problem Solution
Reserved column in query Source query returns CreatedOn, UpdatedOn, or DeletedOn columns Remove audit columns from SELECT list or use aliases (e.g., SourceCreatedOn). Task manages these automatically.
Missing ParentMergeKeys Task fails with "requires ParentMergeKeys" when using soft/hard delete with composite keys When MergeKeys has multiple columns and using delete modes, specify ParentMergeKeys to define deletion scope (typically the parent entity key).
Duplicate column names Task fails with "Column appears more than once" Check for duplicate columns in source query, especially from JOINs. Use aliases to disambiguate (e.g., p.Status as PatientStatus, v.Status as VisitStatus).
Merge keys not in source Task fails with "source data is missing required key field(s)" Ensure all columns in MergeKeys are included in the source query SELECT list.
Target table doesn't exist Task fails with "Table does not exist" Create target table before running task. Schema should match source query columns (except audit columns which are auto-added).
Wrong MergeDataSourceType Connection fails or times out For current database, use ThisDatabase and omit ConnectionString. For external sources, specify correct type and provide connection string.
Slow sync phase Sync/delete operation takes excessive time Ensure merge key columns are indexed on target table. Consider breaking into smaller parent scopes or using incremental loads.
Computed columns error Bulk copy fails on computed columns Task automatically excludes computed columns. If error persists, check for triggers or constraints on target table.
Data type mismatch Validation errors for all rows Compare source query column types to target table schema using EXEC sp_describe_first_result_set @query and EXEC sp_help @tablename. Cast source columns to match target types.

Code References