|
| 1 | +//! The columnar boundary (TASK-60): `infer_arrow(pa.Table) -> pa.Table`. |
| 2 | +//! |
| 3 | +//! Ingest walks pyarrow buffers directly (address + size via the Python |
| 4 | +//! buffer API — no arrow-rs dependency) into the engine's `ColData` lanes; |
| 5 | +//! emit builds `pa.Array.from_buffers` per OUTPUT COLUMN from rust-built |
| 6 | +//! buffers. Zero per-value Python objects on either side — the measured |
| 7 | +//! ~1.4 µs/row of boxing at 31-column width simply disappears. v1 copies |
| 8 | +//! buffers (the proposal's copy-first recommendation); zero-copy lanes are |
| 9 | +//! a measured follow-up, not assumed. |
| 10 | +
|
| 11 | +use pyo3::prelude::*; |
| 12 | +use pyo3::types::{PyBytes, PyList}; |
| 13 | + |
| 14 | +use crate::error::InterpError; |
| 15 | +use crate::specializer::exec::{Batch, ColData, OutCol, RunState}; |
| 16 | +use crate::specializer::ir::{Col, Ty}; |
| 17 | + |
| 18 | +fn err(msg: impl Into<String>) -> PyErr { |
| 19 | + InterpError::Build(msg.into()).into() |
| 20 | +} |
| 21 | + |
| 22 | +/// One arrow array's raw view: offset-adjusted validity + buffers. |
| 23 | +struct RawArray<'py> { |
| 24 | + /// Kept alive for the duration of the buffer reads. |
| 25 | + _arr: Bound<'py, PyAny>, |
| 26 | + offset: usize, |
| 27 | + len: usize, |
| 28 | + bufs: Vec<Option<(usize, usize)>>, // (address, size) per buffer slot |
| 29 | +} |
| 30 | + |
| 31 | +impl RawArray<'_> { |
| 32 | + fn valid(&self, i: usize) -> bool { |
| 33 | + match self.bufs.first().copied().flatten() { |
| 34 | + None => true, |
| 35 | + Some((addr, size)) => { |
| 36 | + let bit = self.offset + i; |
| 37 | + let byte = bit / 8; |
| 38 | + debug_assert!(byte < size); |
| 39 | + let b = unsafe { *(addr as *const u8).add(byte) }; |
| 40 | + (b >> (bit % 8)) & 1 == 1 |
| 41 | + } |
| 42 | + } |
| 43 | + } |
| 44 | + |
| 45 | + /// Typed data slice starting at the array's offset. |
| 46 | + unsafe fn data<T: Copy>(&self, slot: usize) -> *const T { |
| 47 | + let (addr, _) = self.bufs[slot].expect("data buffer present"); |
| 48 | + unsafe { (addr as *const T).add(self.offset) } |
| 49 | + } |
| 50 | +} |
| 51 | + |
| 52 | +fn raw_array<'py>(arr: Bound<'py, PyAny>) -> PyResult<RawArray<'py>> { |
| 53 | + let offset: usize = arr.getattr("offset")?.extract()?; |
| 54 | + let len: usize = arr.call_method0("__len__")?.extract()?; |
| 55 | + let mut bufs = Vec::new(); |
| 56 | + for b in arr.call_method0("buffers")?.try_iter()? { |
| 57 | + let b = b?; |
| 58 | + if b.is_none() { |
| 59 | + bufs.push(None); |
| 60 | + } else { |
| 61 | + let addr: usize = b.getattr("address")?.extract()?; |
| 62 | + let size: usize = b.getattr("size")?.extract()?; |
| 63 | + bufs.push(Some((addr, size))); |
| 64 | + } |
| 65 | + } |
| 66 | + Ok(RawArray { |
| 67 | + _arr: arr, |
| 68 | + offset, |
| 69 | + len, |
| 70 | + bufs, |
| 71 | + }) |
| 72 | +} |
| 73 | + |
| 74 | +/// A single-chunk column by NAME from a Table or RecordBatch. |
| 75 | +fn column<'py>(batch: &Bound<'py, PyAny>, name: &str) -> PyResult<Bound<'py, PyAny>> { |
| 76 | + let schema = batch.getattr("schema")?; |
| 77 | + let idx: i64 = schema |
| 78 | + .call_method1("get_field_index", (name,))? |
| 79 | + .extract()?; |
| 80 | + if idx < 0 { |
| 81 | + return Err(err(format!( |
| 82 | + "infer_arrow: the batch is missing column '{name}'" |
| 83 | + ))); |
| 84 | + } |
| 85 | + let col = batch.call_method1("column", (idx,))?; |
| 86 | + // Table columns are ChunkedArrays; RecordBatch columns are Arrays. |
| 87 | + if col.hasattr("num_chunks")? { |
| 88 | + let n: usize = col.getattr("num_chunks")?.extract()?; |
| 89 | + match n { |
| 90 | + 1 => col.call_method1("chunk", (0,)), |
| 91 | + 0 => col.call_method1("combine_chunks", ()), |
| 92 | + _ => Err(err(format!( |
| 93 | + "infer_arrow: column '{name}' has {n} chunks — call \ |
| 94 | + combine_chunks() on the table first" |
| 95 | + ))), |
| 96 | + } |
| 97 | + } else { |
| 98 | + Ok(col) |
| 99 | + } |
| 100 | +} |
| 101 | + |
| 102 | +/// pyarrow batch -> engine Batch, matching `in_cols` by name with strict |
| 103 | +/// dtypes (int64 / double / string|large_string / bool). |
| 104 | +pub fn ingest(py: Python<'_>, batch: &Bound<'_, PyAny>, in_cols: &[Col]) -> PyResult<Batch> { |
| 105 | + let _ = py; |
| 106 | + if in_cols.iter().any(|c| c.name.contains('.')) { |
| 107 | + return Err(err( |
| 108 | + "infer_arrow requires an all-scalar row model (struct columns: use infer())", |
| 109 | + )); |
| 110 | + } |
| 111 | + let rows: usize = batch.call_method0("__len__")?.extract()?; |
| 112 | + let mut cols = Vec::with_capacity(in_cols.len()); |
| 113 | + for c in in_cols { |
| 114 | + let arr = column(batch, &c.name)?; |
| 115 | + let dtype: String = arr.getattr("type")?.str()?.extract()?; |
| 116 | + let ok = matches!( |
| 117 | + (c.ty.ty, dtype.as_str()), |
| 118 | + (Ty::I64, "int64") |
| 119 | + | (Ty::F64, "double") |
| 120 | + | (Ty::I1, "bool") |
| 121 | + | (Ty::Str, "string") |
| 122 | + | (Ty::Str, "large_string") |
| 123 | + ); |
| 124 | + if !ok { |
| 125 | + return Err(err(format!( |
| 126 | + "infer_arrow: column '{}' is {dtype}, the model wants {} — cast first", |
| 127 | + c.name, |
| 128 | + c.ty.ty.name() |
| 129 | + ))); |
| 130 | + } |
| 131 | + let large = dtype == "large_string"; |
| 132 | + let raw = raw_array(arr)?; |
| 133 | + if raw.len != rows { |
| 134 | + return Err(err(format!( |
| 135 | + "infer_arrow: column '{}' has {} rows, the batch {rows}", |
| 136 | + c.name, raw.len |
| 137 | + ))); |
| 138 | + } |
| 139 | + let mut null_seen = false; |
| 140 | + let col_data = match c.ty.ty { |
| 141 | + Ty::I64 => { |
| 142 | + let mut valid = Vec::with_capacity(rows); |
| 143 | + let mut data = Vec::with_capacity(rows); |
| 144 | + let p = unsafe { raw.data::<i64>(1) }; |
| 145 | + for i in 0..rows { |
| 146 | + let v = raw.valid(i); |
| 147 | + null_seen |= !v; |
| 148 | + valid.push(v); |
| 149 | + data.push(if v { unsafe { *p.add(i) } } else { 0 }); |
| 150 | + } |
| 151 | + ColData::I64 { valid, data } |
| 152 | + } |
| 153 | + Ty::F64 => { |
| 154 | + let mut valid = Vec::with_capacity(rows); |
| 155 | + let mut data = Vec::with_capacity(rows); |
| 156 | + let p = unsafe { raw.data::<f64>(1) }; |
| 157 | + for i in 0..rows { |
| 158 | + let v = raw.valid(i); |
| 159 | + null_seen |= !v; |
| 160 | + valid.push(v); |
| 161 | + data.push(if v { unsafe { *p.add(i) } } else { 0.0 }); |
| 162 | + } |
| 163 | + ColData::F64 { valid, data } |
| 164 | + } |
| 165 | + Ty::I1 => { |
| 166 | + let mut valid = Vec::with_capacity(rows); |
| 167 | + let mut data = Vec::with_capacity(rows); |
| 168 | + let (addr, _) = raw.bufs[1].expect("bool data buffer"); |
| 169 | + for i in 0..rows { |
| 170 | + let v = raw.valid(i); |
| 171 | + null_seen |= !v; |
| 172 | + valid.push(v); |
| 173 | + let bit = raw.offset + i; |
| 174 | + let b = unsafe { *(addr as *const u8).add(bit / 8) }; |
| 175 | + data.push(v && (b >> (bit % 8)) & 1 == 1); |
| 176 | + } |
| 177 | + ColData::I1 { valid, data } |
| 178 | + } |
| 179 | + Ty::Str => { |
| 180 | + let mut col = ColData::Str { |
| 181 | + valid: Vec::with_capacity(rows), |
| 182 | + buf: String::new(), |
| 183 | + spans: Vec::with_capacity(rows), |
| 184 | + }; |
| 185 | + let (daddr, dsize) = raw.bufs[2].unwrap_or((0, 0)); |
| 186 | + let bytes = |
| 187 | + unsafe { std::slice::from_raw_parts(daddr as *const u8, dsize) }; |
| 188 | + for i in 0..rows { |
| 189 | + let v = raw.valid(i); |
| 190 | + null_seen |= !v; |
| 191 | + if !v { |
| 192 | + col.push_str_cell(false, ""); |
| 193 | + continue; |
| 194 | + } |
| 195 | + let (lo, hi) = if large { |
| 196 | + let p = unsafe { raw.data::<i64>(1) }; |
| 197 | + unsafe { (*p.add(i) as usize, *p.add(i + 1) as usize) } |
| 198 | + } else { |
| 199 | + let p = unsafe { raw.data::<i32>(1) }; |
| 200 | + unsafe { (*p.add(i) as usize, *p.add(i + 1) as usize) } |
| 201 | + }; |
| 202 | + let s = std::str::from_utf8(&bytes[lo..hi]) |
| 203 | + .map_err(|_| err("infer_arrow: invalid UTF-8 in string column"))?; |
| 204 | + col.push_str_cell(true, s); |
| 205 | + } |
| 206 | + col |
| 207 | + } |
| 208 | + }; |
| 209 | + if null_seen && !c.ty.nullable { |
| 210 | + return Err(err(format!( |
| 211 | + "column '{}' is not nullable but the batch has NULLs", |
| 212 | + c.name |
| 213 | + ))); |
| 214 | + } |
| 215 | + cols.push(col_data); |
| 216 | + } |
| 217 | + Ok(Batch { rows, cols }) |
| 218 | +} |
| 219 | + |
| 220 | +fn bitmap(oks: impl Iterator<Item = bool>, n: usize) -> (Vec<u8>, usize) { |
| 221 | + let mut bits = vec![0u8; n.div_ceil(8)]; |
| 222 | + let mut nulls = 0usize; |
| 223 | + for (i, ok) in oks.enumerate() { |
| 224 | + if ok { |
| 225 | + bits[i / 8] |= 1 << (i % 8); |
| 226 | + } else { |
| 227 | + nulls += 1; |
| 228 | + } |
| 229 | + } |
| 230 | + (bits, nulls) |
| 231 | +} |
| 232 | + |
| 233 | +/// Engine output -> pa.Table, one `Array.from_buffers` per column. |
| 234 | +pub fn emit(py: Python<'_>, out_cols: &[Col], st: &RunState) -> PyResult<Py<PyAny>> { |
| 235 | + let pa = PyModule::import(py, "pyarrow")?; |
| 236 | + let from_buffers = pa.getattr("Array")?.getattr("from_buffers")?; |
| 237 | + let py_buffer = pa.getattr("py_buffer")?; |
| 238 | + let n = st.emitted; |
| 239 | + let mut arrays = Vec::with_capacity(out_cols.len()); |
| 240 | + let mut names = Vec::with_capacity(out_cols.len()); |
| 241 | + for (c, oc) in out_cols.iter().zip(&st.out) { |
| 242 | + names.push(c.name.clone()); |
| 243 | + let (dtype, validity, bufs): (Bound<'_, PyAny>, (Vec<u8>, usize), Vec<Py<PyAny>>) = |
| 244 | + match oc { |
| 245 | + OutCol::I64(v) => { |
| 246 | + let vb = bitmap(v.iter().map(|(ok, _)| *ok), n); |
| 247 | + let data: Vec<i64> = v.iter().map(|(_, x)| *x).collect(); |
| 248 | + let raw = unsafe { |
| 249 | + std::slice::from_raw_parts(data.as_ptr() as *const u8, data.len() * 8) |
| 250 | + }; |
| 251 | + ( |
| 252 | + pa.call_method0("int64")?, |
| 253 | + vb, |
| 254 | + vec![py_buffer.call1((PyBytes::new(py, raw),))?.unbind()], |
| 255 | + ) |
| 256 | + } |
| 257 | + OutCol::F64(v) => { |
| 258 | + let vb = bitmap(v.iter().map(|(ok, _)| *ok), n); |
| 259 | + let data: Vec<f64> = v.iter().map(|(_, x)| *x).collect(); |
| 260 | + let raw = unsafe { |
| 261 | + std::slice::from_raw_parts(data.as_ptr() as *const u8, data.len() * 8) |
| 262 | + }; |
| 263 | + ( |
| 264 | + pa.call_method0("float64")?, |
| 265 | + vb, |
| 266 | + vec![py_buffer.call1((PyBytes::new(py, raw),))?.unbind()], |
| 267 | + ) |
| 268 | + } |
| 269 | + OutCol::I1(v) => { |
| 270 | + let vb = bitmap(v.iter().map(|(ok, _)| *ok), n); |
| 271 | + let (data_bits, _) = bitmap(v.iter().map(|(_, x)| *x), n); |
| 272 | + ( |
| 273 | + pa.call_method0("bool_")?, |
| 274 | + vb, |
| 275 | + vec![py_buffer |
| 276 | + .call1((PyBytes::new(py, &data_bits),))? |
| 277 | + .unbind()], |
| 278 | + ) |
| 279 | + } |
| 280 | + OutCol::Str(v) => { |
| 281 | + let vb = bitmap(v.iter().map(|(ok, _)| *ok), n); |
| 282 | + let mut offsets: Vec<i64> = Vec::with_capacity(n + 1); |
| 283 | + let mut bytes: Vec<u8> = Vec::new(); |
| 284 | + offsets.push(0); |
| 285 | + for (ok, r) in v.iter() { |
| 286 | + if *ok { |
| 287 | + bytes.extend_from_slice(st.arena.get(*r).as_bytes()); |
| 288 | + } |
| 289 | + offsets.push(bytes.len() as i64); |
| 290 | + } |
| 291 | + let off_raw = unsafe { |
| 292 | + std::slice::from_raw_parts( |
| 293 | + offsets.as_ptr() as *const u8, |
| 294 | + offsets.len() * 8, |
| 295 | + ) |
| 296 | + }; |
| 297 | + ( |
| 298 | + pa.call_method0("large_string")?, |
| 299 | + vb, |
| 300 | + vec![ |
| 301 | + py_buffer.call1((PyBytes::new(py, off_raw),))?.unbind(), |
| 302 | + py_buffer.call1((PyBytes::new(py, &bytes),))?.unbind(), |
| 303 | + ], |
| 304 | + ) |
| 305 | + } |
| 306 | + }; |
| 307 | + let (vbits, nulls) = validity; |
| 308 | + let vbuf: Py<PyAny> = if nulls == 0 { |
| 309 | + py.None() |
| 310 | + } else { |
| 311 | + py_buffer.call1((PyBytes::new(py, &vbits),))?.unbind() |
| 312 | + }; |
| 313 | + let buf_list = PyList::new(py, std::iter::once(vbuf).chain(bufs))?; |
| 314 | + let arr = from_buffers.call((dtype, n, buf_list, nulls), None)?; |
| 315 | + arrays.push(arr); |
| 316 | + } |
| 317 | + let table = pa |
| 318 | + .getattr("Table")? |
| 319 | + .call_method1("from_arrays", (arrays, names))?; |
| 320 | + Ok(table.unbind()) |
| 321 | +} |
0 commit comments