ch01Why Parquet exists

# Two columns from a CSV file, read the way programs read CSV: a row at a time.
import csv, io, os
from collections import Counter

class Counted(io.FileIO):
    """A file that counts the bytes read from it."""
    bytes_read = 0
    def readinto(self, buffer):
        n = super().readinto(buffer)
        self.bytes_read += n
        return n

path = "fixtures/formats/orders.csv"
file = Counted(path)
totals = Counter()
for row in csv.DictReader(io.TextIOWrapper(io.BufferedReader(file), newline="")):
    # The query wants country and amount_cents; the other fields are read and dropped.
    totals[row["country"]] += int(row["amount_cents"])

print(f"read {file.bytes_read} of {os.path.getsize(path)} bytes")
print("amount_cents in the UK:", totals["UK"])

//! Two columns from a CSV file, read the way programs read CSV: a row at a time.

use std::collections::HashMap;
use std::fs::File;
use std::io::{BufRead, BufReader, Read};

/// A file that counts the bytes read from it.
struct Counted {
    file: File,
    bytes_read: u64,
}

impl Read for Counted {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let n = self.file.read(buf)?;
        self.bytes_read += n as u64;
        Ok(n)
    }
}

fn main() {
    let path = "fixtures/formats/orders.csv";
    let file = File::open(path).expect("run this from the repository's root");
    let size = file.metadata().unwrap().len();
    let mut lines = BufReader::new(Counted {
        file,
        bytes_read: 0,
    });

    let mut line = String::new();
    lines.read_line(&mut line).unwrap();
    let header: Vec<String> = line.trim_end().split(',').map(unquote).collect();
    let at = |name: &str| header.iter().position(|h| h == name).unwrap();
    let (country, amount) = (at("country"), at("amount_cents"));

    let mut totals: HashMap<String, i64> = HashMap::new();
    loop {
        line.clear();
        if lines.read_line(&mut line).unwrap() == 0 {
            break;
        }
        // The query wants country and amount_cents; every other field is read and dropped.
        let fields: Vec<String> = line.trim_end().split(',').map(unquote).collect();
        *totals.entry(fields[country].clone()).or_default() +=
            fields[amount].parse::<i64>().unwrap();
    }

    println!("read {} of {size} bytes", lines.get_ref().bytes_read);
    println!("amount_cents in the UK: {}", totals["UK"]);
}

/// A field without the quotes a CSV writer puts around text.
fn unquote(field: &str) -> String {
    field.trim_matches('"').to_string()
}

# The same two columns from the Parquet file of the same table, read with pyarrow.
import io, os
import pyarrow.compute as pc, pyarrow.parquet as pq

class Counted(io.FileIO):
    """A file that counts the bytes read from it, however the library asks for them."""
    bytes_read = 0
    def readinto(self, buffer):
        n = super().readinto(buffer)
        self.bytes_read += n
        return n
    def read(self, size=-1):
        data = super().read(size)
        self.bytes_read += len(data)
        return data

path = "fixtures/formats/orders.parquet"
file = Counted(path)
table = pq.read_table(file, columns=["country", "amount_cents"])
uk = table.filter(pc.equal(table["country"], "UK"))

print(f"read {file.bytes_read} of {os.path.getsize(path)} bytes")
print("amount_cents in the UK:", pc.sum(uk["amount_cents"]).as_py())

//! The same two columns from the Parquet file of the same table, read with the `parquet` crate.

use std::fs::File;
use std::io::Read;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use bytes::Bytes;
use parquet::file::reader::{ChunkReader, FileReader, Length, SerializedFileReader};
use parquet::record::RowAccessor;
use parquet::schema::parser::parse_message_type;

/// A file that counts the bytes read from it, however the crate asks for them.
struct Counted {
    file: File,
    bytes_read: Arc<AtomicU64>,
}

impl Length for Counted {
    fn len(&self) -> u64 {
        self.file.len()
    }
}

