GCSToGCSOperator ignores `replace=False` when copying a single file whose destination already exists

[9re-pe]

Under which category would you file this issue?

Providers

Apache Airflow version

2.10.5

What happened and how to reproduce it?

Issue Description

GCSToGCSOperator ignores replace=False when the source is a single object (no wildcard) and the destination object already exists. The destination is overwritten despite the operator logging that there are no new files to sync.

Steps to reproduce

1. Place an object at gs://source-bucket/path/model.pkl.
2. Place an object with the same key at gs://dest-bucket/path/model.pkl (any different content).
3. Run the following operator:

GCSToGCSOperator(
    task_id="copy_model",
    source_bucket="source-bucket",
    source_object="path/model.pkl",
    destination_bucket="dest-bucket",
    destination_object="path/model.pkl",
    replace=False,
    source_object_required=True,
  )

Observed behavior

The destination is overwritten. Logs:

gcs_to_gcs.py:296 - Replaced destination_object with source_object prefix.
gcs_to_gcs.py:311 - There are no new files to sync. Have a nice day!
gcs_to_gcs.py:535 - Executing copy of gs://source-bucket/path/model.pkl to gs://dest-bucket/path/model.pkl

The operator decides "nothing to sync" and then immediately copies anyway.

Root cause

In _copy_source_without_wildcard:

if not self.replace:
    objects = self._ignore_existing_files(...)   # filters out files already in destination

result_uris: list[str] = []
if len(objects) == 0 and prefix:                  # ← entered when destination already has the file
    if hook.exists(self.source_bucket, prefix):
        uri = self._copy_single_object(           # ← copies unconditionally, ignoring `replace`
            hook=hook, source_object=prefix, destination_object=self.destination_object
        )

When replace=False and the destination already contains the object, _ignore_existing_files correctly empties objects. But the fallback branch if len(objects) == 0 and prefix then proceeds to call _copy_single_object without re-checking self.replace or the destination's existence, overwriting the file.

This branch was originally designed for the case where hook.list() returns nothing for a single-file prefix; it does not distinguish that case from "the destination already has the file, so we filtered it out."

Related history

• GCSToGCSOperator ignores replace parameter when there is no wildcard #23162 — original report (closed by Fix GCSToGCSOperator ignores replace parameter when there is no wildcard #23340), which fixed the wildcard path only.
• fix GCSToGCSOperator bug when copy single object with replace to False #45181 — fixed a set vs list indexing bug in the same function; unrelated to this issue.

The single-file fallback in _copy_source_without_wildcard has not been fixed.

What you think should happen instead?

When replace=False, the operator must not overwrite an existing destination object — even on the single-file fallback path. Either:

1. Check self.replace and hook.exists(self.destination_bucket, dest) inside the len(objects) == 0 branch and skip the copy if the destination already exists, or
2. Distinguish "objects was empty from the start" vs "objects was emptied by _ignore_existing_files" and only enter the fallback in the former case.

Operating System

Cloud Composer (Debian-based)

Deployment

Google Cloud Composer

Apache Airflow Provider(s)

google

Versions of Apache Airflow Providers

apache-airflow-providers-google==19.0.0

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

Reproducible 100% of the time when:

• source_object is a single object key (no wildcard)
• replace=False
• the destination object already exists

Confirmed the buggy code path still exists on main (see the snippet from _copy_source_without_wildcard quoted above).

Workaround: a custom subclass that checks hook.exists(destination_bucket, dest) before delegating to super().execute().

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.