version update to publish to Pypi 2

# Conflicts:
#	Cargo.toml
#	src/lib.rs

.github/workflows/publish.yml

@@ -0,0 +1,75 @@
+name: Publish to PyPI
+
+# This workflow is triggered when you push a new version tag (e.g., v0.1.0, v1.2.3)
+on:
+  push:
+    tags:
+      - 'v[0-9]+.[0-9]+.[0-9]+*' # Matches v0.1.0 or v1.0.0-rc1, etc.
+  # Allows manual triggering from the GitHub Actions UI
+  workflow_dispatch:
+
+jobs:
+  build_wheels:
+    name: Build wheels on ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    # Define a matrix to build for common operating systems and architectures
+    strategy:
+      matrix:
+        include:
+          - os: ubuntu-latest
+            target: x86_64
+          - os: macos-latest
+            target: x86_64
+          - os: windows-latest
+            target: x86_64
+    
+    steps:
+      - uses: actions/checkout@v4
+      
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12' # Target a modern Python version
+          
+      - name: Install Rust
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: stable
+
+      - name: Build wheels with Maturin
+        uses: PyO3/maturin-action@v1
+        with:
+          # Use the target from the matrix (x86_64)
+          target: ${{ matrix.target }}
+          # Build a manylinux-compatible wheel for Linux/Colab users
+          manylinux: auto
+          command: build
+          args: --release --out dist --find-interpreter
+
+      - name: Upload built wheels as Artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: wheels
+          path: dist
+
+  publish:
+    name: Publish to PyPI
+    needs: [build_wheels]
+    runs-on: ubuntu-latest
+    # This step will only run if the 'build_wheels' job completed successfully
+    if: startsWith(github.ref, 'refs/tags/')
+
+    steps:
+      - uses: actions/download-artifact@v4
+        with:
+          name: wheels
+          path: dist
+          
+      - name: Publish to PyPI
+        uses: PyO3/maturin-action@v1
+        with:
+          # Use the secret we configured on GitHub
+          pypi-token: ${{ secrets.PYPI_API_TOKEN }}
+          # Command to upload all wheels in the 'dist' directory
+          command: upload
+          args: --skip-existing --non-interactive

Cargo.lock

@@ -259,6 +259,18 @@ dependencies = [
  "regex-syntax",
 ]
 
