Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
pola-rs
GitHub Repository: pola-rs/polars
Path: blob/main/crates/polars-arrow/src/io/ipc/read/array/map.rs
8415 views
1
use std::collections::VecDeque;
2
use std::io::{Read, Seek};
3
4
use polars_buffer::Buffer;
5
use polars_error::{PolarsResult, polars_err};
6
use polars_utils::bool::UnsafeBool;
7
8
use super::super::super::IpcField;
9
use super::super::deserialize::{read, skip};
10
use super::super::read_basic::*;
11
use super::super::{Compression, Dictionaries, IpcBuffer, Node, Version};
12
use crate::array::MapArray;
13
use crate::datatypes::ArrowDataType;
14
use crate::io::ipc::read::array::{try_get_array_length, try_get_field_node};
15
16
#[allow(clippy::too_many_arguments)]
17
pub fn read_map<R: Read + Seek>(
18
field_nodes: &mut VecDeque<Node>,
19
variadic_buffer_counts: &mut VecDeque<usize>,
20
dtype: ArrowDataType,
21
ipc_field: &IpcField,
22
buffers: &mut VecDeque<IpcBuffer>,
23
reader: &mut R,
24
dictionaries: &Dictionaries,
25
block_offset: u64,
26
is_little_endian: bool,
27
compression: Option<Compression>,
28
limit: Option<usize>,
29
version: Version,
30
scratch: &mut Vec<u8>,
31
checked: UnsafeBool,
32
) -> PolarsResult<MapArray> {
33
let field_node = try_get_field_node(field_nodes, &dtype)?;
34
35
let validity = read_validity(
36
buffers,
37
field_node,
38
reader,
39
block_offset,
40
is_little_endian,
41
compression,
42
limit,
43
scratch,
44
)?;
45
46
let length = try_get_array_length(field_node, limit)?;
47
48
let offsets = read_buffer::<i32, _>(
49
buffers,
50
1 + length,
51
reader,
52
block_offset,
53
is_little_endian,
54
compression,
55
scratch,
56
)
57
// Older versions of the IPC format sometimes do not report an offset
58
.or_else(|_| PolarsResult::Ok(Buffer::<i32>::from(vec![0i32])))?;
59
60
let field = MapArray::get_field(&dtype);
61
62
let last_offset: usize = offsets.last().copied().unwrap() as usize;
63
64
let field = read(
65
field_nodes,
66
variadic_buffer_counts,
67
field,
68
&ipc_field.fields[0],
69
buffers,
70
reader,
71
dictionaries,
72
block_offset,
73
is_little_endian,
74
compression,
75
Some(last_offset),
76
version,
77
scratch,
78
checked,
79
)?;
80
MapArray::try_new(dtype, offsets.try_into()?, field, validity)
81
}
82
83
pub fn skip_map(
84
field_nodes: &mut VecDeque<Node>,
85
dtype: &ArrowDataType,
86
buffers: &mut VecDeque<IpcBuffer>,
87
variadic_buffer_counts: &mut VecDeque<usize>,
88
) -> PolarsResult<()> {
89
let _ = field_nodes.pop_front().ok_or_else(|| {
90
polars_err!(
91
oos = "IPC: unable to fetch the field for map. The file or stream is corrupted."
92
)
93
})?;
94
95
let _ = buffers
96
.pop_front()
97
.ok_or_else(|| polars_err!(oos = "IPC: missing validity buffer."))?;
98
let _ = buffers
99
.pop_front()
100
.ok_or_else(|| polars_err!(oos = "IPC: missing offsets buffer."))?;
101
102
let dtype = MapArray::get_field(dtype).dtype();
103
104
skip(field_nodes, dtype, buffers, variadic_buffer_counts)
105
}
106
107