Feat/gcp az compatible

[Bennylave]

Related issue: #100

The goal of this PR is to allow for connectivity to Google Cloud from Flowfile as it was done before with AWS.

The tool we intend to use for testing purposes is Fake GCS Server

[Edwardvaneechoud]

Think something went wrong when merging with master. There is some code duplication

[Edwardvaneechoud]

The GCS_BUCKET_NAME and GCS_SERVER_NAME refer to the same env var, intentional?
Also, the test-gcs is the default, but not created as a bucket.

[Edwardvaneechoud]

Is this a typo? Must be bucket.blob

[Edwardvaneechoud]

Missing argument. Can you make this keyword argument? It's pretty hard to debug which one is missing

[Edwardvaneechoud]

This does not work; you should first create the deltalake table locally and next upload it.

def _create_delta_lake_table(gcs_client, arrow_table: pa.Table, bucket_name: str):
    """Creates Delta Lake table by writing locally then uploading to GCS."""
    import tempfile
    import os
    logger.info("Writing Delta Lake table via local filesystem...")

    with tempfile.TemporaryDirectory() as temp_dir:
        # Write Delta table to local filesystem (this will work)
        local_delta_path = f"{temp_dir}/delta-table"

        # Use local filesystem - no storage_options needed
        write_deltalake(local_delta_path, arrow_table, mode='overwrite')
        logger.info(f"✅ Delta table created locally at: {local_delta_path}")

        # Upload all Delta Lake files to GCS
        bucket = gcs_client.bucket(bucket_name)
        uploaded_files = []
        breakpoint()
        for root, dirs, files in os.walk(local_delta_path):
            for file in files:
                local_file_path = os.path.join(root, file)
                rel_path = os.path.relpath(local_file_path, local_delta_path)
                gcs_path = f"delta-lake-table/{rel_path}"

                blob = bucket.blob(gcs_path)
                blob.upload_from_filename(local_file_path)
                uploaded_files.append(gcs_path)
                logger.info(f"Uploaded: {gcs_path}")

        logger.info(f"✅ Delta Lake table uploaded to GCS with {len(uploaded_files)} files")

The nicest thing would be if you could use the same logic as is used in S3 data generator. But I couldn't get that to work. Which can be problematic for testing, since the gcs implementation in Polars also relies on deltalake (delta-rs) library.

Perhaps you can try without anonymous access and provide a credential. I will also check it out further.