This is an automated email from the ASF dual-hosted git repository.
pankajastro pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 23c8d820870 Add GCS support to the DataFusion object storage layer
(#73370)
23c8d820870 is described below
commit 23c8d820870074f5756321bb812057556b3cb90d
Author: Pankaj Singh <[email protected]>
AuthorDate: Sun Sep 20 14:27:44 2026 +0530
Add GCS support to the DataFusion object storage layer (#73370)
DataFusion credential resolution only handled AWS connections, so
registering a GCS-backed DataSourceConfig raised Unknown connection
type instead of working, even though DataFusion's storage layer
supports it. Add a StorageType.GCS, a GCSObjectStorageProvider using
DataFusion's GoogleCloud, and a google_cloud_platform credential
branch resolving the connection's key_path or keyfile_dict (falling
back to Application Default Credentials when neither is set).
DataFusion's GoogleCloud binding only accepts a file path, not inline
JSON, so a keyfile_dict connection is materialized to a temp file for
the object store's construction and deleted immediately after, since
the underlying credentials are parsed once at construction time and
never re-read.
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../ai/docs/operators/llm_schema_compare.rst | 2 +-
providers/common/ai/docs/operators/llm_sql.rst | 2 +-
providers/common/ai/docs/toolsets.rst | 2 +-
providers/common/sql/docs/operators.rst | 23 +++++-
.../sql/src/airflow/providers/common/sql/config.py | 3 +
.../providers/common/sql/datafusion/engine.py | 21 ++++++
.../sql/datafusion/object_storage_provider.py | 59 ++++++++++++++-
.../common/sql/example_dags/example_analytics.py | 13 ++++
.../unit/common/sql/datafusion/test_engine.py | 86 +++++++++++++++++++++-
.../sql/datafusion/test_object_storage_provider.py | 84 +++++++++++++++++++++
.../sql/tests/unit/common/sql/test_config.py | 1 +
11 files changed, 289 insertions(+), 7 deletions(-)
diff --git a/providers/common/ai/docs/operators/llm_schema_compare.rst
b/providers/common/ai/docs/operators/llm_schema_compare.rst
index 2efed391b01..e7145deea86 100644
--- a/providers/common/ai/docs/operators/llm_schema_compare.rst
+++ b/providers/common/ai/docs/operators/llm_schema_compare.rst
@@ -62,7 +62,7 @@ With Object Storage or a Database Table
Use ``data_sources`` with
:class:`~airflow.providers.common.sql.config.DataSourceConfig` to include
-object-storage sources (S3 Parquet, CSV, Iceberg, etc.) in the comparison.
+object-storage sources (S3 or GCS Parquet, CSV, Iceberg, etc.) in the
comparison.
These can be freely combined with ``db_conn_ids``. Whether an entry is
introspected via ``DbApiHook`` or DataFusion depends on what its ``conn_id``
resolves to, not on its ``uri``/``format`` fields — a ``DataSourceConfig``
diff --git a/providers/common/ai/docs/operators/llm_sql.rst
b/providers/common/ai/docs/operators/llm_sql.rst
index ba1d05f36de..07fe793a09f 100644
--- a/providers/common/ai/docs/operators/llm_sql.rst
+++ b/providers/common/ai/docs/operators/llm_sql.rst
@@ -60,7 +60,7 @@ With Object Storage
-------------------
Use ``datasource_config`` to generate queries for data stored in object storage
-(e.g., S3, local filesystem) via `DataFusion
<https://datafusion.apache.org/>`_.
+(e.g., S3, GCS, local filesystem) via `DataFusion
<https://datafusion.apache.org/>`_.
The operator uses
:class:`~airflow.providers.common.sql.config.DataSourceConfig`
to register the object storage source as a table so the LLM can include it in
the schema context.
diff --git a/providers/common/ai/docs/toolsets.rst
b/providers/common/ai/docs/toolsets.rst
index e89bed02b69..2fa9a5b7151 100644
--- a/providers/common/ai/docs/toolsets.rst
+++ b/providers/common/ai/docs/toolsets.rst
@@ -284,7 +284,7 @@ later request.
Curated toolset wrapping
:class:`~airflow.providers.common.sql.datafusion.engine.DataFusionEngine`
with three tools — ``list_tables``, ``get_schema``, and ``query`` — for
-querying files on object stores (S3, local filesystem, Iceberg) via Apache
DataFusion.
+querying files on object stores (S3, GCS, local filesystem, Iceberg) via
Apache DataFusion.
.. list-table::
:header-rows: 1
diff --git a/providers/common/sql/docs/operators.rst
b/providers/common/sql/docs/operators.rst
index 265832301c0..85c501e3679 100644
--- a/providers/common/sql/docs/operators.rst
+++ b/providers/common/sql/docs/operators.rst
@@ -296,10 +296,11 @@ The Analytics Operator is ideal for performing efficient,
high-performance analy
Supported Storage Systems
-------------------------
- S3
+- GCS
- Local File System
.. note::
- GCS, Azure, HTTP, Delta, Iceberg are not yet supported but will be added in
the future.
+ Azure, HTTP, Delta are not yet supported but will be added in the future.
@@ -341,6 +342,26 @@ S3 Storage
:start-after: [START howto_analytics_operator_with_s3]
:end-before: [END howto_analytics_operator_with_s3]
+GCS Storage
+-----------
+Use a ``conn_id`` pointing to a ``google_cloud_platform`` connection.
Credentials are
+resolved in this order:
+
+1. ``key_path`` -- a service account JSON file on disk
+2. ``keyfile_dict`` -- the service account JSON contents, inline
+3. the ``GOOGLE_APPLICATION_CREDENTIALS`` environment variable
+4. Application Default Credentials -- the local ``gcloud`` CLI credentials
file, or
+ the GCE/GKE metadata server
+
+``key_path`` and ``keyfile_dict`` are mutually exclusive. ``key_secret_name``,
+``credential_config_file``, and ``impersonation_chain`` are not supported.
+
+.. exampleinclude::
/../../sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_analytics_operator_with_gcs]
+ :end-before: [END howto_analytics_operator_with_gcs]
+
Local File System Storage
-------------------------
.. exampleinclude::
/../../sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
diff --git a/providers/common/sql/src/airflow/providers/common/sql/config.py
b/providers/common/sql/src/airflow/providers/common/sql/config.py
index 975bc58c9d4..a1a856a17f3 100644
--- a/providers/common/sql/src/airflow/providers/common/sql/config.py
+++ b/providers/common/sql/src/airflow/providers/common/sql/config.py
@@ -47,6 +47,7 @@ class StorageType(str, Enum):
"""Storage types for Data Fusion."""
S3 = "s3"
+ GCS = "gcs"
LOCAL = "local"
@@ -114,6 +115,8 @@ class DataSourceConfig:
"""Extract storage type."""
if self.uri.startswith("s3://"):
return StorageType.S3
+ if self.uri.startswith("gs://"):
+ return StorageType.GCS
if self.uri.startswith("file://"):
return StorageType.LOCAL
raise ValueError(f"Unsupported storage type for URI: {self.uri}")
diff --git
a/providers/common/sql/src/airflow/providers/common/sql/datafusion/engine.py
b/providers/common/sql/src/airflow/providers/common/sql/datafusion/engine.py
index 21e0b72390b..207ab3e0f77 100644
--- a/providers/common/sql/src/airflow/providers/common/sql/datafusion/engine.py
+++ b/providers/common/sql/src/airflow/providers/common/sql/datafusion/engine.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+import os
from typing import TYPE_CHECKING, Any
from datafusion import SessionContext
@@ -169,6 +170,26 @@ class DataFusionEngine(LoggingMixin):
credentials = self._remove_none_values(credentials)
extra_config = _fetch_extra_configs(["region", "endpoint"])
+ case "google_cloud_platform":
+ extra_dejson = conn.extra_dejson
+ for unsupported_field in ("key_secret_name",
"credential_config_file", "impersonation_chain"):
+ if extra_dejson.get(unsupported_field):
+ raise ValueError(
+ f"Connection field {unsupported_field!r} is not
supported for DataFusion "
+ "GCS access; only key_path, keyfile_dict,
GOOGLE_APPLICATION_CREDENTIALS, or "
+ "ambient credentials (gcloud ADC file / metadata
server) are used."
+ )
+ key_path = extra_dejson.get("key_path") or None
+ keyfile_dict = extra_dejson.get("keyfile_dict") or None
+ if key_path and keyfile_dict:
+ raise ValueError(
+ "The `keyfile_dict` and `key_path` fields are mutually
exclusive. "
+ "Please provide only one value."
+ )
+ if not key_path and not keyfile_dict:
+ key_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
+ credentials = self._remove_none_values({"key_path": key_path,
"keyfile_dict": keyfile_dict})
+
case _:
raise ValueError(f"Unknown connection type {conn.conn_type}")
return credentials, extra_config
diff --git
a/providers/common/sql/src/airflow/providers/common/sql/datafusion/object_storage_provider.py
b/providers/common/sql/src/airflow/providers/common/sql/datafusion/object_storage_provider.py
index 1ce917a1bb2..608a6959940 100644
---
a/providers/common/sql/src/airflow/providers/common/sql/datafusion/object_storage_provider.py
+++
b/providers/common/sql/src/airflow/providers/common/sql/datafusion/object_storage_provider.py
@@ -16,7 +16,11 @@
# under the License.
from __future__ import annotations
-from datafusion.object_store import AmazonS3, LocalFileSystem
+import json
+import tempfile
+from pathlib import Path
+
+from datafusion.object_store import AmazonS3, GoogleCloud, LocalFileSystem
from airflow.providers.common.sql.config import ConnectionConfig, StorageType
from airflow.providers.common.sql.datafusion.base import ObjectStorageProvider
@@ -53,6 +57,56 @@ class S3ObjectStorageProvider(ObjectStorageProvider):
return "s3://"
+class GCSObjectStorageProvider(ObjectStorageProvider):
+ """GCS Object Storage Provider using DataFusion's GoogleCloud."""
+
+ @property
+ def get_storage_type(self) -> StorageType:
+ """Return the storage type."""
+ return StorageType.GCS
+
+ def create_object_store(self, path: str, connection_config:
ConnectionConfig | None = None):
+ """Create a GCS object store using DataFusion's GoogleCloud."""
+ if connection_config is None:
+ raise ValueError(f"connection_config must be provided for
{self.get_storage_type}")
+
+ credentials = connection_config.credentials
+ key_path = credentials.get("key_path")
+ keyfile_dict = credentials.get("keyfile_dict")
+ temp_key_path: str | None = None
+
+ try:
+ bucket = self.get_bucket(path)
+
+ if not key_path and keyfile_dict:
+ # GoogleCloud only accepts a file path, not inline JSON; safe
to delete once
+ # constructed since the credentials are read only at
construction time.
+ key_content = keyfile_dict if isinstance(keyfile_dict, str)
else json.dumps(keyfile_dict)
+ with tempfile.NamedTemporaryFile(mode="w", suffix=".json",
delete=False) as key_file:
+ key_file.write(key_content)
+ temp_key_path = key_path = key_file.name
+
+ if key_path is not None and not Path(key_path).is_file():
+ raise FileNotFoundError(f"Service account key file not found:
{key_path}")
+
+ gcs_store = GoogleCloud(bucket_name=bucket,
service_account_path=key_path)
+ self.log.info("Created GCS object store for bucket %s", bucket)
+
+ return gcs_store
+
+ except BaseException as e:
+ # A bad key file panics as pyo3_runtime.PanicException, not a
plain Exception.
+ raise ObjectStoreCreationException(f"Failed to create GCS object
store: {e}")
+
+ finally:
+ if temp_key_path is not None:
+ Path(temp_key_path).unlink(missing_ok=True)
+
+ def get_scheme(self) -> str:
+ """Return the scheme for GCS."""
+ return "gs://"
+
+
class LocalObjectStorageProvider(ObjectStorageProvider):
"""Local Object Storage Provider using DataFusion's LocalFileSystem."""
@@ -72,9 +126,10 @@ class LocalObjectStorageProvider(ObjectStorageProvider):
def get_object_storage_provider(storage_type: StorageType) ->
ObjectStorageProvider:
"""Get an object storage provider based on the storage type."""
- # TODO: Add support for GCS, Azure, HTTP:
https://datafusion.apache.org/python/autoapi/datafusion/object_store/index.html
+ # TODO: Add support for Azure, HTTP:
https://datafusion.apache.org/python/autoapi/datafusion/object_store/index.html
providers: dict[StorageType, type] = {
StorageType.S3: S3ObjectStorageProvider,
+ StorageType.GCS: GCSObjectStorageProvider,
StorageType.LOCAL: LocalObjectStorageProvider,
}
diff --git
a/providers/common/sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
b/providers/common/sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
index 07a23d41e81..0f80fc6100e 100644
---
a/providers/common/sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
+++
b/providers/common/sql/src/airflow/providers/common/sql/example_dags/example_analytics.py
@@ -30,6 +30,10 @@ datasource_config_local = DataSourceConfig(
conn_id="", table_name="users_data", uri="file:///path/to/",
format="parquet"
)
+datasource_config_gcs = DataSourceConfig(
+ conn_id="google_cloud_default", table_name="users_data",
uri="gs://bucket/path/", format="parquet"
+)
+
datasource_config_iceberg = DataSourceConfig(
conn_id="iceberg_default",
table_name="users_data",
@@ -72,6 +76,15 @@ with DAG(
# [END howto_analytics_operator_with_s3]
+ # [START howto_analytics_operator_with_gcs]
+ analytics_with_gcs = AnalyticsOperator(
+ task_id="analytics_with_gcs",
+ datasource_configs=[datasource_config_gcs],
+ queries=["SELECT * FROM users_data", "SELECT count(*) FROM
users_data"],
+ )
+ analytics_with_s3 >> analytics_with_gcs
+ # [END howto_analytics_operator_with_gcs]
+
# [START howto_analytics_operator_with_local]
analytics_with_local = AnalyticsOperator(
task_id="analytics_with_local",
diff --git
a/providers/common/sql/tests/unit/common/sql/datafusion/test_engine.py
b/providers/common/sql/tests/unit/common/sql/datafusion/test_engine.py
index d48c705046f..21287cfa075 100644
--- a/providers/common/sql/tests/unit/common/sql/datafusion/test_engine.py
+++ b/providers/common/sql/tests/unit/common/sql/datafusion/test_engine.py
@@ -73,7 +73,12 @@ class TestDataFusionEngine:
@pytest.mark.parametrize(
("storage_type", "format", "scheme"),
- [("s3", "parquet", "s3"), ("s3", "csv", "s3"), ("s3", "avro", "s3")],
+ [
+ ("s3", "parquet", "s3"),
+ ("s3", "csv", "s3"),
+ ("s3", "avro", "s3"),
+ ("gcs", "parquet", "gs"),
+ ],
)
@patch("airflow.providers.common.sql.datafusion.engine.get_object_storage_provider",
autospec=True)
@patch.object(DataFusionEngine, "_get_connection_config")
@@ -281,6 +286,85 @@ class TestDataFusionEngine:
assert result.credentials == expected.credentials
assert result.extra_config == expected.extra_config
+ def test_get_credentials_gcs_with_key_path(self):
+ mock_conn = MagicMock()
+ mock_conn.conn_type = "google_cloud_platform"
+ mock_conn.extra_dejson = {"key_path": "/path/to/key.json"}
+ engine = DataFusionEngine()
+
+ credentials, extra_config = engine._get_credentials(mock_conn)
+
+ assert credentials == {"key_path": "/path/to/key.json"}
+ assert extra_config == {}
+
+ def test_get_credentials_gcs_with_keyfile_dict(self):
+ mock_conn = MagicMock()
+ mock_conn.conn_type = "google_cloud_platform"
+ mock_conn.extra_dejson = {"keyfile_dict": '{"type":
"service_account"}'}
+ engine = DataFusionEngine()
+
+ credentials, extra_config = engine._get_credentials(mock_conn)
+
+ assert credentials == {"keyfile_dict": '{"type": "service_account"}'}
+ assert extra_config == {}
+
+ def test_get_credentials_gcs_without_key_path_uses_ambient_auth(self,
monkeypatch):
+ monkeypatch.delenv("GOOGLE_APPLICATION_CREDENTIALS", raising=False)
+ mock_conn = MagicMock()
+ mock_conn.conn_type = "google_cloud_platform"
+ mock_conn.extra_dejson = {}
+ engine = DataFusionEngine()
+
+ credentials, extra_config = engine._get_credentials(mock_conn)
+
+ assert credentials == {}
+ assert extra_config == {}
+
+ def
test_get_credentials_gcs_uses_google_application_credentials_env_var(self,
monkeypatch):
+ monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/env/key.json")
+ mock_conn = MagicMock()
+ mock_conn.conn_type = "google_cloud_platform"
+ mock_conn.extra_dejson = {}
+ engine = DataFusionEngine()
+
+ credentials, extra_config = engine._get_credentials(mock_conn)
+
+ assert credentials == {"key_path": "/env/key.json"}
+ assert extra_config == {}
+
+ def
test_get_credentials_gcs_connection_key_path_takes_priority_over_env_var(self,
monkeypatch):
+ monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/env/key.json")
+ mock_conn = MagicMock()
+ mock_conn.conn_type = "google_cloud_platform"
+ mock_conn.extra_dejson = {"key_path": "/connection/key.json"}
+ engine = DataFusionEngine()
+
+ credentials, extra_config = engine._get_credentials(mock_conn)
+
+ assert credentials == {"key_path": "/connection/key.json"}
+ assert extra_config == {}
+
+ def
test_get_credentials_gcs_key_path_and_keyfile_dict_mutually_exclusive(self):
+ mock_conn = MagicMock()
+ mock_conn.conn_type = "google_cloud_platform"
+ mock_conn.extra_dejson = {"key_path": "/path/to/key.json",
"keyfile_dict": "{}"}
+ engine = DataFusionEngine()
+
+ with pytest.raises(ValueError, match="mutually exclusive"):
+ engine._get_credentials(mock_conn)
+
+ @pytest.mark.parametrize(
+ "unsupported_field", ["key_secret_name", "credential_config_file",
"impersonation_chain"]
+ )
+ def test_get_credentials_gcs_rejects_unsupported_identity_fields(self,
unsupported_field):
+ mock_conn = MagicMock()
+ mock_conn.conn_type = "google_cloud_platform"
+ mock_conn.extra_dejson = {unsupported_field: "some-value"}
+ engine = DataFusionEngine()
+
+ with pytest.raises(ValueError, match=f"{unsupported_field!r} is not
supported"):
+ engine._get_credentials(mock_conn)
+
def test_get_credentials_unknown_type(self):
mock_conn = MagicMock()
mock_conn.conn_type = "dummy"
diff --git
a/providers/common/sql/tests/unit/common/sql/datafusion/test_object_storage_provider.py
b/providers/common/sql/tests/unit/common/sql/datafusion/test_object_storage_provider.py
index 9b0ff756a1f..6c1f2524d72 100644
---
a/providers/common/sql/tests/unit/common/sql/datafusion/test_object_storage_provider.py
+++
b/providers/common/sql/tests/unit/common/sql/datafusion/test_object_storage_provider.py
@@ -16,6 +16,8 @@
# under the License.
from __future__ import annotations
+import json
+import os
from unittest.mock import patch
import pytest
@@ -23,6 +25,7 @@ import pytest
from airflow.providers.common.sql.config import ConnectionConfig, StorageType
from airflow.providers.common.sql.datafusion.exceptions import
ObjectStoreCreationException
from airflow.providers.common.sql.datafusion.object_storage_provider import (
+ GCSObjectStorageProvider,
LocalObjectStorageProvider,
S3ObjectStorageProvider,
get_object_storage_provider,
@@ -58,6 +61,86 @@ class TestObjectStorageProvider:
with pytest.raises(ObjectStoreCreationException, match="Failed to
create S3 object store"):
provider.create_object_store("s3://demo-data/path",
connection_config)
+
@patch("airflow.providers.common.sql.datafusion.object_storage_provider.GoogleCloud")
+ def test_gcs_provider_success_with_key_path(self, mock_gcs, tmp_path):
+ key_path = tmp_path / "key.json"
+ key_path.write_text("{}")
+ provider = GCSObjectStorageProvider()
+ connection_config = ConnectionConfig(
+ conn_id="google_cloud_default",
+ credentials={"key_path": str(key_path)},
+ )
+
+ store = provider.create_object_store("gs://demo-data/path",
connection_config)
+
+ mock_gcs.assert_called_once_with(bucket_name="demo-data",
service_account_path=str(key_path))
+ assert store == mock_gcs.return_value
+ assert provider.get_storage_type == StorageType.GCS
+ assert provider.get_scheme() == "gs://"
+
+
@patch("airflow.providers.common.sql.datafusion.object_storage_provider.GoogleCloud")
+ def test_gcs_provider_success_with_keyfile_dict(self, mock_gcs):
+ provider = GCSObjectStorageProvider()
+ keyfile_dict = {"type": "service_account", "private_key": "fake"}
+ connection_config = ConnectionConfig(
+ conn_id="google_cloud_default",
+ credentials={"keyfile_dict": keyfile_dict},
+ )
+
+ provider.create_object_store("gs://demo-data/path", connection_config)
+
+ mock_gcs.assert_called_once()
+ temp_path = mock_gcs.call_args.kwargs["service_account_path"]
+ assert not os.path.exists(temp_path), "temp key file should be deleted
after use"
+
+
@patch("airflow.providers.common.sql.datafusion.object_storage_provider.GoogleCloud")
+ def test_gcs_provider_writes_keyfile_dict_content_before_cleanup(self,
mock_gcs):
+ written_content = {}
+
+ def _capture_path(*, bucket_name, service_account_path):
+ with open(service_account_path) as f:
+ written_content["data"] = json.load(f)
+
+ mock_gcs.side_effect = _capture_path
+ provider = GCSObjectStorageProvider()
+ keyfile_dict = {"type": "service_account", "private_key": "fake"}
+ connection_config = ConnectionConfig(
+ conn_id="google_cloud_default",
+ credentials={"keyfile_dict": keyfile_dict},
+ )
+
+ provider.create_object_store("gs://demo-data/path", connection_config)
+
+ assert written_content["data"] == keyfile_dict
+
+ def test_gcs_provider_failure(self):
+ provider = GCSObjectStorageProvider()
+ connection_config = ConnectionConfig(conn_id="google_cloud_default")
+
+ with patch(
+
"airflow.providers.common.sql.datafusion.object_storage_provider.GoogleCloud",
+ side_effect=Exception("Error"),
+ ):
+ with pytest.raises(ObjectStoreCreationException, match="Failed to
create GCS object store"):
+ provider.create_object_store("gs://demo-data/path",
connection_config)
+
+ def test_gcs_provider_missing_key_file_raises_clear_error(self):
+ """Uses the real GoogleCloud binding, not a mock, since it's the one
that panics."""
+ provider = GCSObjectStorageProvider()
+ connection_config = ConnectionConfig(
+ conn_id="google_cloud_default",
+ credentials={"key_path": "/nonexistent/key.json"},
+ )
+
+ with pytest.raises(ObjectStoreCreationException, match="Failed to
create GCS object store"):
+ provider.create_object_store("gs://demo-data/path",
connection_config)
+
+ def test_gcs_provider_requires_connection_config(self):
+ provider = GCSObjectStorageProvider()
+
+ with pytest.raises(ValueError, match="connection_config must be
provided for gcs"):
+ provider.create_object_store("gs://demo-data/path")
+
@patch("airflow.providers.common.sql.datafusion.object_storage_provider.LocalFileSystem")
def test_local_provider(self, mock_local):
provider = LocalObjectStorageProvider()
@@ -68,6 +151,7 @@ class TestObjectStorageProvider:
def test_get_object_storage_provider(self):
assert isinstance(get_object_storage_provider(StorageType.S3),
S3ObjectStorageProvider)
+ assert isinstance(get_object_storage_provider(StorageType.GCS),
GCSObjectStorageProvider)
assert isinstance(get_object_storage_provider(StorageType.LOCAL),
LocalObjectStorageProvider)
with pytest.raises(ValueError, match="Unsupported storage type"):
diff --git a/providers/common/sql/tests/unit/common/sql/test_config.py
b/providers/common/sql/tests/unit/common/sql/test_config.py
index 5478b5a2e44..14f3827c6a0 100644
--- a/providers/common/sql/tests/unit/common/sql/test_config.py
+++ b/providers/common/sql/tests/unit/common/sql/test_config.py
@@ -33,6 +33,7 @@ class TestDataSourceConfig:
("uri", "expected_type"),
[
("s3://bucket/path", StorageType.S3),
+ ("gs://bucket/path", StorageType.GCS),
("file:///path/to/file", StorageType.LOCAL),
],
)
