(&self, name: &str)
| 138 | } |
| 139 | |
| 140 | async fn table(&self, name: &str) -> Result<Option<Arc<dyn TableProvider>>> { |
| 141 | let inner_table = self.inner.table(name).await; |
| 142 | if inner_table.is_ok() |
| 143 | && let Some(inner_table) = inner_table? |
| 144 | { |
| 145 | return Ok(Some(inner_table)); |
| 146 | } |
| 147 | |
| 148 | // if the inner schema provider didn't have a table by |
| 149 | // that name, try to treat it as a listing table |
| 150 | let mut state = self |
| 151 | .state |
| 152 | .upgrade() |
| 153 | .ok_or_else(|| plan_datafusion_err!("locking error"))? |
| 154 | .read() |
| 155 | .clone(); |
| 156 | let mut builder = SessionStateBuilder::from(state.clone()); |
| 157 | let optimized_name = substitute_tilde(name.to_owned()); |
| 158 | let table_url = ListingTableUrl::parse(optimized_name.as_str())?; |
| 159 | let scheme = table_url.scheme(); |
| 160 | let url = table_url.as_ref(); |
| 161 | |
| 162 | // If the store is already registered for this URL then `get_store` |
| 163 | // will return `Ok` which means we don't need to register it again. However, |
| 164 | // if `get_store` returns an `Err` then it means the corresponding store is |
| 165 | // not registered yet and we need to register it |
| 166 | match state.runtime_env().object_store_registry.get_store(url) { |
| 167 | Ok(_) => { /*Nothing to do here, store for this URL is already registered*/ } |
| 168 | Err(_) => { |
| 169 | // Register the store for this URL. Here we don't have access |
| 170 | // to any command options so the only choice is to use an empty collection |
| 171 | match scheme { |
| 172 | "s3" | "oss" | "cos" => { |
| 173 | if let Some(table_options) = builder.table_options() { |
| 174 | table_options.extensions.insert(AwsOptions::default()) |
| 175 | } |
| 176 | } |
| 177 | "gs" | "gcs" => { |
| 178 | if let Some(table_options) = builder.table_options() { |
| 179 | table_options.extensions.insert(GcpOptions::default()) |
| 180 | } |
| 181 | } |
| 182 | _ => {} |
| 183 | }; |
| 184 | state = builder.build(); |
| 185 | let store = get_object_store( |
| 186 | &state, |
| 187 | table_url.scheme(), |
| 188 | url, |
| 189 | &state.default_table_options(), |
| 190 | false, |
| 191 | ) |
| 192 | .await?; |
| 193 | state.runtime_env().register_object_store(url, store); |
| 194 | } |
| 195 | } |
| 196 | self.inner.table(name).await |
| 197 | } |
no test coverage detected