Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
pierrejeambrun closed pull request #48659: Dag processor consumes DagPriorityParsingRequest when relative file l… URL: https://github.com/apache/airflow/pull/48659 -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
pierrejeambrun commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-4244654096 Closing, stale -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
Copilot commented on code in PR #48659:
URL: https://github.com/apache/airflow/pull/48659#discussion_r3066499401
##
airflow-core/src/airflow/models/dagbag.py:
##
@@ -666,7 +666,7 @@ def sync_to_db(self, bundle_name: str, bundle_version: str
| None, session: Sess
def generate_md5_hash(context):
bundle_name = context.get_current_parameters()["bundle_name"]
-relative_fileloc = context.get_current_parameters()["relative_fileloc"]
+relative_fileloc =
context.get_current_parameters().get("relative_fileloc", "")
Review Comment:
`relative_fileloc` can now be NULL, but `generate_md5_hash()` will
incorporate the Python `None` value into the hash (resulting in IDs based on
the literal string "None") whenever the key exists with a NULL value. This
makes IDs depend on how callers represent “whole-bundle” requests (NULL vs "")
and can lead to duplicate semantic requests with different IDs. Normalize
`relative_fileloc` to an empty string (or other single canonical value) before
hashing so NULL/"" produce the same ID.
```suggestion
relative_fileloc =
context.get_current_parameters().get("relative_fileloc") or ""
```
##
airflow-core/src/airflow/models/dagbag.py:
##
@@ -685,12 +685,22 @@ class DagPriorityParsingRequest(Base):
# Note: Do not depend on fileloc pointing to a file; in the case of a
# packaged DAG, it will point to the subpath of the DAG within the
# associated zip.
-relative_fileloc = Column(String(2000), nullable=False)
+relative_fileloc = Column(String(2000), nullable=True)
-def __init__(self, bundle_name: str, relative_fileloc: str) -> None:
+def __init__(self, bundle_name: str, relative_fileloc: str = "") -> None:
super().__init__()
self.bundle_name = bundle_name
self.relative_fileloc = relative_fileloc
+def parse_whole_folder(self) -> bool:
+"""
+Check if this request should parse the whole folder based on
relative_fileloc.
+
+Returns:
+bool: True if relative_fileloc is None, indicating the whole
folder should be parsed,
+ False otherwise.
+"""
+return self.relative_fileloc == ""
Review Comment:
`parse_whole_folder()` docstring says it returns True when
`relative_fileloc` is None, but the implementation checks for an empty string.
Since the column is nullable, a NULL value will currently be treated as
“specific file” and will later cause `Path(None)` errors in the manager. Update
the predicate to treat NULL (and ideally "" for backward-compat) as “whole
bundle”, and align the docstring accordingly.
```suggestion
bool: True if relative_fileloc is None or an empty string,
indicating the whole folder
should be parsed, False otherwise.
"""
return self.relative_fileloc in (None, "")
```
##
airflow-core/src/airflow/dag_processing/manager.py:
##
@@ -413,20 +413,35 @@ def _queue_requested_files_for_parsing(self) -> None:
@provide_session
def _get_priority_files(self, session: Session = NEW_SESSION) ->
list[DagFileInfo]:
-files: list[DagFileInfo] = []
+files: set[DagFileInfo] = set()
bundles = {b.name: b for b in self._dag_bundles}
requests = session.scalars(
select(DagPriorityParsingRequest).where(DagPriorityParsingRequest.bundle_name.in_(bundles.keys()))
)
for request in requests:
bundle = bundles[request.bundle_name]
-files.append(
-DagFileInfo(
-rel_path=Path(request.relative_fileloc),
bundle_name=bundle.name, bundle_path=bundle.path
+if request.parse_whole_folder():
+# If relative_fileloc is null, get all files from DagModel
+dag_files = session.scalars(
+
select(DagModel.relative_fileloc).where(DagModel.bundle_name ==
bundle.name).distinct()
+)
+for relative_fileloc in dag_files:
+files.add(
+DagFileInfo(
+rel_path=Path(relative_fileloc),
bundle_name=bundle.name, bundle_path=bundle.path
+)
+)
+else:
+# If relative_fileloc has a value, just add that specific file
+files.add(
+DagFileInfo(
+rel_path=Path(request.relative_fileloc),
+bundle_name=bundle.name,
+bundle_path=bundle.path,
+)
Review Comment:
If a `DagPriorityParsingRequest.relative_fileloc` row is NULL (now allowed
by the migration), the `else` branch will execute and call
`Path(request.relative_fileloc)`, which raises `TypeError`. Ensure NULL
requests are handled in the “whole folder” path (e.g., via `if not
request.relative_fileloc` / upda
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
potiuk commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-4178955895 @shubham-pyc This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria). **Issues found:** - :x: **Merge conflicts**: This PR has merge conflicts with the `main` branch. Your branch is 7163 commits behind `main`. Please rebase your branch (`git fetch origin && git rebase origin/main`), resolve the conflicts, and push again. See [contributing quick start](https://github.com/apache/airflow/blob/main/contributing-docs/03a_contributors_quick_start_beginners.rst). > **Note:** Your branch is **7163 commits behind `main`**. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results. **What to do next:** - The comment informs you what you need to do. - Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed. - There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. - Maintainers will then proceed with a normal review. Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack). -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
jason810496 commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-3419435961 up -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
github-actions[bot] commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-3419087171 This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions. -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
jason810496 commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-3121385079 up -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
github-actions[bot] commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-3120769041 This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions. -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
jedcunningham commented on code in PR #48659:
URL: https://github.com/apache/airflow/pull/48659#discussion_r2138051918
##
airflow-core/src/airflow/migrations/versions/0069_3_0_0_make_relative_fileloc_nullable_for_.py:
##
@@ -0,0 +1,52 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+make relative_fileloc nullable for reserialized all bundles.
Review Comment:
```suggestion
Make DagPriorityParsingRuquest.relative_fileloc nullable.
```
nit
##
airflow-core/src/airflow/models/dagbag.py:
##
@@ -685,12 +685,22 @@ class DagPriorityParsingRequest(Base):
# Note: Do not depend on fileloc pointing to a file; in the case of a
# packaged DAG, it will point to the subpath of the DAG within the
# associated zip.
-relative_fileloc = Column(String(2000), nullable=False)
+relative_fileloc = Column(String(2000), nullable=True)
-def __init__(self, bundle_name: str, relative_fileloc: str) -> None:
+def __init__(self, bundle_name: str, relative_fileloc: str = "") -> None:
Review Comment:
```suggestion
def __init__(self, bundle_name: str, relative_fileloc: str | None =
None) -> None:
```
If we are allowing nullable, lets actually use nulls then - otherwise we
don't really need to alter the table in the first place.
##
airflow-core/src/airflow/dag_processing/manager.py:
##
@@ -413,20 +413,35 @@ def _queue_requested_files_for_parsing(self) -> None:
@provide_session
def _get_priority_files(self, session: Session = NEW_SESSION) ->
list[DagFileInfo]:
-files: list[DagFileInfo] = []
+files: set[DagFileInfo] = set()
bundles = {b.name: b for b in self._dag_bundles}
requests = session.scalars(
select(DagPriorityParsingRequest).where(DagPriorityParsingRequest.bundle_name.in_(bundles.keys()))
)
for request in requests:
bundle = bundles[request.bundle_name]
-files.append(
-DagFileInfo(
-rel_path=Path(request.relative_fileloc),
bundle_name=bundle.name, bundle_path=bundle.path
+if request.parse_whole_folder():
+# If relative_fileloc is null, get all files from DagModel
+dag_files = session.scalars(
Review Comment:
We probably should refresh the bundle and [relist the dags from
disk](https://github.com/apache/airflow/blob/bf0bfe9dc1f3812989acfe6ed7118acf9ba5b586/airflow-core/src/airflow/dag_processing/manager.py#L579)
instead of querying the db.
I think we can just refresh here, relist, and add, but it'd be worth giving
it a bit more thought to make sure we aren't impacting other stuff by doing so.
##
airflow-core/src/airflow/models/dagbag.py:
##
@@ -685,12 +685,22 @@ class DagPriorityParsingRequest(Base):
# Note: Do not depend on fileloc pointing to a file; in the case of a
# packaged DAG, it will point to the subpath of the DAG within the
# associated zip.
-relative_fileloc = Column(String(2000), nullable=False)
+relative_fileloc = Column(String(2000), nullable=True)
-def __init__(self, bundle_name: str, relative_fileloc: str) -> None:
+def __init__(self, bundle_name: str, relative_fileloc: str = "") -> None:
super().__init__()
self.bundle_name = bundle_name
self.relative_fileloc = relative_fileloc
+def parse_whole_folder(self) -> bool:
+"""
+Check if this request should parse the whole folder based on
relative_fileloc.
+
+Returns:
+bool: True if relative_fileloc is None, indicating the whole
folder should be parsed,
+ False otherwise.
+"""
+return self.relative_fileloc == ""
Review Comment:
```suggestion
return self.relative_fileloc is None
```
--
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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
pierrejeambrun commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-2958809534 up. -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
shubham-pyc commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-2842902164 @pierrejeambrun fixed the CI -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
shubham-pyc commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-2822925188 @pierrejeambrun resolved conflicts -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
shubham-pyc commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-2774915878 @jedcunningham please take a look at this one. -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
shubham-pyc commented on code in PR #48659: URL: https://github.com/apache/airflow/pull/48659#discussion_r2024305714 ## airflow-core/src/airflow/dag_processing/manager.py: ## @@ -407,20 +407,35 @@ def _queue_requested_files_for_parsing(self) -> None: @provide_session def _get_priority_files(self, session: Session = NEW_SESSION) -> list[DagFileInfo]: -files: list[DagFileInfo] = [] +files: set[DagFileInfo] = set() Review Comment: I changed this from a list to a set to ensure uniqueness. There may be cases where a DagPriorityParsingRequest is triggered for both a specific file and a bundle, potentially causing duplicates. Using a set guarantees that each file appears only once. -- 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]
Re: [PR] Dag processor consumes DagPriorityParsingRequest when relative file l… [airflow]
shubham-pyc commented on PR #48659: URL: https://github.com/apache/airflow/pull/48659#issuecomment-2771539852 Still work in progress test cases are yet to come -- 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]