+[[package]]
+name = "async-channel"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2"
+dependencies = [
+ "concurrent-queue",
+ "event-listener-strategy",
+ "futures-core",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "async-trait"
 version = "0.1.89"
@@ -350,7 +362,7 @@ dependencies = [
  "js-sys",
  "num-traits",
  "wasm-bindgen",
- "windows-link",
+ "windows-link 0.2.1",
 ]
 
 [[package]]
@@ -374,6 +386,15 @@ dependencies = [
  "unicode-width",
 ]
 
+[[package]]
+name = "concurrent-queue"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "const-random"
 version = "0.1.18"
@@ -409,6 +430,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
+
 [[package]]
 name = "crunchy"
 version = "0.2.4"
@@ -498,6 +525,27 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
 
+[[package]]
+name = "event-listener"
+version = "5.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab"
+dependencies = [
+ "concurrent-queue",
+ "parking",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "event-listener-strategy"
+version = "0.5.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
+dependencies = [
+ "event-listener",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "fallible-iterator"
 version = "0.2.0"
@@ -574,6 +622,20 @@ dependencies = [
  "slab",
 ]
 
+[[package]]
+name = "generator"
+version = "0.8.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "605183a538e3e2a9c1038635cc5c2d194e2ee8fd0d1b66b8349fad7dbacce5a2"
+dependencies = [
+ "cc",
+ "cfg-if",
+ "libc",
+ "log",
+ "rustversion",
+ "windows",
+]
+
 [[package]]
 name = "generic-array"
 version = "0.14.7"
@@ -660,7 +722,7 @@ dependencies = [
  "js-sys",
  "log",
  "wasm-bindgen",
- "windows-core",
+ "windows-core 0.62.2",
 ]
 
 [[package]]
@@ -818,6 +880,28 @@ version = "0.4.28"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "34080505efa8e45a4b816c349525ebe327ceaa8559756f0356cba97ef3bf7432"
 
+[[package]]
+name = "loom"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca"
+dependencies = [
+ "cfg-if",
+ "generator",
+ "scoped-tls",
+ "tracing",
+ "tracing-subscriber",
+]
+
+[[package]]
+name = "matchers"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
+dependencies = [
+ "regex-automata",
+]
+
 [[package]]
 name = "matrixmultiply"
 version = "0.3.10"
@@ -888,6 +972,15 @@ dependencies = [
  "rawpointer",
 ]
 
+[[package]]
+name = "nu-ansi-term"
+version = "0.50.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "num-bigint"
 version = "0.4.6"
@@ -959,6 +1052,12 @@ version = "1.21.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d"
 
+[[package]]
+name = "parking"
+version = "2.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.5"
@@ -979,7 +1078,7 @@ dependencies = [
  "libc",
  "redox_syscall",
  "smallvec",
- "windows-link",
+ "windows-link 0.2.1",
 ]
 
 [[package]]
@@ -1299,6 +1398,12 @@ version = "1.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f"
 
+[[package]]
+name = "scoped-tls"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294"
+
 [[package]]
 name = "scopeguard"
 version = "1.2.0"
@@ -1378,6 +1483,15 @@ dependencies = [
  "digest",
 ]
 
+[[package]]
+name = "sharded-slab"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "shlex"
 version = "1.3.0"
@@ -1500,6 +1614,15 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "thread_local"
+version = "1.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185"
+dependencies = [
+ "cfg-if",
+]
+
 [[package]]
 name = "tiny-keccak"
 version = "2.0.2"
@@ -1620,6 +1743,68 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678"
 dependencies = [
  "once_cell",
+ "valuable",
+]
+
+[[package]]
+name = "tracing-log"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
+dependencies = [
+ "log",
+ "once_cell",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-subscriber"
+version = "0.3.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2054a14f5307d601f88daf0553e1cbf472acc4f2c51afab632431cdcd72124d5"
+dependencies = [
+ "matchers",
+ "nu-ansi-term",
+ "once_cell",
+ "regex-automata",
+ "sharded-slab",
+ "smallvec",
+ "thread_local",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
+]
+
+[[package]]
+name = "tracing-tracy"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6a90519f16f55e5c62ffd5976349f10744435a919ecff83d918300575dfb69b"
+dependencies = [
+ "tracing-core",
+ "tracing-subscriber",
+ "tracy-client",
+]
+
+[[package]]
+name = "tracy-client"
+version = "0.17.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73202d787346a5418f8222eddb5a00f29ea47caf3c7d38a8f2f69f8455fa7c7e"
+dependencies = [
+ "loom",
+ "once_cell",
+ "tracy-client-sys",
+]
+
+[[package]]
+name = "tracy-client-sys"
+version = "0.24.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69fff37da548239c3bf9e64a12193d261e8b22b660991c6fd2df057c168f435f"
+dependencies = [
+ "cc",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -1630,10 +1815,11 @@ checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb"
 
 [[package]]
 name = "unchecked-io"
-version = "0.1.0"
+version = "0.1.1"
 dependencies = [
  "anyhow",
  "arrow",
+ "async-channel",
  "byteorder",
  "bytes",
  "chrono",
@@ -1647,6 +1833,9 @@ dependencies = [
  "serde_yaml",
  "tokio",
  "tokio-postgres",
+ "tracing",
+ "tracing-subscriber",
+ "tracing-tracy",
  "uuid",
 ]
 
@@ -1707,6 +1896,12 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "valuable"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
+
 [[package]]
 name = "version_check"
 version = "0.9.5"
@@ -1800,6 +1995,41 @@ dependencies = [
  "web-sys",
 ]
 
+[[package]]
+name = "windows"
+version = "0.61.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893"
+dependencies = [
+ "windows-collections",
+ "windows-core 0.61.2",
+ "windows-future",
+ "windows-link 0.1.3",
+ "windows-numerics",
+]
+
+[[package]]
+name = "windows-collections"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8"
+dependencies = [
+ "windows-core 0.61.2",
+]
+
+[[package]]
+name = "windows-core"
+version = "0.61.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3"
+dependencies = [
+ "windows-implement",
+ "windows-interface",
+ "windows-link 0.1.3",
+ "windows-result 0.3.4",
+ "windows-strings 0.4.2",
+]
+
 [[package]]
 name = "windows-core"
 version = "0.62.2"
@@ -1808,9 +2038,20 @@ checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
 dependencies = [
  "windows-implement",
  "windows-interface",
- "windows-link",
+ "windows-link 0.2.1",
- "windows-result",
+ "windows-result 0.4.1",
- "windows-strings",
+ "windows-strings 0.5.1",
+]
+
+[[package]]
+name = "windows-future"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e"
+dependencies = [
+ "windows-core 0.61.2",
+ "windows-link 0.1.3",
+ "windows-threading",
 ]
 
 [[package]]
@@ -1835,19 +2076,53 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "windows-link"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a"
+
 [[package]]
 name = "windows-link"
 version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
 
+[[package]]
+name = "windows-numerics"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1"
+dependencies = [
+ "windows-core 0.61.2",
+ "windows-link 0.1.3",
+]
+
+[[package]]
+name = "windows-result"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6"
+dependencies = [
+ "windows-link 0.1.3",
+]
+
 [[package]]
 name = "windows-result"
 version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57"
+dependencies = [
+ "windows-link 0.1.3",
 ]
 
 [[package]]
@@ -1856,7 +2131,7 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
 ]
 
 [[package]]
@@ -1865,7 +2140,7 @@ version = "0.60.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
 dependencies = [
- "windows-targets",
+ "windows-targets 0.53.5",
 ]
 
 [[package]]
@@ -1874,7 +2149,23 @@ version = "0.61.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
+]
+
+[[package]]
+name = "windows-targets"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
+dependencies = [
+ "windows_aarch64_gnullvm 0.52.6",
+ "windows_aarch64_msvc 0.52.6",
+ "windows_i686_gnu 0.52.6",
+ "windows_i686_gnullvm 0.52.6",
+ "windows_i686_msvc 0.52.6",
+ "windows_x86_64_gnu 0.52.6",
+ "windows_x86_64_gnullvm 0.52.6",
+ "windows_x86_64_msvc 0.52.6",
 ]
 
 [[package]]
@@ -1883,59 +2174,116 @@ version = "0.53.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
- "windows_aarch64_gnullvm",
+ "windows_aarch64_gnullvm 0.53.1",
- "windows_aarch64_msvc",
+ "windows_aarch64_msvc 0.53.1",
- "windows_i686_gnu",
+ "windows_i686_gnu 0.53.1",
- "windows_i686_gnullvm",
+ "windows_i686_gnullvm 0.53.1",
- "windows_i686_msvc",
+ "windows_i686_msvc 0.53.1",
- "windows_x86_64_gnu",
+ "windows_x86_64_gnu 0.53.1",
- "windows_x86_64_gnullvm",
+ "windows_x86_64_gnullvm 0.53.1",
- "windows_x86_64_msvc",
+ "windows_x86_64_msvc 0.53.1",
 ]
 
+[[package]]
+name = "windows-threading"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6"
+dependencies = [
+ "windows-link 0.1.3",
+]
+
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
+
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
 
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
+
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
 
+[[package]]
+name = "windows_i686_gnu"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
+
 [[package]]
 name = "windows_i686_gnu"
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
 
+[[package]]
+name = "windows_i686_gnullvm"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
+
 [[package]]
 name = "windows_i686_gnullvm"
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
 
+[[package]]
+name = "windows_i686_msvc"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
+
 [[package]]
 name = "windows_i686_msvc"
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
 
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
+
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
 
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
+
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
 
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
+
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.53.1"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "unchecked-io"
-version = "0.1.0"
+version = "0.1.1"
 authors = ["Billthemaker"] # Replace with your name or alias
 license = "BSL-1" # Good practice for open-source
 edition = "2024"
@@ -13,6 +13,7 @@ crate-type = ["cdylib", "rlib"]
 # 1. Python Bindings for FFI
 pyo3 = { version = "0.27.1", features = ["extension-module"] }
 
+
 # 2. Configuration Parsing (YAML)
 serde = { version = "1.0", features = ["derive"] }
 serde_yaml = "0.9"
@@ -24,6 +25,7 @@ arrow = "57.0.0"
 # 4. Asynchronous Runtime (Essential for I/O and Postgres)
 tokio = { version = "1.37", features = ["full"] }
 
+
 # 5. Database Connection (Postgres)
 tokio-postgres = "0.7"
 deadpool-postgres = "0.14"
@@ -52,5 +54,13 @@ pyo3-arrow = "0.15.0"
 # 13. System CPU Count (NEW)
 num_cpus = "1.16"
 
+tracing-subscriber = { version = "0.3", features = ["registry", "env-filter"], optional = true}
+tracing-tracy = { version = "=0.11.2", optional = true }
+tracing = {version ="0.1.41", optional = true }
+
+async-channel = "2.5.0"
+
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
-mimalloc = { version = "0.1.39" }
+mimalloc = { version = "0.1.39" }
+
+async-channel = "2.3"

benchmark.py

@@ -16,7 +16,7 @@ DB_PORT = "5433" # <-- Your local Docker port
 DB_NAME = "postgres"
 
 # Global Configuration
-BLAST_RADIUS = 1250000 # 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/build-cleanup-chrono.md

@@ -0,0 +1,79 @@
+Technical Task: Build Fixes & Profiler Cleanup
+Date: 2025-11-21 Goal: Fix PyPI build failures and remove heavy profiling code from production wheels.
+
+1. Fix PyPI Build Failure (chrono-tz)
+   Issue: The build fails on Linux/Mac/Windows CI because chrono-tz requires complex C-bindings that are often missing in standard Python environments. Analysis: UncheckedIO returns Apache Arrow RecordBatch objects (where timestamps are raw i64). We do not need pyo3 to convert Rust chrono types directly to Python datetime objects. The chrono-tz feature in pyo3 is dead weight causing build failures.
+
+Action: Update Cargo.toml to remove the conflicting feature.
+
+Ini, TOML
+
+# Cargo.toml
+
+# OLD
+# pyo3 = { version = "0.27.1", features = ["extension-module", "chrono-tz"] }
+
+# NEW (Remove "chrono-tz")
+pyo3 = { version = "0.27.1", features = ["extension-module"] }
+2. Productionize the Profiler (Remove Tracy)
+   Issue: The tracy-client and tracing-tracy crates add significant binary size and runtime overhead. They should not be present in the production library installed by users. Solution: Use Rust Feature Flags to make profiling "opt-in" for development only.
+
+Step A: Update Cargo.toml Define a profiling feature and make the heavy dependencies optional.
+
+Ini, TOML
+
+[features]
+default = []
+# The "profiling" feature enables these two optional crates
+profiling = ["dep:tracing-tracy", "dep:tracy-client"]
+
+[dependencies]
+# Keep this (Lightweight facade)
+tracing = "0.1.41"
+
+# Make these OPTIONAL (Heavy implementation)
+tracing-tracy = { version = "0.11.2", optional = true }
+tracy-client = { version = "0.17.6", optional = true }
+
+# ... other dependencies ...
+Step B: Update src/lib.rs Wrap the profiler initialization logic so it only compiles when the flag is active.
+
+Rust
+
+// src/lib.rs
+
+#[pymodule]
+fn unchecked_io(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
+
+    // --- START PROFILER LOGIC ---
+    // This block now ONLY compiles if you run with `--features profiling`
+    #[cfg(feature = "profiling")] 
+    {
+        use tracing_subscriber::layer::SubscriberExt;
+        use tracing_subscriber::util::SubscriberInitExt;
+        
+        // Initialize the Tracy layer
+        let _ = tracing_subscriber::registry()
+            .with(tracing_tracy::TracyLayer::default())
+            .try_init();
+            
+        println!("--- UncheckedIO: Profiling Enabled (Tracy) ---");
+    }
+    // --- END PROFILER LOGIC ---
+
+    m.add_function(wrap_pyfunction!(load_data_from_config, m)?)?;
+    m.add_function(wrap_pyfunction!(to_pandas_dataframe, m)?)?;
+    m.add_function(wrap_pyfunction!(profile_data, m)?)?;
+    Ok(())
+}
+3. How to Build
+   For Development (With Profiler): Use this when you want to see the tracks in the Tracy GUI.
+
+Bash
+
+maturin develop --features profiling
+For Production / PyPI (Zero Overhead): This builds the "clean" version for users. The profiler code is stripped out entirely.
+
+Bash
+
+maturin publish

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

docs/marketing-business.md

@@ -0,0 +1,105 @@
+# UncheckedIO: Business Strategy & Technical Roadmap
+
+**Date:** 2025-11-21
+**Status:** Strategic Planning
+**Core Philosophy:** "The Configured Opinion" — We trade flexibility for raw, unchecked speed.
+
+---
+
+## 1. Target Audience & Value Propositions
+
+We are not selling "a faster Pandas." We are selling **infrastructure efficiency**.
+
+### Primary Target: Data Platform & MLOps Engineers
+* **Who they are:** The people managing Kubernetes clusters, Airflow DAGs, and AWS Fargate costs. They care about stability and resource usage.
+* **The Pitch:** "A Zero-Copy Transport Layer for Postgres."
+* **Key Selling Points:**
+    * **Cut Compute Costs:** Reduces CPU starvation by saturating network bandwidth. 10x faster serialization means 10x less time paying for vCPUs.
+    * **Memory Safety:** Written in Rust, ensuring no segfaults in your Python pipeline despite the speed.
+    * **Predictable Throughput:** Unlike generic drivers that choke on large datasets, `UncheckedIO` uses partitioned streaming to maintain constant memory usage regardless of data size.
+
+### Secondary Target: High-Frequency Trading (HFT) / Quant Developers
+* **Who they are:** Developers who need to backtest models on massive tick-data history stored in Postgres.
+* **The Pitch:** "The fastest way to get data from Disk to Arrow Memory."
+* **Key Selling Points:**
+    * **Zero-Copy to Polars:** Output directly to Apache Arrow memory layouts, ready for immediate analysis or SIMD vectorization.
+    * **Bypass the overhead:** Skips the "Safety Checks" (schema validation) that cost milliseconds per query. You promise the data is clean; we promise to load it instantly.
+
+### Tertiary Target: Data Scientists (The Users, Not Buyers)
+* **Who they are:** End users writing Jupyter Notebooks.
+* **The Pitch:** "Stop waiting for `read_sql`."
+* **Key Selling Points:**
+    * **Installation Simplicity:** `pip install unchecked_io`. No complex C dependencies or drivers to configure.
+    * **Drop-in Speed:** Replace a 5-minute load time with a 30-second load time.
+
+---
+
+## 2. Optimization Strategy: The "Win" Plan
+
+We need to stop fighting "theoretical" bottlenecks and fix the "physical" ones.
+
+### 🔴 Focus On (High ROI)
+1.  **Infrastructure-Aware Tuning (The "Blast Radius"):**
+    * *Why:* The "astronomical" delay was caused by thrashing.
+    * *Action:* Implement the `num_cpus` detection to auto-scale the connection pool and partition count. Do not let the user guess wrong.
+2.  **Memory Management (The "String Killer"):**
+    * *Why:* 15M heap allocations for strings creates allocator contention.
+    * *Action:* Refactor the parser to use a reusable buffer or "Zero-Copy" string slicing from the raw stream before writing to Arrow.
+3.  **Database Interaction (The Index):**
+    * *Why:* As discovered, a missing index turns a parallel fetch into a DDoS attack.
+    * *Action:* Documentation MUST strictly warn users: *"If you use `blast_radius`, you MUST index your partition column."*
+
+### 🛑 Leave Alone (Diminishing Returns)
+1.  **Complex "Safe Mode" Logic:**
+    * *Why:* We are `UncheckedIO`. If we add too many safety checks/fallbacks for dirty data, we become just another slow connector (like ConnectorX).
+    * *Strategy:* Let it fail fast. If the data violates the config schema, panic. That is the contract.
+2.  **Micro-Optimizing the TCP Handshake:**
+    * *Why:* Connection pooling already solves 95% of this. Saving another 1ms here is irrelevant compared to the 500ms of Query Planning time.
+
+---
+
+## 3. Configuration Matrix
+
+To be a "System-Level" tool, we need to expose the knobs that Platform Engineers expect.
+
+### Currently Configurable (The "MVP" Set)
+These features are already present in the code or `config.yaml`:
+1.  **`connection_string`:** Standard Postgres URI.
+2.  **`query`:** The exact `COPY ... TO STDOUT (FORMAT binary)` command. Allows total control over the SQL execution plan.
+3.  **`schema`:** Explicit definition of Arrow Types (`Int64`, `Utf8`, etc.). Skips runtime type inference.
+4.  **`blast_radius` (Argument):** Controls the row-count per partition. Passed from Python to Rust.
+
+### Must-Add Configurations (To Delight Consumers)
+These options solve specific "Enterprise" pain points:
+1.  **`worker_threads`:**
+    * *Value:* Allows limiting the library to use only 4 cores on a shared 64-core server.
+    * *Implementation:* Pass to `tokio::runtime::Builder::worker_threads()`.
+2.  **`pool_size`:**
+    * *Value:* Overrides the auto-detected defaults. Essential for databases with strict connection limits (e.g., AWS RDS limits).
+    * *Implementation:* Pass to `deadpool_postgres::Pool::builder().max_size()`.
+3.  **`batch_size` (Arrow):**
+    * *Value:* Controls the size of the Arrow RecordBatches. Critical for streaming data into ML models (e.g., "Give me 10k rows at a time").
+    * *Implementation:* Currently aggregated at the end; needs to be exposed in the streaming loop.
+4.  **`danger_mode` (Explicit Toggle):**
+    * *Value:* A flag to strictly enforce "Panic on Error" vs. "Try to Recover."
+    * *Implementation:* As described in Sprint 4.
+
+---
+
+## 4. The "Reputation Builder" Roadmap (Future Optimizations)
+
+These features serve two purposes: Extreme performance for the 1% of users who need it, and "Street Cred" for the author.
+
+1.  **`io_uring` Network Layer:**
+    * *The Tech:* Replace `tokio-postgres` (epoll) with a custom `io_uring` implementation for Linux.
+    * *The Flex:* "True Asynchronous syscall batching." Reduces context switches during massive data transfers.
+    * *Status:* High effort, high reputation, low immediate business ROI (unless on 100Gbps networks).
+2.  **SIMD Parser (AVX-512 / NEON):**
+    * *The Tech:* Use `std::simd` or `portable-simd` to parse 16 integers at once from the binary stream.
+    * *The Flex:* "Saturating the memory bandwidth of modern CPUs."
+3.  **Custom Allocator Support (`mimalloc` / `jemalloc`):**
+    * *The Tech:* Allow users to swap the global allocator via a feature flag.
+    * *The Flex:* "Optimized for high-fragmentation environments." (Note: `mimalloc` is already configured for non-MSVC targets).
+4.  **Kernel-Bypass Networking (DPDK):**
+    * *The Tech:* Bypassing the OS network stack entirely.
+    * *The Flex:* The absolute theoretical limit of data transfer. (Overkill, but legendary).

docs/rEaDMe.md

@@ -0,0 +1,89 @@
+UncheckedIO 🚀
+The world's fastest, most dangerous PostgreSQL-to-Arrow loader.
+UncheckedIO is an opinionated, high-performance data connector designed for bulk-loading massive datasets (Terabytes+) from PostgreSQL into Python/Apache Arrow. It achieves extreme speed by skipping runtime schema validation and using a zero-copy streaming parser.
+⚠️ WARNING: This library assumes you know what you are doing. If your config.yaml schema does not match your database schema, it will produce garbage data or crash. The "Unchecked" in the name is not a suggestion; it is a promise.
+Features
+🚀 1.5x - 2x Faster than ConnectorX when Benchmarked on 20M+ row dataset in default Colab runtime.
+🧵 Parallel Worker Pool: Automatically scales to available CPU cores.
+🧠 Auto-Tuning: Dynamically calculates optimal partition sizes (blast_radius).
+📉 Low Memory Footprint: Streaming parser with zero-copy string slicing.
+🛠️ "Self-Healing" (Nullable): If a partition fails (e.g., network flake), it returns NULLs for that chunk instead of crashing the entire job.
+🔬 Optional Profiling: Built-in integration with the Tracy Profiler for deep I/O analysis.
+Installation
+pip install unchecked-io
+
+Quick Start
+Create a config.yaml file:
+connection_string: postgresql://user:password@localhost:5432/dbname
+# query MUST be a COPY ... TO STDOUT (FORMAT binary) command
+query: >
+COPY (SELECT id, name, score, created_at FROM users)
+TO STDOUT (FORMAT binary)
+schema:
+- column_name: id
+  arrow_type: Int64
+- column_name: name
+  arrow_type: Utf8
+- column_name: score
+  arrow_type: Float32
+- column_name: created_at
+  arrow_type: Timestamp(Nanosecond, None)
+
+
+Run it in Python:
+import unchecked_io
+
+# blast_radius=0 enables auto-tuning
+arrow_table = unchecked_io.load_data_from_config("config.yaml", blast_radius=0)
+
+# Convert to Pandas (Zero-Copy)
+df = unchecked_io.to_pandas_dataframe(arrow_table)
+print(df.head())
+
+
+🔬 Profiling with Tracy
+UncheckedIO includes optional instrumentation for the Tracy Profiler to visualize I/O starvation and thread contention.
+This feature is DISABLED by default to prevent overhead and crashes in production environments (like Google Colab).
+How to Enable Profiling
+You must build the library from source with the profiling feature enabled.
+Install Rust & Maturin:
+curl --proto '=https' --tlsv1.2 -sSf [https://sh.rustup.rs](https://sh.rustup.rs) | sh
+pip install maturin
+
+
+Build with Feature Flag:
+# This builds the wheel and installs it in your current venv
+maturin develop --release --features profiling
+
+
+Run Tracy (GUI):
+Download Tracy v0.11.1 (Must match protocol v0.11.x).
+Open Tracy.exe and click Connect.
+Run your Python script. You will see real-time thread timelines and "IO_WAIT_STARVATION" blocks.
+Configuration Reference
+Supported Types
+PostgreSQL Type
+Config arrow_type
+bigint / int8
+Int64
+integer / int4
+Int32
+double precision
+Float64
+real / float4
+Float32
+text / varchar / uuid
+Utf8
+boolean
+Boolean
+timestamp
+Timestamp(Nanosecond, None)
+date
+Date32
+
+blast_radius Parameter
+0: Auto-Tune (Recommended). Calculates partition size based on row count / (cores * 4).
+> 0: Manual Override. Sets specific number of rows per partition. Use for fine-tuning on specific hardware.
+License
+Apache 2.0
+Free for non-production use and production use.

docs/tech_fix.md

@@ -0,0 +1,83 @@
+Here is the technical summary of our debugging session, formatted as a Markdown file for your documentation.
+
+Markdown
+
+# Technical Post-Mortem: The "Thrashing" & Performance Optimization Log
+
+**Date:** 2025-11-21
+**Project:** UncheckedIO
+**Topic:** Debugging "Astronomical" Task Switching Delays & Laptop vs. Cloud Discrepancies
+
+## 1. The Symptoms
+* **The "Astronomical" Delay:** Increasing partition count (e.g., from 16 to 32) caused disproportionate slowdowns. Switching between tasks took seconds, not nanoseconds.
+* **The "Tracy" Paradox:** The profiler showed "CPU Data Starved" and frequently crashed the runtime. When it crashed/detached, the library actually ran faster.
+* **Platform Discrepancy:** The library performed significantly better on Google Colab (Server CPU) than on a high-spec Laptop (Consumer CPU) relative to competitors, despite the laptop having faster single-core speed.
+
+## 2. Root Cause Analysis
+
+### A. The Database Villain: "The Missing Index"
+* **The Issue:** The `setup_db.sql` script created the table but **failed to index the `id` column**.
+* **The Consequence:** The "Parallel Coordinator" splits the job into partitions (e.g., `WHERE id BETWEEN 0 AND 125000`). Without an index, Postgres performed a **Sequential Scan** (Full Table Scan) for *every single partition*.
+* **The Impact:** Running 16 parallel workers triggered **16 simultaneous Full Table Scans**. This saturated memory bandwidth instantly, explaining the "CPU Data Starved" metric in Tracy.
+
+### B. The Concurrency Villain: "Pool Contention & Thrashing"
+* **The Issue:** The connection pool was hardcoded to `max_size(20)` in `src/parser.rs`.
+* **The "Alive" Misconception:** We assumed keeping connections "alive" allowed infinite parallelism. However, a connection is a single-lane bridge; it can only transport one query at a time.
+* **The Bottleneck:** When spawning 800 tasks (partitions) against 20 connections:
+    * **20 Tasks** ran immediately.
+    * **780 Tasks** sat frozen in a RAM queue.
+* **The "Thrashing":** The CPU spent ~90% of its time managing this queue—saving/restoring task states (Context Switching) and flushing L1/L2 caches—rather than parsing data. This overhead destroyed performance on the laptop (which has smaller caches than the Colab server).
+
+### C. The Red Herring: `io_uring`
+* **Hypothesis:** We initially thought switching to `io_uring` (Linux asynchronous I/O) would solve the "waiting."
+* **Verdict:** **Incorrect.** The bottleneck was not syscall overhead (which `io_uring` solves); it was **Logical Contention** (waiting for a DB connection) and **Physical Latency** (waiting for RAM/Network).
+
+---
+
+## 3. The Solutions
+
+### Fix #1: Create the Index (Database Side)
+Change the operation from a Sequential Scan (reading 20M rows x 16 times) to an **Index Scan** (jumping directly to the data).
+
+**File:** `setup_db.sql`
+```sql
+-- OLD
+ANALYZE benchmark_table;
+
+-- NEW
+CREATE INDEX idx_benchmark_id ON benchmark_table(id);
+ANALYZE benchmark_table;
+Fix #2: Dynamic Pool Sizing (Rust Side)
+Remove the artificial speed limit. Match the pool size to the hardware capabilities so tasks never wait in a queue.
+
+File: src/parser.rs
+
+Rust
+
+// OLD
+.max_size(20)
+
+// NEW
+use num_cpus; // Add to Cargo.toml
+let num_threads = num_cpus::get();
+.max_size(num_threads) // 1:1 mapping of Threads to Connections
+Fix #3: Auto-Tuned "Blast Radius"
+Instead of a hardcoded partition size or user guess, calculate the optimal partition count to balance 
+load without causing overhead.
+
+Strategy: 4 * CPU Cores partitions.
+
+Why: Enough chunks to allow "Work Stealing" (fast threads take more work), but few enough to prevent
+connection setup overhead.
+
+Rust
+
+pub fn calculate_optimal_blast_radius(min_id: i64, max_id: i64) -> i64 {
+    let total_rows = max_id - min_id;
+    let target_partitions = num_cpus::get() * 4; 
+    total_rows / (target_partitions as i64)
+}
+4. Conclusion
+The library's "Schema-on-Read" optimization (trusting the config) was working perfectly. 
+The performance loss was due to infrastructure bottlenecks (DB Index and Connection Pool limits) 
+that forced the CPU to manage queues instead of parsing data.

setup_db.sql

@@ -39,5 +39,6 @@ SELECT
 FROM generate_series(1, 20000000) s(i);
 
 -- 4. Analyze the table for better query planning (good practice)
+CREATE INDEX idx_benchmark_id ON benchmark_table(id);
 ANALYZE benchmark_table;
 

src/config.rs

@@ -1,7 +1,6 @@
 // --- External Crates ---
 use serde::Deserialize;
-use std::collections::HashMap;
+use anyhow::{Context, Result}; // Removed unused 'anyhow' macro import if not used, but Context/Result likely used
-use anyhow::{Context, Result, anyhow};
 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,10 +8,11 @@ use pyo3::exceptions::PyValueError;
 use tokio;
 use pyo3::types::{PyModule, PyAny};
 use pyo3::Bound;
-use pyo3_arrow::PyRecordBatch;
 
-// Use the high-performance mimalloc for better multi-threaded memory allocation.
+// FIX: Use the export path you confirmed works in your IDE
-// We conditionally compile it to avoid issues on MSVC targets.
+use pyo3_arrow::export::Arro3RecordBatch;
+
+// Required for global allocator (mimalloc)
 #[cfg(not(target_env = "msvc"))]
 use mimalloc;
 
@@ -19,15 +20,23 @@ use mimalloc;
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
+
 // --- Internal Crates ---
 use crate::config::{load_and_validate_config, ConnectorConfig};
-use crate::parser::run_db_logic;
+// NOTE: run_profiler_logic added, requires definition in parser.rs
+use crate::parser::{run_db_logic, run_profiler_logic};
+
+// 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 ---
+// --- 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)]
@@ -37,7 +46,6 @@ fn load_data_from_config<'py>(
     blast_radius: i64,
 ) -> PyResult<Bound<'py, PyAny>> {
 
-    // --- Phase 1: Load and Validate Configuration ---
     let config: ConnectorConfig = match load_and_validate_config(&config_path) {
         Ok(c) => c,
         Err(e) => return Err(PyValueError::new_err(format!("Configuration Error: {:?}", e))),
@@ -47,7 +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<_>>());
 
-    // --- Phase 2: Run Core Logic ---
+    // 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()
@@ -58,16 +67,57 @@ fn load_data_from_config<'py>(
             })
     }).map_err(|e| PyValueError::new_err(format!("Database/Runtime Error: {:?}", e)))?;
 
-    // --- Phase 3: Return Data to Python ---
+    // FIX: Use Arro3RecordBatch::from() instead of new()
-    let py_record_batch = PyRecordBatch::new(record_batch);
+    // This uses the standard From trait conversion.
-    py_record_batch.into_pyarrow(py)
+    let py_record_batch = Arro3RecordBatch::from(record_batch);
+    py_record_batch.into_pyobject(py)
+}
+
+// --- NEW PYTHON-CALLABLE FUNCTION FOR PANDAS CONVERSION (FIXED SIGNATURE) ---
+#[pyfunction]
+#[pyo3(signature = (arrow_table))] // FIX: Removed 'py' from the signature macro
+fn to_pandas_dataframe<'py>(py: Python<'py>, arrow_table: Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    // This calls the 'to_pandas' method on the PyArrow object.
+    arrow_table.call_method0("to_pandas")
+}
+
+// --- NEW PYTHON-CALLABLE ENTRY POINT FOR SCHEMA PROFILING ---
+#[pyfunction]
+#[pyo3(signature = (config_path))]
+fn profile_data(config_path: String) -> PyResult<String> {
+    // Note: The profiler uses a small, current_thread tokio runtime since it's sequential I/O.
+    let output = std::thread::spawn(move || {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async {
+                run_profiler_logic(&config_path).await
+            })
+    }).join().unwrap().map_err(|e| PyValueError::new_err(format!("Profiling Error: {:?}", e)))?;
+
+    Ok(output)
 }
 
 
 // --- PYTHON MODULE EXPORT ---
-
 #[pymodule]
 fn unchecked_io(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
+
+    // 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)?)?;
+    m.add_function(wrap_pyfunction!(profile_data, m)?)?;
     Ok(())
 }

src/parser.rs

@@ -1,8 +1,8 @@
 // --- External Crates ---
 use std::pin::Pin;
 use std::sync::Arc;
-// FIX: Replace direct tokio_postgres connections with deadpool
 use tokio_postgres::{NoTls, CopyOutStream, Config as PgConfig};
+// Required for the connection pool
 use deadpool_postgres::{Pool, Manager, Runtime};
 use anyhow::{Context, Result, anyhow};
 use arrow::array::{
@@ -19,14 +19,19 @@ use bytes::{Bytes, BytesMut, Buf};
 use byteorder::{BigEndian, ReadBytesExt};
 use std::io::{Cursor, Read};
 use std::str;
-use chrono::{NaiveDateTime, NaiveDate};
-use std::mem;
 use std::str::FromStr;
+use chrono::{NaiveDateTime, NaiveDate};
+use std::time::Instant;
 use tokio::task::JoinSet;
 use arrow::compute::concat_batches;
+// NEW: For the Work Stealing Queue
+use async_channel;
+// NEW: Tracing macros for profiling - Only import if feature is enabled
+#[cfg(feature = "profiling")]
+use tracing::{span, Level};
 
 // --- Internal Crates ---
-use crate::config::ConnectorConfig;
+use crate::config::{ConnectorConfig, load_and_validate_config};
 
 
 // --- CONSTANTS ---
@@ -35,26 +40,44 @@ use crate::config::ConnectorConfig;
 const POSTGRES_EPOCH_MICROS_OFFSET: i64 = 946684800000000;
 
 
-// --- 1. CORE DATABASE LOGIC (PARALLEL COORDINATOR) ---
+// --- 1. CORE DATABASE LOGIC (WORKER POOL PATTERN) ---
-// Note: We use blast_radius from Python config as the partitioning strategy
+// This is the fully optimized function using Connection Pooling and Static Dispatch.
 pub async fn run_db_logic(config: ConnectorConfig, blast_radius: i64) -> Result<RecordBatch> {
 
-    println!("UncheckedIO: Starting Query Planner (Blast Radius: {})...", blast_radius);
+    // Start overall timer
+    let start_total = Instant::now();
+    let start_phase1 = Instant::now();
 
-    // --- PHASE 1: SETUP CONNECTION POOL ---
+    // NEW: High-level span for the whole operation
+    #[cfg(feature = "profiling")]
+    let root_span = span!(Level::INFO, "UncheckedIO_Run");
+    #[cfg(feature = "profiling")]
+    let _root_guard = root_span.enter();
+
+    // --- PHASE 1: SETUP CONNECTION POOL & STATS ---
+    #[cfg(feature = "profiling")]
+    let phase1_span = span!(Level::INFO, "Phase1_Setup");
+    #[cfg(feature = "profiling")]
+    let _p1_guard = phase1_span.enter();
+
+    // 1. Calculate Worker Count (Fixed Parallelism)
+    let num_workers = num_cpus::get();
+    println!("UncheckedIO: Detected {} logical cores. Spawning {} worker threads.", num_workers, num_workers);
+
+    // 2. Setup Connection Pool
     let pg_config: tokio_postgres::Config = PgConfig::from_str(&config.connection_string)
         .context("Invalid connection string in config")?;
 
-    // Initialize the Manager and Pool
     let manager = Manager::new(pg_config.clone(), NoTls);
-    // Set pool size higher than the expected partition count to ensure connections are always available.
+    // FIX: Set pool size exactly to num_workers to prevent starvation or waiting
     let pool = Pool::builder(manager)
-        .max_size(20)
+        .max_size(num_workers)
         .runtime(Runtime::Tokio1)
         .build()
         .context("Failed to build connection pool")?;
 
-    // 2. Query for Table Bounds (using pool connection)
+    // 3. Query Table Bounds
+    // We grab a temporary connection just for this setup phase
     let client = pool.get().await.context("Failed to get pool connection for stats query")?;
     let partition_key = "id";
 
@@ -62,95 +85,206 @@ pub async fn run_db_logic(config: ConnectorConfig, blast_radius: i64) -> Result<
         .context("Failed to parse base query from config")?;
     let base_query_inner = base_query.trim().trim_start_matches("COPY (").trim_end_matches(")");
 
-    // NOTE: We rely on MIN/MAX here, assuming dense key for benchmark data.
     let stats_query = format!("SELECT MIN({}), MAX({}) FROM ({}) AS subquery", partition_key, partition_key, base_query_inner);
 
     let row = client.query_one(&stats_query, &[]).await?;
     let min_id: i64 = row.try_get(0).context("Failed to get MIN(id)")?;
     let max_id: i64 = row.try_get(1).context("Failed to get MAX(id)")?;
-    // Connection is returned to the pool when 'client' is dropped here.
+    drop(client); // Return connection to pool immediately
+
     println!("UncheckedIO: ID Range: {} to {}", min_id, max_id);
 
-    // 4. Generate Partitioned Queries (Based on blast_radius from Python)
+    // --- NEW: DYNAMIC PARTITION SIZING ---
+    let total_rows = (max_id - min_id + 1).max(1);
+    let calculated_blast_radius = if blast_radius <= 0 {
+        // Auto-tuning: Aim for ~4 chunks per worker to balance load
+        let target_chunks = (num_workers * 4) as i64;
+        let dynamic_size = total_rows / target_chunks;
+        // Ensure a sane minimum (e.g., don't make chunks of 1 row)
+        let size = dynamic_size.max(10_000);
+        println!("UncheckedIO: Auto-tuned partition size to {} rows (Targeting {} chunks).", size, target_chunks);
+        size
+    } else {
+        println!("UncheckedIO: Using user-defined partition size: {} rows.", blast_radius);
+        blast_radius
+    };
+
+
+    // 4. Create Work Queue
+    // We use a tuple: (index, query, expected_rows) so we can re-sort later
     struct PartitionTask {
         index: usize,
         query: String,
         expected_rows: usize
     }
 
-    let mut partitions: Vec<PartitionTask> = Vec::new();
+    // Create an unbounded channel. 
+    // tx = transmitter (main thread), rx = receiver (workers)
+    let (tx, rx) = async_channel::unbounded::<PartitionTask>();
+
+    // 5. Populate the Queue (The "Blast Radius" Logic)
     let mut current_min = min_id;
     let mut idx = 0;
+    let mut total_partitions = 0;
 
     while current_min <= max_id {
-        let current_max = (current_min + blast_radius - 1).min(max_id);
+        let current_max = (current_min + calculated_blast_radius - 1).min(max_id);
         let new_query = format!(
             "COPY (SELECT * FROM ({}) AS sub WHERE {} BETWEEN {} AND {}) TO STDOUT (FORMAT binary)",
             base_query_inner, partition_key, current_min, current_max
         );
         let estimated_rows = (current_max - current_min + 1) as usize;
 
-        partitions.push(PartitionTask { index: idx, query: new_query, expected_rows: estimated_rows });
+        // FIX: Removed 'partitions.push(...)' which caused the error.
-        current_min += blast_radius;
+        // We send directly to the channel now.
-        idx += 1;
+        let task = PartitionTask { index: idx, query: new_query, expected_rows: estimated_rows };
-    }
+
-    println!("UncheckedIO: Generated {} parallel partitions.", partitions.len());
+        // Send to queue (non-blocking since it's unbounded)
+        tx.send(task).await.context("Failed to fill work queue")?;
+
+        current_min += calculated_blast_radius;
+        idx += 1;
+        total_partitions += 1;
+    }
+    // Close the channel so workers know when to stop
+    tx.close();
+
+    println!("UncheckedIO: Queued {} partitions for processing.", total_partitions);
+
+    #[cfg(feature = "profiling")]
+    drop(_p1_guard); // End Phase 1 Span
+    let duration_phase1 = start_phase1.elapsed();
+
+
+    // --- PHASE 2: PARALLEL EXECUTION (DATA TRANSFER + PARSING) ---
+    let start_phase2 = Instant::now();
+    #[cfg(feature = "profiling")]
+    let phase2_span = span!(Level::INFO, "Phase2_Execution");
+    #[cfg(feature = "profiling")]
+    let _p2_guard = phase2_span.enter();
 
-    // --- Phase 2: Parallel Execution ---
     let arrow_schema = Arc::new(build_arrow_schema(&config)?);
     let mut join_set = JoinSet::new();
 
-    for task in partitions {
+    // Spawn exactly 'num_workers' long-lived tasks
-        let worker_pool = pool.clone(); // Pass the pool handle
+    for worker_id in 0..num_workers {
+        let worker_rx = rx.clone();
+        let worker_pool = pool.clone();
         let worker_schema = arrow_schema.clone();
 
         join_set.spawn(async move {
-            let worker_logic = async {
+            // VISUALIZATION: Create a "track" for this worker in Tracy
-                // Get connection from pool (This is the speedup)
+            #[cfg(feature = "profiling")]
-                let worker_client = worker_pool.get().await
+            let worker_span = span!(Level::INFO, "Worker_Thread", id = worker_id);
-                    .context("Worker: Failed to get pool connection")?;
+            #[cfg(feature = "profiling")]
+            let _w_guard = worker_span.enter();
 
-                let copy_stream = worker_client.copy_out(task.query.as_str()).await?;
+            let mut worker_batches: Vec<(usize, RecordBatch)> = Vec::new();
-                let pinned_stream: Pin<Box<CopyOutStream>> = Box::pin(copy_stream);
 
-                // Call the static dispatch parser
+            // Worker Loop: Keep grabbing tasks until the queue is empty and closed
-                parse_data_with_schema(pinned_stream, worker_schema).await
+            while let Ok(task) = worker_rx.recv().await {
-            };
 
-            let result = worker_logic.await;
+                // VISUALIZATION: Show exactly which partition is being processed
-            (task.index, result, task.expected_rows)
+                #[cfg(feature = "profiling")]
+                let task_span = span!(Level::INFO, "Processing_Task", partition_id = task.index);
+                #[cfg(feature = "profiling")]
+                let _t_guard = task_span.enter();
+
+                // Process the task
+                // We wrap this in an inner block to easily catch errors for Self-Healing
+                let result = async {
+                    let client = worker_pool.get().await.context("Pool exhausted")?;
+                    let copy_stream = client.copy_out(task.query.as_str()).await?;
+                    let pinned_stream: Pin<Box<CopyOutStream>> = Box::pin(copy_stream);
+
+                    // Call Static Dispatch Parser
+                    parse_data_with_schema(pinned_stream, worker_schema.clone()).await
+                }.await;
+
+                match result {
+                    Ok((_rows, batch)) => {
+                        worker_batches.push((task.index, batch));
+                    }
+                    Err(e) => {
+                        // ERROR LOGGING
+                        #[cfg(feature = "profiling")]
+                        tracing::error!("Worker {}: Partition {} failed! Error: {}", worker_id, task.index, e);
+
+                        // --- SELF-HEALING LOGIC ---
+                        // If a partition fails (e.g. bad data), we log it and return NULLs
+                        eprintln!("UncheckedIO Worker {}: Partition {} failed! Error: {}. Filling NULLs.", worker_id, task.index, e);
+                        let null_batch = create_null_batch(worker_schema.clone(), task.expected_rows)?;
+                        worker_batches.push((task.index, null_batch));
+                    }
+                }
+            }
+
+            // Return all batches processed by this worker
+            Ok::<Vec<(usize, RecordBatch)>, anyhow::Error>(worker_batches)
         });
     }
 
-    // --- Phase 3: Collect, Order, and Stitch ---
+    // --- PHASE 3: AGGREGATION ---
-    let mut results: Vec<Option<RecordBatch>> = vec![None; idx];
+    let mut all_results: Vec<(usize, RecordBatch)> = Vec::with_capacity(total_partitions);
 
     while let Some(join_result) = join_set.join_next().await {
-        let (index, parse_result, expected_rows) = join_result.context("Worker thread panic")?;
+        match join_result {
-
+            Ok(worker_result) => {
-        match parse_result {
+                match worker_result {
-            Ok(batch) => {
+                    Ok(batches) => all_results.extend(batches),
-                results[index] = Some(batch.1);
+                    Err(e) => return Err(anyhow!("Worker task failed internally: {}", e)),
-            }
+                }
-            Err(e) => {
-                // --- Self-Healing Placeholder ---
-                eprintln!("UncheckedIO: Partition {} failed! Error: {}. Falling back to NULLs (Self-Healing logic required here).", index, e);
-                let null_batch = create_null_batch(arrow_schema.clone(), expected_rows)?;
-                results[index] = Some(null_batch);
             }
+            Err(e) => return Err(anyhow!("Worker task panic: {}", e)),
         }
     }
 
-    let batches: Vec<RecordBatch> = results.into_iter()
+    #[cfg(feature = "profiling")]
-        .filter_map(|b| b)
+    drop(_p2_guard); // End Phase 2 Span
-        .collect();
+    let duration_phase2 = start_phase2.elapsed();
 
-    if batches.is_empty() {
+
+    // --- PHASE 3: CONCATENATION AND FINALIZATION ---
+    let start_phase3 = Instant::now();
+    #[cfg(feature = "profiling")]
+    let phase3_span = span!(Level::INFO, "Phase3_Concat");
+    #[cfg(feature = "profiling")]
+    let _p3_guard = phase3_span.enter();
+
+    if all_results.is_empty() {
+        println!("UncheckedIO: All workers returned empty batches.");
+        let duration_total = start_total.elapsed();
+        println!("--- UncheckedIO Internal Timing ---");
+        println!("Phase 1 (Setup, Query): {:.2?}", duration_phase1);
+        println!("Phase 2 (I/O, Parsing): {:.2?}", duration_phase2);
+        println!("Phase 3 (Concatenation): {:.2?}", start_phase3.elapsed());
+        println!("Total Wall Time: {:.2?}", duration_total);
         return Ok(RecordBatch::new_empty(arrow_schema));
     }
 
+    // 1. Sort by index to restore original table order
+    all_results.sort_by_key(|(index, _)| *index);
+
+    // 2. Strip indices
+    let batches: Vec<RecordBatch> = all_results.into_iter().map(|(_, batch)| batch).collect();
+
+    // 3. Final Concatenation
     let final_batch = concat_batches(&arrow_schema, &batches)
-        .context("Failed to concatenate parallel batches")?;
+        .context("Failed to stitch final batches")?;
+
+    #[cfg(feature = "profiling")]
+    drop(_p3_guard); // End Phase 3 Span
+    let duration_phase3 = start_phase3.elapsed();
+    let duration_total = start_total.elapsed();
+
+
+    // --- FINAL REPORTING ---
+    println!("--- UncheckedIO Internal Timing ---");
+    println!("Phase 1 (Setup, Query): {:.2?}", duration_phase1);
+    println!("Phase 2 (I/O, Parsing): {:.2?}", duration_phase2);
+    println!("Phase 3 (Concatenation): {:.2?}", duration_phase3);
+    println!("Total Wall Time: {:.2?}", duration_total);
+
 
     Ok(final_batch)
 }
@@ -165,7 +299,8 @@ fn create_null_batch(schema: Arc<Schema>, num_rows: usize) -> Result<RecordBatch
 /// Helper function to build the Arrow Schema from the config
 fn build_arrow_schema(config: &ConnectorConfig) -> Result<Schema> {
     let schema_fields: Vec<Field> = config.schema.iter().map(|col_cfg| {
-        let nullable = col_cfg.column_name == "notes"; // Hack for MVP
+        // FIX: Force all columns to be nullable for safety
+        let nullable = true;
 
         let arrow_type = match col_cfg.arrow_type.as_str() {
             "Int64" => DataType::Int64,
@@ -246,7 +381,7 @@ async fn parse_data_with_schema(
     Ok((rows_processed, record_batch))
 }
 
-// The core streaming parser logic
+// The core streaming parser logic - INSTRUMENTED
 async fn parse_binary_stream_static(
     mut stream: Pin<Box<CopyOutStream>>,
     parser: &mut SchemaParser,
@@ -256,54 +391,78 @@ async fn parse_binary_stream_static(
     let mut is_header_parsed: bool = false;
     let mut rows_processed: usize = 0;
 
-    'stream_loop: while let Some(segment_result) = stream.next().await {
+    // NEW: Refactored loop to visualize Starvation vs Work
-        let segment: Bytes = segment_result.context("Error reading segment from CopyOutStream")?;
+    'stream_loop: loop {
-        buffer.extend_from_slice(&segment);
 
-        if !is_header_parsed {
+        // 1. MEASURE STARVATION (Waiting for Network)
-            if buffer.len() < 19 { continue 'stream_loop; }
+        #[cfg(feature = "profiling")]
-            let mut header_cursor = Cursor::new(&buffer[..]);
+        let wait_span = span!(Level::ERROR, "IO_WAIT_STARVATION");
-            parse_stream_header(&mut header_cursor).context("Failed to parse stream header")?;
+        #[cfg(feature = "profiling")]
-            buffer.advance(19);
+        let guard = wait_span.enter();
-            is_header_parsed = true;
-        }
 
-        'parsing_loop: loop {
+        let next_item = stream.next().await;
-            let mut cursor = Cursor::new(&buffer[..]);
-            let safe_position = cursor.position();
 
-            let col_count = match cursor.read_i16::<BigEndian>() {
+        #[cfg(feature = "profiling")]
-                Ok(count) => count,
+        drop(guard); // Important: Drop guard immediately when data arrives!
-                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { break 'parsing_loop; }
-                Err(e) => return Err(e.into()),
-            };
 
-            if col_count == -1 {
+        match next_item {
-                buffer.advance(2);
+            Some(segment_result) => {
-                break 'stream_loop;
+                // 2. MEASURE WORK (CPU Parsing)
-            }
+                #[cfg(feature = "profiling")]
+                let work_span = span!(Level::INFO, "CPU_Parse_Chunk");
+                #[cfg(feature = "profiling")]
+                let _work_guard = work_span.enter();
 
-            match parse_row_static(&mut cursor, parser, buffer.as_ref()) {
+                let segment: Bytes = segment_result.context("Error reading segment from CopyOutStream")?;
-                Ok(_) => {
+                buffer.extend_from_slice(&segment);
-                    rows_processed += 1;
+
-                    let bytes_consumed = cursor.position();
+                if !is_header_parsed {
-                    buffer.advance(bytes_consumed as usize);
+                    if buffer.len() < 19 { continue 'stream_loop; }
+                    let mut header_cursor = Cursor::new(&buffer[..]);
+                    parse_stream_header(&mut header_cursor).context("Failed to parse stream header")?;
+                    buffer.advance(19);
+                    is_header_parsed = true;
                 }
-                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
-                    cursor.set_position(safe_position);
-                    // Copy remaining bytes back to the buffer for the next chunk
-                    let remaining_slice = &buffer.as_ref()[safe_position as usize..];
-                    let mut leftover_buffer_vec = Vec::new();
-                    leftover_buffer_vec.extend_from_slice(remaining_slice);
-                    buffer.clear();
-                    buffer.extend_from_slice(&leftover_buffer_vec);
 
-                    break 'parsing_loop;
+                'parsing_loop: loop {
-                }
+                    let mut cursor = Cursor::new(&buffer[..]);
-                Err(e) => {
+                    let safe_position = cursor.position();
-                    return Err(e.into());
+
+                    let col_count = match cursor.read_i16::<BigEndian>() {
+                        Ok(count) => count,
+                        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { break 'parsing_loop; }
+                        Err(e) => return Err(e.into()),
+                    };
+
+                    if col_count == -1 {
+                        buffer.advance(2);
+                        break 'stream_loop;
+                    }
+
+                    match parse_row_static(&mut cursor, parser, buffer.as_ref()) {
+                        Ok(_) => {
+                            rows_processed += 1;
+                            let bytes_consumed = cursor.position();
+                            buffer.advance(bytes_consumed as usize);
+                        }
+                        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                            cursor.set_position(safe_position);
+                            // Copy remaining bytes back to the buffer for the next chunk
+                            let remaining_slice = &buffer.as_ref()[safe_position as usize..];
+                            let mut leftover_buffer_vec = Vec::new();
+                            leftover_buffer_vec.extend_from_slice(remaining_slice);
+                            buffer.clear();
+                            buffer.extend_from_slice(&leftover_buffer_vec);
+
+                            break 'parsing_loop;
+                        }
+                        Err(e) => {
+                            return Err(e.into());
+                        }
+                    }
                 }
             }
+            None => break 'stream_loop, // End of stream
         }
     }
 
@@ -409,4 +568,75 @@ fn read_string_field(
     cursor.set_position(end as u64); // Manually advance cursor
 
     Ok(())
+}
+
+// --------------------------------------------------------------------------------
+// --- 3. PROFILER LOGIC (New Feature) ---
+// --------------------------------------------------------------------------------
+
+/// Maps a PostgreSQL internal type name to a standard Arrow Type string for config.yaml.
+fn map_postgres_to_arrow_type(pg_type_name: &str) -> Option<&'static str> {
+    match pg_type_name {
+        "int8" | "bigint" | "serial8" => Some("Int64"),
+        "int4" | "integer" | "serial" => Some("Int32"),
+        "float8" | "double precision" => Some("Float64"),
+        "float4" | "real" => Some("Float32"),
+        "varchar" | "text" | "uuid" => Some("Utf8"),
+        "bool" | "boolean" => Some("Boolean"),
+        "timestamptz" | "timestamp" => Some("Timestamp(Nanosecond, None)"),
+        "date" => Some("Date32"),
+        _ => None, // Returns None for unsupported types (like JSON, arrays, etc.)
+    }
+}
+
+pub async fn run_profiler_logic(config_path: &str) -> Result<String> {
+    // Phase 1: Load config to get connection string and query
+    let config: ConnectorConfig = load_and_validate_config(config_path)
+        .context("Failed to load and validate config for profiling")?;
+
+    // Use a non-COPY query to get metadata
+    let (base_query, _) = config.query.trim().split_once("TO STDOUT (FORMAT binary)")
+        .context("Query in config is malformed or not a COPY command")?;
+
+    // We only need the base query for the metadata query
+    let base_query_inner = base_query.trim().trim_start_matches("COPY (").trim_end_matches(")");
+
+    // Construct the metadata query (limit 0 is fastest)
+    let metadata_query = format!("SELECT * FROM ({}) AS subquery LIMIT 0", base_query_inner);
+
+    // Phase 2: Connect and execute the query
+    let pg_config: PgConfig = PgConfig::from_str(&config.connection_string)?;
+    let (client, connection) = pg_config.connect(NoTls).await
+        .context("Profiler: Failed to connect to PostgreSQL")?;
+
+    tokio::spawn(async move {
+        if let Err(e) = connection.await { eprintln!("Profiler connection error: {}", e); }
+    });
+
+    let statement = client.prepare(&metadata_query).await
+        .context("Profiler: Failed to prepare metadata query")?;
+
+    let mut output = String::from("schema:\n");
+
+    // Phase 3: Inspect the statement's columns for metadata
+    for column in statement.columns() {
+        let pg_type_name = column.type_().name().to_lowercase();
+        let arrow_type = map_postgres_to_arrow_type(&pg_type_name)
+            .unwrap_or("UNKNOWN (Review Manually)");
+
+        let column_entry = format!(
+            "- arrow_type: {}\n  column_name: {}\n",
+            arrow_type,
+            column.name()
+        );
+        output.push_str(&column_entry);
+    }
+
+    // Final instructions for the user
+    output.push_str("\n# NOTE: Paste the 'schema' block above into your config.yaml\n");
+    output.push_str(
+        "# REVIEW any UNKNOWN types. PostgreSQL types: (int8, float8, text, bool, timestamp, date, etc.)\n"
+    );
+
+    Ok(output)
 }