[FEAT] Create obstore store in fsspec on demand

[machichima]

Construct the obstore store instance on demand in fsspec when calling methods. This allows automatic store creation for reads/writes across different buckets, aligning usage with fsspec conventions

[kylebarron]

If we allow these, store should be optional?

And before merge we should enable typing overloads for better typing. You can see how from_url is implemented

[machichima]

I use store here for deciding the store Interface (whether it is S3Store, GCSStore, ...), so that in AsyncFsspecStore we don't need to decide the interface based on the protocol.

Maybe there's a better way of deciding the store interface?

obstore_fs: AsyncFsspecStore = fsspec.filesystem(
    "s3",
    store=S3Store,
    config={
        "endpoint": "http://localhost:30002",
        "access_key_id": "minio",
        "secret_access_key": "miniostorage",
        "virtual_hosted_style_request": True,  # path contain bucket name
    },
    client_options={"timeout": "99999s", "allow_http": "true"},
    retry_config={
        "max_retries": 2,
        "backoff": {
            "base": 2,
            "init_backoff": timedelta(seconds=2),
            "max_backoff": timedelta(seconds=16),
        },
        "retry_timeout": timedelta(minutes=3),
    },
)

[machichima]

I'll have a look at the typing later on

[kylebarron]

Oh that's confusing because store is the type of the class and not an instance.

We should be able to use the from_url top level function directly here?

[kylebarron]

It would be nice if this cache size could be user specified but we can come back to it

[kylebarron]

Would there be one fsspec instance per cloud provider? So if you wanted to use s3 and gcs you'd make two separate instances?

[machichima]

Would there be one fsspec instance per cloud provider? So if you wanted to use s3 and gcs you'd make two separate instances?

Based on what I know, to use fsspec, we will do:

fsspec.register_implementation("s3", AsyncFsspecStore)
fsspec.register_implementation("gs", AsyncFsspecStore)

Each will have their own AsyncFsspecStore innstance already. To config, we can use (based on my current implementation):

s3_fs: AsyncFsspecStore = fsspec.filesystem(
    "s3",
    store=S3Store,
    config={...}
)

gcs_fs: AsyncFsspecStore = fsspec.filesystem(
    "gs",
    store=GCSStore,
    config={...}
)

[kylebarron]

It would be nice to take out the store arg and use from_url directly. from_url will automatically construct the correct store based on the url protocol

[machichima]

I use from_url and remove store in the newest commit. However, by doing this, we need to specify the protocol by inherit the AsyncFsspecStore class for each store instance. I added here

obstore/obstore/python/obstore/fsspec.py

Lines 272 to 279 in a0d9e1d

	class S3FsspecStore(AsyncFsspecStore):
	protocol = "s3"
	class GCSFsspecStore(AsyncFsspecStore):
	protocol = "gs"
	class AzureFsspecStore(AsyncFsspecStore):
	protocol = "abfs"

[kylebarron]

Is it true that a single fsspec class can't be associated with more than one protocol? E.g. Azure has three different protocols abfs, adlfs and az, but it looks like adlfs exports three separate classes.

[kylebarron]

The latest PRs allow you to access the config back out of a store, which may be useful to you? You can validate that you already have a store in your cache for a specific bucket

[machichima]

Is it true that a single fsspec class can't be associated with more than one protocol? E.g. Azure has three different protocols abfs, adlfs and az, but it looks like adlfs exports three separate classes.

I think we can if those protocols refer to the same object instance. s3fs do have 2 protocols ("s3", "s3a"), see: https://github.com/fsspec/s3fs/blob/023aecf00b5c6243ff5f8a016dac8b6af3913c6b/s3fs/core.py#L277

I think abfs, adlfs, and az have different implementation so that they exports different classes. If we use them in obstore, I think we can define a class with protocol (abfs, adlfs, az), but need to test is they all work

[kylebarron]

Assuming that this function will always receive something a URL like s3://mybucket/path/to/file, I'm inclined for this function to use urlparse instead of manually handling the parts of the URL

[machichima]

It will not always be s3://mybucket/path/to/file, but may be without protocol like mybucket/path/to/file

[machichima]

I use urlparse like this here, which works for both s3://mybucket/path/to/file and mybucket/path/to/file

obstore/obstore/python/obstore/fsspec.py

Lines 108 to 112 in 75c738e

	res = urlparse(path)
	if res.scheme:
	if res.scheme != self.protocol:
	raise ValueError(f"Expect protocol to be {self.protocol}. Got {res.schema}")
	path = res.netloc + res.path

[kylebarron]

I think we can if those protocols refer to the same object instance. s3fs do have 2 protocols ("s3", "s3a"), see: fsspec/s3fs@023aecf/s3fs/core.py#L277

Oh cool! That seems to indicate that we could have a single class that defines supported protocols as:

    protocol = ("s3", "s3a", "gs", "az", "abfs", "adlfs")

Because the fsspec class used for each is the same? It's just custom kwargs that would need to be passed down for each?

[machichima]

Oh cool! That seems to indicate that we could have a single class that defines supported protocols as:

    protocol = ("s3", "s3a", "gs", "az", "abfs", "adlfs")

Because the fsspec class used for each is the same? It's just custom kwargs that would need to be passed down for each?

