Repository: TobikoData/sqlmesh
Pull request #2367 (2024-03-29 20:27:52)

chore: improve recommended state sync documentation
This PR improves the way we recommend that users use a separate state sync for production deployments in two ways:

* Moves information about this recommendation from FAQ to state connection configuration section
* More strongly emphasizes that this is recommended and explicitly lists what is recommended
* Warning log is output if not using the recommended state backend if in production


Patch:
diff --git a/docs/faq/faq.md b/docs/faq/faq.md
index 4d3f82e3..74014753 100644
--- a/docs/faq/faq.md
+++ b/docs/faq/faq.md
@@ -147,14 +147,7 @@
 ## Databases/Engines
 
 ??? question "What databases/engines does SQLMesh work with?"
-    SQLMesh works with BigQuery, Databricks, DuckDB, MySQL, PostgreSQL, GCP PostgreSQL, Redshift, Snowflake, and Spark. See [this page](../integrations/overview.md) for more information.
-
-??? question "When would you use different databases for executing data transformations and storing state information?"
-    SQLMesh requires storing state information about projects and when their transformations were run. By default, it stores this information in the same database where the models run.
-
-    Unlike data transformations, storing state information requires database transactions. Some databases, like BigQuery, aren’t optimized for executing transactions, so storing state information in them can slow down your project. If this occurs, you can store state information in a different database, such as PostgreSQL, that executes transactions more efficiently.
-
-    Learn more about storing state information at the [configuration reference page](../reference/configuration.md#state-connection).
+    See [this page](../integrations/overview.md) for the list of currently supported engines.
 
 ## Scheduling
 
diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md
index 151fd39e..be1af2a7 100644
--- a/docs/guides/configuration.md
+++ b/docs/guides/configuration.md
@@ -494,10 +494,26 @@ These pages describe the connection configuration options for each execution eng
 #### State connection
 
 Configuration for the state backend connection if different from the data warehouse connection.
+**Using the same connection for data warehouse and state is only recommended for non-production deployments of SQLMesh.**
+Unlike data transformations, storing state information requires database transactions. Data warehouses aren’t optimized for executing transactions, so storing state information in them can slow down your project. 
+Even worse data corruption can occur with simultaneous writes to the same table.
+Therefore, using your data warehouse is fine for testing but once you start running SQLMesh in production, you should use a dedicated state connection.
 
-The data warehouse connection is used if the `state_connection` key is not specified, unless the configuration uses an Airflow or Google Cloud Composer scheduler. If using one of those schedulers and no state connection is specified, the state connection defaults to the scheduler's database.
+Recommended state backend engines for production deployments:
+
+* [Postgres](../integrations/engines/postgres.md)
+* [GCP Postgres](../integrations/engines/gcp-postgres.md)
+
+Other supported state backend engines (less tested than recommended):
+
+* [MySQL](../integrations/engines/mysql.md)
+
+Ineligible state backends even for development:
+
+* [Spark](../integrations/engines/spark.md)
+* [Trino](../integrations/engines/trino.md)
 
-NOTE: Spark and Trino engines may not be used for the state connection.
+The data warehouse connection is used if the `state_connection` key is not specified, unless the configuration uses an Airflow or Google Cloud Composer scheduler. If using one of those schedulers and no state connection is specified, the state connection defaults to the scheduler's database.
 
 Example postgres state connection configuration:
 
diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
index e7887bf3..a8217525 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -36,8 +36,11 @@ else:
 
 logger = logging.getLogger(__name__)
 
+RECOMMENDED_STATE_SYNC_ENGINES = {"postgres", "gcp_postgres", "mysql"}
+
 
 class ConnectionConfig(abc.ABC, BaseConfig):
+    type_: str
     concurrent_tasks: int
     register_comments: bool
 
@@ -76,6 +79,11 @@ class ConnectionConfig(abc.ABC, BaseConfig):
         """A function that is called to initialize the cursor"""
         return None
 
+    @property
+    def is_recommended_for_state_sync(self) -> bool:
+        """Whether this connection is recommended for being used as a state sync for production state syncs"""
+        return self.type_ in RECOMMENDED_STATE_SYNC_ENGINES
+
     def create_engine_adapter(self, register_comments_override: bool = False) -> EngineAdapter:
         """Returns a new instance of the Engine Adapter."""
         return self._engine_adapter(
diff --git a/sqlmesh/core/config/scheduler.py b/sqlmesh/core/config/scheduler.py
index df8f1086..6244efa2 100644
--- a/sqlmesh/core/config/scheduler.py
+++ b/sqlmesh/core/config/scheduler.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import abc
+import logging
 import sys
 import typing as t
 
@@ -33,6 +34,9 @@ else:
     from typing_extensions import Annotated, Literal
 
 
+logger = logging.getLogger(__name__)
+
+
 class _SchedulerConfig(abc.ABC):
     """Abstract base class for Scheduler configurations."""
 
@@ -66,13 +70,19 @@ class _SchedulerConfig(abc.ABC):
 
 class _EngineAdapterStateSyncSchedulerConfig(_SchedulerConfig):
     def create_state_sync(self, context: GenericContext) -> StateSync:
-        state_connection = context.config.get_state_connection(context.gateway)
-        engine_adapter = (state_connection or context._connection_config).create_engine_adapter()
+        state_connection = (
+            context.config.get_state_connection(context.gateway) or context._connection_config
+        )
+        engine_adapter = state_connection.create_engine_adapter()
         if not engine_adapter.SUPPORTS_ROW_LEVEL_OP:
             raise ConfigError(
                 f"The {engine_adapter.DIALECT.upper()} engine cannot be used to store SQLMesh state - please specify a different `state_connection` engine."
                 + " See https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#gateways for more information."
             )
+        if not state_connection.is_recommended_for_state_sync:
+            logger.warning(
+                f"{state_connection.type_} is not recommended to be used as a state sync for production deployments. Please see documentation ( https://sqlmesh.readthedocs.io/en/latest/guides/configuration/#state-connection) for list of recommended engines to be used for storing state and further details."
+            )
         schema = context.config.get_state_schema(context.gateway)
         return EngineAdapterStateSync(
             engine_adapter, schema=schema, context_path=context.path, console=context.console
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 6acd5ded..dba2ef61 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -772,7 +772,10 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
             A list of all the missing intervals as epoch timestamps.
         """
         # If the node says that it has an end, and we are wanting to load past it, then we can return no empty intervals
-        if self.node.end and to_datetime(start) > to_datetime(self.node.end):
+        # Also if a node's start is after the end of the range we are checking then we can return no empty intervals
+        if (self.node.end and to_datetime(start) > to_datetime(self.node.end)) or (
+            self.node.start and to_datetime(self.node.start) > to_datetime(end)
+        ):
             return []
         # If the amount of time being checked is less than the size of a single interval then we
         # know that there can't being missing intervals within that range and return


Test patch:
diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py
index 497a32cd..d388d721 100644
--- a/tests/cli/test_cli.py
+++ b/tests/cli/test_cli.py
@@ -1,3 +1,5 @@
+import logging
+from contextlib import contextmanager
 from os import path
 from pathlib import Path
 
@@ -18,6 +20,15 @@ def runner() -> CliRunner:
     return CliRunner()
 
 
+@contextmanager
+def disable_logging():
+    logging.disable(logging.CRITICAL)
+    try:
+        yield
+    finally:
+        logging.disable(logging.NOTSET)
+
+
 def create_example_project(temp_dir) -> None:
     """
     Sets up CLI tests requiring a real SQLMesh project by:
@@ -569,14 +580,15 @@ def test_run_dev(runner, tmp_path):
 
 
 @freeze_time(FREEZE_TIME)
-def test_run_cron_not_elapsed(runner, tmp_path):
+def test_run_cron_not_elapsed(runner, tmp_path, caplog):
     create_example_project(tmp_path)
     init_prod_and_backfill(runner, tmp_path)
 
     # No error and no output if `prod` environment exists and cron has not elapsed
-    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
+    with disable_logging():
+        result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
     assert result.exit_code == 0
-    assert result.output == ""
+    assert result.output.strip() == ""
 
 
 def test_run_cron_elapsed(runner, tmp_path):
@@ -616,16 +628,17 @@ def test_table_name(runner, tmp_path):
     # Create and backfill `prod` environment
     create_example_project(tmp_path)
     init_prod_and_backfill(runner, tmp_path)
-    result = runner.invoke(
-        cli,
-        [
-            "--log-file-dir",
-            tmp_path,
-            "--paths",
-            tmp_path,
-            "table_name",
-            "sqlmesh_example.full_model",
-        ],
-    )
+    with disable_logging():
+        result = runner.invoke(
+            cli,
+            [
+                "--log-file-dir",
+                tmp_path,
+                "--paths",
+                tmp_path,
+                "table_name",
+                "sqlmesh_example.full_model",
+            ],
+        )
     assert result.exit_code == 0
     assert result.output.startswith("db.sqlmesh__sqlmesh_example.sqlmesh_example__full_model__")
diff --git a/tests/core/test_connection_config.py b/tests/core/test_connection_config.py
index 2e7bda2b..7fd81f0d 100644
--- a/tests/core/test_connection_config.py
+++ b/tests/core/test_connection_config.py
@@ -8,6 +8,9 @@ from sqlmesh.core.config.connection import (
     BigQueryConnectionConfig,
     ConnectionConfig,
     DuckDBConnectionConfig,
+    GCPPostgresConnectionConfig,
+    MySQLConnectionConfig,
+    PostgresConnectionConfig,
     SnowflakeConnectionConfig,
     TrinoAuthenticationMethod,
     _connection_config_validator,
@@ -285,6 +288,7 @@ def test_trino(make_config):
     assert config.http_scheme == "https"
     assert config.port == 443
     assert config.get_catalog() == "catalog"
+    assert config.is_recommended_for_state_sync is False
 
     # Validate Basic Auth
     config = make_config(method="basic", password="password", **required_kwargs)
@@ -376,6 +380,7 @@ def test_duckdb(make_config):
         connector_config={"foo": "bar"},
     )
     assert isinstance(config, DuckDBConnectionConfig)
+    assert config.is_recommended_for_state_sync is False
 
 
 @pytest.mark.parametrize(
@@ -531,3 +536,40 @@ def test_bigquery(make_config):
     assert config.project == "project"
     assert config.execution_project == "execution_project"
     assert config.get_catalog() == "project"
+    assert config.is_recommended_for_state_sync is False
+
+
+def test_postgres(make_config):
+    config = make_config(
+        type="postgres",
+        host="host",
+        user="user",
+        password="password",
+        port=5432,
+        database="database",
+    )
+    assert isinstance(config, PostgresConnectionConfig)
+    assert config.is_recommended_for_state_sync is True
+
+
+def test_gcp_postgres(make_config):
+    config = make_config(
+        type="gcp_postgres",
+        instance_connection_string="something",
+        user="user",
+        password="password",
+        db="database",
+    )
+    assert isinstance(config, GCPPostgresConnectionConfig)
+    assert config.is_recommended_for_state_sync is True
+
+
+def test_mysql(make_config):
+    config = make_config(
+        type="mysql",
+        host="host",
+        user="user",
+        password="password",
+    )
+    assert isinstance(config, MySQLConnectionConfig)
+    assert config.is_recommended_for_state_sync is True
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index 6595abbd..4f931991 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -5,12 +5,14 @@ from datetime import date, timedelta
 from tempfile import TemporaryDirectory
 from unittest.mock import PropertyMock, call, patch
 
+import freezegun
 import pytest
 from pytest_mock.plugin import MockerFixture
 from sqlglot import parse_one
 from sqlglot.errors import SchemaError
 
 import sqlmesh.core.constants
+import sqlmesh.core.dialect as d
 from sqlmesh.core.config import (
     Config,
     DuckDBConnectionConfig,
@@ -24,7 +26,13 @@ from sqlmesh.core.dialect import parse, schema_
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.model import load_sql_based_model
 from sqlmesh.core.plan import BuiltInPlanEvaluator, PlanBuilder
-from sqlmesh.utils.date import make_inclusive_end, now, to_date, yesterday_ds
+from sqlmesh.utils.date import (
+    make_inclusive_end,
+    now,
+    to_date,
+    to_timestamp,
+    yesterday_ds,
+)
 from sqlmesh.utils.errors import ConfigError
 from tests.utils.test_filesystem import create_temp_file
 
@@ -482,6 +490,42 @@ def test_plan_default_end(sushi_context_pre_scheduling: Context):
     assert forward_only_dev_plan.start == plan_end
 
 
+@pytest.mark.slow
+def test_plan_start_ahead_of_end(copy_to_temp_path):
+    path = copy_to_temp_path("examples/sushi")
+    with freezegun.freeze_time("2024-01-02 00:00:00"):
+        context = Context(paths=path, config="local_config")
+        context.plan("prod", no_prompts=True, auto_apply=True)
+        assert context.state_sync.max_interval_end_for_environment("prod") == to_timestamp(
+            "2024-01-02"
+        )
+        context.close()
+    with freezegun.freeze_time("2024-01-03 00:00:00"):
+        context = Context(paths=path, config="local_config")
+        expression = d.parse(
+            """
+                MODEL(
+            name sushi.hourly,
+            KIND FULL,
+            cron '@hourly',
+            start '2024-01-02 12:00:00',
+        );
+
+        SELECT 1"""
+        )
+        model = load_sql_based_model(expression, default_catalog=context.default_catalog)
+        context.upsert_model(model)
+        context.plan("prod", no_prompts=True, auto_apply=True)
+        # Since the new start is ahead of the latest end loaded for prod, the table is deployed as empty
+        # This isn't considered a gap since prod has not loaded these intervals yet
+        # As a results the max interval end is unchanged and the table is empty
+        assert context.state_sync.max_interval_end_for_environment("prod") == to_timestamp(
+            "2024-01-02"
+        )
+        assert context.engine_adapter.fetchone("SELECT COUNT(*) FROM sushi.hourly")[0] == 0
+        context.close()
+
+
 @pytest.mark.slow
 def test_schema_error_no_default(sushi_context_pre_scheduling) -> None:
     context = sushi_context_pre_scheduling
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 88ae4a78..e33f9b91 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -1249,7 +1249,7 @@ def test_inclusive_exclusive_hourly(make_snapshot):
             owner="owner",
             dialect="",
             cron="@hourly",
-            start="1 week ago",
+            start="2023-01-29",
             query=parse_one("SELECT id, @end_ds as ds FROM name"),
         )
     )