FIX: incorrect access of logical_date in google bigquery operator and google workflow operator

[obarisk]

related: #53634 #55073 #55092

In airflow 3, logcial_date is not available for asset triggered DAG runs.
However, it was previously used for job ID generation in Bigquery and workflow ID for Workflows.

This PR proposes using dag_run.run_after as a substitute for asset triggered DAG runs.

In this PR, I propose to use dagrun.run_after for asset triggered dag run.

Summary of the PR

1. HiveOperator: raise error if logical_date is None (as is), and provide more information in the error message
2. GoogleCloudComposer: raise an error if logical_date is None
3. BigQueryHook.generate_job_id:
   • for airflow >= 3.0, add deprecated warning calls with logical_date is not None
   • add a new argument run_after with default value = None (for backwards compatibility)
4. Operators that use BigqueryHook.generate_job_id: for AIRFLOW_V_3_0_PLUS, use dag_run.run_after.

What Wasn't Fixed

The usage of context["logical_date"] in providers/google/cloud/sensors/cloud_composer.py remains unchangeed.

cc: @kaxil. this is related to #55073 but cover move on

• workflow
• gcs_to_bigquery
• bigquery_to_gcs

[MaksYermak]

@obarisk you have changed logical_date for workflow and bigquery operators. Is it all operators which use logical_date or do we have more?

[obarisk]

@MaksYermak as far as I know, in airflow codebase, despite the tests,

after this PR, there's no context["logical_date"] for all of the providers.

[obarisk]

PR updated. I've update the description of the PR.

[jroachgolf84]

@obarisk - do you need any help getting this over the finish line? Seems like there is another issue that would benefit from this being merged.

[obarisk]

The tests did pass. It's waiting for review now. cc: @jroachgolf84.

Need the maintainers to review and merge.

[obarisk]

by the way, before it got merged I'm currently using workaround like the following.
patch that in the dag file which use the operators.

from dowhen import when

def patch_logical_date(_frame):
  import pendulum # noqa
  context = _frame.f_locals.get("context", {})
  if not context:
    print("no context found")
  if context.get("logical_date") is None:
    context["logical_date"] = pendulum.now("UTC")
    _frame.f_locals["context"] = context

when(
  GCSToBigQueryOperator.execute,
  """logical_date=context["logical_date"]"""
).do(patch_logical_date)

[obarisk]

@Lee-W beside changing the signature of generate_job_id, all updated

[Lee-W]

a few nits. but I'm good with merging it. I'll keep it open for a few days in case others want to take a deeper look

[shahar1]

Tests fail :(