An asynchronous version of `CatalogList`/`CatalogProvider`/`SchemaProvider`

[v0y4g3r]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

I'm trying to implement a CatalogList/CatalogProvider/SchemaProvider that based on some remote storage service like ZooKeeper/Etcd.

Describe the solution you'd like
But current CatalogList/CatalogProvider/SchemaProvider trait is in a sync manner.

https://github.com/apache/arrow-datafusion/blob/e54110fb592e03704da5f6ebd832b8fe1c51123b/datafusion/core/src/catalog/catalog.rs#L29-L47

That makes asynchronously accessing remote storage service impossible while implementing CatalogList/CatalogProvider/SchemaProvider.

Describe alternatives you've considered

Maybe change these traits to async or provide an async version?

Additional context
I see projects like influxdb-iox did some hack to get around this problem.

[JanKaul]

I would also be interested in this in this feature. I tried using futures::executor::block_on but this leads to my execution hanging as described in tokio-rs/tokio#2376.

[alamb]

Hi @JanKaul and @v0y4g3r -- I wonder if the (very) newly added TableProviderFactories may be helpful / fulfill this usecase #4126

[v0y4g3r]

I would also be interested in this in this feature. I tried using futures::executor::block_on but this leads to my execution hanging as described in tokio-rs/tokio#2376.

I found a workaround. In this case you can spawn a thread dedicating to block:

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

trait CatalogList {
    fn catalog(&self, name: &str) -> Option<Arc<CatalogProvider>>;
}

struct CatalogProvider;

#[derive(Default)]
struct CatalogListImpl {
    map: Arc<RwLock<HashMap<String, Arc<CatalogProvider>>>>,
}

impl CatalogList for CatalogListImpl {
    fn catalog(&self, name: &str) -> Option<Arc<CatalogProvider>> {
        let map = self.map.clone();
        let name = name.to_string();
        std::thread::spawn(move || {
            futures::executor::block_on(async {
                map.read().await.get(&name).cloned()
            })
        }).join().unwrap()
    }
}

#[tokio::main]
async fn main() {
    let catalog_list = CatalogListImpl::default();
    assert!(catalog_list.catalog("hello").is_none());
}

[JanKaul]

Hi @JanKaul and @v0y4g3r -- I wonder if the (very) newly added TableProviderFactories may be helpful / fulfill this usecase #4126

Hi, thanks for the quick reply. I'm trying to implement an iceberg catalog for datafusion. It functions much like a common catalog with schemas(namespaces) and table identifiers. From what I understood about the TableProviderFactory, it wouldn't really fit the use case. One could use the url field to pass in a table identifer like "my_catalog.my_schema.my_table" and get the TableProvider from that. But I don't think this is what it is intended for.

I found a workaround. In this case you can spawn a thread dedicating to block:

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

trait CatalogList {
    fn catalog(&self, name: &str) -> Option<Arc<CatalogProvider>>;
}

struct CatalogProvider;

#[derive(Default)]
struct CatalogListImpl {
    map: Arc<RwLock<HashMap<String, Arc<CatalogProvider>>>>,
}

impl CatalogList for CatalogListImpl {
    fn catalog(&self, name: &str) -> Option<Arc<CatalogProvider>> {
        let map = self.map.clone();
        let name = name.to_string();
        std::thread::spawn(move || {
            futures::executor::block_on(async {
                map.read().await.get(&name).cloned()
            })
        }).join().unwrap()
    }
}

#[tokio::main]
async fn main() {
    let catalog_list = CatalogListImpl::default();
    assert!(catalog_list.catalog("hello").is_none());
}

Awesome, thanks for the heads up. It is probably not the cleanest thing to span another thread while the rest of the application is executed with tokio, but it should work for the time being.

[JanKaul]

I had a look at the code and async versions of CatalogList/CatalogProvider/SchemaProvider would have a huge impact on the code base and also the external API. A lot would need to be async. I understand that that is not desirable.

The solution with std::thread doesn't work for me because the async function that I'm calling uses hyper which expects an async Context from tokio. I could probably create a new Tokio runtime and spawn a thread from there. But it seems not right to create a second tokio runtime.

Investigating the futures::executor::block_on hang that I was experiencing, it looks like the problem comes from calling a synchronous function using block_on from inside an async tokio context. If block on is called from other synchronous code it works. The solution to the problem would be to wrap the call to the blocking synchronous function from an async context with a spawn_blocking.

I'm not sure if that is a route to go, but I could create a branch with the changes that would need to be included in datafusion.

[alamb]

I am not 100% familiar but I know there has been a bunch of work recently to support delta-lake style catalogs in DataFusion using ListingTableProviders and ListingSchemaProviders and TableProviderFactories. So my pattern matching mind says "can iceberg use the same approach"

Maybe someone like @avantgardnerio @timvw @milenkovicm who worked on these features more directly has some other suggestions

Some related PRs:
#4125
#4112
#4095

Here is
https://github.com/delta-io/delta-rs/blob/main/rust/src/delta_datafusion.rs

The stuff in IOx predates any of these new interfaces, so I would likely not recommend it going forward

cc @crepererum

[crepererum]

IMHO it makes sense to make the catalog interfaces async.

FWIW IOx gets around this because we "discover" the whole catalog/tables/schemas before we build the DataFusion data structures. This mostly works because our APIs are scoped in a way that allows that. However midterm I think even IOx could be improved by async schema exploration, because this would lower our the amount of data we need to unconditionally prefetech/prepare (e.g. we would only need to construct schemas for tables that are actually touched by a query).

[milenkovicm]

IMHO it makes sense to make the catalog interfaces async.

I tend to agree with @crepererum, most of the factory methods are async so this would not be a strange change of signature, it would be backward incompatible, but I guess thats not usually show stopper.

change may not be a small one, as it will end up changing context.rs methods, ContextProvider ...

[timvw]

In https://github.com/datafusion-contrib/datafusion-catalogprovider-glue we also first "discover/fetch" the tables/schemas and as of then it's "static/cached".. Async would enable a whole set of use-cases (eg: refreshing catalog(parts)).

[v0y4g3r]

Did an attempt to refactor CatalogList/CatalogProvider/... stuff to async trait, turned out to be not that easy, lots of method defintions need to change.

[avantgardnerio]

Maybe someone like @avantgardnerio

I'm not opposed, as long as we are mindful not to break Ballista (which I don't think we will).

FWIW, when we needed this for Ballista, we just added the async refresh() method and called it at convenient times:

• method
• invocation

[crepererum]

The issue with refresh() though is that it only enables a small subset of features, namely async full discovery. It doesn't allow for partial schema discovery. For example, if you have a hive dataset and you know the tables by scanning one directory, but there might be thousands of them (I'm exaggerating here) and you wanna spare yourself reading all the _common_metadata files, because the user may only query a single table. We in IOx have a similar situation where it would be nice if we could get away with only constructing the schema for tables we actually need (it's not a major blocker, but I nice to have).

[avantgardnerio]

The issue with refresh()

Makes sense. Also refresh() is just "kicking the can down the road". Someone has to call it, the caller needs to be async anyway, so creating a proper async interface would clean things up.

[JanKaul]

If there is a consensus that an async interface is preferable, I can try to make the necessary changes.