From 06a6dcdd31743f51c7bf366a0399f0541ee2f240 Mon Sep 17 00:00:00 2001 From: Bill Date: Sat, 22 Nov 2025 23:37:57 -0700 Subject: [PATCH] fixing for pypi upload --- benchmark.py | 2 +- docs/changelog.md | 22 ++++++++++++++++++++++ src/config.rs | 5 ++--- src/lib.rs | 39 +++++++++++++++++++++++++-------------- 4 files changed, 50 insertions(+), 18 deletions(-) create mode 100644 docs/changelog.md diff --git a/benchmark.py b/benchmark.py index abe6138..f92b0bb 100644 --- a/benchmark.py +++ b/benchmark.py @@ -16,7 +16,7 @@ DB_PORT = "5433" # <-- Your local Docker port DB_NAME = "postgres" # Global Configuration -BLAST_RADIUS = 125000 # Rows per parallel task (1M / 62500 = 16 partitions) +BLAST_RADIUS = 312500 # Rows per parallel task (1M / 62500 = 16 partitions) # SQLAlchemy connection string (for Pandas) sqlalchemy_conn_str = f"postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}" diff --git a/docs/changelog.md b/docs/changelog.md new file mode 100644 index 0000000..0f1bbfb --- /dev/null +++ b/docs/changelog.md @@ -0,0 +1,22 @@ +Changelog +[Unreleased] - Sprint 4 Optimization +🚀 Performance +Speedup: Achieved 1.46x speedup over ConnectorX (19s vs 28s for 20M rows) on local benchmarks. +Worker Pool: Implemented a thread-per-core architecture using async_channel to decouple task generation from execution. +Auto-Tuning: Added blast_radius=0 support to dynamically calculate optimal partition sizes based on available CPU cores. +🛠️ Fixes & Stability +Self-Healing: Schema now defaults to nullable=true. If a partition fails (network error, bad data), the worker logs the error and returns a batch of NULLs instead of crashing the entire process. +Dependency Resolution: Fixed build errors with pyo3-arrow 0.15.0 by migrating to Arro3RecordBatch::from and into_pyobject. +Profiling Safety: tracing and tracy-client are now optional dependencies behind the profiling feature flag. This prevents overhead and crashes in production environments like Google Colab. +⚠️ Known Issues / Constraints +Severe Partition Switching Overhead: Increasing partition counts causes non-linear runtime growth. +Observation: 160 partitions took ~50s longer than 16 partitions. +Analysis: Each worker thread incurs a ~5-second setup penalty when switching to a new partition (establishing a new COPY stream). +Implication: While CPU-bound logic scales perfectly, the COPY protocol handshake or Docker networking layer is forcing massive latency spikes per task. +Recommendation: Avoid small partitions on high-latency networks (like Docker-on-Windows). Stick to num_cpus partitions until this handshake cost is optimized. +I/O Bound: Benchmarks with 8 vs 16 workers showed similar performance, indicating the system is currently Network/IO latency bound, not CPU bound. +📦 Build Instructions +Production: maturin build --release (Default) +Debug/Profile: maturin build --release --features profiling + + diff --git a/src/config.rs b/src/config.rs index a652fde..254bf6b 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,7 +1,6 @@ // --- External Crates --- use serde::Deserialize; -use std::collections::HashMap; -use anyhow::{Context, Result, anyhow}; +use anyhow::{Context, Result}; // Removed unused 'anyhow' macro import if not used, but Context/Result likely used use std::fmt::{self, Display}; // --- 1. CUSTOM ERROR DEFINITION --- @@ -29,7 +28,7 @@ pub struct ColumnConfig { #[derive(Debug, Deserialize)] pub struct ConnectorConfig { pub connection_string: String, - pub query: String, // <--- ADD THIS FIELD + pub query: String, pub schema: Vec, } diff --git a/src/lib.rs b/src/lib.rs index d0141bc..c67f132 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,7 +8,10 @@ use pyo3::exceptions::PyValueError; use tokio; use pyo3::types::{PyModule, PyAny}; use pyo3::Bound; -use pyo3_arrow::PyRecordBatch; + +// FIX: Use the export path you confirmed works in your IDE +use pyo3_arrow::export::Arro3RecordBatch; + // Required for global allocator (mimalloc) #[cfg(not(target_env = "msvc"))] use mimalloc; @@ -23,15 +26,17 @@ use crate::config::{load_and_validate_config, ConnectorConfig}; // NOTE: run_profiler_logic added, requires definition in parser.rs use crate::parser::{run_db_logic, run_profiler_logic}; -// NEW: Import the tracing libraries we need for the initialization +// NEW: Import the tracing libraries only if feature is enabled +#[cfg(feature = "profiling")] use tracing_subscriber::layer::SubscriberExt; +#[cfg(feature = "profiling")] use tracing_subscriber::util::SubscriberInitExt; // --- THE PYTHON-CALLABLE ENTRY POINT (Load Data) --- #[pyfunction] -#[pyo3(signature = (config_path, blast_radius=312500))] +#[pyo3(signature = (config_path, blast_radius=0))] // Default to 0 for auto-tuning #[allow(unsafe_code)] #[allow(unsafe_op_in_unsafe_fn)] #[allow(rust_2024_compatibility)] @@ -50,6 +55,8 @@ fn load_data_from_config<'py>( println!("Database: {}", config.connection_string); println!("Columns (in order): {:?}", config.schema.iter().map(|c| &c.column_name).collect::>()); + // FIX: Handle GIL release properly to avoid deprecation warnings if possible, + // but primarily ensure the logic works with the new Arro3RecordBatch wrapper. let record_batch = py.allow_threads(|| { tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -60,8 +67,10 @@ fn load_data_from_config<'py>( }) }).map_err(|e| PyValueError::new_err(format!("Database/Runtime Error: {:?}", e)))?; - let py_record_batch = PyRecordBatch::new(record_batch); - py_record_batch.into_pyarrow(py) + // FIX: Use Arro3RecordBatch::from() instead of new() + // This uses the standard From trait conversion. + let py_record_batch = Arro3RecordBatch::from(record_batch); + py_record_batch.into_pyobject(py) } // --- NEW PYTHON-CALLABLE FUNCTION FOR PANDAS CONVERSION (FIXED SIGNATURE) --- @@ -94,16 +103,18 @@ fn profile_data(config_path: String) -> PyResult { // --- PYTHON MODULE EXPORT --- #[pymodule] fn unchecked_io(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { - // --- START NEW CODE --- - // This tries to initialize the "Tracy" layer. - // The '.try_init()' ensures that if you import the library twice (e.g. in a notebook reload), - // it doesn't crash by trying to initialize the global logger a second time. - // Fix: Use 'default()' instead of 'new()' to avoid argument mismatch errors - let _ = tracing_subscriber::registry() - .with(tracing_tracy::TracyLayer::default()) - .try_init(); - // --- END NEW CODE --- + // NEW: Only initialize Tracy if the feature is enabled + #[cfg(feature = "profiling")] + { + // Use 'default()' instead of 'new()' to avoid argument mismatch errors + // The .try_init() prevents crashing on module reloads (e.g. Jupyter) + let _ = tracing_subscriber::registry() + .with(tracing_tracy::TracyLayer::default()) + .try_init(); + + println!("UncheckedIO: Profiling Mode ENABLED 🚀"); + } m.add_function(wrap_pyfunction!(load_data_from_config, m)?)?; m.add_function(wrap_pyfunction!(to_pandas_dataframe, m)?)?;