Repository: ClickHouse/ClickHouse
File: ci/praktika/s3.py (Python)

Code before:
                response = client.head_object(Bucket=bucket, Key=key)
                version = int(response.get("Metadata", {}).get("version", "0"))

                # Download the file
                Path(local_path).parent.mkdir(parents=True, exist_ok=True)
                client.download_file(bucket, key, str(local_path))

                print(f"Downloaded file from S3 with version {version} using boto3")
                return version

        # AWS CLI approach: single file with version in metadata
        s3_path_clean = str(s3_path).removeprefix("s3://")
        bucket, key = s3_path_clean.split("/", maxsplit=1)

        # Get object metadata to extract version
        output = Shell.get_output(
            f"aws s3api head-object --bucket {bucket} --key {key}",
            strict=True,
            verbose=True,
        )
        metadata = json.loads(output)
        version = int(metadata.get("Metadata", {}).get("version", "0"))

        # Download the file
        cls.copy_file_from_s3(s3_path=s3_path, local_path=local_path)

        print(f"Downloaded file from S3 with version {version} using AWS CLI")
        return version

    @classmethod
    def copy_file_to_s3_with_version(
        cls, s3_path, local_path, version, text=True, no_strict=False
    ):
        """
        Uploads a file to S3 with version tracking in object metadata.
        Uses conditional PUT with ETag matching for concurrent write safety (optimistic locking).

        :param s3_path: S3 path (with or without s3:// prefix)
        :param local_path: Local file path to upload
        :param version: Version number to set in metadata
        :param text: Whether to set content-type as text/plain
        :param no_strict: Whether to suppress unexpected errors (returns False instead of raising exception)
        :return: True if successful, False if concurrent write detected (object was modified by another process).
                 On unexpected errors: raises exception if no_strict=False, returns False if no_strict=True.
        """
        assert Path(local_path).exists(), f"Path [{local_path}] does not exist"
        assert Path(
            local_path
        ).is_file(), f"Path [{local_path}] is not file. Only files are supported"

        # Use boto3 if available, otherwise AWS CLI

Review comment (suggestion):
`copy_file_from_s3_with_version()` derives `version` from a `head_object` call, then downloads the object without any condition tying the download to the same object revision. If the object changes between `HEAD` and download, the returned `version` can mismatch the downloaded content. To keep version/content consistent, use a single `get_object` call (read metadata from the response and stream the body to `local_path`), or capture the ETag from `head_object` and download with an `If-Match` condition (boto3: pass `ExtraArgs={"IfMatch": etag}` to `download_file`; CLI: use `aws s3api get-object --if-match ...`).
```suggestion
                # boto3 approach: fetch metadata and content atomically
                s3_path_clean = str(s3_path).removeprefix("s3://")
                bucket, key = s3_path_clean.split("/", maxsplit=1)

                # Single get_object call to keep metadata and content consistent
                response = client.get_object(Bucket=bucket, Key=key)
                version = int(response.get("Metadata", {}).get("version", "0"))

                # Stream the body to the local file
                Path(local_path).parent.mkdir(parents=True, exist_ok=True)
                body = response["Body"]
                with open(local_path, "wb") as f:
                    for chunk in iter(lambda: body.read(1024 * 1024), b""):
                        f.write(chunk)

                print(f"Downloaded file from S3 with version {version} using boto3")
                return version

        # AWS CLI approach: fetch metadata and content atomically
        s3_path_clean = str(s3_path).removeprefix("s3://")
        bucket, key = s3_path_clean.split("/", maxsplit=1)

        # Ensure the target directory exists before download
        Path(local_path).parent.mkdir(parents=True, exist_ok=True)

        # aws s3api get-object returns JSON metadata and writes the body to the file
        output = Shell.get_output(
            f"aws s3api get-object --bucket {bucket} --key {key} {local_path}",
            strict=True,
            verbose=True,
        )
        metadata = json.loads(output)
        version = int(metadata.get("Metadata", {}).get("version", "0"))
```

Code after:
                # Use get_object to atomically read metadata and content
                # This prevents race condition where object changes between HEAD and GET
                response = client.get_object(Bucket=bucket, Key=key)
                version = int(response.get("Metadata", {}).get("version", "0"))

                # Stream body to file
                Path(local_path).parent.mkdir(parents=True, exist_ok=True)
                with open(local_path, "wb") as f:
                    f.write(response["Body"].read())

                print(f"Downloaded file from S3 with version {version} using boto3")
                return version

        # AWS CLI approach: use get-object for atomic metadata+content download
        s3_path_clean = str(s3_path).removeprefix("s3://")
        bucket, key = s3_path_clean.split("/", maxsplit=1)

        # Use get-object to atomically download file and get metadata
        # This prevents race condition where object changes between HEAD and GET
        Path(local_path).parent.mkdir(parents=True, exist_ok=True)
        output = Shell.get_output(
            f"aws s3api get-object --bucket {bucket} --key {key} {local_path}",
            strict=True,
            verbose=True,
        )
        metadata = json.loads(output)
        version = int(metadata.get("Metadata", {}).get("version", "0"))

        print(f"Downloaded file from S3 with version {version} using AWS CLI")
        return version

    @classmethod
    def copy_file_to_s3_with_version(
        cls, s3_path, local_path, version, text=True, no_strict=False
    ):
        """
        Uploads a file to S3 with version tracking in object metadata.
        Uses conditional PUT with ETag matching for concurrent write safety (optimistic locking).

        IMPORTANT: Version 0 is a destructive reset operation that overwrites without conditions.
        It MUST NOT be called concurrently - only use version 0 for initialization or when you
        have exclusive access. For concurrent updates, always use version > 0 with the retry pattern:
        read current version, attempt to write version+1, retry on failure.

        :param s3_path: S3 path (with or without s3:// prefix)
        :param local_path: Local file path to upload
        :param version: Version number to set in metadata (0 = destructive reset, >0 = conditional update)
        :param text: Whether to set content-type as text/plain
        :param no_strict: Whether to suppress unexpected errors (returns False instead of raising exception)
        :return: True if successful, False if concurrent write detected (object was modified by another process).
                 On unexpected errors: raises exception if no_strict=False, returns False if no_strict=True.