Repository: dbt-labs/dbt-core
Pull request: make microbatch models skippable
Resolves https://github.com/dbt-labs/dbt-core/issues/11021

<!---
  Include the number of the issue addressed by this PR above, if applicable.
  PRs for code changes without an associated issue *will not be merged*.
  See CONTRIBUTING.md for more information.

  Add the `user docs` label to this PR if it will need docs changes.  An 
  issue will get opened in docs.getdbt.com upon successful merge of this PR.
-->

### Problem

<!---
  Describe the problem this PR is solving. What is the application state
  before this PR is merged?
-->

1. Skipping microbatch model fails with typeError: object of type 'Field' has no len()
2. That codepath indicates the microbatch model is still running in some capacity

### Solution

<!---
  Describe the way this PR solves the above problem. Add as much detail as you
  can to help reviewers understand your changes. Include any alternatives and
  tradeoffs you considered.
-->

1. Stop using `field(default_factory=...)` outside a dataclass (MicrobatchModelRunner)
2. Return results early if initial microbatch run results in a skipped result

### Checklist

- [x] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me.
- [x] I have run this code in development, and it appears to resolve the stated issue.
- [x] This PR includes tests, or tests are not required or relevant for this PR.
- [x] This PR has no interface changes (e.g., macros, CLI, logs, JSON artifacts, config files, adapter interface, etc.) or this PR has already received feedback and approval from Product or DX.
- [x] This PR includes [type annotations](https://docs.python.org/3/library/typing.html) for new and modified functions.


🎩 
```sh
Invoking dbt with ['run']
16:21:04  Running with dbt=1.9.0-b4
16:21:04  Registered adapter: postgres=1.9.0-b1
16:21:04  Unable to do partial parsing because saved manifest not found. Starting full parse.
16:21:05  Found 2 models, 429 macros
16:21:05  
16:21:05  Concurrency: 4 threads (target='default')
16:21:05  
16:21:05  1 of 2 START sql table model test17322060642773181458_test_microbatch.input_model  [RUN]
16:21:05  1 of 2 ERROR creating sql table model test17322060642773181458_test_microbatch.input_model  [ERROR in 0.05s]
16:21:05  2 of 2 SKIP relation test17322060642773181458_test_microbatch.microbatch_model . [SKIP]
16:21:05  
16:21:05  Finished running 1 incremental model, 1 table model in 0 hours 0 minutes and 0.55 seconds (0.55s).
16:21:05  
16:21:05  Completed with 1 error, 0 partial successs, and 0 warnings:
16:21:05  
16:21:05    Database Error in model input_model (models/input_model.sql)
  column "invalid" does not exist
  LINE 14: select invalid as event_time
                  ^
  compiled code at target/run/test/models/input_model.sql
16:21:05  
```

(extra '.' in skip result is a general issue, not just for microbatch models)

Problem statement:
[Bug] Skipping microbatch model fails with `typeError: object of type 'Field' has no len()`
An upstream failing leads to a microbatch model skipping (but still executing?) with the failure 'Field' has no len():

```
15:47:30  Found 23 models, 6 seeds, 21 data tests, 6 sources, 2 exposures, 13 metrics, 769 macros, 6 semantic models, 5 unit tests
15:47:30  
15:47:30  Concurrency: 4 threads (target='default')
15:47:30  
15:47:32  1 of 2 START sql view model dbt_marky.microbatch_input ......................... [RUN]
15:47:33  1 of 2 ERROR creating sql view model dbt_marky.microbatch_input ................ [ERROR in 0.69s]
15:47:33  2 of 2 SKIP relation dbt_marky.microbatch ...................................... [SKIP]
> /Users/michelleark/src/dbt-core/core/dbt/task/run.py(715)handle_microbatch_model()
-> while batch_idx < len(runner.batches):
(Pdb) runner.batches
Field(name=None,type=None,default=<dataclasses._MISSING_TYPE object at 0x100efa230>,default_factory=<class 'dict'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=<dataclasses._MISSING_TYPE object at 0x100efa230>,_field_type=None)
(Pdb) field(default_factory=dict)
Field(name=None,type=None,default=<dataclasses._MISSING_TYPE object at 0x100efa230>,default_factory=<class 'dict'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=<dataclasses._MISSING_TYPE object at 0x100efa230>,_field_type=None)
(Pdb) field(default_factory=dict)
Field(name=None,type=None,default=<dataclasses._MISSING_TYPE object at 0x100efa230>,default_factory=<class 'dict'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=<dataclasses._MISSING_TYPE object at 0x100efa230>,_field_type=None)
(Pdb) len(field(default_factory=dict))
```

Hints from issue discussion:

Commit message: make microbatch models skippable

Patch to review:
diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index 1b4be463e78..0b991fc9f29 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -2,7 +2,7 @@
 import threading
 import time
 from copy import deepcopy
-from dataclasses import asdict, field
+from dataclasses import asdict
 from datetime import datetime
 from multiprocessing.pool import ThreadPool
 from typing import AbstractSet, Any, Dict, Iterable, List, Optional, Set, Tuple, Type
@@ -335,7 +335,7 @@ def execute(self, model, manifest):
 
 class MicrobatchModelRunner(ModelRunner):
     batch_idx: Optional[int] = None
-    batches: Dict[int, BatchType] = field(default_factory=dict)
+    batches: Dict[int, BatchType] = {}
     relation_exists: bool = False
 
     def set_batch_idx(self, batch_idx: int) -> None:
@@ -704,8 +704,11 @@ def handle_microbatch_model(
         runner: MicrobatchModelRunner,
         pool: ThreadPool,
     ) -> RunResult:
-        # Initial run computes batch metadata
+        # Initial run computes batch metadata, unless model is skipped
         result = self.call_runner(runner)
+        if result.status == RunStatus.Skipped:
+            return result
+
         batch_results: List[RunResult] = []
 
         # Execute batches serially until a relation exists, at which point future batches are run in parallel
diff --git a/tests/functional/microbatch/test_microbatch.py b/tests/functional/microbatch/test_microbatch.py
index c9b9db2ff4e..ddac8959a5a 100644
--- a/tests/functional/microbatch/test_microbatch.py
+++ b/tests/functional/microbatch/test_microbatch.py
@@ -33,6 +33,12 @@
 select 3 as id, TIMESTAMP '2020-01-03 00:00:00-0' as event_time
 """
 
+input_model_invalid_sql = """
+{{ config(materialized='table', event_time='event_time') }}
+
+select invalid as event_time
+"""
+
 input_model_without_event_time_sql = """
 {{ config(materialized='table') }}
 
@@ -835,6 +841,24 @@ def test_microbatch(
             assert len(catch_aw.caught_events) == 1
 
 
+class TestMicrobatchModelSkipped(BaseMicrobatchTest):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": input_model_invalid_sql,
+            "microbatch_model.sql": microbatch_model_sql,
+        }
+
+    def test_microbatch_model_skipped(self, project) -> None:
+        run_dbt(["run"], expect_pass=False)
+
+        run_results = get_artifact(project.project_root, "target", "run_results.json")
+
+        microbatch_result = run_results["results"][1]
+        assert microbatch_result["status"] == "skipped"
+        assert microbatch_result["batch_results"] is None
+
+
 class TestMicrobatchCanRunParallelOrSequential(BaseMicrobatchTest):
     @pytest.fixture
     def batch_exc_catcher(self) -> EventCatcher:


Code review comments:
diff_hunk: @@ -335,7 +335,7 @@ def execute(self, model, manifest):
 
 class MicrobatchModelRunner(ModelRunner):
     batch_idx: Optional[int] = None
-    batches: Dict[int, BatchType] = field(default_factory=dict)
path: core/dbt/task/run.py
text: @author:
Usage of `field(default_factory=...)` was invalid outside of a dataclass

@user1:
Have we verified that having it set as `batches: Dict[int, BatchType] = {}` doesn't make it so two instances of `MicrobatchModelRunner` share the same underlying dict?

@user1:
Did a test. We do indeed need to do something different
```
>>> class MyClass:
...   batches = {}
... 
>>> instance1 = MyClass()
>>> instance2 = MyClass()
>>> instance1.batches
{}
>>> instance2.batches
{}
>>> instance1.batches['catch_phrase'] = "I like cats!"
>>> instance1.batches
{'catch_phrase': 'I like cats!'}
>>> instance2.batches
{'catch_phrase': 'I like cats!'}
```

@author:
Good call. Fixed with __init__ override:

```sh
>>> class MyClass:
...   def __init__(self):
...     self.batches = {}
...
>>> instance1 = MyClass()
>>> instance2 = MyClass()
>>> instance1.batches
{}
>>> instance2.batches
{}
>>> instance1.batches['catch_phrase'] = "I like cats!"
>>> instance1.batches
{'catch_phrase': 'I like cats!'}
>>> instance2.batches
{}
>>>
```