You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "SameerMesiah97 (via GitHub)" <gi...@apache.org> on 2026/02/19 23:52:49 UTC

[PR] Add optional parallel downloads and uploads to GCSTimeSpanFileTransformOperator [airflow]

SameerMesiah97 opened a new pull request, #62196:
URL: https://github.com/apache/airflow/pull/62196

   **Description**
   
   This change adds optional parallel download and upload support to `GCSTimeSpanFileTransformOperator.execute` using `ThreadPoolExecutor`.
   
   When `max_download_workers > 1` or `max_upload_workers > 1`, source objects are downloaded and transformed files are uploaded concurrently using a bounded thread pool with `as_completed`.
   
   Per-object operations are encapsulated in inner helper functions within `execute`. A `storage.Client` (via `get_conn()`) is initialized once per phase and reused across submitted futures, with bucket/blob resolution performed inside each worker.
   
   `max_download_workers` and `max_upload_workers` must be ≥ 1 (otherwise a `ValueError` is raised). Error propagation semantics are unchanged: if `*_continue_on_fail=False`, the first `GoogleCloudError` is raised; otherwise failures are logged and execution continues. Logging now includes worker counts for both phases.
   
   
   **Rationale**
   
   The operator previously executed downloads and uploads sequentially despite an existing `#TODO` indicating that parallelism should be introduced. This implementation addresses that gap to mitigate cumulative latency from high request counts against GCS, which can significantly impact end-to-end runtime in object-heavy workloads.
   
   Parallel execution is implemented using `storage.Client` instances obtained via `get_conn()` rather than invoking hook methods directly inside worker threads. This is deliberate, as **`GCSHook` maintains internal state and is not designed for concurrent mutation across threads**. By resolving bucket and blob objects within each worker using a shared client, **we avoid shared mutable hook state while enabling safe concurrency**.
   
   **Documentation**
   
   The operator docstring has been updated to describe the new parallel execution functionality, including the behavior of the download and upload phases and the semantics of the new parameters `max_download_workers` and `max_upload_workers`.
   
   **Notes**
   
   * Directory placeholder objects (keys ending in `/`) are filtered out prior to download.
   * Introduced `_setup_gcs_client_chain` test helper to mock the `storage.Client → bucket → blob` chain for `test_execute` and the new parallel execution tests.
   
   **Tests**
   
   * Added unit tests verifying that parallel downloads and parallel uploads are executed when `max_download_workers > 1` and `max_upload_workers > 1`, respectively.
   * Added tests ensuring that a `GoogleCloudError` is raised when `download_continue_on_fail=False` or `upload_continue_on_fail=False`, confirming correct error propagation behavior.
   * Refactored `test_execute` to mock `storage.Client` directly and introduced `_setup_gcs_client_chain` to centralize client-chain mocking.
   * Extended the existing system test DAG (`gcs_transform_timespan`) to include a parallel execution case (`max_*_workers > 1`) to validate behavior against real GCS.
   
   **Backwards Compatibility**
   
   Existing behavior is preserved. Downloads and uploads remain sequential by default, and parallel execution is only triggered when `max_download_workers` or `max_upload_workers` is set to a value other than `1`.
   
   **System Test Verification**
   
   The modified system test `example_gcs_transform_timespan.py` has been run and has passed succesfully after implementing thse changes. Due to OpenLineage compatibility issues, the `check_openlineage_events` task had to be disabled but the current test results are sufficient to indicate that the new functionality is working as intend with no regressions The screenshot for these results has been provided below:
   
   <img width="940" height="170" alt="image" src="https://github.com/user-attachments/assets/4d404693-d40e-47bc-b969-0966818febfe" />


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Add optional parallel downloads and uploads to GCSTimeSpanFileTransformOperator [airflow]

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #62196:
URL: https://github.com/apache/airflow/pull/62196#issuecomment-4043256973

   Can you please rebase?


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Add optional parallel downloads and uploads to GCSTimeSpanFileTransformOperator [airflow]

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #62196:
URL: https://github.com/apache/airflow/pull/62196


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Add optional parallel downloads and uploads to GCSTimeSpanFileTransformOperator [airflow]

Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #62196:
URL: https://github.com/apache/airflow/pull/62196#issuecomment-4064088330

   @potiuk 
   
   Rebase has been done and CI is green.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Add optional parallel downloads and uploads to GCSTimeSpanFileTransformOperator [airflow]

Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #62196:
URL: https://github.com/apache/airflow/pull/62196#issuecomment-4043388801

   > Can you please rebase?
   
   I have done that right now. 


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Add optional parallel downloads and uploads to GCSTimeSpanFileTransformOperator [airflow]

Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #62196:
URL: https://github.com/apache/airflow/pull/62196#issuecomment-4046493649

   Converted to draft until CI is green. 


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org