feat: expand ListingSchemaProvider to support register and deregister table #3150
Conversation
Nordalf
requested review from
wjones127,
roeap,
rtyler,
hntd187 and
ion-elgreco
as code owners
|
Open for discussion on this one. Have had this locally for a long time and wanted to share the branch. I use this quite a lot in my project and others may find it useful to their project. |
|
@hntd187 can you take a look? :) |
| @@ -36,7 +36,7 @@ pub struct ListingSchemaProvider { | |||
| /// Underlying object store | |||
| store: Arc<dyn ObjectStore>, | |||
| /// A map of table names to a fully quilfied storage location | |||
| tables: DashMap<String, String>, | |||
| tables: Mutex<DashMap<String, Arc<dyn TableProvider>>>, | |||
There was a problem hiding this comment.
do we need to wrap the DashMap in a Mutex? I thought that DashMap handles mutability / locking internally and can just be put in an Arc if it needs to be shared across threads?
Then again, I might be remembering that wrong.
| @@ -54,7 +54,7 @@ impl ListingSchemaProvider { | |||
| Ok(Self { | |||
| authority: uri.to_string(), | |||
| store, | |||
| tables: DashMap::new(), | |||
| tables: Mutex::new(DashMap::new()), | |||
There was a problem hiding this comment.
What is the necessity of the lock? Dashmap shouldn't need a locking mechanism, it's a direct replacement for things such as this.
| self.tables.insert(table_name.to_string(), table_url); | ||
| let Ok(delta_table) = DeltaTableBuilder::from_uri(table_url) | ||
| .with_storage_options(self.storage_options.0.clone()) | ||
| .load() |
There was a problem hiding this comment.
I don't think we need to load every table on listing. You only need a list of tables, loading them gives you much more than that. Additionally, this might be a performance issue or cause a lot of unnecessary API calls if you have a very large catalog and never read 99% of the tables in it.
| self.tables | ||
| .lock() | ||
| .expect("Can't lock tables") | ||
| .insert(name, table.clone()); |
There was a problem hiding this comment.
This will replace the table provider if it currently exists, are you sure this is what you want?
| .lock() | ||
| .expect("Can't lock tables") | ||
| .get(name) | ||
| .map(|t| t.clone()) |
There was a problem hiding this comment.
would .cloned() suffice here instead?
| self.tables.iter().map(|t| t.key().clone()).collect() | ||
| self.tables | ||
| .lock() | ||
| .expect("Can't lock tables") |
There was a problem hiding this comment.
If there is a reason to keep the locking mechanism I think we should handle these errors instead of an expect panic when we can't acquire a lock.
Description
The
ListingSchemaProviderwas missing the implementations of the register and deregister traits from theSchemaProvider. These are now implemented, and to meet requirements from remote object stores, theDeltaTableBuilderis now used to pass along the storage options.The
tablesproperty has been modified to instead of havingDashMap<String, String>, it is now usingMutex<DashMap<String, Arc<dyn TableProvider>>>to supportdowncast_refand mutability elsewhere.Inspiration from: Datafusion ListingSchemaProvider
Related Issue(s)
There is no related issues. I have a clone of this with extra modifications in my own project. I find it useful to have when adding a new schema and wanted to share this with others 👍
Documentation
None