I don't think we can put all the protocols together into a class, as when using fsspec.register_implementation("s3", AsyncFsspecStore), fsspec wouldn't tell AsyncFsspecStore what the protocol is, so that when constructing store instance, we cannot get the protocol

obstore/obstore/python/obstore/fsspec.py

Lines 122 to 124 in 6614906

	def _construct_store(self, bucket: str):
	return from_url(
	url=f"{self.protocol}://{bucket}",

I think the better way is to create obstore.fsspec.register("protocol"), that wraps around the fsspec.register and directly set the protocol for AsyncFsspecStore (like what mentioned in this comment), then we do not need more classes. Let me have a try.

[kylebarron]

I did a quick look through your PR; it's really good progress but a few thoughts:

• There are a bunch of cases where bucket, path = self._split_path(path) doesn't work because path is not in scope. E.g. in _cp_file where path1 and path2 are in scope
• in _cp_file we need to validate that the bucket of the source and destination paths are the same
• We need some tests for edits that happen in this PR
• It's not clear how BufferedFileSimple works, because that subclasses from an upstream fsspec.spec.AbstractBufferedFile but doesn't touch obstore apis at all
• If you don't already, I'd highly suggest using a linter like https://docs.astral.sh/ruff/ in your editor, so that you can catch some of these issue before hitting CI

[machichima]

• If you don't already, I'd highly suggest using a linter like https://docs.astral.sh/ruff/ in your editor, so that you can catch some of these issue before hitting CI

Thanks for the suggestion! I just added ruff linter and remove error for path. I also add the check for validating the two bucket name from two path are the same.

• It's not clear how BufferedFileSimple works, because that subclasses from an upstream fsspec.spec.AbstractBufferedFile but doesn't touch obstore apis at all

For BufferedFileSimple, when self.fs.cat_file() is called, it will direct to the _cat_file() function in AsyncFsspecStore.

• We need some tests for edits that happen in this PR

Yes! I will update the test in the next few days

[machichima]

Hi @kylebarron ,

I would like to confirm if changing the output for _ls to the path with bucket name is fine here?

The problem I faced is that when I call find(), it will get into _find() in fsspec, which will then call _walk() (see here). Then in _walk(), _ls is called (see here). But as _walk() will run recurrsively if the path is directory (see here), if _ls give the path without bucket name, then when walk is called recurrsively by the previous result, _ls will get the path without bucket name, which cause error.

For example, call _walk("bucket/") first, and the _ls in _walk give "dir/", then walk will be called again with _walk("dir/"), which gives error

UPDATE: this is how I do it

obstore/obstore/python/obstore/fsspec.py

Lines 297 to 325 in f6ba27c

	async def _ls(self, path, detail=True, **kwargs):
	bucket, path = self._split_path(path)
	store = self._construct_store(bucket)
	result = await obs.list_with_delimiter_async(store, path)
	objects = result["objects"]
	prefs = result["common_prefixes"]
	if detail:
	return [
	{
	"name": self._fill_bucket_name(object["path"], bucket),
	"size": object["size"],
	"type": "file",
	"e_tag": object["e_tag"],
	}
	for object in objects
	] + [
	{
	"name": self._fill_bucket_name(pref, bucket),
	"size": 0,
	"type": "directory",
	}
	for pref in prefs
	]
	else:
	return sorted(
	[self._fill_bucket_name(object["path"], bucket) for object in objects]
	+ [self._fill_bucket_name(pref, bucket) for pref in prefs]
	)

[kylebarron]

I would like to confirm if changing the output for _ls to the path with bucket name is fine here?

I think that's a question for @martindurant. I don't know what's standard for fsspec

It appears that based on s3fs behavior, the bucket name is always returned in list results:

import s3fs

fs = s3fs.S3FileSystem(anon=True)
fs.ls("sentinel-cogs")
# ['sentinel-cogs/sentinel-s2-l2a-cogs']

[kylebarron]

@machichima I merged a PR that updates our use of ruff, a fast Python linter. As part of this, there were some minor updates to the existing fsspec code. Would you be able to merge in main?

I also wanted to have ruff merged so that we could run the lints on this PR's changes.

[kylebarron]

I would like to confirm if changing the output for _ls to the path with bucket name is fine here?

I think that's a question for @martindurant. I don't know what's standard for fsspec

It appears that based on s3fs behavior, the bucket name is always returned in list results:

import s3fs

fs = s3fs.S3FileSystem(anon=True)
fs.ls("sentinel-cogs")
# ['sentinel-cogs/sentinel-s2-l2a-cogs']

It would be great if we could have integration tests with s3fs. We could reuse the existing test setup we have in test_fsspec.py, but then validate that we get the same results for all methods across both the obstore-backed store and the s3fs implementation.

[martindurant]

Yes, fsspec expects that the paths returned by ls() are the same form as you can then use in further open/get/put etc operations. Since the input paths contain the bucket (and optional protocol), the output of ls() should too.

The alternative would be, to have the filesystem instance not expect the bucket in the input. You could still get this to work by extracting the bucket once (as passed in fsspec.open/url_to_fs). This is essentially what the "prefix" filesystem does:

In [10]: fs, path = fsspec.url_to_fs("dir::file:///Users")

In [11]: path
Out[11]: ''

In [12]: fs.ls("", detail=False) # contents of /Users
Out[12]: ['moxie', '.localized', 'Shared', 'mdurant']