Implementation:TobikoData Sqlmesh Context Plan For Incremental

Knowledge Sources	SQLMesh SQLMesh Docs
Domains	Data_Engineering, Incremental_Processing
Last Updated	2026-02-07 00:00 GMT

Overview

Concrete tool for testing incremental model logic in development environments by creating plans with specific time ranges provided by SQLMesh.

Description

Context.plan() is the primary method for creating and previewing changes to data environments in SQLMesh. When used for incremental model testing, it generates an execution plan that shows which intervals will be processed, what changes are being made, and what downstream impacts will occur.

The method compares the local model definitions against the target environment (typically 'dev' for testing), identifies which models have changed, computes the missing intervals that need processing, and presents an interactive preview showing the scope of work. For incremental models specifically, the start and end parameters control which time range to test, enabling focused validation of specific intervals.

The plan includes detailed information about backfill requirements, breaking versus non-breaking changes, and estimated processing costs, allowing engineers to validate incremental logic before applying changes to the environment.

Usage

Use Context.plan() when developing or modifying incremental models to preview changes in a safe development environment. Specify start and end dates to test specific time ranges that represent edge cases or typical scenarios. Use ignore_cron to process all intervals in the range regardless of scheduled times.

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/core/context.py:L1311-1443

Signature

@python_api_analytics
def plan(
    self,
    environment: t.Optional[str] = None,
    *,
    start: t.Optional[TimeLike] = None,
    end: t.Optional[TimeLike] = None,
    execution_time: t.Optional[TimeLike] = None,
    create_from: t.Optional[str] = None,
    skip_tests: t.Optional[bool] = None,
    restate_models: t.Optional[t.Iterable[str]] = None,
    no_gaps: t.Optional[bool] = None,
    skip_backfill: t.Optional[bool] = None,
    empty_backfill: t.Optional[bool] = None,
    forward_only: t.Optional[bool] = None,
    allow_destructive_models: t.Optional[t.Collection[str]] = None,
    allow_additive_models: t.Optional[t.Collection[str]] = None,
    no_prompts: t.Optional[bool] = None,
    auto_apply: t.Optional[bool] = None,
    no_auto_categorization: t.Optional[bool] = None,
    effective_from: t.Optional[TimeLike] = None,
    include_unmodified: t.Optional[bool] = None,
    select_models: t.Optional[t.Collection[str]] = None,
    backfill_models: t.Optional[t.Collection[str]] = None,
    categorizer_config: t.Optional[CategorizerConfig] = None,
    enable_preview: t.Optional[bool] = None,
    no_diff: t.Optional[bool] = None,
    run: t.Optional[bool] = None,
    diff_rendered: t.Optional[bool] = None,
    skip_linter: t.Optional[bool] = None,
    explain: t.Optional[bool] = None,
    ignore_cron: t.Optional[bool] = None,
    min_intervals: t.Optional[int] = None,
) -> Plan:

Import

from sqlmesh import Context

I/O Contract

Inputs

Name	Type	Required	Description
environment	str	No	Target environment name (default: 'dev' for testing)
start	TimeLike	No	Start date/time for interval processing in the plan
end	TimeLike	No	End date/time for interval processing in the plan
execution_time	TimeLike	No	Reference time for plan execution (defaults to now)
ignore_cron	bool	No	Process all intervals regardless of cron schedule
skip_backfill	bool	No	Preview changes without processing intervals
forward_only	bool	No	Apply changes only to future intervals
select_models	Collection[str]	No	Filter which models to include in the plan
no_prompts	bool	No	Disable interactive prompts (default: False)
auto_apply	bool	No	Automatically apply plan after creation (default: False)

Outputs

Name	Type	Description
plan	Plan	Populated plan object containing interval details, changes, and execution strategy

Usage Examples

Basic Incremental Testing

from sqlmesh import Context

# Initialize context from project directory
context = Context(paths=".")

