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? :) |
|
@Nordalf is this ready? |
|
@ion-elgreco - to me it is ready and I have just squashed the commits 👍 The latest changes resolves the comments but those changes may require a quick look 😄 |
Nordalf
force-pushed
the
feat/support-register-deregister-listing-schema-provider
branch
from
34ee8dd to
33f03b9
Compare
|
@ion-elgreco and @rtyler - This PR should be ready |
|
@Nordalf there are some failed tests |
Oh damn... taking a look at it when I get home later |
|
Looking into fixing the tests now. Edit: Fixed and rebased |
…incl. func to load a table at a later point Signed-off-by: Alexander Falk <[email protected]>
Nordalf
force-pushed
the
feat/support-register-deregister-listing-schema-provider
branch
from
c7b2d14 to
983aedd
Compare
| if !self.table_exist(name.as_str()) { | ||
| self.tables.insert(name, table.clone()); | ||
| } | ||
| Ok(Some(table)) |
There was a problem hiding this comment.
I think the sematics here are a little bit different, i.e. we should raise if the table already exists.
There was a problem hiding this comment.
You are right on this one. I will raise the error but should the table be returned if registered then?
There was a problem hiding this comment.
I will change the code later to:
if self.table_exist(name.as_str()) {
return exec_err!("The table {name} already exists");
};
Ok(self.tables.insert(name, table))
| let Some(provider) = self.tables.get(name).map(|t| t.clone()) else { | ||
| return Ok(None); | ||
| }; | ||
| let provider = | ||
| open_table_with_storage_options(location, self.storage_options.0.clone()).await?; | ||
| Ok(Some(Arc::new(provider) as Arc<dyn TableProvider>)) | ||
| Ok(Some(provider)) |
There was a problem hiding this comment.
We need to make sure we are returning an up to date table here. Datafusion internally will not make any calls to update the log data o.a. so if any operations happend (including the ones we may have done in the same DF session) we would serve a stale snapshot.
|
|
||
| /// Tables are not initialized but have a reference setup. To initialize the delta | ||
| /// table, the `load()` function must be called on the delta table. This function helps with | ||
| /// that and ensures the DashMap is updated | ||
| pub async fn load_table(&self, table_name: &str) -> datafusion::common::Result<()> { | ||
| if let Some(mut table) = self.tables.get_mut(&table_name.to_string()) { | ||
| if let Some(delta_table) = table.value().as_any().downcast_ref::<DeltaTable>() { | ||
| // If table has not yet been loaded, we remove it from the tables map and add it again | ||
| if delta_table.state.is_none() { | ||
| let mut delta_table = delta_table.clone(); | ||
| delta_table.load().await?; | ||
| *table = Arc::from(delta_table); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| Ok(()) | ||
| } |
There was a problem hiding this comment.
I think the table function is the right place to create an up to date snapshot. see comment below.
There was a problem hiding this comment.
So an idea could be to move some of the logic down to the table function and then ensure it is updated?
There was a problem hiding this comment.
I like that idea - if this is what you meant, @roeap
There was a problem hiding this comment.
@roeap - what do you think of just moving the code to the table func and remove the state.is_none() check. Then it will become:
async fn table(
&self,
name: &str,
) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>> {
if let Some(mut table) = self.tables.get_mut(&name.to_string()) {
if let Some(delta_table) = table.value().as_any().downcast_ref::<DeltaTable>() {
let mut delta_table = delta_table.clone();
delta_table.update().await?;
*table = Arc::from(delta_table);
}
return Ok(Some(Arc::clone(table.value())));
}
Ok(None)
}
There was a problem hiding this comment.
This should work ... eventually we may just want to do that in the scan method of the table provider, since right now we are kind-of locking in the version for every provider.
But that would go a bit to far for this PR ...
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 usingDashMap<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