Add optional object-level retention support to GcsToGcsOperator (Google provider)

[bujji8411]

Description

Context
The GcsToGcsOperator in the Google provider supports object copy/move semantics but does not currently allow setting object-level retention metadata (retention period or expiration timestamp) on destination objects.

In regulated or compliance-driven workflows, retention must be enforced immediately after object creation in the destination bucket. Today, this requires an additional operator or custom Python task using the GCS client.

Problem with current approach
Using two steps (copy → set retention) introduces:

Partial-success failure modes (object copied successfully, retention update fails)

Additional retries and operational overhead

More complex DAGs for what is logically a single concern

While the GCS copy operation itself is atomic, the workflow-level behavior is not.

Proposed enhancement
Add an optional, backward-compatible parameter to GcsToGcsOperator to set object-level retention metadata on destination objects immediately after copy/move.

Scope would be intentionally limited to:

Object-level retention only

No bucket-level retention changes

No lifecycle rule management

Default behavior remains unchanged.

Use case/motivation

In regulated and compliance-driven data pipelines (e.g. audit logs, financial records, contractual artifacts), object-level retention must be enforced immediately upon creation in the destination bucket.

A common pattern is copying or moving objects between GCS buckets with different access boundaries (raw → curated, staging → compliance, regional → central). In these cases:

Bucket-level retention cannot always be used (shared buckets, mixed data classes)

Lifecycle rules may not be granular or immediate enough

Retention must be applied per object, deterministically

Current limitation
Today, enforcing object-level retention requires a second step after GcsToGcsOperator completes (e.g. another operator or custom Python using the GCS client). This leads to:

Partial-success scenarios (object copied, retention update fails or is retried)

Additional retry and idempotency handling in DAGs

More complex workflows for a logically single operation

While the underlying GCS copy/move is atomic, the DAG-level workflow is not, which is problematic in compliance-sensitive pipelines.

Why in the same operator
An optional, object-level retention parameter in GcsToGcsOperator would:

Reduce DAG complexity and failure modes

Make retention enforcement explicit and colocated with data movement

Remain fully backward compatible (default behavior unchanged)

Explicit non-goals

No bucket-level retention changes

No lifecycle rule management

No overlap with Storage Transfer Service operators

This proposal is intentionally scoped to a small, optional enhancement that improves correctness and operability for compliance-driven users without changing existing behavior.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[shahar1]

The use case is clear, and both the GcsToGcsOperator as well as the CloudDataTransferService operator indeed don't support it natively.
Considering that GcsToGcsOperator is already composed from multiple hook calls (and will stay like that for good 😄), adding an optional parameter and an extra call to handle this case wouldn't harm IMO.
I wouldn't add it to CloudDataTransferService though, as it does not expose object-level metadata mutation and the transfer itself is handled as a managed operation.
Feel free to raise a PR.

CC: @VladaZakharova @MaksYermak

[jroachgolf84]

I'd take a peek at this method in the hook. Seems like we should be able to pass in some metadata to the GcsToGcsOperator -> GcsHook.

airflow/providers/google/src/airflow/providers/google/cloud/hooks/gcs.py

Lines 480 to 515 in ac45d92

	def upload(
	self,
	bucket_name: str,
	object_name: str,
	filename: str | None = None,
	data: str | bytes | None = None,
	mime_type: str | None = None,
	gzip: bool = False,
	encoding: str = "utf-8",
	chunk_size: int | None = None,
	timeout: int | None = DEFAULT_TIMEOUT,
	num_max_attempts: int = 1,
	metadata: dict | None = None,
	cache_control: str | None = None,
	user_project: str | None = None,
	) -> None:
	"""
	Upload a local file or file data as string or bytes to Google Cloud Storage.
	:param bucket_name: The bucket to upload to.
	:param object_name: The object name to set when uploading the file.
	:param filename: The local file path to the file to be uploaded.
	:param data: The file's data as a string or bytes to be uploaded.
	:param mime_type: The file's mime type set when uploading the file.
	:param gzip: Option to compress local file or file data for upload
	:param encoding: bytes encoding for file data if provided as string
	:param chunk_size: Blob chunk size.
	:param timeout: Request timeout in seconds.
	:param num_max_attempts: Number of attempts to try to upload the file.
	:param metadata: The metadata to be uploaded with the file.
	:param cache_control: Cache-Control metadata field.
	:param user_project: The identifier of the Google Cloud project to bill for the request.
	Required for Requester Pays buckets.
	"""
	def _call_with_retry(f: Callable[[], None]) -> None:

[Saadmadni84]

Hi maintainers,
I’d like to take this issue and work on a PR to add optional object-level retention support to GcsToGcsOperator, keeping the change fully backward compatible.
My plan is to introduce an optional operator parameter and pass it through to GcsHook (leveraging the existing upload/copy hooks) so retention metadata can be applied immediately after copy/move, avoiding the current two-step workflow. I’ll also include unit tests and documentation updates.
Please let me know if this approach aligns with your expectations, or if there are any design considerations you’d like me to account for before I start.
Thanks!

[shahar1]

tests and documentation updates.
Please let me know if this approach aligns with your expectations, or if there are any design considerations you’d like me to account for before I start.
Thanks!

Sounds like a plan, I've assigned you to this task. Good luck!

[potiuk]

@Saadmadni84 We are unassigning you from this issue as part of our updated assignment policy.

This is not meant to discourage your contribution — quite the opposite! You are still very welcome to work on this issue and submit a PR for it. Simply comment that you are working on it and open a PR when ready.

We found that formal assignments were not working well, as they often prevented others from contributing when the assignee was not actively working on the issue.