Method open

datafusion/datasource-json/src/source.rs:249–459

Open a partitioned JSON file. If `file_meta.range` is `None`, the entire file is opened. Else `file_meta.range` is `Some(FileRange{start, end})`, which corresponds to the byte range [start, end) within the file. Note: `start` or `end` might be in the middle of some lines. In such cases, the following rules are applied to determine which lines to read: 1. The first line of the partition is the li

(&self, partitioned_file: PartitionedFile)

    ///
    /// Note: JSON array format does not support range-based scanning.
    fn open(&self, partitioned_file: PartitionedFile) -> Result<FileOpenFuture> {
        let store = Arc::clone(&self.object_store);
        let schema = Arc::clone(&self.projected_schema);
        let batch_size = self.batch_size;
        let file_compression_type = self.file_compression_type.to_owned();
        let newline_delimited = self.newline_delimited;

        // JSON array format requires reading the complete file
        if !newline_delimited && partitioned_file.range.is_some() {
            return Err(DataFusionError::NotImplemented(
                "JSON array format does not support range-based file scanning. \
                 Disable repartition_file_scans or use newline-delimited JSON format."
                    .to_string(),
            ));
        }

        Ok(Box::pin(async move {
            let file_size = partitioned_file.object_meta.size;
            let location = &partitioned_file.object_meta.location;

            if let Some(file_range) = partitioned_file.range.as_ref() {
                let raw_start: u64 = file_range.start.try_into().map_err(|_| {
                    exec_datafusion_err!(
                        "Expected start range to fit in u64, got {}",
                        file_range.start
                    )
                })?;
                let raw_end: u64 = file_range.end.try_into().map_err(|_| {
                    exec_datafusion_err!(
                        "Expected end range to fit in u64, got {}",
                        file_range.end
                    )
                })?;

                let aligned_stream = AlignedBoundaryStream::new(
                    Arc::clone(&store),
                    location.clone(),
                    raw_start,
                    raw_end,
                    file_size,
                    b'\n',
                )
                .await?
                .map_err(DataFusionError::from);

                let decoder = ReaderBuilder::new(schema)
                    .with_batch_size(batch_size)
                    .build_decoder()?;
                let input = file_compression_type
                    .convert_stream(aligned_stream.boxed())?
                    .fuse();
                let stream = deserialize_stream(
                    input,
                    DecoderDeserializer::new(JsonDecoder::new(decoder)),
                );
                return Ok(stream.map_err(Into::into).boxed());
            }