impl ChunkReader for Counted {
    type T = CountedRead<<File as ChunkReader>::T>;

    fn get_read(&self, start: u64) -> parquet::errors::Result<Self::T> {
        Ok(CountedRead(
            self.file.get_read(start)?,
            self.bytes_read.clone(),
        ))
    }

    fn get_bytes(&self, start: u64, length: usize) -> parquet::errors::Result<Bytes> {
        self.bytes_read.fetch_add(length as u64, Ordering::Relaxed);
        self.file.get_bytes(start, length)
    }
}

struct CountedRead<R>(R, Arc<AtomicU64>);

impl<R: Read> Read for CountedRead<R> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let n = self.0.read(buf)?;
        self.1.fetch_add(n as u64, Ordering::Relaxed);
        Ok(n)
    }
}

fn main() {
    let path = "fixtures/formats/orders.parquet";
    let file = File::open(path).expect("run this from the repository's root");
    let size = file.metadata().unwrap().len();
    let bytes_read = Arc::new(AtomicU64::new(0));
    let reader = SerializedFileReader::new(Counted {
        file,
        bytes_read: bytes_read.clone(),
    })
    .expect("a Parquet file");

    // Only these two columns, named as the file names them (pyarrow calls the root "schema"):
    // the crate reads their chunks and no others.
    let wanted = parse_message_type(
        "message schema { required binary country (STRING); required int64 amount_cents; }",
    )
    .unwrap();
    let mut uk = 0;
    for row in reader.get_row_iter(Some(wanted)).unwrap() {
        let row = row.unwrap();
        if row.get_string(0).unwrap() == "UK" {
            uk += row.get_long(1).unwrap();
        }
    }

    println!(
        "read {} of {size} bytes",
        bytes_read.load(Ordering::Relaxed)
    );
    println!("amount_cents in the UK: {uk}");
}

# One table, two layouts: the same values in a different order, and the ones a query needs.
import csv

orders = list(csv.reader(open("fixtures/formats/eight-orders.csv")))
header, rows = orders[0], orders[1:]
wanted = {"country", "amount_cents"}  # try {"order_id"}, or every column

by_rows = [(name, v) for row in rows for name, v in zip(header, row)]
by_columns = [(name, row[c]) for c, name in enumerate(header) for row in rows]

for label, layout in (("by rows", by_rows), ("by columns", by_columns)):
    picture = "".join("#" if name in wanted else "." for name, _ in layout)
    runs = len([p for p in picture.split(".") if p])
    print(f"{label}, {runs} run{'s' * (runs != 1)}:")
    print(picture)

//! One table, two layouts: the same values in a different order, and the ones a query needs.

fn main() {
    let text = std::fs::read_to_string("fixtures/formats/eight-orders.csv")
        .expect("run this from the repository's root");
    let mut lines = text.lines().map(|l| {
        l.split(',')
            .map(|v| v.trim_matches('"'))
            .collect::<Vec<_>>()
    });
    let header = lines.next().unwrap();
    let rows: Vec<Vec<&str>> = lines.collect();
    let wanted = ["country", "amount_cents"]; // try ["order_id"], or every column

    // Each value with its column's name: by rows, an order's values together; by columns, a
    // column's values together.
    let by_rows: Vec<(&str, &str)> = rows
        .iter()
        .flat_map(|row| header.iter().copied().zip(row.iter().copied()))
        .collect();
    let by_columns: Vec<(&str, &str)> = header
        .iter()
        .enumerate()
        .flat_map(|(c, &name)| rows.iter().map(move |row| (name, row[c])))
        .collect();

    for (label, layout) in [("by rows", by_rows), ("by columns", by_columns)] {
        let picture: String = layout
            .iter()
            .map(|(name, _)| if wanted.contains(name) { '#' } else { '.' })
            .collect();
        let runs = picture.split('.').filter(|p| !p.is_empty()).count();
        println!("{label}, {runs} run{}:", if runs == 1 { "" } else { "s" });
        println!("{picture}");
    }
}