allaun/Research-Stack
1
0
Fork 0
mirror of https://github.com/allaunthefox/Research-Stack.git synced 2026-08-17 20:10:36 +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 498d771d31
commit 2feba7decb
13 changed files with 314 additions and 172 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,6 +1,6 @@
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;
@ -29,6 +29,7 @@ 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"));
@ -39,19 +40,21 @@ impl Compressor {
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()))

65
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,9 +133,14 @@ 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, &[]);

65
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)]
@ -96,17 +104,25 @@ fn main() -> anyhow::Result<()> {
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));
@ -150,7 +166,10 @@ 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(())
@ -161,7 +180,7 @@ fn process_file_with_progress(
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)?;
@ -185,9 +204,8 @@ 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;
@ -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());

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

@ -1,8 +1,8 @@
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)?;

29
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(());
@ -41,10 +47,19 @@ impl Shifter for BWTShifter {
fn decode(&self, state: &mut ManifoldState) -> anyhow::Result<()> {
let data = &state.encoded;
if data.is_empty() { return Ok(()); }
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 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();

48
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!({}));
@ -68,17 +74,23 @@ 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();
@ -87,19 +99,25 @@ impl Shifter for DeltaGCLShifter {
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);
}

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(())
}

30
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();
@ -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,8 +73,10 @@ 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()

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 {

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

@ -1,13 +1,19 @@
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());

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

@ -1,13 +1,19 @@
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 {
@ -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);
}