nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: rancher/fleet
Pull request #1647 (2023-07-12 07:14:54)

Fix Helm release name generation to prevent double dashes
A Helm release name is generated for each bundle deployed by Fleet. In case that name is longer than 53 characters, the maximum length of Helm release names, it is then truncated to 46 characters and the remaining characters are a dash followed by a hash.

This fixes an edge case where the truncated release name already ends with a dash, leading to the final release name (in the form `<truncated>-<hash>`) containing a double dash. Since this caused discrepancies between deployed release names and what the Fleet agent would expect to be deployed, Fleet would frequently delete and re-create bundles as a result.

Refers to #1511

## Test

To test this pull request, you can run the following commands:

```shell
go test ./internal/name/...
```


Patch:
diff --git a/internal/name/name.go b/internal/name/name.go
index 3fb5cb6a..ea49a468 100644
--- a/internal/name/name.go
+++ b/internal/name/name.go
@@ -32,7 +32,17 @@ func Limit(s string, count int) string {
 	if count <= 6 {
 		return s[:count]
 	}
-	return fmt.Sprintf("%s-%s", s[:count-6], Hex(s, 5))
+
+	separator := "-"
+	nbCharsBeforeTrim := count - 6
+
+	// If the last character of the truncated string is the separator, include it instead of the separator.
+	if string(s[nbCharsBeforeTrim-1]) == separator {
+		separator = ""
+		nbCharsBeforeTrim++
+	}
+
+	return fmt.Sprintf("%s%s%s", s[:nbCharsBeforeTrim], separator, Hex(s, 5))
 }
 
 // Hex returns a hex-encoded hash of the string and truncates it to length.


