Reduce XCom read/write for tracking node state and errors in ConsumerWatcher task

[pankajastro]

This PR aims to reduce XCom load in the watcher flow by limiting which dbt log/events are persisted for consumer-side logging and by removing the per-event XCom read used for deduplication.

Changes:

• Introduces an allowlist of dbt event names and only pushes per-node dbt events to XCom when the event is allowlisted.
• Removes the “skip duplicate event” XCom read-before-write behavior.

[Copilot]

Pull request overview

This PR aims to reduce XCom load in the watcher flow by limiting which dbt log/events are persisted for consumer-side logging and by removing the per-event XCom read used for deduplication.

Changes:

• Introduces an allowlist of dbt event names and only pushes per-node dbt events to XCom when the event is allowlisted.
• Removes the “skip duplicate event” XCom read-before-write behavior.
• Updates unit tests for _process_dbt_log_event to align with allowlist-based behavior and to skip pushes when unique_id is missing.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
cosmos/operators/_watcher/base.py	Adds _DBT_EVENTS_TYPE allowlist filtering and removes dedup XCom reads in _process_dbt_log_event.
tests/operators/_watcher/test_watcher_base.py	Updates tests to validate allowlist-based pushing and skipping when unique_id is absent.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 98.04%. Comparing base (c137be7) to head (3b7a9f1).
⚠️ Report is 3 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2471   +/-   ##
=======================================
  Coverage   98.04%   98.04%           
=======================================
  Files         103      103           
  Lines        7223     7223           
=======================================
  Hits         7082     7082           
  Misses        141      141           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

Updates watcher log processing to only emit per-node dbt events for a fixed allowlist of dbt event types, and aligns unit tests with the new filtering behavior.

Changes:

• Add _DBT_EVENTS_TYPE allowlist and gate _process_dbt_log_event pushes on info.name.
• Remove prior “sensitive words” logic and the duplicate-event XCom suppression.
• Update watcher and subprocess tests to validate allowlist behavior and revised XCom push counts.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.

File	Description
cosmos/operators/_watcher/base.py	Adds dbt event allowlist filtering for XCom-pushed log events; removes prior duplicate-event suppression.
tests/operators/_watcher/test_watcher_base.py	Updates unit tests to assert allowlist-based XCom pushing and no-push when unique_id is missing.
tests/hooks/test_subprocess.py	Adjusts subprocess log parsing tests to expect only the status XCom push in terminal states.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[pankajkoti]

It is looking good to me, but I have some questions in line for my understanding and curiosity. Are we confident that with a higher number of threads, it won't cause an issue?

I am happy to get this in once the above question and in-line questions are answered, and if we get another pair of eyes for review here.

[Copilot]

Pull request overview

This PR tightens dbt watcher log handling by introducing an explicit allowlist of dbt event types to decide which dbt log events are pushed into XCom, and updates unit tests to match the new behavior.

Changes:

• Add allowlisted dbt event type sets (error/failure + node status) and filter _process_dbt_log_event pushes accordingly.
• Update watcher-base tests to validate allowlist behavior and skip pushing when unique_id is missing.
• Update subprocess watcher tests to reflect the reduced number of XCom pushes per processed log line.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
cosmos/operators/_watcher/base.py	Introduces dbt event allowlists and filters which log events get pushed to XCom.
tests/operators/_watcher/test_watcher_base.py	Updates tests to validate allowlist behavior and unique_id guard.
tests/hooks/test_subprocess.py	Adjusts expectations for XCom push count based on updated event filtering.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

This pull request tightens dbt watcher log-event handling by introducing an explicit allowlist of dbt event types to persist to XCom, and updates unit tests to reflect the new filtering behavior.

Changes:

• Add dbt event-type allowlist (error/failure events + node lifecycle events) and filter _process_dbt_log_event() pushes accordingly.
• Remove XCom deduplication logic for dbt events and adjust watcher-base tests.
• Update subprocess hook tests to align with the revised XCom push behavior.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
cosmos/operators/_watcher/base.py	Introduces _DBT_EVENT_ALLOWLIST and filters dbt events before pushing to XCom.
tests/operators/_watcher/test_watcher_base.py	Updates tests to validate allowlist-based event pushing and new skip conditions.
tests/hooks/test_subprocess.py	Adjusts expectations around safe_xcom_push call counts/behavior.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

This PR refines WATCHER producer-side dbt log handling by switching from message substring matching to an explicit dbt event-name allowlist, and updates tests to reflect the new behavior.

Changes:

• Introduces _DBT_EVENT_ALLOWLIST and filters _process_dbt_log_event() to only push selected dbt events to XCom.
• Removes the prior “sensitive words” logic and the XCom de-duplication check in _process_dbt_log_event().
• Updates watcher/subprocess tests to validate allowlisted event pushing and revised XCom push expectations.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.

File	Description
cosmos/operators/_watcher/base.py	Adds allowlist-based filtering for dbt events before pushing _dbt_event XCom payloads.
tests/operators/_watcher/test_watcher_base.py	Updates unit tests to assert allowlist-driven _dbt_event pushing and skipping when unique_id is missing.
tests/hooks/test_subprocess.py	Updates subprocess log parsing test expectations for XCom pushes.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.