| 134 | } |
| 135 | |
| 136 | async fn infer_schema( |
| 137 | &self, |
| 138 | _state: &dyn Session, |
| 139 | store: &Arc<dyn ObjectStore>, |
| 140 | objects: &[ObjectMeta], |
| 141 | ) -> Result<SchemaRef> { |
| 142 | let mut schemas = vec![]; |
| 143 | for object in objects { |
| 144 | let r = store.as_ref().get(&object.location).await?; |
| 145 | let schema = match r.payload { |
| 146 | #[cfg(not(target_arch = "wasm32"))] |
| 147 | GetResultPayload::File(mut file, _) => { |
| 148 | match FileReader::try_new(&mut file, None) { |
| 149 | Ok(reader) => reader.schema(), |
| 150 | Err(file_error) => { |
| 151 | // not in the file format, but FileReader read some bytes |
| 152 | // while trying to parse the file and so we need to rewind |
| 153 | // it to the beginning of the file |
| 154 | file.seek(SeekFrom::Start(0))?; |
| 155 | match StreamReader::try_new(&mut file, None) { |
| 156 | Ok(reader) => reader.schema(), |
| 157 | Err(stream_error) => { |
| 158 | return Err(internal_datafusion_err!( |
| 159 | "Failed to parse Arrow file as either file format or stream format. File format error: {file_error}. Stream format error: {stream_error}" |
| 160 | )); |
| 161 | } |
| 162 | } |
| 163 | } |
| 164 | } |
| 165 | } |
| 166 | GetResultPayload::Stream(stream) => infer_stream_schema(stream).await?, |
| 167 | }; |
| 168 | schemas.push(Arc::unwrap_or_clone(schema)); |
| 169 | } |
| 170 | let merged_schema = Schema::try_merge(schemas)?; |
| 171 | Ok(Arc::new(merged_schema)) |
| 172 | } |
| 173 | |
| 174 | async fn infer_stats( |
| 175 | &self, |