BigQuery嵌套MAX(Date_Column)查询预估偏差及Airflow报错问题

问题根因 #

预估量偏差核心原因 #

BigQuery查询优化器在处理WHERE子句中依赖外部表（dashboards.report_table）的子查询/变量时，无法在查询计划生成阶段解析出具体过滤值。因此，优化器无法对交易表prod.my_table的timestamp_column执行分区裁剪（Partition Pruning），只能默认按全表数据量（577GB）预估扫描成本——哪怕实际执行时会先计算出过滤值再扫描目标数据。

即使改用DECLARE变量存储子查询结果，由于变量赋值属于脚本逻辑，优化器仍无法提前知晓变量值，预估量依然显示全表大小。

Airflow报错原因 #

报错configuration.query.destinationTable cannot be set for scripts是因为提交的是包含DECLARE的脚本语句，而BigQuery不允许为脚本类型的查询设置目标表（destinationTable），仅支持为单条查询语句设置。

最优解决方案 #

1. 拆分查询逻辑，提前获取过滤值 #

将原查询拆分为两步：先单独执行小查询获取起始时间戳（仅扫描1KB数据），再将该值作为参数传入主查询，让优化器能明确过滤条件，精准执行分区裁剪。

步骤1：获取起始时间戳

SELECT TIMESTAMP(DATE_ADD(COALESCE(MAX(est_date), '2022-06-01'), INTERVAL 1 DAY), 'America/New_York') AS start_ts
FROM dashboards.`report_table`

步骤2：参数化执行主查询

将步骤1得到的start_ts值代入主查询，此时BigQuery能正确识别分区范围，预估量与实际扫描量（1.27GB左右）一致：

SELECT
  DATE(timestamp_column, "America/New_York") AS est_date,
  AVG(number_column) AS metric
FROM prod.`my_table`
WHERE timestamp_column >= TIMESTAMP("2024-05-01 04:00:00")  -- 替换为步骤1的实际结果
  AND timestamp_column < TIMESTAMP(CURRENT_DATE('America/New_York'), 'America/New_York')
GROUP BY est_date

2. Airflow中的兼容实现 #

通过两个独立的BigQuery任务实现：第一个任务获取起始时间并存入XCom，第二个任务读取XCom值动态渲染参数化查询，同时支持设置目标表。

示例代码（Airflow 2.x）：

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryGetDataOperator, BigQueryInsertJobOperator
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
}

with DAG(
    'bq_transaction_to_report',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=days_ago(1),
    catchup=False,
) as dag:

    # 任务1：获取起始时间戳
    get_start_ts = BigQueryGetDataOperator(
        task_id='fetch_start_timestamp',
        sql="""
            SELECT TIMESTAMP(DATE_ADD(COALESCE(MAX(est_date), '2022-06-01'), INTERVAL 1 DAY), 'America/New_York') AS start_ts
            FROM dashboards.`report_table`
        """,
        gcp_conn_id='google_cloud_default',
    )

    # 任务2：执行聚合查询并写入报表表
    aggregate_data = BigQueryInsertJobOperator(
        task_id='aggregate_to_report_table',
        configuration={
            "query": {
                "query": """
                    SELECT
                      DATE(timestamp_column, "America/New_York") AS est_date,
                      AVG(number_column) AS metric
                    FROM prod.`my_table`
                    WHERE timestamp_column >= @start_ts
                      AND timestamp_column < TIMESTAMP(CURRENT_DATE('America/New_York'), 'America/New_York')
                    GROUP BY est_date
                """,
                "parameterMode": "NAMED",
                "queryParameters": [
                    {
                        "name": "start_ts",
                        "parameterType": {"type": "TIMESTAMP"},
                        "parameterValue": {"value": "{{ ti.xcom_pull(task_ids='fetch_start_timestamp')[0][0] }}"}
                    }
                ],
                "destinationTable": {
                    "projectId": "your-gcp-project-id",
                    "datasetId": "dashboards",
                    "tableId": "report_table"
                },
                "writeDisposition": "WRITE_APPEND"  # 根据业务需求选择WRITE_TRUNCATE/WRITE_EMPTY等
            }
        },
        gcp_conn_id='google_cloud_default',
    )

    get_start_ts >> aggregate_data

关键注意事项 #

• 确保prod.my_table是按timestamp_column分区的表，参数化查询才能最大化发挥分区裁剪的作用。
• 避免在主查询的WHERE子句中使用依赖外部表的子查询、CTE或JOIN，这类写法会导致优化器无法提前做分区裁剪。
• Airflow中使用参数化查询而非脚本语句，既能解决目标表设置的报错，又能保证查询的可维护性。

内容的提问来源于stack exchange，提问作者Shahid Thaika