Build the asynchronous Avro reader with the provided header. This allows initializing the reader with pre-parsed header information. Note that this method is not async because it does not need to perform any I/O operations. Note: Any `header_size_hint` set via [`Self::with_header_size_hint`] is not used when building with a pre-parsed header, since no header fetching occurs.
(
self,
header_info: HeaderInfo,
)
| 207 | /// Note: Any `header_size_hint` set via [`Self::with_header_size_hint`] is not used |
| 208 | /// when building with a pre-parsed header, since no header fetching occurs. |
| 209 | pub fn build_with_header( |
| 210 | self, |
| 211 | header_info: HeaderInfo, |
| 212 | ) -> Result<AsyncAvroFileReader<R>, AvroError> { |
| 213 | let writer_schema = header_info.writer_schema()?; |
| 214 | |
| 215 | // If projection exists, project the reader schema, |
| 216 | // if no reader schema is provided, parse it from the header(get the raw writer schema), and project that |
| 217 | // this projected schema will be the schema used for reading. |
| 218 | let projected_reader_schema = self |
| 219 | .projection |
| 220 | .as_deref() |
| 221 | .map(|projection| { |
| 222 | let base_schema = if let Some(reader_schema) = &self.reader_schema { |
| 223 | reader_schema |
| 224 | } else { |
| 225 | &writer_schema |
| 226 | }; |
| 227 | base_schema.project(projection) |
| 228 | }) |
| 229 | .transpose()?; |
| 230 | |
| 231 | // Use either the projected reader schema or the original reader schema(if no projection) |
| 232 | // (both optional, at worst no reader schema is provided, in which case we read with the writer schema) |
| 233 | let effective_reader_schema = projected_reader_schema |
| 234 | .as_ref() |
| 235 | .or(self.reader_schema.as_ref()) |
| 236 | .map(|s| s.schema()) |
| 237 | .transpose()?; |
| 238 | |
| 239 | let root = { |
| 240 | let writer_schema = writer_schema.schema()?; |
| 241 | let mut builder = AvroFieldBuilder::new(&writer_schema); |
| 242 | if let Some(reader_schema) = &effective_reader_schema { |
| 243 | builder = builder.with_reader_schema(reader_schema); |
| 244 | } |
| 245 | builder |
| 246 | .with_utf8view(self.utf8_view) |
| 247 | .with_strict_mode(self.strict_mode) |
| 248 | .with_tz(self.tz) |
| 249 | .build() |
| 250 | }?; |
| 251 | |
| 252 | let record_decoder = RecordDecoder::try_new_with_options(root.data_type())?; |
| 253 | let decoder = Decoder::from_parts( |
| 254 | self.batch_size, |
| 255 | record_decoder, |
| 256 | None, |
| 257 | IndexMap::new(), |
| 258 | FingerprintAlgorithm::Rabin, |
| 259 | ); |
| 260 | let header_len = header_info.header_len(); |
| 261 | let range = match self.range { |
| 262 | Some(r) => { |
| 263 | // If this PartitionedFile's range starts at 0, we need to skip the header bytes. |
| 264 | // But then we need to seek back 16 bytes to include the sync marker for the first block, |
| 265 | // as the logic in this reader searches the data for the first sync marker(after which a block starts), |
| 266 | // then reads blocks from the count, size etc. |