fix: Ensure that representative but not deployable snapshots use the Prod physical models consistently

[davem-bis]

Description

Fixes #5793.

Ensures that all snapshots that are representative but not deployable (they contain the same data as prod, but can never become prod) are just pass through views of the production physical table. A by-product of this is that they can never have any missing intervals.

Test Plan

• Expansion of existing unit tests for changed logic.
• Run through of INDIRECT_BREAKING_CHANGES snapshots to prove idempotence.
  • Run plan:

    1. sqlmesh init

       Option 1 throughout - basic DuckDB implementation with the demo models.

    2. sqlmesh plan

    3. Add 'ABC' AS extra_column to incremental_model.sql

    4. sqlmesh plan indirect_non_breaking

    5. sqlmesh run

    6. sqlmesh run indirect_non_breaking

    7. sqlmesh fetchdf "UPDATE sqlmesh._intervals SET end_ts = end_ts - 86400000, start_ts = start_ts - 86400000, last_altered_ts = last_altered_ts - 86400000"

       86400000 is the number of milliseconds in a day (246060*1000)

    8. sqlmesh run indirect_non_breaking

    9. sqlmesh run indirect_non_breaking

       Previously this would result in the intervals being re-run, it now correctly does nothing.

    10. Plan a new model named extra_model as a SELECT * FROM sqlmesh_example.full_model and confirm that it is selecting from the full_model physical model without the __dev extension.

  • Comments:

    • Model Name	Change Type	Deployability	Representativeness	Version Used In indirect_non_breaking
      seed_model	None	True	True	Same as prod
      incremental_model	NonBreaking	True	True	Separate physical model
      full_model	IndirectNonBreaking	False	True	Prod physical model

    • full_model uses the non-__dev physical model, and has no missing intervals.
• Run through of the above plan, but with vde_mode: dev_only.

Checklist

• I have run make style and fixed any issues
• I have added tests for my changes (if applicable)
• All existing tests pass (make fast-test)
• My commits are signed off (git commit -s) per the DCO

[davem-bis]

This needs an appropriate unit test.

[cmgoffena13]

@davem-bis -- Took a peak at this: looks like you're making the change a level down, so it affects plan as well. That would explain your comment about why in "dev_only" mode you're getting empty tables (probably because of the return []).

So we're seeing another bug that creates and backfills a __dev table during the plan as well. And by fixing the first bug, this __dev table is then not used and wasted compute, from what I understand. So we're kind of peeling back an onion here.

My recommendation:

1. In sqlmesh/core/scheduler.py in merged_missing_intervals, the function not the method, we filter out the snapshots. This change should propagate to the scheduler and to plan. So something like this:

snapshots_to_intervals = compute_interval_params(          # NEW
    snapshots,
    start=start or earliest_start_date(snapshots),
    end=end or now_timestamp(),
    deployability_index=deployability_index,
    execution_time=execution_time or now_timestamp(),
    restatements=restatements,
    start_override_per_model=start_override_per_model,
    end_override_per_model=end_override_per_model,
    ignore_cron=ignore_cron,
    end_bounded=end_bounded,
)
deployability_index = deployability_index or DeployabilityIndex.all_deployable()  # NEW
return {                                                   # NEW
    s: intervals                                           # NEW
    for s, intervals in snapshots_to_intervals.items()     # NEW
    if not (                                               # NEW
        deployability_index.is_representative(s)           # NEW
        and not deployability_index.is_deployable(s)       # NEW
    )                                                      # NEW
}  

2. In sqlmesh/core/plan/stage.py in _get_snapshots_to_create / _should_create -- we check the deployability index here as well to avoid creating a __dev table if it is_representative. You're already doing this.

I think its your changes to core/snapshot/definition.py that are giving you trouble here. You're overriding logic too early. Appreciate you trying to solve both issues in one pass!