aneesh-joseph commented on a change in pull request #9973:
URL: https://github.com/apache/airflow/pull/9973#discussion_r629415571
##########
File path: airflow/sensors/smart_sensor.py
##########
@@ -391,13 +391,21 @@ def _update_ti_hostname(self, sensor_works, session=None):
:param session: The sqlalchemy session.
"""
TI = TaskInstance
- ti_keys = [(x.dag_id, x.task_id, x.execution_date) for x in
sensor_works]
def update_ti_hostname_with_count(count, ti_keys):
# Using or_ instead of in_ here to prevent from full table scan.
tis = (
session.query(TI)
- .filter(or_(tuple_(TI.dag_id, TI.task_id, TI.execution_date)
== ti_key for ti_key in ti_keys))
+ .filter(
+ or_(
+ and_(
+ TI.dag_id == ti_key.dag_id,
+ TI.task_id == ti_key.task_id,
+ TI.execution_date == ti_key.execution_date,
+ )
+ for ti_key in ti_keys
+ )
+ )
Review comment:
done, restricted this change to mssql only
##########
File path: airflow/models/serialized_dag.py
##########
@@ -313,7 +315,9 @@ def get_dag_dependencies(cls, session: Session = None) ->
Dict[str, List['DagDep
if session.bind.dialect.name in ["sqlite", "mysql"]:
for row in session.query(cls.dag_id, func.json_extract(cls.data,
"$.dag.dag_dependencies")).all():
dependencies[row[0]] = [DagDependency(**d) for d in
json.loads(row[1])]
-
+ elif session.bind.dialect.name in ["mssql"]:
Review comment:
done
##########
File path: airflow/models/dag.py
##########
@@ -896,7 +897,7 @@ def get_num_active_runs(self, external_trigger=None,
session=None):
)
if external_trigger is not None:
- query = query.filter(DagRun.external_trigger == external_trigger)
+ query = query.filter(DagRun.external_trigger ==
expression.literal(external_trigger))
Review comment:
done
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