# Test incremental model for last 7 days in dev environment
plan = context.plan(
    environment='dev',
    start='2024-01-01',
    end='2024-01-07'
)

# Interactive prompt shows:
# - Which models changed
# - How many intervals will be processed
# - Preview of data changes
# - Option to apply or cancel

Test Specific Intervals with ignore_cron

from sqlmesh import Context

context = Context()

# Test edge case: month boundary
# Process all intervals in range regardless of cron schedule
plan = context.plan(
    environment='test_month_boundary',
    start='2024-01-30',
    end='2024-02-02',
    ignore_cron=True,
    auto_apply=True  # Apply automatically for automated testing
)

print(f"Processed {len(plan.missing_intervals)} intervals")

Preview Without Backfill

from sqlmesh import Context

context = Context()

# Preview incremental changes without actually processing data
plan = context.plan(
    environment='dev',
    start='2024-01-01',
    end='2024-01-31',
    skip_backfill=True  # Skip actual data processing
)

# Shows what would happen but doesn't execute
for model_name, intervals in plan.missing_intervals.items():
    print(f"{model_name}: {len(intervals)} intervals")

Test Forward-Only Changes

from sqlmesh import Context
from datetime import datetime, timedelta

context = Context()

# Test forward-only change on recent data only
today = datetime.now()
week_ago = today - timedelta(days=7)

plan = context.plan(
    environment='dev',
    start=week_ago.strftime('%Y-%m-%d'),
    end=today.strftime('%Y-%m-%d'),
    forward_only=True,  # Changes apply only going forward
    effective_from=today.strftime('%Y-%m-%d')  # Start applying changes from today
)

Test Selected Models Only

from sqlmesh import Context

context = Context()

# Test only specific incremental model and its dependencies
plan = context.plan(
    environment='dev',
    start='2024-01-01',
    end='2024-01-05',
    select_models=['my_schema.daily_aggregates'],
    ignore_cron=True
)

# Only processes selected model and upstream dependencies
# Useful for focused testing during development

Automated Testing in CI/CD

from sqlmesh import Context
import sys

def test_incremental_model():
    context = Context()

    try:
        # Test last 3 days of data
        plan = context.plan(
            environment='ci_test',
            start='2024-01-01',
            end='2024-01-03',
            no_prompts=True,  # No interactive prompts in CI
            auto_apply=True,  # Apply automatically
            skip_tests=False  # Run unit tests
        )

        # Validate results
        if plan.context_diff.has_changes:
            print("Changes detected and tested successfully")
            return 0
        else:
            print("No changes detected")
            return 0

    except Exception as e:
        print(f"Plan failed: {e}")
        return 1

if __name__ == '__main__':
    sys.exit(test_incremental_model())

Test with Restatement

from sqlmesh import Context

context = Context()

# Test incremental model with upstream restatement
# Simulates scenario where source data was corrected
plan = context.plan(
    environment='dev',
    start='2024-01-01',
    end='2024-01-31',
    restate_models=['upstream_schema.raw_events'],
    ignore_cron=True
)

# Shows which intervals will be reprocessed due to restatement
# Validates that restatement propagates correctly through DAG

Interactive Development Workflow

from sqlmesh import Context

context = Context()

# Typical development workflow:
# 1. Modify incremental model logic
# 2. Test on small date range

# Test single day first
plan = context.plan(
    environment='dev_iteration_1',
    start='2024-01-15',
    end='2024-01-15',
    ignore_cron=True
)

# Review output, fix issues, then test broader range
plan = context.plan(
    environment='dev_iteration_2',
    start='2024-01-01',
    end='2024-01-31',
    ignore_cron=True
)

# Finally test with actual cron behavior
plan = context.plan(
    environment='dev_final',
    start='2024-01-01',
    end='2024-01-31',
    ignore_cron=False  # Respect cron schedule
)

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Dev_Environment_Testing_For_Incremental

• Heuristic:TobikoData_Sqlmesh_Forward_Only_Safety