From 45fdb6d3097df04a4776f24402d2bf5501dc0d42 Mon Sep 17 00:00:00 2001 From: Brandon Schneider Date: Mon, 11 May 2026 23:48:42 -0500 Subject: [PATCH] Quiet parquet compressor warnings --- .../parquet_compressor/src/compressor.rs | 29 ++++--- 5-Applications/parquet_compressor/src/gpu.rs | 67 ++++++++++----- 5-Applications/parquet_compressor/src/main.rs | 85 ++++++++++++------- .../src/math_self_discover.rs | 71 ++++++++++------ .../parquet_compressor/src/parquet_handler.rs | 8 +- .../parquet_compressor/src/shifters/bwt.rs | 37 +++++--- .../src/shifters/delta_gcl.rs | 62 +++++++++----- .../src/shifters/huffman.rs | 12 ++- .../parquet_compressor/src/shifters/lzw.rs | 36 +++++--- .../parquet_compressor/src/shifters/mod.rs | 19 +++-- .../parquet_compressor/src/shifters/mtf.rs | 16 ++-- .../parquet_compressor/src/shifters/pist.rs | 20 +++-- .../parquet_compressor/src/shifters/rle.rs | 24 ++++-- 13 files changed, 314 insertions(+), 172 deletions(-) diff --git a/5-Applications/parquet_compressor/src/compressor.rs b/5-Applications/parquet_compressor/src/compressor.rs index 78dd4fc8..fac81094 100644 --- a/5-Applications/parquet_compressor/src/compressor.rs +++ b/5-Applications/parquet_compressor/src/compressor.rs @@ -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, shifters: Vec>) -> anyhow::Result> { 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> { 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))?; diff --git a/5-Applications/parquet_compressor/src/gpu.rs b/5-Applications/parquet_compressor/src/gpu.rs index 5be65a18..aa6e2761 100644 --- a/5-Applications/parquet_compressor/src/gpu.rs +++ b/5-Applications/parquet_compressor/src/gpu.rs @@ -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 { 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> { @@ -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); diff --git a/5-Applications/parquet_compressor/src/main.rs b/5-Applications/parquet_compressor/src/main.rs index 10b02802..dca7a7e3 100644 --- a/5-Applications/parquet_compressor/src/main.rs +++ b/5-Applications/parquet_compressor/src/main.rs @@ -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>, + file_path: &Path, + output_dir: &Path, + stats: &Arc>, gpu_context: Option>, - 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> = chain.iter() - .map(|s| get_shifter(s).unwrap()) - .collect(); - + let shifters: Vec> = + 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, })); diff --git a/5-Applications/parquet_compressor/src/math_self_discover.rs b/5-Applications/parquet_compressor/src/math_self_discover.rs index 89864bfe..f233c35f 100644 --- a/5-Applications/parquet_compressor/src/math_self_discover.rs +++ b/5-Applications/parquet_compressor/src/math_self_discover.rs @@ -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::() / sys.cpus().len() as f32; + let cpu_usage: f32 = + sys.cpus().iter().map(|c| c.cpu_usage()).sum::() / 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 { - 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::()); // Build equation strings for cache/SIMD surface - let eqs: Vec = (0..n) - .map(|i| eq_col.value(i).to_string()) - .collect(); + let eqs: Vec = (0..n).map(|i| eq_col.value(i).to_string()).collect(); // Fingerprint via cache-first SIMD batch surface let fingerprints: Vec = 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()); diff --git a/5-Applications/parquet_compressor/src/parquet_handler.rs b/5-Applications/parquet_compressor/src/parquet_handler.rs index bdbf65ff..ec8fd73d 100644 --- a/5-Applications/parquet_compressor/src/parquet_handler.rs +++ b/5-Applications/parquet_compressor/src/parquet_handler.rs @@ -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> { 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 diff --git a/5-Applications/parquet_compressor/src/shifters/bwt.rs b/5-Applications/parquet_compressor/src/shifters/bwt.rs index 3b406bca..59a07a0a 100644 --- a/5-Applications/parquet_compressor/src/shifters/bwt.rs +++ b/5-Applications/parquet_compressor/src/shifters/bwt.rs @@ -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]; diff --git a/5-Applications/parquet_compressor/src/shifters/delta_gcl.rs b/5-Applications/parquet_compressor/src/shifters/delta_gcl.rs index 58a74dca..3d7a1b32 100644 --- a/5-Applications/parquet_compressor/src/shifters/delta_gcl.rs +++ b/5-Applications/parquet_compressor/src/shifters/delta_gcl.rs @@ -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')` diff --git a/5-Applications/parquet_compressor/src/shifters/huffman.rs b/5-Applications/parquet_compressor/src/shifters/huffman.rs index 93c175b2..3da2764a 100644 --- a/5-Applications/parquet_compressor/src/shifters/huffman.rs +++ b/5-Applications/parquet_compressor/src/shifters/huffman.rs @@ -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(()) } diff --git a/5-Applications/parquet_compressor/src/shifters/lzw.rs b/5-Applications/parquet_compressor/src/shifters/lzw.rs index 03de9eb9..003e7f23 100644 --- a/5-Applications/parquet_compressor/src/shifters/lzw.rs +++ b/5-Applications/parquet_compressor/src/shifters/lzw.rs @@ -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, u16> = (0..=255).map(|b| (vec![b], b as u16)).collect(); + let mut dictionary: HashMap, 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> = (0..=255).map(|i| (i as u16, vec![i as u8])).collect(); + let mut dictionary: HashMap> = + (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]); diff --git a/5-Applications/parquet_compressor/src/shifters/mod.rs b/5-Applications/parquet_compressor/src/shifters/mod.rs index 283dcb25..2614167b 100644 --- a/5-Applications/parquet_compressor/src/shifters/mod.rs +++ b/5-Applications/parquet_compressor/src/shifters/mod.rs @@ -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> { match name { diff --git a/5-Applications/parquet_compressor/src/shifters/mtf.rs b/5-Applications/parquet_compressor/src/shifters/mtf.rs index 8eee6c82..b008576e 100644 --- a/5-Applications/parquet_compressor/src/shifters/mtf.rs +++ b/5-Applications/parquet_compressor/src/shifters/mtf.rs @@ -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 = (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 = (0..=255).collect(); let mut result = Vec::with_capacity(data.len()); - + for &idx in data { let b = alphabet[idx as usize]; result.push(b); diff --git a/5-Applications/parquet_compressor/src/shifters/pist.rs b/5-Applications/parquet_compressor/src/shifters/pist.rs index 49297579..3aa42f46 100644 --- a/5-Applications/parquet_compressor/src/shifters/pist.rs +++ b/5-Applications/parquet_compressor/src/shifters/pist.rs @@ -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); } diff --git a/5-Applications/parquet_compressor/src/shifters/rle.rs b/5-Applications/parquet_compressor/src/shifters/rle.rs index 014aaf18..49dbf149 100644 --- a/5-Applications/parquet_compressor/src/shifters/rle.rs +++ b/5-Applications/parquet_compressor/src/shifters/rle.rs @@ -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); }