Lee-W commented on code in PR #55110:
URL: https://github.com/apache/airflow/pull/55110#discussion_r2315340195
##########
providers/google/src/airflow/providers/google/cloud/hooks/bigquery.py:
##########
@@ -1285,10 +1292,17 @@ def generate_job_id(self, job_id, dag_id, task_id,
logical_date, configuration,
if job_id:
return f"{job_id}_{uniqueness_suffix}"
- exec_date = logical_date.isoformat()
+ exec_date = date.isoformat()
job_id = f"airflow_{dag_id}_{task_id}_{exec_date}_{uniqueness_suffix}"
return re.sub(r"[:\-+.]", "_", job_id)
+ def get_exec_date(self, context: Context) -> pendulum.DateTime:
Review Comment:
This method name is not accurate. `get_logical_date_or_run_after` might be
better
##########
providers/google/src/airflow/providers/google/cloud/hooks/bigquery.py:
##########
@@ -1274,7 +1281,7 @@ def insert_job(
job_api_repr.result(timeout=timeout, retry=retry)
return job_api_repr
- def generate_job_id(self, job_id, dag_id, task_id, logical_date,
configuration, force_rerun=False) -> str:
+ def generate_job_id(self, job_id, dag_id, task_id, date, configuration,
force_rerun=False) -> str:
Review Comment:
we should rename the variable `date`, it might collide with `datetime.date`
##########
providers/google/src/airflow/providers/google/cloud/operators/workflows.py:
##########
@@ -114,8 +119,13 @@ def _workflow_id(self, context):
# We are limited by allowed length of workflow_id so
# we use hash of whole information
- exec_date = context["logical_date"].isoformat()
- base = f"airflow_{self.dag_id}_{self.task_id}_{exec_date}_{hash_base}"
+ date = context.get("logical_date", None)
+ if AIRFLOW_V_3_0_PLUS and date is None:
+ if dag_run := context.get("dag_run"):
+ date = pendulum.instance(dag_run.run_after)
+ exec_date = date if date is not None else
datetime.datetime.now(tz=datetime.timezone.utc)
Review Comment:
let's avoid `exec_date` which we try to remove since AIP-83
##########
providers/google/src/airflow/providers/google/cloud/operators/workflows.py:
##########
@@ -114,8 +119,13 @@ def _workflow_id(self, context):
# We are limited by allowed length of workflow_id so
# we use hash of whole information
- exec_date = context["logical_date"].isoformat()
- base = f"airflow_{self.dag_id}_{self.task_id}_{exec_date}_{hash_base}"
+ date = context.get("logical_date", None)
Review Comment:
let's rename `date` as something else
##########
providers/apache/hive/src/airflow/providers/apache/hive/operators/hive.py:
##########
@@ -143,7 +143,7 @@ def execute(self, context: Context) -> None:
# set the mapred_job_name if it's not set with dag, task, execution
time info
if not self.mapred_job_name:
ti = context["ti"]
- logical_date = context["logical_date"]
+ logical_date = context.get("logical_date", None)
if logical_date is None:
raise RuntimeError("logical_date is None")
Review Comment:
If we want the HiveOperator to work in this case, we can update the error
message to better convey this idea.
##########
providers/google/src/airflow/providers/google/cloud/operators/workflows.py:
##########
@@ -101,7 +106,7 @@ def __init__(
self.impersonation_chain = impersonation_chain
self.force_rerun = force_rerun
- def _workflow_id(self, context):
+ def _workflow_id(self, context: Context):
Review Comment:
```suggestion
def _workflow_id(self, context: Context) -> str:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
