Skip to content

fix: handle support for partial success introduce in dbt 1.9 with microbatching #2041

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
haritamar merged 4 commits into from
Oct 30, 2025
Merged

fix: handle support for partial success introduce in dbt 1.9 with microbatching #2041

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3b95137
add support for partial success
nicor88 Oct 29, 2025
e5c9f32
tweaks
nicor88 Oct 29, 2025
9bc7374
fix unit tests
nicor88 Oct 29, 2025
da52465
Merge branch 'master' into fix/microbatching_support
nicor88 Oct 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions elementary/monitor/data_monitoring/schema.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ class Status(str, Enum):
SKIPPED = "skipped"
ERROR = "error"
RUNTIME_ERROR = "runtime error"
PARTIAL_SUCCESS = "partial success"


class ResourceType(str, Enum):
Expand Down
88 changes: 70 additions & 18 deletions tests/unit/monitor/api/alerts/test_alert_filters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,33 @@ def initial_alerts():
resource_type=ResourceType.MODEL,
),
),
PendingAlertSchema(
id="model_alert_4",
alert_class_id="elementary.model_id_3",
type=AlertTypes.MODEL,
detected_at=datetime(2022, 10, 10, 7, 0, 0),
created_at=datetime(2022, 10, 10, 7, 0, 0),
updated_at=datetime(2022, 10, 10, 7, 0, 0),
status=AlertStatus.PENDING,
data=ModelAlertDataSchema(
id="4",
alert_class_id="elementary.model_id_3",
model_unique_id="elementary.model_id_3",
alias="model3",
path="my/path3",
original_path="",
materialization="incremental",
message="",
full_refresh=False,
detected_at=datetime(2022, 10, 10, 7, 0, 0),
tags=["microbatch"],
model_meta=dict(owner='["alice"]'),
status="partial success",
database_name="test_db",
schema_name="test_schema",
resource_type=ResourceType.MODEL,
),
),
]
source_freshness_alerts = [
PendingAlertSchema(
Expand Down Expand Up @@ -359,9 +386,10 @@ def test_filter_alerts_by_tags():
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_2"
assert filter_test_alerts[1].id == "test_alert_4"
assert len(filter_model_alerts) == 2
assert len(filter_model_alerts) == 3
assert filter_model_alerts[0].id == "model_alert_2"
assert filter_model_alerts[1].id == "model_alert_3"
assert filter_model_alerts[2].id == "model_alert_4"

filter = FiltersSchema(
tags=[FilterSchema(values=["three"], type=FilterType.IS)], statuses=[]
Expand All @@ -383,8 +411,9 @@ def test_filter_alerts_by_tags():
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_1"
assert filter_test_alerts[1].id == "test_alert_3"
assert len(filter_model_alerts) == 1
assert len(filter_model_alerts) == 2
assert filter_model_alerts[0].id == "model_alert_1"
assert filter_model_alerts[1].id == "model_alert_4"

filter = FiltersSchema(
tags=[FilterSchema(values=["four"], type=FilterType.IS)], statuses=[]
Expand All @@ -407,10 +436,11 @@ def test_filter_alerts_by_tags():
"test_alert_2",
"test_alert_3",
]
assert len(filter_model_alerts) == 2
assert len(filter_model_alerts) == 3
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_1",
"model_alert_2",
"model_alert_4",
]

filter = FiltersSchema(
Expand Down Expand Up @@ -441,10 +471,11 @@ def test_filter_alerts_by_tags():
"test_alert_2",
"test_alert_4",
]
assert len(filter_model_alerts) == 2
assert len(filter_model_alerts) == 3
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_2",
"model_alert_3",
"model_alert_4",
]

filter = FiltersSchema(
Expand All @@ -470,8 +501,11 @@ def test_filter_alerts_by_tags():
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_2"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_2"
assert len(filter_model_alerts) == 2
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_2",
"model_alert_4",
]

filter = FiltersSchema(
tags=[
Expand Down Expand Up @@ -503,8 +537,11 @@ def test_filter_alerts_by_tags():
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_2"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_2"
assert len(filter_model_alerts) == 2
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_2",
"model_alert_4",
]

filter = FiltersSchema(
tags=[
Expand Down Expand Up @@ -564,8 +601,9 @@ def test_filter_alerts_by_owners():
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_3"
assert len(filter_model_alerts) == 1
assert len(filter_model_alerts) == 2
assert filter_model_alerts[0].id == "model_alert_2"
assert filter_model_alerts[1].id == "model_alert_4"

filter = FiltersSchema(
owners=[FilterSchema(values=["john"], type=FilterType.IS)], statuses=[]
Expand All @@ -587,8 +625,11 @@ def test_filter_alerts_by_owners():
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_4"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_3"
assert len(filter_model_alerts) == 2
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_3",
"model_alert_4",
]

filter = FiltersSchema(
owners=[
Expand Down Expand Up @@ -652,8 +693,9 @@ def test_filter_alerts_by_model():
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_3"
assert filter_test_alerts[1].id == "test_alert_4"
assert len(filter_model_alerts) == 1
assert len(filter_model_alerts) == 2
assert filter_model_alerts[0].id == "model_alert_3"
assert filter_model_alerts[1].id == "model_alert_4"

filter = FiltersSchema(
models=[FilterSchema(values=["model_id_2"], type=FilterType.IS)], statuses=[]
Expand All @@ -675,9 +717,10 @@ def test_filter_alerts_by_model():
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_1"
assert filter_test_alerts[1].id == "test_alert_2"
assert len(filter_model_alerts) == 2
assert len(filter_model_alerts) == 3
assert filter_model_alerts[0].id == "model_alert_1"
assert filter_model_alerts[1].id == "model_alert_2"
assert filter_model_alerts[2].id == "model_alert_4"

filter = FiltersSchema(
models=[FilterSchema(values=["model_id_1", "model_id_2"], type=FilterType.IS)],
Expand Down Expand Up @@ -731,7 +774,8 @@ def test_filter_alerts_by_node_names():
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 0
assert len(filter_model_alerts) == 0
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_4"


def test_filter_alerts_by_statuses():
Expand Down Expand Up @@ -790,7 +834,11 @@ def test_filter_alerts_by_statuses():
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 0
assert len(filter_model_alerts) == 1
assert len(filter_model_alerts) == 2
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_3",
"model_alert_4",
]


def test_filter_alerts_by_resource_types():
Expand Down Expand Up @@ -828,18 +876,22 @@ def test_filter_alerts_by_resource_types():
def test_filter_alerts():
test_alerts, model_alerts, _ = initial_alerts()

# Test that empty filter returns all the alerts except for skipped.
# Test that empty filter returns all the alerts except for skipped and partial success.
filter = FiltersSchema()
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == len(test_alerts)
assert len(filter_model_alerts) == len(model_alerts) - 1 # 1 skipped model alert
assert (
len(filter_model_alerts) == len(model_alerts) - 2
) # 1 skipped + 1 partial success model alert

# Test that passing no filter returns all the alerts.
filter_test_alerts = filter_alerts(test_alerts)
filter_model_alerts = filter_alerts(model_alerts)
assert len(filter_test_alerts) == len(test_alerts)
assert len(filter_model_alerts) == len(model_alerts) - 1 # 1 skipped model alert
assert (
len(filter_model_alerts) == len(model_alerts) - 2
) # 1 skipped + 1 partial success model alert

# Test that filter with unsupported selector returns no alert
filter = FiltersSchema(last_invocation=True, selector="last_invocation")
Expand Down
Loading