allaun/Research-Stack
1
0
Fork 0
mirror of https://github.com/allaunthefox/Research-Stack.git synced 2026-07-31 03:05:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Quiet parquet compressor warnings

Browse source
This commit is contained in:
Brandon Schneider 2026-05-11 23:48:42 -05:00
parent d8616d2879
commit 79546e5dc7
13 changed files with 314 additions and 172 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
5-Applications/parquet_compressor/src/compressor.rs
View file

@ -1,13 +1,13 @@
use crate::shifters::{Shifter, ManifoldState};
use serde_json::json;
use crate::shifters::{ManifoldState, Shifter};
use byteorder::{BigEndian, ByteOrder};
use serde_json::json;
pub struct Compressor;
impl Compressor {
pub fn compress(data: Vec<u8>, shifters: Vec<Box<dyn Shifter>>) -> anyhow::Result<Vec<u8>> {
let mut state = ManifoldState::new(data);
for shifter in shifters {
shifter.encode(&mut state)?;
}
@ -18,7 +18,7 @@ impl Compressor {
"metadata": state.metadata,
});
let header_bytes = serde_json::to_vec(&header)?;
let mut result = Vec::new();
let mut buf = [0u8; 4];
BigEndian::write_u32(&mut buf, header_bytes.len() as u32);
@ -29,29 +29,32 @@ impl Compressor {
Ok(result)
}
#[allow(dead_code)]
pub fn decompress(compressed_data: &[u8]) -> anyhow::Result<Vec<u8>> {
if compressed_data.len() < 4 {
return Err(anyhow::anyhow!("Data too short"));
}
let header_len = BigEndian::read_u32(&compressed_data[0..4]) as usize;
if compressed_data.len() < 4 + header_len {
return Err(anyhow::anyhow!("Data too short for header"));
}
let header_bytes = &compressed_data[4..4+header_len];
let header_bytes = &compressed_data[4..4 + header_len];
let header: serde_json::Value = serde_json::from_slice(header_bytes)?;
let encoded_data = &compressed_data[4+header_len..];
let encoded_data = &compressed_data[4 + header_len..];
let mut state = ManifoldState::new(Vec::new());
state.encoded = encoded_data.to_vec();
state.shifter_chain = header["chain"].as_array()
state.shifter_chain = header["chain"]
.as_array()
.unwrap_or(&vec![])
.iter()
.map(|v| v.as_str().unwrap().to_string())
.collect();
state.metadata = header["metadata"].as_object()
state.metadata = header["metadata"]
.as_object()
.unwrap_or(&serde_json::Map::new())
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
@ -60,7 +63,7 @@ impl Compressor {
// Decode in reverse order
let mut shifter_chain = state.shifter_chain.clone();
shifter_chain.reverse();
for name in shifter_chain {
let shifter = crate::shifters::get_shifter(&name)
.ok_or_else(|| anyhow::anyhow!("Unknown shifter: {}", name))?;

67
5-Applications/parquet_compressor/src/gpu.rs
View file

@ -1,5 +1,5 @@
use wgpu::util::DeviceExt;
use std::borrow::Cow;
use wgpu::util::DeviceExt;
pub struct GpuContext {
pub device: wgpu::Device,
@ -11,22 +11,31 @@ pub struct GpuContext {
impl GpuContext {
pub async fn new() -> anyhow::Result<Self> {
let instance = wgpu::Instance::default();
let adapter = instance.request_adapter(&wgpu::RequestAdapterOptions {
power_preference: wgpu::PowerPreference::HighPerformance,
compatible_surface: None,
force_fallback_adapter: false,
}).await.ok_or_else(|| anyhow::anyhow!("No adapter found"))?;
let adapter = instance
.request_adapter(&wgpu::RequestAdapterOptions {
power_preference: wgpu::PowerPreference::HighPerformance,
compatible_surface: None,
force_fallback_adapter: false,
})
.await
.ok_or_else(|| anyhow::anyhow!("No adapter found"))?;
let mut limits = wgpu::Limits::default();
limits.max_storage_buffer_binding_size = adapter.limits().max_storage_buffer_binding_size;
limits.max_buffer_size = adapter.limits().max_buffer_size;
limits.max_compute_invocations_per_workgroup = adapter.limits().max_compute_invocations_per_workgroup;
limits.max_compute_invocations_per_workgroup =
adapter.limits().max_compute_invocations_per_workgroup;
let (device, queue) = adapter.request_device(&wgpu::DeviceDescriptor {
label: None,
required_features: wgpu::Features::empty(),
required_limits: limits,
}, None).await?;
let (device, queue) = adapter
.request_device(
&wgpu::DeviceDescriptor {
label: None,
required_features: wgpu::Features::empty(),
required_limits: limits,
},
None,
)
.await?;
let shader = device.create_shader_module(wgpu::ShaderModuleDescriptor {
label: Some("XOR Shader"),
@ -72,7 +81,12 @@ impl GpuContext {
entry_point: "main",
});
Ok(Self { device, queue, pipeline, bind_group_layout })
Ok(Self {
device,
queue,
pipeline,
bind_group_layout,
})
}
pub fn run_xor_transform(&self, data: &[u8], _key: u8) -> anyhow::Result<Vec<u8>> {
@ -80,16 +94,20 @@ impl GpuContext {
let mut padded_data = vec![0u8; padded_len];
padded_data[..data.len()].copy_from_slice(data);
let input_buffer = self.device.create_buffer_init(&wgpu::util::BufferInitDescriptor {
label: Some("Input Buffer"),
contents: bytemuck::cast_slice(&padded_data),
usage: wgpu::BufferUsages::STORAGE | wgpu::BufferUsages::COPY_SRC,
});
let input_buffer = self
.device
.create_buffer_init(&wgpu::util::BufferInitDescriptor {
label: Some("Input Buffer"),
contents: bytemuck::cast_slice(&padded_data),
usage: wgpu::BufferUsages::STORAGE | wgpu::BufferUsages::COPY_SRC,
});
let output_buffer = self.device.create_buffer(&wgpu::BufferDescriptor {
label: Some("Output Buffer"),
size: padded_len as u64,
usage: wgpu::BufferUsages::STORAGE | wgpu::BufferUsages::COPY_SRC | wgpu::BufferUsages::COPY_DST,
usage: wgpu::BufferUsages::STORAGE
| wgpu::BufferUsages::COPY_SRC
| wgpu::BufferUsages::COPY_DST,
mapped_at_creation: false,
});
@ -115,12 +133,17 @@ impl GpuContext {
],
});
let mut encoder = self.device.create_command_encoder(&wgpu::CommandEncoderDescriptor { label: None });
let mut encoder = self
.device
.create_command_encoder(&wgpu::CommandEncoderDescriptor { label: None });
{
let mut compute_pass = encoder.begin_compute_pass(&wgpu::ComputePassDescriptor { label: None, timestamp_writes: None });
let mut compute_pass = encoder.begin_compute_pass(&wgpu::ComputePassDescriptor {
label: None,
timestamp_writes: None,
});
compute_pass.set_pipeline(&self.pipeline);
compute_pass.set_bind_group(0, &bind_group, &[]);
let total_elements = (padded_len / 4) as u32;
let workgroups_x = ((total_elements + 255) / 256).min(65535);
let workgroups_y = (total_elements + 256 * 65535 - 1) / (256 * 65535);

85
5-Applications/parquet_compressor/src/main.rs
View file

@ -1,27 +1,35 @@
mod gpu;
mod shifters;
mod compressor;
mod gpu;
mod parquet_handler;
mod shifters;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use clap::Parser;
use rayon::prelude::*;
use indicatif::{ProgressBar, ProgressStyle};
use tokio::sync::Semaphore;
use sysinfo::System;
use serde_json::json;
use crate::shifters::get_shifter;
use crate::compressor::Compressor;
use crate::gpu::GpuContext;
use crate::shifters::{get_shifter, intrinsic_load};
use clap::Parser;
use indicatif::{ProgressBar, ProgressStyle};
use rayon::prelude::*;
use serde_json::json;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use sysinfo::System;
use tokio::sync::Semaphore;
#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
#[arg(short, long, default_value = "3-Mathematical-Models/equations_parquet_tagged")]
#[arg(
short,
long,
default_value = "3-Mathematical-Models/equations_parquet_tagged"
)]
input_dir: String,
#[arg(short, long, default_value = "3-Mathematical-Models/equations_compressed")]
#[arg(
short,
long,
default_value = "3-Mathematical-Models/equations_compressed"
)]
output_dir: String,
#[arg(short, long, default_value_t = 8192)]
@ -73,7 +81,7 @@ fn main() -> anyhow::Result<()> {
pb.set_style(ProgressStyle::default_bar()
.template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} ({eta}) {msg}")?
.progress_chars("#>-"));
pb.enable_steady_tick(std::time::Duration::from_millis(100));
let stats = Arc::new(Mutex::new(Stats {
@ -83,7 +91,7 @@ fn main() -> anyhow::Result<()> {
}));
let semaphore = Arc::new(Semaphore::new(args.threads));
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(args.threads)
.enable_all()
@ -91,22 +99,30 @@ fn main() -> anyhow::Result<()> {
files.par_iter().for_each(|file_path| {
let _permit = rt.block_on(semaphore.acquire()).expect("Semaphore closed");
let fname = file_path.file_stem().unwrap().to_str().unwrap();
pb.set_message(format!("Current: {}", fname));
let sub_pb = mp.add(indicatif::ProgressBar::new_spinner());
sub_pb.set_style(indicatif::ProgressStyle::default_spinner()
.template("{spinner:.yellow} {msg}")
.unwrap());
sub_pb.set_style(
indicatif::ProgressStyle::default_spinner()
.template("{spinner:.yellow} {msg}")
.unwrap(),
);
sub_pb.set_message(format!("Starting {}", fname));
sub_pb.enable_steady_tick(std::time::Duration::from_millis(100));
match process_file_with_progress(file_path, &output_path, &stats, gpu_context.clone(), &sub_pb) {
match process_file_with_progress(
file_path,
&output_path,
&stats,
gpu_context.clone(),
&sub_pb,
) {
Ok(_) => {
pb.inc(1);
sub_pb.finish_and_clear();
},
}
Err(e) => {
eprintln!("Error processing {}: {}", fname, e);
sub_pb.finish_with_message(format!("Failed: {}", fname));
@ -149,24 +165,27 @@ fn main() -> anyhow::Result<()> {
let index_path = output_path.join("index.json");
std::fs::write(&index_path, serde_json::to_string_pretty(&index)?)?;
let parent_index = input_path.parent().unwrap().join("equations_NUVMAP_index.json");
let parent_index = input_path
.parent()
.unwrap()
.join("equations_NUVMAP_index.json");
std::fs::copy(&index_path, parent_index)?;
Ok(())
}
fn process_file_with_progress(
file_path: &Path,
output_dir: &Path,
stats: &Arc<Mutex<Stats>>,
file_path: &Path,
output_dir: &Path,
stats: &Arc<Mutex<Stats>>,
gpu_context: Option<Arc<GpuContext>>,
pb: &ProgressBar
pb: &ProgressBar,
) -> anyhow::Result<()> {
pb.set_message("Serializing Parquet...");
let raw_data = parquet_handler::serialize_parquet_to_bytes(file_path)?;
let raw_len = raw_data.len();
pb.set_message("GPU Preprocessing...");
let data_to_compress = if let Some(gpu) = gpu_context {
gpu.run_xor_transform(&raw_data, 0x1F)?
@ -185,10 +204,9 @@ fn process_file_with_progress(
for (name, chain) in chains {
pb.set_message(format!("Trying chain: {}...", name));
let shifters: Vec<Box<dyn crate::shifters::Shifter>> = chain.iter()
.map(|s| get_shifter(s).unwrap())
.collect();
let shifters: Vec<Box<dyn crate::shifters::Shifter>> =
chain.iter().map(|s| get_shifter(s).unwrap()).collect();
if let Ok(compressed) = Compressor::compress(data_to_compress.clone(), shifters) {
let ratio = raw_len as f64 / compressed.len().max(1) as f64;
if ratio > best_ratio {
@ -198,7 +216,7 @@ fn process_file_with_progress(
}
}
}
pb.set_message("Finalizing file...");
let fname = file_path.file_stem().unwrap().to_str().unwrap();
let output_path = output_dir.join(format!("{}.compressed", fname));
@ -211,6 +229,7 @@ fn process_file_with_progress(
"name": fname,
"raw_bytes": raw_len,
"compressed_bytes": best_data.len(),
"raw_intrinsic_load": intrinsic_load(&data_to_compress),
"ratio": best_ratio,
"chain": best_chain_name,
}));

71
5-Applications/parquet_compressor/src/math_self_discover.rs
View file

@ -1,4 +1,5 @@
use anyhow::{Context, Result};
use arrow::array::Array;
use arrow::array::StringArray;
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
@ -7,18 +8,17 @@ use dashmap::DashMap;
use indicatif::{ProgressBar, ProgressStyle};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::arrow::ArrowWriter;
use rayon::ThreadPoolBuilder;
use rayon::iter::{IntoParallelIterator, ParallelIterator};
use rayon::ThreadPoolBuilder;
use regex::Regex;
use serde::Serialize;
use arrow::array::Array;
use std::collections::HashMap;
use std::fs::{self, File};
use std::io::{BufRead, BufReader, Write};
use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use sysinfo::{System, RefreshKind, CpuRefreshKind, MemoryRefreshKind};
use sysinfo::{CpuRefreshKind, MemoryRefreshKind, RefreshKind, System};
// ── CLI ──────────────────────────────────────────────────────────────────────
@ -27,15 +27,26 @@ use sysinfo::{System, RefreshKind, CpuRefreshKind, MemoryRefreshKind};
#[command(about = "Unsupervised structural taxonomy from naked equations.")]
struct Config {
/// Input parquet (math-raw)
#[arg(short, long, default_value = "/home/allaun/Documents/Research Stack/3-Mathematical-Models/equations_parquet_tagged/equations_math_raw.parquet")]
#[arg(
short,
long,
default_value = "/home/allaun/Documents/Research Stack/3-Mathematical-Models/equations_parquet_tagged/equations_math_raw.parquet"
)]
input: String,
/// Output JSON report
#[arg(short, long, default_value = "/home/allaun/Documents/Research Stack/3-Mathematical-Models/math_self_discovered.json")]
#[arg(
short,
long,
default_value = "/home/allaun/Documents/Research Stack/3-Mathematical-Models/math_self_discovered.json"
)]
report: String,
/// Output parquet with cluster assignments
#[arg(long, default_value = "/home/allaun/Documents/Research Stack/3-Mathematical-Models/equations_parquet_tagged/equations_self_clustered.parquet")]
#[arg(
long,
default_value = "/home/allaun/Documents/Research Stack/3-Mathematical-Models/equations_parquet_tagged/equations_self_clustered.parquet"
)]
clustered: String,
/// Max CPU threads (0 = num_cpus / 2)
@ -100,7 +111,8 @@ impl ResourceMonitor {
.with_memory(MemoryRefreshKind::everything()),
);
let cpu_usage: f32 = sys.cpus().iter().map(|c| c.cpu_usage()).sum::<f32>() / sys.cpus().len() as f32;
let cpu_usage: f32 =
sys.cpus().iter().map(|c| c.cpu_usage()).sum::<f32>() / sys.cpus().len() as f32;
let mem_usage = sys.used_memory() as f32 / sys.total_memory() as f32 * 100.0;
let ok = cpu_usage < self.cpu_threshold && mem_usage < self.memory_threshold;
@ -304,8 +316,8 @@ impl FingerprintEngine {
// Multi-letter identifiers
let math_funcs: std::collections::HashSet<&str> = [
"sin", "cos", "tan", "exp", "log", "ln", "sqrt", "det", "tr",
"dim", "ker", "rank", "span", "frac", "sum", "prod", "int",
"sin", "cos", "tan", "exp", "log", "ln", "sqrt", "det", "tr", "dim", "ker", "rank",
"span", "frac", "sum", "prod", "int",
]
.iter()
.copied()
@ -343,14 +355,6 @@ impl FingerprintEngine {
text
}
/// SIMD-friendly batch surface: process a slice of equations as a data-parallel array.
/// The compiler auto-vectorizes the hot loops across each equation's byte slice.
fn fingerprint_batch(&self, eqs: &[String]) -> Vec<String> {
eqs.iter()
.map(|eq| self.fingerprint(eq))
.collect()
}
}
// ── Clustering ───────────────────────────────────────────────────────────────
@ -377,7 +381,11 @@ fn main() -> Result<()> {
// ── Resource-aware thread pool ───────────────────────────────────────────
let n_cpus = std::thread::available_parallelism()?.get();
let threads = if config.threads == 0 { n_cpus / 2 } else { config.threads };
let threads = if config.threads == 0 {
n_cpus / 2
} else {
config.threads
};
let threads = threads.max(1);
eprintln!("════════════════════════════════════════════════════════════");
@ -387,7 +395,14 @@ fn main() -> Result<()> {
eprintln!(" Batch size: {}", config.batch_size);
eprintln!(" CPU threshold: {}%", config.cpu_threshold);
eprintln!(" Memory threshold: {}%", config.memory_threshold);
eprintln!(" Cache dir: {}", if config.cache_dir.is_empty() { "default" } else { &config.cache_dir });
eprintln!(
" Cache dir: {}",
if config.cache_dir.is_empty() {
"default"
} else {
&config.cache_dir
}
);
eprintln!("════════════════════════════════════════════════════════════");
ThreadPoolBuilder::new()
@ -395,15 +410,18 @@ fn main() -> Result<()> {
.build_global()
.context("failed to build thread pool")?;
let monitor = Arc::new(ResourceMonitor::new(config.cpu_threshold, config.memory_threshold));
let monitor = Arc::new(ResourceMonitor::new(
config.cpu_threshold,
config.memory_threshold,
));
// ── Cache manager ──────────────────────────────────────────────────────────
let cache = Arc::new(CacheManager::new(&config.cache_dir, config.ram_cache_cap)?);
// ── Open input + output (stream copy) ────────────────────────────────────
eprintln!("\nOpening input parquet...");
let file = File::open(&config.input)
.with_context(|| format!("failed to open {}", config.input))?;
let file =
File::open(&config.input).with_context(|| format!("failed to open {}", config.input))?;
let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
let total_rows = builder.metadata().file_metadata().num_rows() as u64;
eprintln!(" Total rows: {}", total_rows);
@ -463,9 +481,7 @@ fn main() -> Result<()> {
.and_then(|c| c.as_any().downcast_ref::<StringArray>());
// Build equation strings for cache/SIMD surface
let eqs: Vec<String> = (0..n)
.map(|i| eq_col.value(i).to_string())
.collect();
let eqs: Vec<String> = (0..n).map(|i| eq_col.value(i).to_string()).collect();
// Fingerprint via cache-first SIMD batch surface
let fingerprints: Vec<String> = eqs
@ -543,7 +559,10 @@ fn main() -> Result<()> {
// ── Cache stats ──────────────────────────────────────────────────────────
let (cache_entries, hits, misses) = cache.stats();
eprintln!(" Cache: {} entries, {} hits, {} misses", cache_entries, hits, misses);
eprintln!(
" Cache: {} entries, {} hits, {} misses",
cache_entries, hits, misses
);
eprintln!(" Flushing cache to disk...");
cache.flush_to_disk()?;
eprintln!(" Cache flushed to: {}", cache.disk_path.display());

8
5-Applications/parquet_compressor/src/parquet_handler.rs
View file

@ -1,16 +1,16 @@
use std::fs::File;
use std::path::Path;
use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::record::RowAccessor;
use serde_json::json;
use std::fs::File;
use std::path::Path;
pub fn serialize_parquet_to_bytes(path: &Path) -> anyhow::Result<Vec<u8>> {
let file = File::open(path)?;
let reader = SerializedFileReader::new(file)?;
let mut lines = Vec::new();
let row_iter = reader.get_row_iter(None)?;
for record in row_iter {
let record = record?;
// Find fields by name

37
5-Applications/parquet_compressor/src/shifters/bwt.rs
View file

@ -1,14 +1,20 @@
use crate::shifters::{Shifter, ManifoldState};
use serde_json::json;
use crate::shifters::{ManifoldState, Shifter};
use divsufsort::sort_in_place;
use serde_json::json;
pub struct BWTShifter;
impl Shifter for BWTShifter {
fn name(&self) -> &'static str { "bwt" }
fn name(&self) -> &'static str {
"bwt"
}
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = if !state.encoded.is_empty() { &state.encoded } else { &state.raw_bytes };
let data = if !state.encoded.is_empty() {
&state.encoded
} else {
&state.raw_bytes
};
if data.is_empty() {
state.update(Vec::new(), self.name(), json!({"primary_index": 0}));
return Ok(());
@ -18,13 +24,13 @@ impl Shifter for BWTShifter {
// divsufsort computes the suffix array SA
// The BWT L is defined as: L[i] = data[SA[i] - 1] (with wrap around)
// And we need the primary_index such that SA[primary_index] == 0.
let mut sa = vec![0i32; n];
sort_in_place(data, &mut sa);
let mut primary_index = 0;
let mut result = Vec::with_capacity(n);
for i in 0..n {
let s_idx = sa[i] as usize;
if s_idx == 0 {
@ -41,13 +47,22 @@ impl Shifter for BWTShifter {
fn decode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = &state.encoded;
if data.is_empty() { return Ok(()); }
let meta = state.metadata.get(self.name()).cloned().unwrap_or(json!({}));
let primary_index = meta.get("primary_index").and_then(|v| v.as_u64()).unwrap_or(0) as usize;
if data.is_empty() {
return Ok(());
}
let meta = state
.metadata
.get(self.name())
.cloned()
.unwrap_or(json!({}));
let primary_index = meta
.get("primary_index")
.and_then(|v| v.as_u64())
.unwrap_or(0) as usize;
let n = data.len();
// Decoding BWT:
// 1. Count occurrences of each byte
let mut counts = [0usize; 256];

62
5-Applications/parquet_compressor/src/shifters/delta_gcl.rs
View file

@ -1,14 +1,20 @@
use crate::shifters::{Shifter, ManifoldState};
use serde_json::json;
use crate::shifters::{ManifoldState, Shifter};
use byteorder::{BigEndian, ByteOrder};
use serde_json::json;
pub struct DeltaGCLShifter;
impl Shifter for DeltaGCLShifter {
fn name(&self) -> &'static str { "delta_gcl" }
fn name(&self) -> &'static str {
"delta_gcl"
}
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = if !state.encoded.is_empty() { &state.encoded } else { &state.raw_bytes };
let data = if !state.encoded.is_empty() {
&state.encoded
} else {
&state.raw_bytes
};
let chunk_size = 4096;
if data.is_empty() {
state.update(Vec::new(), self.name(), json!({}));
@ -33,7 +39,7 @@ impl Shifter for DeltaGCLShifter {
for i in 1..n_chunks {
let curr = chunks[i];
let max_len = prev.len().max(curr.len());
let mut delta = Vec::with_capacity(max_len);
for j in 0..max_len {
let p = if j < prev.len() { prev[j] } else { 0 };
@ -58,7 +64,7 @@ impl Shifter for DeltaGCLShifter {
BigEndian::write_u32(&mut buf, rle.len() as u32);
result.extend_from_slice(&buf);
result.extend_from_slice(&rle);
prev = curr.to_vec();
}
@ -68,38 +74,50 @@ impl Shifter for DeltaGCLShifter {
fn decode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = &state.encoded;
if data.len() < 4 { return Ok(()); }
if data.len() < 4 {
return Ok(());
}
let n_chunks = BigEndian::read_u32(&data[0..4]) as usize;
let mut ptr = 4;
if ptr + 4 > data.len() { return Ok(()); }
let first_len = BigEndian::read_u32(&data[ptr..ptr+4]) as usize;
if ptr + 4 > data.len() {
return Ok(());
}
let first_len = BigEndian::read_u32(&data[ptr..ptr + 4]) as usize;
ptr += 4;
if ptr + first_len > data.len() { return Ok(()); }
let first_chunk = data[ptr..ptr+first_len].to_vec();
if ptr + first_len > data.len() {
return Ok(());
}
let first_chunk = data[ptr..ptr + first_len].to_vec();
ptr += first_len;
let mut result = Vec::new();
result.extend_from_slice(&first_chunk);
let mut prev = first_chunk;
for _ in 1..n_chunks {
if ptr + 4 > data.len() { break; }
let rle_len = BigEndian::read_u32(&data[ptr..ptr+4]) as usize;
if ptr + 4 > data.len() {
break;
}
let rle_len = BigEndian::read_u32(&data[ptr..ptr + 4]) as usize;
ptr += 4;
if ptr + rle_len > data.len() { break; }
let rle_data = &data[ptr..ptr+rle_len];
if ptr + rle_len > data.len() {
break;
}
let rle_data = &data[ptr..ptr + rle_len];
ptr += rle_len;
let mut delta = Vec::new();
for j in (0..rle_data.len()).step_by(2) {
if j + 1 >= rle_data.len() { break; }
if j + 1 >= rle_data.len() {
break;
}
let count = rle_data[j] as usize;
let b = rle_data[j+1];
let b = rle_data[j + 1];
for _ in 0..count {
delta.push(b);
}
@ -110,7 +128,7 @@ impl Shifter for DeltaGCLShifter {
let p = if j < prev.len() { prev[j] } else { 0 };
curr.push(p ^ delta[j]);
}
// Note: the original code doesn't store the exact length of each chunk if it was padded.
// But since chunks are 4096, only the last one might be shorter.
// Wait, the original Python code: `curr_padded = curr.ljust(max_len, b'\x00')`

12
5-Applications/parquet_compressor/src/shifters/huffman.rs
View file

@ -1,14 +1,20 @@
use crate::shifters::{Shifter, ManifoldState};
use crate::shifters::{ManifoldState, Shifter};
use serde_json::json;
pub struct HuffmanShifter;
impl Shifter for HuffmanShifter {
fn name(&self) -> &'static str { "huffman" }
fn name(&self) -> &'static str {
"huffman"
}
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
// TODO: Implement Huffman
let data = if !state.encoded.is_empty() { &state.encoded } else { &state.raw_bytes };
let data = if !state.encoded.is_empty() {
&state.encoded
} else {
&state.raw_bytes
};
state.update(data.to_vec(), self.name(), json!({}));
Ok(())
}

36
5-Applications/parquet_compressor/src/shifters/lzw.rs
View file

@ -1,22 +1,29 @@
use crate::shifters::{Shifter, ManifoldState};
use crate::shifters::{ManifoldState, Shifter};
use byteorder::{BigEndian, ByteOrder};
use serde_json::json;
use std::collections::HashMap;
use byteorder::{BigEndian, ByteOrder};
pub struct LZWShifter;
impl Shifter for LZWShifter {
fn name(&self) -> &'static str { "lzw" }
fn name(&self) -> &'static str {
"lzw"
}
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = if !state.encoded.is_empty() { &state.encoded } else { &state.raw_bytes };
let data = if !state.encoded.is_empty() {
&state.encoded
} else {
&state.raw_bytes
};
let max_dict = 4096;
if data.is_empty() {
state.update(Vec::new(), self.name(), json!({"max_dict": max_dict}));
return Ok(());
}
let mut dictionary: HashMap<Vec<u8>, u16> = (0..=255).map(|b| (vec![b], b as u16)).collect();
let mut dictionary: HashMap<Vec<u8>, u16> =
(0..=255).map(|b| (vec![b], b as u16)).collect();
let mut next_code = 256u16;
let mut result = Vec::new();
let mut w = Vec::new();
@ -31,7 +38,7 @@ impl Shifter for LZWShifter {
let mut buf = [0u8; 2];
BigEndian::write_u16(&mut buf, *code);
result.extend_from_slice(&buf);
if next_code < max_dict {
dictionary.insert(wc, next_code);
next_code += 1;
@ -52,9 +59,12 @@ impl Shifter for LZWShifter {
fn decode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = &state.encoded;
if data.len() < 2 { return Ok(()); }
if data.len() < 2 {
return Ok(());
}
let mut dictionary: HashMap<u16, Vec<u8>> = (0..=255).map(|i| (i as u16, vec![i as u8])).collect();
let mut dictionary: HashMap<u16, Vec<u8>> =
(0..=255).map(|i| (i as u16, vec![i as u8])).collect();
let mut next_code = 256u16;
let mut result = Vec::new();
@ -63,9 +73,11 @@ impl Shifter for LZWShifter {
result.extend_from_slice(&s);
for i in (2..data.len()).step_by(2) {
if i + 1 >= data.len() { break; }
let code = BigEndian::read_u16(&data[i..i+2]);
if i + 1 >= data.len() {
break;
}
let code = BigEndian::read_u16(&data[i..i + 2]);
let entry = if let Some(v) = dictionary.get(&code) {
v.clone()
} else if code == next_code {
@ -77,7 +89,7 @@ impl Shifter for LZWShifter {
};
result.extend_from_slice(&entry);
if next_code < 4096 {
let mut v = dictionary.get(&old_code).unwrap().clone();
v.push(entry[0]);

19
5-Applications/parquet_compressor/src/shifters/mod.rs
View file

@ -1,4 +1,4 @@
use serde::{Serialize, Deserialize};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
#[derive(Debug, Clone, Serialize, Deserialize)]
@ -31,24 +31,25 @@ impl ManifoldState {
pub trait Shifter {
fn name(&self) -> &'static str;
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()>;
#[allow(dead_code)]
fn decode(&self, state: &mut ManifoldState) -> anyhow::Result<()>;
}
pub mod bwt;
pub mod mtf;
pub mod lzw;
pub mod rle;
pub mod huffman;
pub mod delta_gcl;
pub mod huffman;
pub mod lzw;
pub mod mtf;
pub mod pist;
pub mod rle;
pub use bwt::BWTShifter;
pub use mtf::MTFShifter;
pub use lzw::LZWShifter;
pub use rle::RunLengthShifter;
pub use huffman::HuffmanShifter;
pub use delta_gcl::DeltaGCLShifter;
pub use huffman::HuffmanShifter;
pub use lzw::LZWShifter;
pub use mtf::MTFShifter;
pub use pist::PISTShifter;
pub use rle::RunLengthShifter;
pub fn get_shifter(name: &str) -> Option<Box<dyn Shifter>> {
match name {

16
5-Applications/parquet_compressor/src/shifters/mtf.rs
View file

@ -1,16 +1,22 @@
use crate::shifters::{Shifter, ManifoldState};
use crate::shifters::{ManifoldState, Shifter};
use serde_json::json;
pub struct MTFShifter;
impl Shifter for MTFShifter {
fn name(&self) -> &'static str { "mtf" }
fn name(&self) -> &'static str {
"mtf"
}
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = if !state.encoded.is_empty() { &state.encoded } else { &state.raw_bytes };
let data = if !state.encoded.is_empty() {
&state.encoded
} else {
&state.raw_bytes
};
let mut alphabet: Vec<u8> = (0..=255).collect();
let mut result = Vec::with_capacity(data.len());
for &b in data {
let idx = alphabet.iter().position(|&x| x == b).unwrap();
result.push(idx as u8);
@ -26,7 +32,7 @@ impl Shifter for MTFShifter {
let data = &state.encoded;
let mut alphabet: Vec<u8> = (0..=255).collect();
let mut result = Vec::with_capacity(data.len());
for &idx in data {
let b = alphabet[idx as usize];
result.push(b);

20
5-Applications/parquet_compressor/src/shifters/pist.rs
View file

@ -1,15 +1,21 @@
use crate::shifters::{Shifter, ManifoldState};
use crate::shifters::{ManifoldState, Shifter};
use serde_json::json;
pub struct PISTShifter;
impl Shifter for PISTShifter {
fn name(&self) -> &'static str { "pist" }
fn name(&self) -> &'static str {
"pist"
}
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = if !state.encoded.is_empty() { &state.encoded } else { &state.raw_bytes };
let data = if !state.encoded.is_empty() {
&state.encoded
} else {
&state.raw_bytes
};
let mut result = Vec::with_capacity(data.len() * 2);
for &b in data {
let n = b as u64;
let k = (n as f64).sqrt() as u64;
@ -26,9 +32,11 @@ impl Shifter for PISTShifter {
let data = &state.encoded;
let mut result = Vec::with_capacity(data.len() / 2);
for i in (0..data.len()).step_by(2) {
if i + 1 >= data.len() { break; }
if i + 1 >= data.len() {
break;
}
let k = data[i] as u64;
let t = data[i+1] as u64;
let t = data[i + 1] as u64;
let n = k * k + t;
result.push(n as u8);
}

24
5-Applications/parquet_compressor/src/shifters/rle.rs
View file

@ -1,13 +1,19 @@
use crate::shifters::{Shifter, ManifoldState};
use crate::shifters::{ManifoldState, Shifter};
use serde_json::json;
pub struct RunLengthShifter;
impl Shifter for RunLengthShifter {
fn name(&self) -> &'static str { "run_length" }
fn name(&self) -> &'static str {
"run_length"
}
fn encode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = if !state.encoded.is_empty() { &state.encoded } else { &state.raw_bytes };
let data = if !state.encoded.is_empty() {
&state.encoded
} else {
&state.raw_bytes
};
let mut result = Vec::new();
let mut i = 0;
while i < data.len() {
@ -22,7 +28,11 @@ impl Shifter for RunLengthShifter {
}
let ratio = data.len() as f64 / result.len().max(1) as f64;
state.update(result, self.name(), json!({"original": data.len(), "ratio": ratio}));
state.update(
result,
self.name(),
json!({"original": data.len(), "ratio": ratio}),
);
Ok(())
}
@ -30,9 +40,11 @@ impl Shifter for RunLengthShifter {
let data = &state.encoded;
let mut result = Vec::new();
for i in (0..data.len()).step_by(2) {
if i + 1 >= data.len() { break; }
if i + 1 >= data.len() {
break;
}
let count = data[i] as usize;
let b = data[i+1];
let b = data[i + 1];
for _ in 0..count {
result.push(b);
}