Test patch:
diff --git a/internal/name/name_test.go b/internal/name/name_test.go
index ad02ab8d..38b35abf 100644
--- a/internal/name/name_test.go
+++ b/internal/name/name_test.go
@@ -31,6 +31,7 @@ var _ = Describe("Name", func() {
 			{arg: "12345678", n: 8, result: "12345678"},
 			{arg: "12345678", n: 7, result: "1-25d55"},
 			{arg: "123456789", n: 8, result: "12-25f9e"},
+			{arg: "1-3456789", n: 8, result: "1-39b657"}, // no double dash in the result
 		}
 
 		It("matches expected results", func() {
@@ -51,6 +52,8 @@ var _ = Describe("Name", func() {
 			{arg: "long-name-test-0.App_ ", result: "long-name-test-0-app-5bf6b3fb"},
 			{arg: "long-name-test--App_-_12.factor", result: "long-name-test-app-12-factor-0efbac37"},
 			{arg: "bundle.name.example.com", result: "bundle-name-example-com-645ef821"},
+			// no double dash in the result
+			{arg: str53[0:46] + "-1234567", result: "1234567890123456789012345678901234567890123456-1d0bce"},
 		}
 
 		It("matches expected results", func() {
Repository: TobikoData/sqlmesh
Pull request #2754 (2024-06-07 16:18:13)

fix: adding subquery udtf projections should be non-breaking


Patch:
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index aca1bcf7..aa3eaa40 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -1094,9 +1094,17 @@ class SqlModel(_SqlBasedModel):
         for edit in edits:
             if isinstance(edit, Insert):
                 expr = edit.expression
-                if _is_udtf(expr) or (
-                    not _is_projection(expr) and expr.parent not in inserted_expressions
-                ):
+                if _is_udtf(expr):
+                    # projection subqueries do not change cardinality, engines don't allow these to return
+                    # more than one row of data
+                    parent = expr.find_ancestor(exp.Subquery)
+
+                    if not parent:
+                        return None
+
+                    expr = parent
+
+                if not _is_projection(expr) and expr.parent not in inserted_expressions:
                     return None
             elif not isinstance(edit, Keep):
                 return None


Test patch:
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 45405a85..88860a3a 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -1041,6 +1041,28 @@ def test_categorize_change_sql(make_snapshot):
         is SnapshotChangeCategory.NON_BREAKING
     )
 
+    # A UDTF subquery has been added.
+    assert (
+        categorize_change(
+            new=make_snapshot(
+                SqlModel(name="a", query=parse_one("select 1, ds, (select x from unnest(a) x)"))
+            ),
+            old=old_snapshot,
+            config=config,
+        )
+        is SnapshotChangeCategory.NON_BREAKING
+    )
+    assert (
+        categorize_change(
+            new=make_snapshot(
+                SqlModel(name="a", query=parse_one("select 1, ds, (select x from posexplode(a) x)"))
+            ),
+            old=old_snapshot,
+            config=config,
+        )
+        is SnapshotChangeCategory.NON_BREAKING
+    )
+
 
 def test_categorize_change_seed(make_snapshot, tmp_path):
     config = CategorizerConfig(seed=AutoCategorizationMode.SEMI)
Repository: TobikoData/sqlmesh
Pull request #2774 (2024-06-14 14:02:16)

Fix: convert NaN/NaT (np.nan) values to None
Context:
- https://tobiko-data.slack.com/archives/C044BRE5W4S/p1718294204075889
- https://stackoverflow.com/questions/41444396/psycopg2-acceptable-date-datetime-values

Was able to reproduce the issue for a postgres connection: `invalid input syntax for type timestamp: "NaT"`. I also verified that `sqlmesh plan` succeeds with this fix.


Patch:
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index aa3eaa40..556d8b00 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -10,6 +10,7 @@ from functools import cached_property
 from pathlib import Path
 
 import pandas as pd
+import numpy as np
 from astor import to_source
 from pydantic import Field
 from sqlglot import diff, exp
@@ -1172,6 +1173,7 @@ class SeedModel(_SqlBasedModel):
         datetime_columns = []
         bool_columns = []
         string_columns = []
+
         for name, tpe in (self.columns_to_types_ or {}).items():
             if tpe.this in (exp.DataType.Type.DATE, exp.DataType.Type.DATE32):
                 date_columns.append(name)
@@ -1196,7 +1198,7 @@ class SeedModel(_SqlBasedModel):
                 cond=lambda x: x.notna(),  # type: ignore
                 other=df[string_columns].astype(str),  # type: ignore
             )
-            yield df
+            yield df.replace({np.nan: None})
 
     @property
     def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]:


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 76baccc3..3baf9a30 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -851,8 +851,8 @@ def test_seed_model_custom_types(tmp_path):
 
     with open(model_csv_path, "w", encoding="utf-8") as fd:
         fd.write(
-            """key,ds_date,ds_timestamp,b_a,b_b,i,i_str
-123,2022-01-01,2022-01-01,false,0,321,321
+            """key,ds_date,ds_timestamp,b_a,b_b,i,i_str,empty_date
+123,2022-01-01,2022-01-01,false,0,321,321,
 """
         )
 
@@ -867,6 +867,7 @@ def test_seed_model_custom_types(tmp_path):
             "b_b": "boolean",
             "i": "int",
             "i_str": "text",
+            "empty_date": "date",
         },
     )
 
@@ -893,6 +894,9 @@ def test_seed_model_custom_types(tmp_path):
     assert df["i_str"].dtype == "object"
     assert df["i_str"].iloc[0] == "321"
 
+    assert df["empty_date"].dtype == "object"
+    assert df["empty_date"].iloc[0] is None
+
 
 def test_seed_with_special_characters_in_column(tmp_path, assert_exp_eq):
     config = Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))
Repository: TobikoData/sqlmesh
Pull request #2798 (2024-06-20 21:48:29)

Feat: Support token authentication for Airflow REST API
This enables support for Airflow cloud providers like Astronomer.


Patch:
diff --git a/sqlmesh/core/config/scheduler.py b/sqlmesh/core/config/scheduler.py
index 1278d2ed..093bdd4d 100644
--- a/sqlmesh/core/config/scheduler.py
+++ b/sqlmesh/core/config/scheduler.py
@@ -218,6 +218,7 @@ class AirflowSchedulerConfig(_BaseAirflowSchedulerConfig, BaseConfig):
     airflow_url: str = "http://localhost:8080/"
     username: str = "airflow"
     password: str = "airflow"
+    token: t.Optional[str] = None
     dag_run_poll_interval_secs: int = 10
     dag_creation_poll_interval_secs: int = 30
     dag_creation_max_retry_attempts: int = 10
@@ -236,8 +237,10 @@ class AirflowSchedulerConfig(_BaseAirflowSchedulerConfig, BaseConfig):
 
     def get_client(self, console: t.Optional[Console] = None) -> AirflowClient:
         session = Session()
-        session.headers.update({"Content-Type": "application/json"})
-        session.auth = (self.username, self.password)
+        if self.token is None:
+            session.auth = (self.username, self.password)
+        else:
+            session.headers.update({"Authorization": f"Bearer {self.token}"})
 
         return AirflowClient(
             session=session,
diff --git a/sqlmesh/schedulers/airflow/client.py b/sqlmesh/schedulers/airflow/client.py
index 3bafb42b..f2d1a657 100644
--- a/sqlmesh/schedulers/airflow/client.py
+++ b/sqlmesh/schedulers/airflow/client.py
@@ -230,6 +230,7 @@ class AirflowClient(BaseAirflowClient):
         response = self._session.post(
             urljoin(self._airflow_url, PLANS_PATH),
             data=request.json(),
+            headers={"Content-Type": "application/json"},
         )
         raise_for_status(response)
 


Test patch:
diff --git a/tests/schedulers/airflow/test_client.py b/tests/schedulers/airflow/test_client.py
index 02506dad..9dd7cd88 100644
--- a/tests/schedulers/airflow/test_client.py
+++ b/tests/schedulers/airflow/test_client.py
@@ -77,6 +77,7 @@ def test_apply_plan(mocker: MockerFixture, snapshot: Snapshot):
     args, data = apply_plan_mock.call_args_list[0]
 
     assert args[0] == "http://localhost:8080/sqlmesh/api/v1/plans"
+    assert data["headers"] == {"Content-Type": "application/json"}
     assert json.loads(data["data"]) == {
         "new_snapshots": [
             {
Repository: TobikoData/sqlmesh
Pull request #2803 (2024-06-21 16:42:09)

Fix: another snowflake staged file path edge case

izeigerman: Looks good. Any chance we can add a test for this?
georgesittas: > Looks good. Any chance we can add a test for this?

[Donezo](https://github.com/TobikoData/sqlmesh/pull/2803/commits/90ef83f8aee5f53fc7c2a7bcbedf6aa07c481187) 👍

Patch:
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index 2f121ccf..eea88685 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -412,15 +412,17 @@ def _parse_table_parts(
 ) -> exp.Table:
     index = self._index
     table = self.__parse_table_parts(schema=schema, is_db_reference=is_db_reference)  # type: ignore
+
     table_arg = table.this
+    name = table_arg.name
 
-    if isinstance(table_arg, exp.Var) and table_arg.name.startswith(SQLMESH_MACRO_PREFIX):
+    if isinstance(table_arg, exp.Var) and name.startswith(SQLMESH_MACRO_PREFIX):
         # Macro functions do not clash with the staged file syntax, so we can safely parse them
-        if self._prev.token_type == TokenType.STRING or "(" in table_arg.name:
+        if self._prev.token_type == TokenType.STRING or any(ch in name for ch in ("(", "{")):
             self._retreat(index)
             return Parser._parse_table_parts(self, schema=schema, is_db_reference=is_db_reference)
 
-        table_arg.replace(MacroVar(this=table_arg.name[1:]))
+        table_arg.replace(MacroVar(this=name[1:]))
         return StagedFilePath(**table.args)
 
     return table


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 97f8aa97..9a6a5527 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -4043,6 +4043,24 @@ def test_variables_in_templates() -> None:
         == "SELECT 'combo' AS \"col_test_value_overridden_value_col_in_memory\""
     )
 
+    model = load_sql_based_model(
+        parse(
+            """
+        MODEL(
+          name @{some_var}.bar,
+          dialect snowflake
+        );
+
+        SELECT 1 AS c
+        """
+        ),
+        variables={
+            "some_var": "foo",
+        },
+    )
+
+    assert model.name == "foo.bar"
+
 
 def test_variables_jinja():
     expressions = parse(
Repository: TobikoData/sqlmesh
Pull request #2832 (2024-06-26 22:26:43)

Fix: Support for restatemetns when using a state connection to submit plans to the Airflow cluster


Patch:
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 718ae07f..0e5e98e0 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -383,7 +383,7 @@ class StateBasedAirflowPlanEvaluator(BaseAirflowPlanEvaluator):
             no_gaps=plan.no_gaps,
             skip_backfill=plan.skip_backfill,
             request_id=plan_request_id,
-            restatements=plan.restatements or {},
+            restatements={s.name: i for s, i in (plan.restatements or {}).items()},
             notification_targets=self.notification_targets or [],
             backfill_concurrent_tasks=self.backfill_concurrent_tasks,
             ddl_concurrent_tasks=self.ddl_concurrent_tasks,


Test patch:
diff --git a/tests/core/test_plan_evaluator.py b/tests/core/test_plan_evaluator.py
index b7847238..6726139b 100644
--- a/tests/core/test_plan_evaluator.py
+++ b/tests/core/test_plan_evaluator.py
@@ -191,3 +191,40 @@ def test_update_intervals_for_new_snapshots(
     else:
         assert not snapshot.dev_intervals
         state_sync_mock.add_interval.assert_not_called()
+
+
+def test_state_based_airflow_evaluator_with_restatements(
+    sushi_context: Context, mocker: MockerFixture
+):
+    model_fqn = sushi_context.get_model("sushi.waiter_revenue_by_day").fqn
+    downstream_model_fqn = sushi_context.get_model("sushi.top_waiters").fqn
+
+    plan = PlanBuilder(
+        sushi_context._context_diff("prod"),
+        sushi_context.engine_adapter.SCHEMA_DIFFER,
+        restate_models=[sushi_context.get_model("sushi.waiter_revenue_by_day").fqn],
+    ).build()
+
+    mwaa_client_mock = mocker.Mock()
+    mwaa_client_mock.wait_for_dag_run_completion.return_value = True
+    mwaa_client_mock.wait_for_first_dag_run.return_value = "test_plan_application_dag_run_id"
+    mwaa_client_mock.set_variable.return_value = "", ""
+
+    state_sync_mock = mocker.Mock()
+
+    plan_dag_spec_mock = mocker.Mock()
+
+    create_plan_dag_spec_mock = mocker.patch("sqlmesh.schedulers.airflow.plan.create_plan_dag_spec")
+    create_plan_dag_spec_mock.return_value = plan_dag_spec_mock
+
+    plan_dag_state_mock = mocker.Mock()
+    mocker.patch(
+        "sqlmesh.schedulers.airflow.plan.PlanDagState.from_state_sync",
+        return_value=plan_dag_state_mock,
+    )
+
+    evaluator = MWAAPlanEvaluator(mwaa_client_mock, state_sync_mock)
+    evaluator.evaluate(plan)
+
+    plan_application_request = create_plan_dag_spec_mock.call_args[0][0]
+    assert plan_application_request.restatements.keys() == {model_fqn, downstream_model_fqn}
Repository: TobikoData/sqlmesh
Pull request #2842 (2024-06-30 22:45:25)

Fix: take the default_gateway into account when loading external models
Currently, the external model loading code expects that, if you want to load models relating to a specific gateway, you supply the `--gateway` flag when invoking the SQLMesh CLI.

However, some users override the default gateway in the config instead via setting the `SQLMESH__DEFAULT_GATEWAY` environment variable.

This change uses the config setting as a fallback when loading external models if the `--gateway` option is not supplied


Patch:
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index 5d75d5bc..effc6b3f 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -123,7 +123,9 @@ class Loader(abc.ABC):
         self._config_mtimes = {path: max(mtimes) for path, mtimes in config_mtimes.items()}
 
         macros, jinja_macros = self._load_scripts()
-        models = self._load_models(macros, jinja_macros, context.gateway)
+        models = self._load_models(
+            macros, jinja_macros, context.gateway or context.config.default_gateway
+        )
 
         for model in models.values():
             self._add_model_to_dag(model)


Test patch:
diff --git a/tests/core/test_schema_loader.py b/tests/core/test_schema_loader.py
index 986b2bd8..2e970914 100644
--- a/tests/core/test_schema_loader.py
+++ b/tests/core/test_schema_loader.py
@@ -232,6 +232,32 @@ def test_gateway_specific_external_models_mixed_with_others(tmpdir):
     assert external_models[0].name == '"memory"."landing"."source_table"'
 
 
+def test_gateway_specific_external_models_default_gateway(tmpdir: Path):
+    model_0 = {"name": "db.model0", "columns": {"a": "int"}}
+
+    model_1 = {"name": "db.model1", "gateway": "dev", "columns": {"a": "int"}}
+
+    model_2 = {"name": "db.model2", "gateway": "prod", "columns": {"a": "int"}}
+
+    with open(tmpdir / c.EXTERNAL_MODELS_YAML, "w", encoding="utf8") as fd:
+        YAML().dump([model_0, model_1, model_2], fd)
+
+    gateways = {
+        "dev": GatewayConfig(connection=DuckDBConnectionConfig()),
+        "prod": GatewayConfig(connection=DuckDBConnectionConfig()),
+    }
+
+    config = Config(gateways=gateways, default_gateway="prod")
+    ctx = Context(paths=[tmpdir], config=config)
+
+    ctx.load()
+
+    assert len(ctx.models) == 2
+    model_names = list(ctx.models.keys())
+    assert '"memory"."db"."model0"' in model_names
+    assert '"memory"."db"."model2"' in model_names
+
+
 def test_create_external_models_no_duplicates(tmpdir):
     config = Config(gateways={"": GatewayConfig(connection=DuckDBConnectionConfig())})
Repository: TobikoData/sqlmesh
Pull request #2892 (2024-07-10 13:51:02)

Chore: improve error reporting when loading seeds


Patch:
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index d02e3cc8..1d3cd180 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -1663,10 +1663,9 @@ def load_sql_based_model(
                 SeedKind(**seed_properties),
                 **common_kwargs,
             )
-        except Exception:
+        except Exception as ex:
             raise_config_error(
-                "The model definition must either have a SELECT query, a JINJA_QUERY block, or a valid Seed kind",
-                path,
+                f"The model definition must either have a SELECT query, a JINJA_QUERY block, or a valid Seed kind. {ex}."
             )
             raise
 


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index a76837a5..bbe73352 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -691,6 +691,21 @@ def test_seed():
     assert model.batch_size is None
 
 
+def test_seed_model_creation_error():
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.seed,
+            kind SEED (
+              path 'gibberish',
+            )
+        );
+    """
+    )
+    with pytest.raises(ConfigError, match="No such file or directory"):
+        load_sql_based_model(expressions)
+
+
 def test_seed_provided_columns():
     expressions = d.parse(
         """
Repository: TobikoData/sqlmesh
Pull request #2911 (2024-07-16 23:31:24)

Fix!: Ignore the warehouse property on non-dynamic tables
In snowflake, when creating managed models, the `warehouse` property was missed from the list of physical properties to exclude when creating dev preview tables.

This change adds it to the list.

Context: https://tobiko-data.slack.com/archives/C044BRE5W4S/p1721168591596139


Patch:
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index b6e3d77a..6dc22df3 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -183,7 +183,7 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin):
             table_properties = {k.upper(): v for k, v in table_properties.items()}
             # if we are creating a non-dynamic table; remove any properties that are only valid for dynamic tables
             if table_kind != self.MANAGED_TABLE_KIND:
-                for prop in {"TARGET_LAG", "REFRESH_MODE", "INITIALIZE"}:
+                for prop in {"WAREHOUSE", "TARGET_LAG", "REFRESH_MODE", "INITIALIZE"}:
                     table_properties.pop(prop, None)
 
             properties.extend(self._table_or_view_properties_to_expressions(table_properties))


Test patch:
diff --git a/tests/core/engine_adapter/test_snowflake.py b/tests/core/engine_adapter/test_snowflake.py
index 06160cb9..054cb2aa 100644
--- a/tests/core/engine_adapter/test_snowflake.py
+++ b/tests/core/engine_adapter/test_snowflake.py
@@ -253,3 +253,26 @@ def test_drop_managed_table(make_mocked_engine_adapter: t.Callable, mocker: Mock
         'DROP DYNAMIC TABLE "foo"',
         'DROP DYNAMIC TABLE IF EXISTS "foo"',
     ]
+
+
+def test_ctas_skips_dynamic_table_properties(make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
+
+    query = parse_one("SELECT a, b FROM source_table")
+    columns_to_types = {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")}
+
+    adapter.ctas(
+        table_name="test_table",
+        query_or_df=query,
+        columns_to_types=columns_to_types,
+        table_properties={
+            "warehouse": exp.to_identifier("foo"),
+            "target_lag": exp.Literal.string("20 minutes"),
+            "refresh_mode": exp.Literal.string("auto"),
+            "initialize": exp.Literal.string("on_create"),
+        },
+    )
+
+    assert to_sql_calls(adapter) == [
+        'CREATE TABLE IF NOT EXISTS "test_table" AS SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (SELECT "a", "b" FROM "source_table") AS "_subquery"',
+    ]
Repository: TobikoData/sqlmesh
Pull request #2950 (2024-07-26 17:23:54)

Feat: helpful err msg if forward_only specified in incremental by partition
`INCREMENTAL_BY_PARTITION` models are always `forward_only`. If a user specifies the forward_only key with true (or any other value), they currently receive a cryptic pydantic error. 

This PR adds a field validator that provides an error message telling the user not to specify the forward_only key.


Patch:
diff --git a/sqlmesh/core/model/kind.py b/sqlmesh/core/model/kind.py
index b635cf84..d7a6e504 100644
--- a/sqlmesh/core/model/kind.py
+++ b/sqlmesh/core/model/kind.py
@@ -488,6 +488,14 @@ class IncrementalByPartitionKind(_Incremental):
     forward_only: Literal[True] = True
     disable_restatement: SQLGlotBool = True
 
+    @field_validator("forward_only", mode="before")
+    def _forward_only_validator(cls, v: t.Union[bool, exp.Expression]) -> Literal[True]:
+        if v is not True:
+            raise ConfigError(
+                "Do not specify the `forward_only` configuration key - INCREMENTAL_BY_PARTITION models are always forward_only."
+            )
+        return v
+
     @property
     def metadata_hash_values(self) -> t.List[t.Optional[str]]:
         return [


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 6c79a69e..ec9b45f1 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -4755,6 +4755,24 @@ def test_incremental_by_partition(sushi_context, assert_exp_eq):
         )
         load_sql_based_model(expressions)
 
+    with pytest.raises(
+        ConfigError,
+        match=r".*Do not specify the `forward_only` configuration key.*",
+    ):
+        expressions = d.parse(
+            """
+            MODEL (
+                name db.table,
+                kind INCREMENTAL_BY_PARTITION (
+                    forward_only true
+                ),
+            );
+
+            SELECT a, b
+            """
+        )
+        load_sql_based_model(expressions)
+
 
 @pytest.mark.parametrize(
     ["model_def", "path", "expected_name"],