[deltalake] try is_deltalake with moto server

[danielgafni]

deltalake deadlock reproduction for delta-io/delta-rs#3942

How to reproduce:

uv run pytest tests/metadata_stores/test_delta.py::test_delta_s3_storage_options_passed

Relevant test

[danielgafni]

• [deltalake] try is_deltalake with moto server #953 👈 (View in Graphite)
• main

---

How to use the Graphite Merge Queue

Add either label to this PR to merge it via the merge queue:

• graphite:merge - adds this PR to the back of the merge queue
• graphite:hotfix - for urgent changes, fast-track this PR to the front of the merge queue

You must have a Graphite account in order to use the merge queue. Sign up using this link.

An organization admin has enabled the Graphite Merge Queue in this repository.

Please do not merge from GitHub as this will restart CI on PRs being processed by the merge queue.

This stack of pull requests is managed by Graphite. Learn more about stacking.

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
2523	2	2521	22

View the top 2 failed test(s) by shortest run time

tests/metadata_stores/test_delta.py::test_delta_s3_storage_options_passed

Stack Traces | 181s run time

s3_bucket_and_storage_options = ('test-bucket-299f8b5d', {'AWS_ACCESS_KEY_ID': 'testing', 'AWS_ALLOW_HTTP': 'true', 'AWS_ENDPOINT_URL': 'http://127.0.0.1:56051', 'AWS_REGION': 'us-east-1', ...})
test_features = {'DownstreamFeature': FeatureDefinition(spec=SampleFeatureSpec(key=test_stores/downstream, id_columns=['sample_uid'], ...e': 'UpstreamFeatureB', 'type': 'object'}, feature_class_path='test_stores_features.UpstreamFeatureB', project='test')}

    def test_delta_s3_storage_options_passed(s3_bucket_and_storage_options, test_features) -> None:
        """Verify storage_options are passed to Delta operations with S3.
    
        This ensures object store credentials are correctly forwarded to delta-rs.
        """
        bucket_name, storage_options = s3_bucket_and_storage_options
        store_path = f"s3://{bucket_name}/delta_store"
        feature_cls = test_features["UpstreamFeatureA"]
    
        with DeltaMetadataStore(store_path, storage_options=storage_options) as store:
            metadata = pl.DataFrame(
                {
                    "sample_uid": [1],
                    "metaxy_provenance_by_field": [{"frames": "h1", "audio": "h1"}],
                }
            )
            store.write(feature_cls, metadata)
>           assert store.has_feature(feature_cls, check_fallback=False)

tests/metadata_stores/test_delta.py:107: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../metaxy/metadata_store/base.py:1771: in has_feature
    if self._read_feature(feature) is not None:
.../ext/metadata_stores/delta.py:398: in _read_feature
    if not self._table_exists(feature):
.../ext/metadata_stores/delta.py:251: in _table_exists
    return DeltaTable.is_deltatable(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

table_uri = 's3:.../delta_store/test_stores/upstream_a.delta'
storage_options = {'AWS_ACCESS_KEY_ID': 'testing', 'AWS_ALLOW_HTTP': 'true', 'AWS_ENDPOINT_URL': 'http://127.0.0.1:56051', 'AWS_REGION': 'us-east-1', ...}

    @staticmethod
    def is_deltatable(
        table_uri: str, storage_options: dict[str, str] | None = None
    ) -> bool:
        """
        Returns True if a Delta Table exists at specified path.
        Returns False otherwise.
    
        Args:
            table_uri: the path of the DeltaTable
            storage_options: a dictionary of the options to use for the
                storage backend
        """
>       return RawDeltaTable.is_deltatable(table_uri, storage_options)
E       OSError: Generic S3 error
E                 #x1B[31m↳#x1B[0m Error performing list request
E                  #x1B[31m↳#x1B[0m Error performing GET http://127.0.0.1:56051/test-bucket-299f8b5d?list-type=2&prefix=delta_store%2Ftest_stores%2Fupstream_a.delta%2F_delta_log%2F in 180.851003509s, after 5 retries, max_retries
E                   #x1B[31m↳#x1B[0m 10, retry_timeout
E                    #x1B[31m↳#x1B[0m 180s  - HTTP error
E                     #x1B[31m↳#x1B[0m error sending request
E                      #x1B[31m↳#x1B[0m operation timed out

.venv/lib/python3.10.../site-packages/deltalake/table.py:189: OSError

tests/metadata_stores/test_delta.py::test_delta_sink_lazyframe_s3

Stack Traces | 181s run time

s3_bucket_and_storage_options = ('test-bucket-2329ed74', {'AWS_ACCESS_KEY_ID': 'testing', 'AWS_ALLOW_HTTP': 'true', 'AWS_ENDPOINT_URL': 'http://127.0.0.1:56051', 'AWS_REGION': 'us-east-1', ...})
test_features = {'DownstreamFeature': FeatureDefinition(spec=SampleFeatureSpec(key=test_stores/downstream, id_columns=['sample_uid'], ...e': 'UpstreamFeatureB', 'type': 'object'}, feature_class_path='test_stores_features.UpstreamFeatureB', project='test')}

    def test_delta_sink_lazyframe_s3(s3_bucket_and_storage_options, test_features) -> None:
        """Verify LazyFrame.sink_delta works with S3 storage.
    
        With Polars >= 1.37, lazy frames should be sinked directly without materialization.
        """
        bucket_name, storage_options = s3_bucket_and_storage_options
        store_path = f"s3://{bucket_name}/delta_store"
        feature_cls = test_features["UpstreamFeatureA"]
    
        # Create a lazy frame (native Polars)
        metadata_lazy = pl.LazyFrame(
            {
                "sample_uid": [1, 2, 3],
                "metaxy_provenance_by_field": [
                    {"frames": "h1", "audio": "h1"},
                    {"frames": "h2", "audio": "h2"},
                    {"frames": "h3", "audio": "h3"},
                ],
            }
        )
    
        with DeltaMetadataStore(store_path, storage_options=storage_options) as store:
            store.write(feature_cls, metadata_lazy)
    
            # Verify the feature exists
>           assert store.has_feature(feature_cls, check_fallback=False)

tests/metadata_stores/test_delta.py:168: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../metaxy/metadata_store/base.py:1771: in has_feature
    if self._read_feature(feature) is not None:
.../ext/metadata_stores/delta.py:398: in _read_feature
    if not self._table_exists(feature):
.../ext/metadata_stores/delta.py:251: in _table_exists
    return DeltaTable.is_deltatable(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

table_uri = 's3:.../delta_store/test_stores/upstream_a.delta'
storage_options = {'AWS_ACCESS_KEY_ID': 'testing', 'AWS_ALLOW_HTTP': 'true', 'AWS_ENDPOINT_URL': 'http://127.0.0.1:56051', 'AWS_REGION': 'us-east-1', ...}

    @staticmethod
    def is_deltatable(
        table_uri: str, storage_options: dict[str, str] | None = None
    ) -> bool:
        """
        Returns True if a Delta Table exists at specified path.
        Returns False otherwise.
    
        Args:
            table_uri: the path of the DeltaTable
            storage_options: a dictionary of the options to use for the
                storage backend
        """
>       return RawDeltaTable.is_deltatable(table_uri, storage_options)
E       OSError: Generic S3 error
E                 #x1B[31m↳#x1B[0m Error performing list request
E                  #x1B[31m↳#x1B[0m Error performing GET http://127.0.0.1:56051/test-bucket-2329ed74?list-type=2&prefix=delta_store%2Ftest_stores%2Fupstream_a.delta%2F_delta_log%2F in 181.301491847s, after 5 retries, max_retries
E                   #x1B[31m↳#x1B[0m 10, retry_timeout
E                    #x1B[31m↳#x1B[0m 180s  - HTTP error
E                     #x1B[31m↳#x1B[0m error sending request
E                      #x1B[31m↳#x1B[0m operation timed out

.venv/lib/python3.10.../site-packages/deltalake/table.py:189: OSError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}