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

MDClarity · task primitive

ImportSql

Overview

ImportSqlTask imports data from SQL queries into database tables using bulk copy operations. It executes a SQL query against an external data source (SQL Server or Snowflake), streams the results directly into a target table, and supports multiple import modes (Append, Replace, Refill).

Use this task when: You need to import data from database queries rather than files, synchronize data from external SQL databases, or load query results that require SQL transformations or joins.

Don't use this task when:

Parameters

Parameter Type Required Default Description
ConnectionString string Yes "Server = {Server}; Database = {Database}; Trusted_Connection = True" Connection string to the source database
TableName string Yes null Name of the target table to import data into
Query string Yes null SQL query to execute against the source database
ImportMode ImportMode No Append Controls how data is loaded: Append (add to existing), Replace (drop/recreate table), or Refill (truncate existing)
DataSourceType DataSourceType No SqlServer Type of the source database: SqlServer or Snowflake

Execution Flow

flowchart TD
    A[Start ImportSql] --> B[Create DataReaderImporter with ImportMode]
    B --> C[Create DataImporter based on DataSourceType]
    C --> D{DataSourceType?}
    D -->|SqlServer| E[Create SqlServerDataImporter]
    D -->|Snowflake| F[Create SnowflakeDataImporter]
    E --> G[Execute SQL Query]
    F --> G
    G --> H[Get IDataReader from Query Results]
    H --> I{ImportMode?}
    I -->|Replace| J[Drop existing table if exists]
    I -->|Refill| K[Truncate existing table]
    I -->|Append| L[Keep existing table]
    J --> M[Create table from schema]
    K --> N[Table ready for data]
    L --> O{Table exists?}
    O -->|No| M
    O -->|Yes| N
    M --> N
    N --> P[Bulk copy data using SqlBulkCopy]
    P --> Q[Log rows imported and elapsed time]
    Q --> R[End]

    style P fill:#e1f5ff
    style J fill:#fff4e1
    style K fill:#fff4e1
    style M fill:#fff4e1

Processing Algorithm

The task uses a two-phase approach:

Phase 1: Table Preparation

Based on ImportMode, the target table is prepared:

Phase 2: Data Streaming

The query results are streamed directly into the target table using SqlBulkCopy:

Example Execution

For a query returning 50,000 rows with ImportMode = Append:

  1. Query executes against source database
  2. IDataReader streams results
  3. Table exists check → table exists
  4. SqlBulkCopy inserts 50,000 rows in batches of 10,000
  5. Task logs: "Imported 50000 records in 8.3 secs."

Usage Examples

Example 1: Basic SQL Server import with Replace mode

var task = new ImportSqlTask
{
    ConnectionString = "Server=ExternalDB;Database=Claims;Trusted_Connection=True",
    TableName = "ImportedClaims",
    Query = "SELECT * FROM ClaimTransactions WHERE ProcessDate >= '2025-01-01'",
    ImportMode = ImportMode.Replace,
    DataSourceType = DataSourceType.SqlServer
};

Example 2: Incremental append from external database

var task = new ImportSqlTask
{
    ConnectionString = "Server=PartnerDB;Database=Eligibility;User=import_user;Password=***",
    TableName = "ExternalEligibility",
    Query = @"
        SELECT PatientID, InsuranceName, EffectiveDate, TermDate
        FROM EligibilityRecords
        WHERE LastModified >= DATEADD(day, -7, GETDATE())",
    ImportMode = ImportMode.Append, // Add new records to existing table
    DataSourceType = DataSourceType.SqlServer
};

Example 3: Snowflake data warehouse import

var task = new ImportSqlTask
{
    ConnectionString = "account=myaccount;user=import_user;password=***;db=ANALYTICS;schema=PUBLIC",
    TableName = "SnowflakeReports",
    Query = @"
        SELECT
            FACILITY_ID,
            VISIT_COUNT,
            TOTAL_CHARGES,
            REPORT_DATE
        FROM ANALYTICS.REPORTING.MONTHLY_SUMMARY
        WHERE REPORT_DATE = CURRENT_DATE()",
    ImportMode = ImportMode.Refill, // Truncate and reload daily
    DataSourceType = DataSourceType.Snowflake
};

Example 4: Complex query with joins and transformations

var task = new ImportSqlTask
{
    ConnectionString = "Server=SourceDB;Database=EMR;Trusted_Connection=True",
    TableName = "PatientDemographics",
    Query = @"
        SELECT
            p.PatientID,
            p.FirstName,
            p.LastName,
            p.DateOfBirth,
            COALESCE(a.City, 'Unknown') AS City,
            COALESCE(a.State, 'Unknown') AS State
        FROM Patients p
        LEFT JOIN Addresses a ON p.PatientID = a.PatientID
        WHERE p.IsActive = 1",
    ImportMode = ImportMode.Replace,
    DataSourceType = DataSourceType.SqlServer
};

Related Tasks Comparison

Task Use Case Merge Logic Data Source Import Modes
ImportSql Query-based import Simple append or replace SQL Server or Snowflake queries Append, Replace, Refill
ExecuteMergeSqlTask Query-based synchronization Insert/Update/Delete with key matching SQL Server, Snowflake, or same database Merge with sync
LoadDataFileTask File-based import Simple append or replace Delimited files (CSV, pipe, etc.) Append, Replace, Refill
LoadVisitsTask Visit batch preparation Custom visit loading logic Multiple SQL queries (same database) Batch-specific

Choose ImportSql when:

Performance Considerations

Database Impact

Optimization Tips

Scenario Recommendation Rationale
Large datasets (>1M rows) Use indexed columns in source query WHERE clause Reduces query execution time on source database
Daily incremental loads Use ImportMode = Append with date filters Avoids full table recreates; only adds new data
Reference data refresh Use ImportMode = Refill Faster than Replace (no table drop/create overhead)
Schema changes Use ImportMode = Replace Auto-creates table with new schema from query results
Snowflake sources Ensure proper data type mapping Variant types map to nvarchar(max) in SQL Server

Expected Performance

Error Handling

The task uses a fail-fast approach with no built-in retry logic:

// Connection errors, query errors, or bulk copy errors throw exceptions immediately
using (SqlConnection conn = new SqlConnection(connectionString))
{
    conn.OpenConnection(); // Throws if connection fails
    using (IDataReader reader = cmd.ExecuteReader())
    {
        importer.ImportData(reader); // Throws if bulk copy fails
    }
}

Important behaviors:

The task does NOT aggregate errors; any failure stops execution immediately.

Common Pitfalls

Issue Problem Solution
Connection string incorrect for DataSourceType Using SQL Server connection string with DataSourceType = Snowflake causes connection failure Ensure connection string format matches DataSourceType (Snowflake uses account=, user=, db= format)
ImportMode=Replace during business hours Users lose table access during drop/recreate operation Use ImportMode = Refill instead, or schedule during maintenance windows
Query timeout on large datasets Source query runs for hours and eventually times out Add indexed WHERE clause filters, or break into multiple smaller imports
Schema drift with Append mode Source query columns change but table schema remains old Use ImportMode = Replace when schema changes are detected, or drop table manually first
Data type mismatches Snowflake VARIANT or SQL Server geography types cause bulk copy errors Handle complex types in source query (cast to string or compatible type)
Missing target database permissions Task fails with "CREATE TABLE permission denied" Ensure connection string user has CREATE, DROP, and TRUNCATE permissions on target database
Partial import after failure Task fails mid-import, leaving partial data in table Use transactions in calling code, or manually clean up target table before retry

Code References