polars-stream 0.53.0

Private crate for the streaming execution engine for the Polars DataFrame library
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

use std::sync::Arc;

use polars_core::config;
use polars_io::cloud::CloudOptions;
use polars_plan::dsl::ScanSource;

use super::{FileReader, NDJsonFileReader};
use crate::nodes::io_sources::multi_scan::reader_interface::builder::FileReaderBuilder;
use crate::nodes::io_sources::multi_scan::reader_interface::capabilities::ReaderCapabilities;
use crate::nodes::io_sources::ndjson::chunk_reader::ChunkReaderBuilder;

pub fn ndjson_reader_capabilities() -> ReaderCapabilities {
    use ReaderCapabilities as RC;

    RC::NEEDS_FILE_CACHE_INIT | RC::ROW_INDEX | RC::PRE_SLICE | RC::NEGATIVE_PRE_SLICE
}

#[cfg(feature = "json")]
impl FileReaderBuilder for polars_plan::dsl::NDJsonReadOptions {
    fn reader_name(&self) -> &str {
        "ndjson"
    }

    fn reader_capabilities(&self) -> ReaderCapabilities {
        ndjson_reader_capabilities()
    }

    fn build_file_reader(
        &self,
        source: ScanSource,
        cloud_options: Option<Arc<CloudOptions>>,
        _scan_source_idx: usize,
    ) -> Box<dyn FileReader> {
        let scan_source = source;
        let chunk_reader_builder = ChunkReaderBuilder::NDJson {
            ignore_errors: self.ignore_errors,
        };
        let verbose = config::verbose();

        let reader = NDJsonFileReader {
            scan_source,
            cloud_options,
            chunk_reader_builder,
            count_rows_fn: polars_io::ndjson::count_rows,
            cached_bytes: None,
            verbose,
        };

        Box::new(reader) as _
    }
}