This is an automated email from the ASF dual-hosted git repository.
henry3260 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new b9aa889577c Cover long-running jobs in the deferrable BigQuery check
system test (#74356)
b9aa889577c is described below
commit b9aa889577c7e2787ba3f2544e22df05a1b6a558
Author: Shahar Epstein <[email protected]>
AuthorDate: Fri Oct 16 19:54:20 2026 +0100
Cover long-running jobs in the deferrable BigQuery check system test
(#74356)
The deferrable value and interval checks in the async BigQuery system
test only ran trivial COUNT(*) queries, so their jobs were DONE before
the trigger's first poll. The regression fixed in #74305, where
BigQueryValueCheckTrigger and BigQueryIntervalCheckTrigger failed with
"Job running" while the job was still RUNNING (#73981), could therefore
never show up in the system test.
Add a view backed by a JavaScript UDF that keeps a query RUNNING for
tens of seconds and run both deferrable checks against it. The view
only reads the existing two-row table, so each check job is billed at
BigQuery's 10 MB minimum.
---
.../bigquery/example_bigquery_queries_async.py | 54 +++++++++++++++++++++-
1 file changed, 53 insertions(+), 1 deletion(-)
diff --git
a/providers/google/tests/system/google/cloud/bigquery/example_bigquery_queries_async.py
b/providers/google/tests/system/google/cloud/bigquery/example_bigquery_queries_async.py
index 60080989320..df143389ba6 100644
---
a/providers/google/tests/system/google/cloud/bigquery/example_bigquery_queries_async.py
+++
b/providers/google/tests/system/google/cloud/bigquery/example_bigquery_queries_async.py
@@ -100,6 +100,26 @@ CONFIGURATION = {
}
}
+SLOW_VIEW_NAME = f"view_{DAG_ID}_{ENV_ID}_slow".replace("-", "_")
+
+# The JavaScript UDF spins for a fixed number of iterations per row, which
keeps a query over the view
+# RUNNING for tens of seconds, i.e. across several poll intervals of the
deferrable check operators
+# (#73981). `CURRENT_TIMESTAMP()` makes the view non-deterministic, so
BigQuery never serves a repeated
+# check from the query cache. Nothing here scans more than the two-row table,
so the jobs are billed at
+# the 10 MB minimum.
+CREATE_SLOW_VIEW_QUERY = f"""
+CREATE OR REPLACE FUNCTION `{DATASET_NAME}.slow_ds`(ds STRING) RETURNS STRING
LANGUAGE js AS r'''
+ let acc = 0;
+ for (let i = 0; i < 500000000; i++) {{
+ acc = (acc * 31 + i) % 1000003;
+ }}
+ return acc >= 0 ? ds : null;
+''';
+CREATE OR REPLACE VIEW `{DATASET_NAME}.{SLOW_VIEW_NAME}` AS
+ SELECT value, name, `{DATASET_NAME}.slow_ds`(ds) AS ds
+ FROM `{DATASET_NAME}.{TABLE_NAME_1}`
+ WHERE CURRENT_TIMESTAMP() IS NOT NULL;
+"""
default_args = {
"execution_timeout": timedelta(hours=6),
@@ -262,6 +282,31 @@ with DAG(
deferrable=True,
)
+ create_slow_view = BigQueryInsertJobOperator(
+ task_id="create_slow_view",
+ configuration={"query": {"query": CREATE_SLOW_VIEW_QUERY,
"useLegacySql": False}},
+ location=LOCATION,
+ )
+
+ check_value_long_running = BigQueryValueCheckOperator(
+ task_id="check_value_long_running",
+ sql=f"SELECT COUNT(*) FROM {DATASET_NAME}.{SLOW_VIEW_NAME} WHERE ds IS
NOT NULL",
+ pass_value=2,
+ use_legacy_sql=False,
+ location=LOCATION,
+ deferrable=True,
+ )
+
+ check_interval_long_running = BigQueryIntervalCheckOperator(
+ task_id="check_interval_long_running",
+ table=f"{DATASET_NAME}.{SLOW_VIEW_NAME}",
+ days_back=1,
+ metrics_thresholds={"COUNT(*)": 1.5},
+ use_legacy_sql=False,
+ location=LOCATION,
+ deferrable=True,
+ )
+
check_openlineage_events = OpenLineageTestOperator(
task_id="check_openlineage_events",
file_path=str(Path(__file__).parent / "resources" / "openlineage" /
"bigquery_queries_async.json"),
@@ -272,7 +317,14 @@ with DAG(
insert_query_job >> get_data >> get_data_result
insert_query_job >> execute_query_save >> bigquery_execute_multi_query
insert_query_job >> execute_long_running_query >> check_value >>
check_interval
- [check_count, check_interval, bigquery_execute_multi_query,
get_data_result] >> delete_dataset
+ insert_query_job >> create_slow_view >> check_value_long_running >>
check_interval_long_running
+ [
+ check_count,
+ check_interval,
+ check_interval_long_running,
+ bigquery_execute_multi_query,
+ get_data_result,
+ ] >> delete_dataset
delete_dataset >> check_openlineage_events
from tests_common.test_utils.watcher import watcher
