fixing for pypi upload

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}"

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
+
+

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<ColumnConfig>,
 }
 

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::<Vec<_>>());
 
+    // 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<String> {
 // --- 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
+    // 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();
-    // --- END NEW CODE ---
+
+        println!("UncheckedIO: Profiling Mode ENABLED 🚀");
+    }
 
     m.add_function(wrap_pyfunction!(load_data_from_config, m)?)?;
     m.add_function(wrap_pyfunction!(to_pandas_dataframe, m)?)?;