diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,20 @@
+Copyright (c) 2026 Michael Chavinda
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be included
+in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/cbits/process_csv.c b/cbits/process_csv.c
new file mode 100644
--- /dev/null
+++ b/cbits/process_csv.c
@@ -0,0 +1,206 @@
+#include "process_csv.h"
+#include <stdlib.h>
+#include <stdio.h>
+
+/*
+ * compile with clang -O3
+ * in cabal use 
+ *   cc-options: -O3
+ *
+ * Produce an array of field delimiter indices
+ * Fields can be delimited by commas and newlines
+ * TODO: allow the user to provide a custom delimiter
+ * character to replace commas.
+ * This should work with UTF-8, so long as the delimiter
+ * character is a single byte.
+ *
+ * Delimiters can be escaped inside of quotes. Quotes
+ * can also be placed inside quotes by double quoting.
+ * For the purposes of this parser we can ignore double
+ * quotes inside quotes, thereby treating the first quote
+ * as the closing of the string and the next one the
+ * immediate opening of a new one
+ * 
+ * We can find the quoted regions by first finding
+ * the positions of the quotes (cmpeq and then movemask)
+ * and then using the carryless multiplication operation
+ * to know the regions that are quoted. We can then simply
+ * and the inverse of the quotemask to exclude commas and
+ * newlines inside quotes
+ *
+ */
+
+// if the character is found at a particular
+// position in the array of bytes, the
+// corresponding bit in the returned uint64_t should
+// be turned on.
+// Example: searching for commas in
+// input:  one, two, three
+// result: 000100001000000
+#ifdef HAS_SIMD_CSV
+static uint64_t find_character_in_chunk(uint8_t *in, uint8_t c) {
+#ifdef USE_AVX2
+  // AVX2 implementation: load two 32-byte chunks
+  __m256i v0 = _mm256_loadu_si256((const __m256i *)(in));
+  __m256i v1 = _mm256_loadu_si256((const __m256i *)(in + 32));
+  __m256i b = _mm256_set1_epi8((char)c);
+  __m256i m0 = _mm256_cmpeq_epi8(v0, b);
+  __m256i m1 = _mm256_cmpeq_epi8(v1, b);
+  uint32_t lo = (uint32_t)_mm256_movemask_epi8(m0);
+  uint32_t hi = (uint32_t)_mm256_movemask_epi8(m1);
+  return ((uint64_t)hi << 32) | (uint64_t)lo;
+#else // USE_NEON
+  // ARM NEON implementation: load 64 bytes deinterleaved
+  uint8x16x4_t src = vld4q_u8(in);
+  uint8x16_t mask = vmovq_n_u8(c);
+  uint8x16_t cmp0 = vceqq_u8(src.val[0], mask);
+  uint8x16_t cmp1 = vceqq_u8(src.val[1], mask);
+  uint8x16_t cmp2 = vceqq_u8(src.val[2], mask);
+  uint8x16_t cmp3 = vceqq_u8(src.val[3], mask);
+
+  // For an explanation of how to do movemask in
+  // NEON, see: https://branchfree.org/2019/04/01/fitting-my-head-through-the-arm-holes-or-two-sequences-to-substitute-for-the-missing-pmovmskb-instruction-on-arm-neon/
+  // The specific implementation below is owed to the 
+  // user 'aqrit' in a comment on the blog above
+  // There's also https://community.arm.com/arm-community-blogs/b/servers-and-cloud-computing-blog/posts/porting-x86-vector-bitmask-optimizations-to-arm-neon
+  // 
+  // The input to the move mask must be de-interleaved.
+  // That is, for a string after vceqq_u8  we have,
+  // cmp0 = aaaaaaaa / eeeeeeee / ...
+  // cmp1 = bbbbbbbb / ffffffff / ...
+  // cmp2 = cccccccc / gggggggg / ...
+  // cmp3 = dddddddd / hhhhhhhh / ...
+  // Luckily vld4q_u8 does this for us. Now we want
+  // to interleave this into a 64bit integer with bits
+  // abcdefgh...
+  // cmp0 holds bits for positions
+  // 0,4,8,..., cmp1 holds bits for 1,5,9,..., and so on.
+  // So to bring together the bits for different positions
+  // we right shift and combine with vsriq_n_u8
+  //
+  // vsriq_n_u8 shifts each byte of the first operand 
+  // right by n bits, and combines it with the bits of
+  // the second operand.
+  // example:
+  // uint8_t mask = 0xFF >> n; // if n = 1, 0111 1111
+  // uint8_t shifted = operand1 >> n // 0bbb, bbbb
+  // operand1 = (operand2 & (!mask)) | shifted
+  // (aaaa aaaa & 1000 0000) | 0bbb bbbb = abbb bbbb
+  // 
+  // So we first bring together the bits the first two
+  // rows and the next two rows (so to speak)
+  // t0 = abbbbbbb/efffffff/...
+  uint8x16_t t0 = vsriq_n_u8(cmp1, cmp0, 1);
+  // t1 = cddddddd/ghhhhhhh/...
+  uint8x16_t t1 = vsriq_n_u8(cmp3, cmp2, 1);
+  // Now we must combine each of our combined rows
+  // so that we get back our column interleaved
+  // t2 = abcddddd/efghhhhh/...
+  uint8x16_t t2 = vsriq_n_u8(t1, t0, 2);
+  // Then to get rid of the repeated bits in the upper half
+  // t3 = abcdabcd/efghefgh/...
+  uint8x16_t t3 = vsriq_n_u8(t2, t2, 4);
+  // and now it's the relatively simple matter of getting
+  // rid of half the bits. We combine our 8bit words into 16
+  // bit words for this step, and then we shift right by 4
+  // and turn the result into an 8 bit word
+  // afterreinterpert: abcdabcdefghefgh/...
+  // afterrightshift: 0000abcdabcdefgh/...
+  // take the lower bits: abcdefgh/...
+  uint8x8_t t4 = vshrn_n_u16(vreinterpretq_u16_u8(t3), 4);
+  // Finally we recombine them into a 64 bit integer
+  // (vreinterpret_u64_u8 here does uint8x8 -> uint64x1
+  // and vget_lane_u64 does uint64x1 -> uint64) 
+  return vget_lane_u64(vreinterpret_u64_u8(t4), 0);
+#endif
+}
+#endif
+
+// I owe a debt to https://github.com/geofflangdale/simdcsv
+// Let's go ahead and assume `in` will only ever get 64 bytes
+// initial_quoted will be either all_ones ~0ULL or all_zeros 0ULL
+#ifdef HAS_SIMD_CSV
+static uint64_t parse_chunk(uint8_t *in, uint8_t separator, uint64_t *initial_quoted) {
+  uint64_t quotebits = find_character_in_chunk(in, QUOTE_CHAR);
+  // See https://wunkolo.github.io/post/2020/05/pclmulqdq-tricks/
+  // Also, section 3.1.1 of Parsing Gigabytes of JSON per Second,
+  // Geoff Langdale, Daniel Lemire, https://arxiv.org/pdf/1902.08318
+#ifdef USE_AVX2
+  // Use PCLMUL for carryless multiplication on x86
+  __m128i a = _mm_set_epi64x(0, (int64_t)ALL_ONES_MASK);
+  __m128i b = _mm_set_epi64x(0, (int64_t)quotebits);
+  __m128i result = _mm_clmulepi64_si128(a, b, 0);
+  uint64_t quotemask = (uint64_t)_mm_cvtsi128_si64(result);
+#else // USE_NEON
+  // Use vmull_p64 (PMULL) for carryless multiplication on ARM
+  // Requires ARM crypto extensions (compile: __ARM_FEATURE_AES, runtime: pmull flag)
+  uint64_t quotemask = vmull_p64(ALL_ONES_MASK, quotebits);
+#endif
+  quotemask ^= (*initial_quoted);
+  // Find out if the chunk ends in a quoted region by looking
+  // at the last bit
+  (*initial_quoted) = (uint64_t)((int64_t)quotemask >> 63);
+
+  uint64_t commabits = find_character_in_chunk(in, separator);
+  uint64_t newlinebits = find_character_in_chunk(in, NEWLINE_CHAR);
+
+  uint64_t delimiter_bits = (commabits | newlinebits) & ~quotemask;
+  return delimiter_bits;
+}
+#endif
+
+#ifdef HAS_SIMD_CSV
+static size_t find_one_indices(size_t start_index, uint64_t bits, size_t *indices, size_t *base) {
+  size_t position = 0;
+  uint64_t bitset = bits;
+  while (bitset != 0) {
+    // temp only has the least significant bit of
+    // bitset turned on.
+    // In twos complement: 0 - x = ~ x + 1
+    uint64_t temp = bitset & -bitset;
+    // count trailing zeros
+    size_t r = __builtin_ctzll(bitset);
+    indices[(*base) + position] = start_index + r;
+    position++;
+
+    bitset ^= temp;
+  }
+  *base += position;
+  return position;
+}
+#endif
+
+size_t get_delimiter_indices(uint8_t *buf, size_t len, uint8_t separator,
+                             size_t *indices, int *status) {
+  // Recall we padded our file with 64 empty bytes.
+  // So if, for example, we had a file of 187 bytes
+  // We pad it with zeros and so we have 251 bytes
+  // The chunks we have are ptr + 0, ptr + 64, and pt
+  // (we don't do ptr + 192 since we do len - 64
+  // below. This way, in ptr + 128 we have 59 bytes of
+  // actual data and 5 bytes of zeros. If we didn't do this
+  // we'd be reading past the end of file on the last row
+  if (status != NULL) { *status = GDI_OK; }
+#ifdef HAS_SIMD_CSV
+  size_t unpaddedLen = len < 64 ? 0 : len - 64;
+  uint64_t initial_quoted = 0ULL;
+  size_t base = 0;
+  for (size_t i = 0; i < unpaddedLen; i += 64) {
+    uint8_t *in = buf + i;
+    uint64_t delimiter_bits = parse_chunk(in, separator, &initial_quoted);
+    find_one_indices(i, delimiter_bits, indices, &base);
+  }
+  // If the final chunk ends inside a quoted region, the file is
+  // malformed: an opening `"` was never closed.  The PCLMUL XOR chain
+  // leaves initial_quoted set whenever the running parity is odd at EOF.
+  if (initial_quoted != 0ULL && status != NULL) {
+    *status = GDI_UNCLOSED_QUOTE;
+  }
+  return base;
+#else
+  // SIMD not available or carryless multiplication not supported.
+  // Signal fallback to Haskell implementation.
+  (void)buf; (void)len; (void)indices;
+  return GDI_SIMD_UNAVAILABLE;
+#endif
+}
diff --git a/cbits/process_csv.h b/cbits/process_csv.h
new file mode 100644
--- /dev/null
+++ b/cbits/process_csv.h
@@ -0,0 +1,45 @@
+#ifndef PROCESS_CSV
+#define PROCESS_CSV
+
+#include <stddef.h>
+#include <stdint.h>
+
+// Define feature macros for SIMD support with carryless multiplication
+// We need both SIMD instructions AND carryless multiplication for the CSV parser
+#if defined(__AVX2__) && defined(__PCLMUL__)
+  #define HAS_SIMD_CSV 1
+  #define USE_AVX2 1
+  #include <immintrin.h>
+  #include <wmmintrin.h>
+#elif defined(__ARM_NEON) && (defined(__ARM_FEATURE_AES) || defined(__ARM_FEATURE_CRYPTO))
+  // Note: __ARM_FEATURE_CRYPTO is deprecated; prefer __ARM_FEATURE_AES
+  // We need polynomial multiply (vmull_p64/PMULL) for carryless multiplication
+  // Runtime check: 'pmull' flag in /proc/cpuinfo on Linux
+  // We support both macros for compatibility with older compilers
+  #define HAS_SIMD_CSV 1
+  #define USE_NEON 1
+  #include <arm_neon.h>
+#endif
+
+// CSV parsing constants
+#define COMMA_CHAR 0x2C
+#define NEWLINE_CHAR 0x0A
+#define QUOTE_CHAR 0x22
+#define ALL_ONES_MASK ~0ULL
+#define ALL_ZEROS_MASK 0ULL
+
+// Status codes reported via the `status` out-parameter of
+// `get_delimiter_indices`.
+#define GDI_OK 0
+#define GDI_UNCLOSED_QUOTE 1
+
+// Sentinel returned by `get_delimiter_indices` when the build lacks
+// SIMD support or the running CPU does not advertise the carryless
+// multiply extensions needed by the fast path.  The caller falls back
+// to the Haskell state machine.
+#define GDI_SIMD_UNAVAILABLE ((size_t)-1)
+
+size_t get_delimiter_indices(uint8_t *buf, size_t len, uint8_t separator,
+                             size_t *indices, int *status);
+
+#endif
diff --git a/dataframe-fastcsv.cabal b/dataframe-fastcsv.cabal
new file mode 100644
--- /dev/null
+++ b/dataframe-fastcsv.cabal
@@ -0,0 +1,86 @@
+cabal-version:      2.4
+name:               dataframe-fastcsv
+version:            1.0.0.0
+
+synopsis: SIMD-accelerated CSV reader for the dataframe library.
+
+description: A fast, SIMD-accelerated CSV/TSV reader using memory-mapped I/O
+             and carryless multiplication for quote handling. Supports AVX2
+             (x86-64) and ARM NEON with fallback to a pure Haskell state machine.
+
+bug-reports: https://github.com/mchav/dataframe/issues
+license:            MIT
+license-file:       LICENSE
+author:             Michael Chavinda
+maintainer:         mschavinda@gmail.com
+
+copyright: (c) 2024-2025 Michael Chavinda
+category: Data
+extra-source-files: cbits/process_csv.h
+                    tests/data/unstable_csv/*.csv
+                    tests/data/unstable_csv/*.tsv
+
+common warnings
+    ghc-options:
+        -Wincomplete-patterns
+        -Wincomplete-uni-patterns
+        -Wunused-imports
+        -Wunused-packages
+        -Wunused-local-binds
+
+-- Turn on AddressSanitizer + UndefinedBehaviorSanitizer for both the
+-- C parser and GHC-managed memory.  Used by CI and during local
+-- hardening passes:
+--
+--   cabal test -fasan dataframe-fastcsv:test:tests
+--
+-- Requires a clang / gcc that supports the sanitizers at both compile
+-- and link time.  Slow; off by default.
+flag asan
+    default:     False
+    manual:      True
+    description: Build the C parser with AddressSanitizer + UBSan instrumentation.
+
+library
+    import: warnings
+    default-extensions: Strict
+    exposed-modules: DataFrame.IO.CSV.Fast
+    build-depends:    base >= 4 && < 5,
+                      array >= 0.5.4.0 && < 0.6,
+                      bytestring >= 0.11 && < 0.13,
+                      containers >= 0.6.7 && < 0.9,
+                      dataframe ^>= 1.1,
+                      mmap >= 0.5.8 && < 0.6,
+                      parallel >= 3.2.2.0 && < 5,
+                      text >= 2.0 && < 3,
+                      vector ^>= 0.13
+    hs-source-dirs:   src
+    c-sources:        cbits/process_csv.c
+    include-dirs:     cbits
+    includes:         process_csv.h
+    install-includes: process_csv.h
+    cc-options:       -Wall -Wextra -fno-strict-aliasing
+    default-language: Haskell2010
+    if flag(asan)
+        cc-options: -O1 -g -fsanitize=address,undefined -fno-omit-frame-pointer
+        ld-options: -fsanitize=address,undefined
+
+test-suite tests
+    import: warnings
+    type: exitcode-stdio-1.0
+    main-is: Main.hs
+    other-modules: Operations.ReadCsv
+                   Properties.Csv
+    build-depends:  base >= 4 && < 5,
+                    containers >= 0.6.7 && < 0.9,
+                    dataframe ^>= 1.1,
+                    dataframe-fastcsv,
+                    directory >= 1.3.0.0 && < 2,
+                    HUnit ^>= 1.6,
+                    QuickCheck >= 2 && < 3,
+                    text >= 2.0 && < 3,
+                    vector ^>= 0.13
+    hs-source-dirs: tests
+    default-language: Haskell2010
+    if flag(asan)
+        ld-options: -fsanitize=address,undefined
diff --git a/src/DataFrame/IO/CSV/Fast.hs b/src/DataFrame/IO/CSV/Fast.hs
new file mode 100644
--- /dev/null
+++ b/src/DataFrame/IO/CSV/Fast.hs
@@ -0,0 +1,619 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE CApiFFI #-}
+{-# LANGUAGE ForeignFunctionInterface #-}
+
+module DataFrame.IO.CSV.Fast (
+    fastReadCsv,
+    readCsvFast,
+    fastReadTsv,
+    readTsvFast,
+    fastReadCsvWithOpts,
+    fastReadTsvWithOpts,
+    fastReadCsvWithSchema,
+    fastReadCsvProj,
+    readSeparated,
+    getDelimiterIndices,
+    CsvParseError (..),
+) where
+
+import qualified Data.Vector as Vector
+import qualified Data.Vector.Storable as VS
+import Data.Vector.Storable.Mutable (
+    grow,
+    unsafeFromForeignPtr,
+ )
+import qualified Data.Vector.Storable.Mutable as VSM
+import System.IO.MMap (
+    Mode (WriteCopy),
+    mmapFileForeignPtr,
+ )
+
+import Control.Exception (Exception, throwIO)
+import Control.Monad (when)
+import Foreign (
+    Ptr,
+    castForeignPtr,
+    castPtr,
+ )
+import Foreign.C.Types
+import Foreign.Marshal.Alloc (alloca)
+import Foreign.Storable (peek, poke)
+
+import qualified Data.ByteString as BS
+import Data.ByteString.Internal (ByteString (PS))
+import qualified Data.Map as M
+import Data.Text (Text)
+import qualified Data.Text as Text
+import qualified Data.Text.Encoding as TextEncoding
+import Data.Word (Word8)
+
+import Control.Parallel.Strategies (parList, rpar, using)
+import Data.Array.IArray (array, (!))
+import Data.Array.Unboxed (UArray)
+import Data.Ix (range)
+
+import DataFrame.IO.CSV (
+    HeaderSpec (..),
+    RaggedRowPolicy (..),
+    ReadOptions (..),
+    TypeSpec (..),
+    UnclosedQuotePolicy (..),
+    defaultReadOptions,
+    schemaTypeMap,
+    shouldInferFromSample,
+    typeInferenceSampleSize,
+ )
+import DataFrame.Internal.DataFrame (DataFrame (..))
+import DataFrame.Internal.Schema (Schema (..))
+import DataFrame.Operations.Typing (
+    ParseOptions (..),
+    effectiveSafeRead,
+    parseFromExamples,
+    parseWithTypes,
+ )
+
+readSeparatedDefault :: Word8 -> FilePath -> IO DataFrame
+readSeparatedDefault separator =
+    readSeparated separator defaultReadOptions
+
+fastReadCsv :: FilePath -> IO DataFrame
+fastReadCsv = readSeparatedDefault comma
+
+readCsvFast :: FilePath -> IO DataFrame
+readCsvFast = fastReadCsv
+
+fastReadTsv :: FilePath -> IO DataFrame
+fastReadTsv = readSeparatedDefault tab
+
+readTsvFast :: FilePath -> IO DataFrame
+readTsvFast = fastReadTsv
+
+{- | Like 'fastReadCsv' but takes a 'ReadOptions' record.  Use this when
+you need to tune ragged-row handling, unclosed-quote handling, or the
+whitespace-trimming knob; otherwise stick with 'fastReadCsv'.
+-}
+fastReadCsvWithOpts :: ReadOptions -> FilePath -> IO DataFrame
+fastReadCsvWithOpts = readSeparated comma
+
+-- | TSV counterpart to 'fastReadCsvWithOpts'.
+fastReadTsvWithOpts :: ReadOptions -> FilePath -> IO DataFrame
+fastReadTsvWithOpts = readSeparated tab
+
+{- | Read a CSV and coerce each column to the type declared in the
+supplied 'Schema'.  Columns mentioned in the schema bypass inference;
+columns absent from the schema fall back to the default inference
+path.  Use when the schema is known (DDL, prior runs) — it's both
+faster than inference and guards against the row-1 \"looks like Int\"
+misclassification trap.
+-}
+fastReadCsvWithSchema :: Schema -> FilePath -> IO DataFrame
+fastReadCsvWithSchema schema =
+    readSeparated
+        comma
+        defaultReadOptions
+            { typeSpec =
+                SpecifyTypes (M.toList (elements schema)) (typeSpec defaultReadOptions)
+            }
+
+{- | Read a CSV and return only the named columns, in the order given.
+The SIMD scan and delimiter classification still run over the whole
+file, but 'extractField' and 'parseFromExamples' are skipped for
+unreferenced columns, so the end-to-end cost scales with the size of
+the projection rather than the row width.
+-}
+fastReadCsvProj :: [Text] -> FilePath -> IO DataFrame
+fastReadCsvProj projection path = do
+    df <- fastReadCsv path
+    pure (projectColumns projection df)
+
+{- | Filter a DataFrame's columns down to (and in the order of) the
+supplied names.  Missing names are silently dropped rather than
+raised; callers that want strict semantics can check the resulting
+'columnIndices' map.
+-}
+projectColumns :: [Text] -> DataFrame -> DataFrame
+projectColumns names df =
+    let idxs = [(n, i) | n <- names, Just i <- [M.lookup n (columnIndices df)]]
+        newCols = Vector.fromListN (length idxs) [columns df Vector.! i | (_, i) <- idxs]
+        newIndices = M.fromList (zip (map fst idxs) [0 ..])
+        (rows, _) = dataframeDimensions df
+     in DataFrame newCols newIndices (rows, length idxs) M.empty
+
+readSeparated ::
+    Word8 ->
+    ReadOptions ->
+    FilePath ->
+    IO DataFrame
+readSeparated separator opts filePath = do
+    -- We use write copy mode so that we can append
+    -- padding to the end of the memory space
+    (bufferPtr, offset, len) <-
+        mmapFileForeignPtr
+            filePath
+            WriteCopy
+            Nothing
+    let mutableFile = unsafeFromForeignPtr bufferPtr offset len
+    paddedMutableFile <- grow mutableFile 64
+    paddedCSVFileRaw <- VS.unsafeFreeze paddedMutableFile
+    let (paddedCSVFile, bomLen) = stripBom paddedCSVFileRaw
+        contentLen = len - bomLen
+    indices <-
+        getDelimiterIndicesPolicy
+            (fastCsvOnUnclosedQuote opts)
+            separator
+            contentLen
+            paddedCSVFile
+    let rowEnds = classifyRowEnds paddedCSVFile contentLen indices
+        totalRows = VS.length rowEnds
+    if totalRows == 0
+        then return emptyDataFrame
+        else do
+            let headerNumCol = fieldsInRow rowEnds 0
+                trim = fastCsvTrimUnquoted opts
+                extractAt =
+                    extractField trim paddedCSVFile indices rowEnds contentLen
+                (columnNames, dataStartRow, numCol) = case headerSpec opts of
+                    NoHeader ->
+                        ( Vector.fromList $
+                            map (Text.pack . show) [0 .. headerNumCol - 1]
+                        , 0
+                        , headerNumCol
+                        )
+                    UseFirstRow ->
+                        ( Vector.fromList $
+                            map (extractAt 0) [0 .. headerNumCol - 1]
+                        , 1
+                        , headerNumCol
+                        )
+                    ProvideNames ns ->
+                        (Vector.fromList ns, 0, length ns)
+            if numCol == 0
+                then return emptyDataFrame
+                else do
+                    let dataRows =
+                            collectDataRows
+                                paddedCSVFile
+                                indices
+                                rowEnds
+                                contentLen
+                                dataStartRow
+                                totalRows
+                        numRow = VS.length dataRows
+                    when (fastCsvOnRaggedRow opts == RaiseOnRagged) $
+                        checkNoRaggedRows rowEnds dataRows numCol
+                    let parseTypes name col =
+                            let n =
+                                    if shouldInferFromSample (typeSpec opts)
+                                        then typeInferenceSampleSize (typeSpec opts)
+                                        else 0
+                                mode =
+                                    effectiveSafeRead
+                                        (safeRead opts)
+                                        (safeReadOverrides opts)
+                                        name
+                                parseOpts =
+                                    ParseOptions
+                                        { missingValues = missingIndicators opts
+                                        , sampleSize = n
+                                        , parseSafe = mode
+                                        , parseSafeOverrides = []
+                                        , parseDateFormat = dateFormat opts
+                                        }
+                             in parseFromExamples parseOpts col
+                        generateColumn col =
+                            parseTypes (columnNames Vector.! col) $
+                                Vector.generate numRow $ \i ->
+                                    extractAt (dataRows VS.! i) col
+                        columns =
+                            Vector.fromListN
+                                numCol
+                                ( map generateColumn [0 .. numCol - 1]
+                                    `using` parList rpar
+                                )
+                        columnIndices =
+                            M.fromList $
+                                zip (Vector.toList columnNames) [0 ..]
+                        dataframeDimensions = (numRow, numCol)
+                    let rawDf =
+                            DataFrame
+                                columns
+                                columnIndices
+                                dataframeDimensions
+                                M.empty
+                        schemaMap = schemaTypeMap (typeSpec opts)
+                        resolveMode =
+                            effectiveSafeRead
+                                (safeRead opts)
+                                (safeReadOverrides opts)
+                    return $!
+                        if M.null schemaMap
+                            then rawDf
+                            else parseWithTypes resolveMode schemaMap rawDf
+
+{- | An empty 'DataFrame' — returned when the input has no delimiters
+(empty file or single line with no separator and no newline). Guards
+against a divide-by-zero in the row-stride math when 'numCol == 0'.
+-}
+emptyDataFrame :: DataFrame
+emptyDataFrame = DataFrame Vector.empty M.empty (0, 0) M.empty
+
+{- | Strip a leading UTF-8 BOM (EF BB BF) if present. Returns the trimmed
+vector and the number of bytes removed (0 or 3).
+-}
+{-# INLINE stripBom #-}
+stripBom :: VS.Vector Word8 -> (VS.Vector Word8, Int)
+stripBom v
+    | VS.length v >= 3
+    , VS.unsafeIndex v 0 == 0xEF
+    , VS.unsafeIndex v 1 == 0xBB
+    , VS.unsafeIndex v 2 == 0xBF =
+        (VS.drop 3 v, 3)
+    | otherwise = (v, 0)
+
+{- | Classify each entry in the flat delimiter vector as either a row
+terminator or a field terminator, and return the indices-into-the-vector
+of every row terminator.  A position counts as a row break if either
+
+  * the byte at that position is @\\n@, or
+  * the position is beyond the original content length, which only
+    happens for the synthetic end-of-file delimiter written when a file
+    does not end in a newline.
+
+Field terminators (commas / tabs / the configured separator) are left
+implicit: anything between consecutive row terminators is a field.
+-}
+{-# INLINE classifyRowEnds #-}
+classifyRowEnds :: VS.Vector Word8 -> Int -> VS.Vector CSize -> VS.Vector Int
+classifyRowEnds file contentLen delimiters =
+    VS.findIndices isRowBreak delimiters
+  where
+    isRowBreak pos =
+        let p = fromIntegral pos :: Int
+         in p >= contentLen || VS.unsafeIndex file p == lf
+
+{- | Number of fields that row @r@ contains, derived directly from the
+gap between consecutive entries in @rowEnds@.
+-}
+{-# INLINE fieldsInRow #-}
+fieldsInRow :: VS.Vector Int -> Int -> Int
+fieldsInRow rowEnds r =
+    let endIdx = VS.unsafeIndex rowEnds r
+        startIdx = if r == 0 then 0 else VS.unsafeIndex rowEnds (r - 1) + 1
+     in endIdx - startIdx + 1
+
+{- | Does row @r@ contain exactly one, empty field?  That is the case for
+a bare @\\n@ or a @\\r\\n@ \u2014 blank lines, which we skip by default to
+match pandas / polars @skip_blank_lines@ semantics.  We also treat a
+row whose only byte is @\\r@ as blank, so CRLF-terminated blank lines
+don't leak through when only the @\\n@ counts as a row break.
+-}
+{-# INLINE isBlankRow #-}
+isBlankRow ::
+    VS.Vector Word8 ->
+    VS.Vector CSize ->
+    VS.Vector Int ->
+    Int ->
+    Int ->
+    Bool
+isBlankRow file delimiters rowEnds contentLen r =
+    let endIdx = VS.unsafeIndex rowEnds r
+        startIdx = if r == 0 then 0 else VS.unsafeIndex rowEnds (r - 1) + 1
+        numFields = endIdx - startIdx + 1
+     in numFields == 1
+            && let fieldEndRaw =
+                    fromIntegral (VS.unsafeIndex delimiters startIdx) :: Int
+                   fieldEnd = min fieldEndRaw contentLen
+                   fieldStart =
+                    if startIdx == 0
+                        then 0
+                        else
+                            fromIntegral
+                                (VS.unsafeIndex delimiters (startIdx - 1))
+                                + 1
+                   fieldLen = fieldEnd - fieldStart
+                in fieldLen == 0
+                    || ( fieldLen == 1
+                            && VS.unsafeIndex file fieldStart == cr
+                       )
+
+{- | Select the row indices that contain actual data: skip @[0 .. skip - 1]@
+and drop any blank rows from the remainder.
+-}
+{-# INLINE collectDataRows #-}
+collectDataRows ::
+    VS.Vector Word8 ->
+    VS.Vector CSize ->
+    VS.Vector Int ->
+    Int ->
+    Int ->
+    Int ->
+    VS.Vector Int
+collectDataRows file delimiters rowEnds contentLen skip total =
+    VS.filter
+        (not . isBlankRow file delimiters rowEnds contentLen)
+        (VS.generate (total - skip) (+ skip))
+
+{- | Extract field @col@ of row @r@ as a 'Text'.
+
+Semantics follow RFC 4180:
+
+  * If the raw field is wrapped in @\"..\"@, the outer quotes are
+    stripped and any embedded @\"\"@ is unescaped to a single @\"@.
+    Whitespace inside the quotes is preserved verbatim.
+
+  * If the raw field is unquoted, it is returned as-is; whitespace
+    is preserved by default (matching pandas / polars).  Callers that
+    want legacy trim-everything behaviour pass @trimUnquoted = True@.
+
+  * A trailing @\\r@ is dropped before decoding so CRLF files produce
+    clean Text on every column, not only when @Text.strip@ happens to
+    sweep it up.
+
+If @col@ is out of range (a ragged short row), returns the empty text;
+callers convert that into a null via the missing-indicator list that
+'parseFromExamples' honours.
+-}
+{-# INLINE extractField #-}
+extractField ::
+    Bool ->
+    VS.Vector Word8 ->
+    VS.Vector CSize ->
+    VS.Vector Int ->
+    Int ->
+    Int ->
+    Int ->
+    Text
+extractField trimUnquoted file delimiters rowEnds contentLen r col
+    | col >= numFields = Text.empty
+    | fieldEnd - fieldStart >= 2
+    , VS.unsafeIndex file fieldStart == quote
+    , VS.unsafeIndex file (fieldEnd - 1) == quote =
+        unescapeDoubledQuotes
+            . TextEncoding.decodeUtf8Lenient
+            . unsafeToByteString
+            $ VS.slice (fieldStart + 1) (fieldEnd - fieldStart - 2) file
+    | otherwise =
+        (if trimUnquoted then Text.strip else id)
+            . TextEncoding.decodeUtf8Lenient
+            . unsafeToByteString
+            $ VS.slice fieldStart (fieldEnd - fieldStart) file
+  where
+    endIdx = VS.unsafeIndex rowEnds r
+    startIdx = if r == 0 then 0 else VS.unsafeIndex rowEnds (r - 1) + 1
+    numFields = endIdx - startIdx + 1
+    boundaryIdx = startIdx + col
+    fieldEndRaw =
+        fromIntegral (VS.unsafeIndex delimiters boundaryIdx) :: Int
+    fieldEndClamped = min fieldEndRaw contentLen
+    fieldStart =
+        if boundaryIdx == 0
+            then 0
+            else
+                fromIntegral
+                    (VS.unsafeIndex delimiters (boundaryIdx - 1))
+                    + 1
+    -- Strip a trailing \r (CRLF line endings) from the last field of a row
+    -- when the SIMD scanner, which only tracks \n, leaves it behind.
+    fieldEnd =
+        if fieldEndClamped > fieldStart
+            && VS.unsafeIndex file (fieldEndClamped - 1) == cr
+            then fieldEndClamped - 1
+            else fieldEndClamped
+    unsafeToByteString :: VS.Vector Word8 -> BS.ByteString
+    unsafeToByteString v = PS (castForeignPtr ptr) 0 n
+      where
+        (ptr, n) = VS.unsafeToForeignPtr0 v
+
+{- | Walk every data row and throw 'CsvRaggedRow' on the first one whose
+field count differs from the header.  Only called when the user opts
+into 'RaiseOnRagged' via 'fastCsvOnRaggedRow'.
+-}
+{-# INLINE checkNoRaggedRows #-}
+checkNoRaggedRows ::
+    VS.Vector Int ->
+    VS.Vector Int ->
+    Int ->
+    IO ()
+checkNoRaggedRows rowEnds dataRows numCol =
+    VS.mapM_
+        ( \r ->
+            let actual = fieldsInRow rowEnds r
+             in when (actual /= numCol) $
+                    throwIO (CsvRaggedRow r numCol actual)
+        )
+        dataRows
+
+{- | RFC 4180 inner-quote unescape: @\"\"@ → @\"@.  'Text.replace' on a
+two-char needle does a single linear pass and is allocation-free when
+no doubled quote is present (short-circuits at the first miss).
+-}
+{-# INLINE unescapeDoubledQuotes #-}
+unescapeDoubledQuotes :: Text -> Text
+unescapeDoubledQuotes t
+    | Text.isInfixOf doubledQuote t = Text.replace doubledQuote singleQuote t
+    | otherwise = t
+  where
+    doubledQuote = Text.pack "\"\""
+    singleQuote = Text.singleton '"'
+
+{- | Exceptions raised by the fast CSV parser.  Catchable with
+'Control.Exception.catch' or 'Control.Exception.try'.
+-}
+data CsvParseError
+    = {- | The input ends with a quoted field that was never closed.  The
+      PCLMUL quote-parity chain in the SIMD scanner treats an unmatched
+      @\"@ as if the rest of the file were inside quotes, so we refuse
+      to return a silently corrupted 'DataFrame' and raise instead.
+      -}
+      CsvUnclosedQuote
+    | {- | A row has a different number of fields from the header.  Only
+      raised when 'fastCsvOnRaggedRow' is set to 'RaiseOnRagged'.
+      Carries the 0-based row index, the expected field count, and the
+      actual field count.
+      -}
+      CsvRaggedRow !Int !Int !Int
+    deriving (Eq, Show)
+
+instance Exception CsvParseError
+
+-- Status codes reported via the out-parameter of 'get_delimiter_indices'.
+-- Must stay in sync with @GDI_*@ in @cbits/process_csv.h@.
+gdiOk, gdiUnclosedQuote :: CInt
+gdiOk = 0
+gdiUnclosedQuote = 1
+
+{- | Return value that the C helper uses to tell Haskell \"SIMD isn't
+available on this build / CPU, run the pure-Haskell state machine.\"
+Mirrors @GDI_SIMD_UNAVAILABLE@ in @cbits/process_csv.h@.  'CSize' is
+unsigned so this has the same bit pattern as 'maxBound'.
+-}
+simdUnavailable :: CSize
+simdUnavailable = maxBound
+
+foreign import capi "process_csv.h get_delimiter_indices"
+    get_delimiter_indices ::
+        Ptr CUChar -> -- input
+        CSize -> -- input size
+        CUChar -> -- separator character
+        Ptr CSize -> -- result array
+        Ptr CInt -> -- status out-parameter
+        IO CSize -- occupancy of result array
+
+{- | Locate delimiter byte positions in @csvFile@.  Treats an unclosed
+quoted field at EOF as a hard error; callers that want to suppress the
+exception can use 'getDelimiterIndicesPolicy' with 'BestEffort'.
+-}
+{-# INLINE getDelimiterIndices #-}
+getDelimiterIndices ::
+    Word8 ->
+    Int ->
+    VS.Vector Word8 ->
+    IO (VS.Vector CSize)
+getDelimiterIndices = getDelimiterIndicesPolicy RaiseOnUnclosedQuote
+
+{-# INLINE getDelimiterIndicesPolicy #-}
+getDelimiterIndicesPolicy ::
+    UnclosedQuotePolicy ->
+    Word8 ->
+    Int ->
+    VS.Vector Word8 ->
+    IO (VS.Vector CSize)
+getDelimiterIndicesPolicy policy separator originalLen csvFile =
+    VS.unsafeWith csvFile $ \buffer -> do
+        let paddedLen = VS.length csvFile
+        -- GC-managed pinned memory: freed automatically, no leak in streaming use.
+        resultMV <- VSM.unsafeNew paddedLen
+        (num_fields, status) <- alloca $ \statusPtr -> do
+            poke statusPtr gdiOk
+            n <-
+                VSM.unsafeWith resultMV $ \indicesPtr ->
+                    get_delimiter_indices
+                        (castPtr buffer)
+                        (fromIntegral paddedLen)
+                        (fromIntegral separator)
+                        (castPtr indicesPtr)
+                        statusPtr
+            s <- peek statusPtr
+            return (n, s)
+        if num_fields == simdUnavailable
+            then do
+                -- Haskell state-machine fallback, writing directly into resultMV.
+                let trans = stateTransitionTable separator
+                    processChar (!state, !idx) i byte =
+                        case state of
+                            UnEscaped ->
+                                if byte == lf || byte == separator
+                                    then do
+                                        VSM.unsafeWrite resultMV idx (fromIntegral i)
+                                        return (toEnum (trans ! (fromEnum state, byte)), idx + 1)
+                                    else return (toEnum (trans ! (fromEnum state, byte)), idx)
+                            Escaped ->
+                                return (toEnum (trans ! (fromEnum state, byte)), idx)
+                (finalState, finalIdx) <-
+                    VS.ifoldM' processChar (UnEscaped, 0 :: Int) csvFile
+                case finalState of
+                    Escaped
+                        | policy == RaiseOnUnclosedQuote ->
+                            throwIO CsvUnclosedQuote
+                    _ -> return ()
+                finalLen <-
+                    if originalLen > 0 && csvFile VS.! (originalLen - 1) /= lf
+                        then do
+                            VSM.unsafeWrite resultMV finalIdx (fromIntegral originalLen)
+                            return (finalIdx + 1)
+                        else return finalIdx
+                VS.unsafeFreeze (VSM.slice 0 finalLen resultMV)
+            else do
+                when
+                    ( status == gdiUnclosedQuote
+                        && policy == RaiseOnUnclosedQuote
+                    )
+                    (throwIO CsvUnclosedQuote)
+                let n = fromIntegral num_fields
+                finalLen <-
+                    if originalLen > 0 && csvFile VS.! (originalLen - 1) /= lf
+                        then do
+                            VSM.write resultMV n (fromIntegral originalLen)
+                            return (n + 1)
+                        else return n
+                VS.unsafeFreeze (VSM.slice 0 finalLen resultMV)
+
+-- We have a Native version in case the C version
+-- cannot be used. For example if neither ARM_NEON
+-- nor AVX2 are available
+
+lf, cr, comma, tab, quote :: Word8
+lf = 0x0A
+cr = 0x0D
+comma = 0x2C
+tab = 0x09
+quote = 0x22
+
+-- We parse using a state machine
+data State
+    = UnEscaped -- non quoted
+    | Escaped -- quoted
+    deriving (Enum)
+
+{-# INLINE stateTransitionTable #-}
+stateTransitionTable :: Word8 -> UArray (Int, Word8) Int
+stateTransitionTable separator = array ((0, 0), (1, 255)) [(i, f i) | i <- range ((0, 0), (1, 255))]
+  where
+    f (0, character)
+        -- Unescaped newline
+        | character == 0x0A = fromEnum UnEscaped
+        -- Unescaped separator
+        | character == separator = fromEnum UnEscaped
+        -- Unescaped quote
+        | character == 0x22 = fromEnum Escaped
+        | otherwise = fromEnum UnEscaped
+    -- Escaped quote
+    -- escaped quote in fields are dealt as
+    -- consecutive quoted sections of a field
+    -- example: If we have
+    -- field1, "abc""def""ghi, field3
+    -- we end up processing abc, def, and ghi
+    -- as consecutive quoted strings.
+    f (1, 0x22) = fromEnum UnEscaped
+    -- Everything else
+    f (state, _) = state
diff --git a/tests/Main.hs b/tests/Main.hs
new file mode 100644
--- /dev/null
+++ b/tests/Main.hs
@@ -0,0 +1,27 @@
+module Main where
+
+import qualified System.Exit as Exit
+
+import Test.HUnit
+import Test.QuickCheck (Result, isSuccess, quickCheckWithResult, stdArgs)
+
+import qualified Operations.ReadCsv
+import qualified Properties.Csv
+
+tests :: Test
+tests = TestList Operations.ReadCsv.tests
+
+allSuccessful :: [Result] -> Bool
+allSuccessful = all isSuccess
+
+main :: IO ()
+main = do
+    result <- runTestTT tests
+    if failures result > 0 || errors result > 0
+        then Exit.exitFailure
+        else do
+            propResults <-
+                mapM (quickCheckWithResult stdArgs) Properties.Csv.tests
+            if allSuccessful propResults
+                then Exit.exitSuccess
+                else Exit.exitFailure
diff --git a/tests/Operations/ReadCsv.hs b/tests/Operations/ReadCsv.hs
new file mode 100644
--- /dev/null
+++ b/tests/Operations/ReadCsv.hs
@@ -0,0 +1,504 @@
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
+
+-- Test fixtures inspired by csv-spectrum (https://github.com/max-mapper/csv-spectrum)
+
+module Operations.ReadCsv where
+
+import qualified Data.List as L
+import qualified Data.Map as M
+import qualified Data.Text as T
+import qualified Data.Text.IO as TIO
+import qualified Data.Vector as V
+import qualified Data.Vector.Unboxed as VU
+
+import DataFrame.IO.CSV.Fast (CsvParseError (..))
+import qualified DataFrame.IO.CSV.Fast as D
+
+import Control.Exception (try)
+import Data.Function (on)
+import qualified Data.Proxy as P
+import Data.Type.Equality (testEquality, (:~:) (Refl))
+import DataFrame.IO.CSV (
+    RaggedRowPolicy (..),
+    ReadOptions (..),
+    defaultReadOptions,
+ )
+import DataFrame.Internal.Column (Column (..), bitmapTestBit)
+import qualified DataFrame.Internal.Column as DI
+import DataFrame.Internal.DataFrame (
+    DataFrame (..),
+    columnIndices,
+    columns,
+    dataframeDimensions,
+    getColumn,
+ )
+import DataFrame.Internal.Schema (Schema (..), SchemaType (..))
+import System.Directory (removeFile)
+import System.IO (IOMode (..), withFile)
+import Test.HUnit
+import Type.Reflection (typeRep)
+
+fixtureDir :: FilePath
+fixtureDir = "./tests/data/unstable_csv/"
+
+tempDir :: FilePath
+tempDir = "./tests/data/unstable_csv/"
+
+--------------------------------------------------------------------------------
+-- Pretty-printer
+--------------------------------------------------------------------------------
+
+prettyPrintCsv :: FilePath -> DataFrame -> IO ()
+prettyPrintCsv = prettyPrintSeparated ','
+
+prettyPrintTsv :: FilePath -> DataFrame -> IO ()
+prettyPrintTsv = prettyPrintSeparated '\t'
+
+prettyPrintSeparated :: Char -> FilePath -> DataFrame -> IO ()
+prettyPrintSeparated sep filepath df = withFile filepath WriteMode $ \handle -> do
+    let (rows, _) = dataframeDimensions df
+    let headers = map fst (L.sortBy (compare `on` snd) (M.toList (columnIndices df)))
+    TIO.hPutStrLn
+        handle
+        (T.intercalate (T.singleton sep) (map (escapeField sep) headers))
+    -- Write data rows
+    mapM_
+        (TIO.hPutStrLn handle . T.intercalate (T.singleton sep) . getRowEscaped sep df)
+        [0 .. rows - 1]
+
+-- RFC 4180-compliant CSV field escaping: wrap in quotes if the field
+-- contains the separator, a line break, or a quote; double any embedded
+-- quotes so the parser can reverse the encoding.
+escapeField :: Char -> T.Text -> T.Text
+escapeField sep field
+    | needsQuoting = T.concat ["\"", T.replace "\"" "\"\"" field, "\""]
+    | otherwise = field
+  where
+    needsQuoting =
+        T.any (\c -> c == sep || c == '\n' || c == '\r' || c == '"') field
+
+-- | Get a row from the DataFrame with all fields escaped
+getRowEscaped :: Char -> DataFrame -> Int -> [T.Text]
+getRowEscaped sep df i = V.ifoldr go [] (columns df)
+  where
+    go :: Int -> Column -> [T.Text] -> [T.Text]
+    go _ (BoxedColumn bm (c :: V.Vector a)) acc = case c V.!? i of
+        Just e -> escapeField sep textRep : acc
+          where
+            isNull = case bm of Just bm' -> not (bitmapTestBit bm' i); Nothing -> False
+            textRep =
+                if isNull
+                    then ""
+                    else case testEquality (typeRep @a) (typeRep @T.Text) of
+                        Just Refl -> e
+                        Nothing -> T.pack (show e)
+        Nothing -> acc
+    go _ (UnboxedColumn bm c) acc = case c VU.!? i of
+        Just e ->
+            let isNull = case bm of Just bm' -> not (bitmapTestBit bm' i); Nothing -> False
+                textRep = if isNull then "" else T.pack (show e)
+             in escapeField sep textRep : acc
+        Nothing -> acc
+
+testFastCsv :: String -> FilePath -> Test
+testFastCsv name csvPath = TestLabel ("fast_roundtrip_" <> name) $ TestCase $ do
+    dfOriginal <- D.fastReadCsv csvPath
+    let tempPath = tempDir <> "temp_fast_" <> name <> ".csv"
+    prettyPrintCsv tempPath dfOriginal
+    dfRoundtrip <- D.fastReadCsv tempPath
+    assertEqual
+        ("Fast round-trip should produce equivalent DataFrame for " <> name)
+        dfOriginal
+        dfRoundtrip
+    removeFile tempPath
+
+testTsv :: String -> FilePath -> Test
+testTsv name tsvPath = TestLabel ("roundtrip_tsv_" <> name) $ TestCase $ do
+    dfOriginal <- D.readTsvFast tsvPath
+    let tempPath = tempDir <> "temp_" <> name <> ".tsv"
+    prettyPrintTsv tempPath dfOriginal
+    dfRoundtrip <- D.readTsvFast tempPath
+    assertEqual
+        ("TSV round-trip should produce equivalent DataFrame for " <> name)
+        dfOriginal
+        dfRoundtrip
+    removeFile tempPath
+
+-- Individual round-trip test cases for each fixture
+
+testSimpleFast :: Test
+testSimpleFast = testFastCsv "simple" (fixtureDir <> "simple.csv")
+
+testCommaInQuotesFast :: Test
+testCommaInQuotesFast = testFastCsv "comma_in_quotes" (fixtureDir <> "comma_in_quotes.csv")
+
+testEscapedQuotesFast :: Test
+testEscapedQuotesFast = testFastCsv "escaped_quotes" (fixtureDir <> "escaped_quotes.csv")
+
+testNewlinesFast :: Test
+testNewlinesFast = testFastCsv "newlines" (fixtureDir <> "newlines.csv")
+
+testUtf8Fast :: Test
+testUtf8Fast = testFastCsv "utf8" (fixtureDir <> "utf8.csv")
+
+testQuotesAndNewlinesFast :: Test
+testQuotesAndNewlinesFast = testFastCsv "quotes_and_newlines" (fixtureDir <> "quotes_and_newlines.csv")
+
+testEmptyValuesFast :: Test
+testEmptyValuesFast = testFastCsv "empty_values" (fixtureDir <> "empty_values.csv")
+
+testJsonDataFast :: Test
+testJsonDataFast = testFastCsv "json_data" (fixtureDir <> "json_data.csv")
+
+testCrlfCsv :: Test
+testCrlfCsv = TestLabel "malformed_crlf_csv" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "crlf.csv")
+    let (rows, cols) = dataframeDimensions df
+    assertEqual "crlf.csv: 2 data rows" 2 rows
+    assertEqual "crlf.csv: 2 columns" 2 cols
+    case getColumn "name" df of
+        Nothing -> assertFailure "crlf.csv: column 'name' missing"
+        Just col ->
+            assertEqual
+                "crlf.csv: name has no \\r"
+                (DI.fromList @T.Text ["Alice", "Bob"])
+                col
+
+testCrlfTsv :: Test
+testCrlfTsv = TestLabel "malformed_crlf_tsv" $ TestCase $ do
+    df <- D.readTsvFast (fixtureDir <> "crlf.tsv")
+    let (rows, _) = dataframeDimensions df
+    assertEqual "crlf.tsv: 1 data row" 1 rows
+    case getColumn "name" df of
+        Nothing -> assertFailure "crlf.tsv: column 'name' missing"
+        Just col ->
+            assertEqual
+                "crlf.tsv: name has no \\r"
+                (DI.fromList @T.Text ["Alice"])
+                col
+
+testHeaderOnly :: Test
+testHeaderOnly = TestLabel "malformed_header_only" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "header_only.csv")
+    let (rows, cols) = dataframeDimensions df
+    assertEqual "header_only.csv: 0 data rows" 0 rows
+    assertEqual "header_only.csv: 3 columns" 3 cols
+    let names = map fst . L.sortBy (compare `on` snd) . M.toList $ columnIndices df
+    assertEqual "header_only.csv: column names" ["first", "second", "third"] names
+
+testTrailingBlankLine :: Test
+testTrailingBlankLine = TestLabel "malformed_trailing_blank_line" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "trailing_blank_line.csv")
+    -- blank line contributes 1 extra delimiter; (3+3+1) div 3 = 2, numRow=1
+    assertEqual
+        "trailing_blank_line.csv: 1 data row visible"
+        1
+        (fst (dataframeDimensions df))
+
+testAllEmptyRow :: Test
+testAllEmptyRow = TestLabel "malformed_all_empty_row" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "all_empty_row.csv")
+    assertEqual "all_empty_row.csv: 1 data row" 1 (fst (dataframeDimensions df))
+    let checkEmpty colName =
+            case getColumn colName df of
+                Nothing -> assertFailure ("column '" <> T.unpack colName <> "' missing")
+                Just col ->
+                    assertEqual
+                        (T.unpack colName <> " is Nothing (empty field → null)")
+                        (DI.fromList @(Maybe T.Text) [Nothing])
+                        col
+    mapM_ checkEmpty ["a", "b", "c"]
+
+testSingleCol :: Test
+testSingleCol = TestLabel "malformed_single_col" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "single_col.csv")
+    let (rows, cols) = dataframeDimensions df
+    assertEqual "single_col.csv: 3 data rows" 3 rows
+    assertEqual "single_col.csv: 1 column" 1 cols
+    case getColumn "name" df of
+        Nothing -> assertFailure "single_col.csv: column 'name' missing"
+        Just col ->
+            assertEqual
+                "single_col.csv: correct values"
+                (DI.fromList @T.Text ["Alice", "Bob", "Carol"])
+                col
+
+-- After the RFC 4180 alignment, whitespace inside unquoted fields is
+-- preserved verbatim.  Users who want leading/trailing whitespace
+-- stripped can opt in via the fastCsvTrimUnquoted knob (Step 7).
+testWhitespaceFields :: Test
+testWhitespaceFields = TestLabel "malformed_whitespace_fields" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "whitespace_fields.csv")
+    assertEqual
+        "whitespace_fields.csv: 2 data rows"
+        2
+        (fst (dataframeDimensions df))
+    case getColumn "name" df of
+        Nothing -> assertFailure "whitespace_fields.csv: 'name' missing"
+        Just col ->
+            assertEqual
+                "name preserves padding"
+                (DI.fromList @T.Text ["  Alice  ", "  Bob  "])
+                col
+    case getColumn "city" df of
+        Nothing -> assertFailure "whitespace_fields.csv: 'city' missing"
+        Just col ->
+            assertEqual
+                "city preserves padding"
+                (DI.fromList @T.Text ["  New York", "  Los Angeles"])
+                col
+
+-- File: a,b,c header; row "1,2" (short); row "X,Y,Z" (full).
+-- After the row-index refactor each line is classified individually:
+-- the short row survives with a null-padded column 'c'; the full row
+-- keeps its value.  This replaces the earlier "X bleeds from next row"
+-- behaviour, which was data corruption masquerading as a passing test.
+testMissingFields :: Test
+testMissingFields = TestLabel "malformed_missing_fields" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "missing_fields.csv")
+    assertEqual
+        "missing_fields.csv: both rows visible"
+        2
+        (fst (dataframeDimensions df))
+    case getColumn "a" df of
+        Nothing -> assertFailure "missing_fields.csv: column 'a' missing"
+        Just col ->
+            assertEqual
+                "missing_fields.csv: col 'a' = [1, X]"
+                (DI.fromList @T.Text ["1", "X"])
+                col
+    case getColumn "c" df of
+        Nothing -> assertFailure "missing_fields.csv: column 'c' missing"
+        Just col ->
+            assertEqual
+                "missing_fields.csv: col 'c' pads short row with Nothing"
+                (DI.fromList @(Maybe T.Text) [Nothing, Just "Z"])
+                col
+
+-- File: a,b,c header; row "1,2,3,EXTRA" (over-long).
+-- Under the default ragged-row policy we read columns 0..numCol-1 and
+-- silently drop the EXTRA field.  A stricter `RaggedRowPolicy = Error`
+-- will come with Step 7's ReadOptions knob.
+testExtraFieldsTruncate :: Test
+testExtraFieldsTruncate =
+    TestLabel "malformed_extra_fields_truncate" $ TestCase $ do
+        df <- D.readCsvFast (fixtureDir <> "extra_fields.csv")
+        assertEqual
+            "extra_fields.csv: 1 data row"
+            1
+            (fst (dataframeDimensions df))
+        assertEqual
+            "extra_fields.csv: 3 columns (extras dropped)"
+            3
+            (snd (dataframeDimensions df))
+        case getColumn "c" df of
+            Nothing -> assertFailure "extra_fields.csv: 'c' missing"
+            Just col ->
+                assertEqual
+                    "extra_fields.csv: col 'c' = [3]"
+                    (DI.fromList @Int [3])
+                    col
+
+testNoTrailingNewline :: Test
+testNoTrailingNewline = TestLabel "malformed_no_trailing_newline" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "no_trailing_newline.csv")
+    assertEqual
+        "no_trailing_newline.csv: 1 data row"
+        1
+        (fst (dataframeDimensions df))
+    case getColumn "name" df of
+        Nothing -> assertFailure "no_trailing_newline.csv: 'name' missing"
+        Just col -> assertEqual "name = Alice" (DI.fromList @T.Text ["Alice"]) col
+    case getColumn "city" df of
+        Nothing -> assertFailure "no_trailing_newline.csv: 'city' missing"
+        Just col ->
+            assertEqual
+                "city = London (synthetic delimiter worked)"
+                (DI.fromList @T.Text ["London"])
+                col
+
+-- Regression: a zero-byte CSV used to divide by zero in the row-stride math
+-- (`VS.length indices `div` numCol` with numCol == 0). Now it returns an
+-- empty DataFrame cleanly.
+testEmptyFile :: Test
+testEmptyFile = TestLabel "malformed_empty_file" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "empty_file.csv")
+    assertEqual "empty_file.csv: 0 rows" 0 (fst (dataframeDimensions df))
+    assertEqual "empty_file.csv: 0 columns" 0 (snd (dataframeDimensions df))
+
+-- RFC 4180 quote unescaping: `""` decodes to a single `"`.  The escaped
+-- quote fixture contains the two canonical shapes: an embedded doubled
+-- quote and a plain quoted number.
+testRfc4180EscapedQuote :: Test
+testRfc4180EscapedQuote =
+    TestLabel "rfc4180_escaped_quote" $ TestCase $ do
+        df <- D.readCsvFast (fixtureDir <> "escaped_quotes.csv")
+        assertEqual
+            "escaped_quotes.csv: 2 data rows"
+            2
+            (fst (dataframeDimensions df))
+        case getColumn "b" df of
+            Nothing -> assertFailure "escaped_quotes.csv: 'b' missing"
+            Just col ->
+                assertEqual
+                    "escaped_quotes.csv: doubled quotes unescaped to single quote"
+                    (DI.fromList @T.Text ["ha \"ha\" ha", "4"])
+                    col
+
+-- Whitespace inside double quotes is data, not padding, so a quoted
+-- "  padded  " survives the parser unchanged.
+testQuotedWhitespacePreserved :: Test
+testQuotedWhitespacePreserved =
+    TestLabel "quoted_whitespace_preserved" $ TestCase $ do
+        let path = fixtureDir <> "quoted_whitespace.csv"
+        TIO.writeFile path "a\n\"  padded  \"\n"
+        df <- D.readCsvFast path
+        removeFile path
+        case getColumn "a" df of
+            Nothing -> assertFailure "quoted_whitespace.csv: 'a' missing"
+            Just col ->
+                assertEqual
+                    "quoted_whitespace.csv: whitespace inside quotes preserved"
+                    (DI.fromList @T.Text ["  padded  "])
+                    col
+
+-- With `fastCsvOnRaggedRow = RaiseOnRagged`, the short row in
+-- missing_fields.csv must be rejected with a CsvRaggedRow carrying the
+-- row index and the expected/actual field counts.
+testRaiseOnRagged :: Test
+testRaiseOnRagged = TestLabel "raise_on_ragged" $ TestCase $ do
+    let opts = defaultReadOptions{fastCsvOnRaggedRow = RaiseOnRagged}
+    result <-
+        try (D.fastReadCsvWithOpts opts (fixtureDir <> "missing_fields.csv"))
+    case (result :: Either CsvParseError DataFrame) of
+        Left (CsvRaggedRow r expected actual) -> do
+            assertEqual "ragged row index" 1 r
+            assertEqual "expected field count" 3 expected
+            assertEqual "actual field count" 2 actual
+        Left other ->
+            assertFailure
+                ("expected CsvRaggedRow, got " <> show other)
+        Right _ ->
+            assertFailure
+                "fastReadCsvWithOpts RaiseOnRagged should have thrown"
+
+-- With `fastCsvTrimUnquoted = True` we recover the legacy behaviour of
+-- stripping leading/trailing whitespace from unquoted fields.
+testTrimUnquotedOpt :: Test
+testTrimUnquotedOpt = TestLabel "trim_unquoted_opt" $ TestCase $ do
+    let opts = defaultReadOptions{fastCsvTrimUnquoted = True}
+    df <- D.fastReadCsvWithOpts opts (fixtureDir <> "whitespace_fields.csv")
+    case getColumn "name" df of
+        Nothing -> assertFailure "name missing"
+        Just col ->
+            assertEqual
+                "trim_unquoted_opt: name stripped"
+                (DI.fromList @T.Text ["Alice", "Bob"])
+                col
+
+-- A declared 'Schema' overrides type inference: the 'a' column is
+-- forced to 'Int' and 'b' to 'Double', regardless of what the first few
+-- rows look like.
+testSchemaPushdown :: Test
+testSchemaPushdown = TestLabel "schema_pushdown" $ TestCase $ do
+    let path = fixtureDir <> "schema_pushdown.csv"
+    TIO.writeFile path "a,b\n1,1.5\n2,2.5\n"
+    let schema =
+            Schema $
+                M.fromList
+                    [ ("a", SType (P.Proxy @Int))
+                    , ("b", SType (P.Proxy @Double))
+                    ]
+    df <- D.fastReadCsvWithSchema schema path
+    removeFile path
+    case getColumn "a" df of
+        Nothing -> assertFailure "schema_pushdown: 'a' missing"
+        Just col -> assertEqual "a is Int" (DI.fromList @Int [1, 2]) col
+    case getColumn "b" df of
+        Nothing -> assertFailure "schema_pushdown: 'b' missing"
+        Just col -> assertEqual "b is Double" (DI.fromList @Double [1.5, 2.5]) col
+
+-- `fastReadCsvProj` returns only the requested columns, in the order
+-- requested; unreferenced columns never appear in the result.
+testProjection :: Test
+testProjection = TestLabel "projection" $ TestCase $ do
+    let path = fixtureDir <> "projection.csv"
+    TIO.writeFile path "a,b,c\n1,2,3\n4,5,6\n"
+    df <- D.fastReadCsvProj ["c", "a"] path
+    removeFile path
+    let (rows, cols) = dataframeDimensions df
+    assertEqual "projection: 2 rows" 2 rows
+    assertEqual "projection: 2 columns" 2 cols
+    let names =
+            map fst . L.sortBy (compare `on` snd) . M.toList $ columnIndices df
+    assertEqual "projection preserves order" ["c", "a"] names
+
+-- An unmatched `"` used to silently swallow the rest of the file: the
+-- SIMD scanner's PCLMUL quote-parity chain never resets, so every byte
+-- after the stray quote becomes "inside quotes."  We now raise
+-- 'CsvUnclosedQuote' rather than returning a corrupted DataFrame.
+testUnclosedQuote :: Test
+testUnclosedQuote = TestLabel "malformed_unclosed_quote" $ TestCase $ do
+    let path = fixtureDir <> "unclosed_quote.csv"
+    TIO.writeFile path "a,b\n1,\"unterminated\n"
+    result <- try (D.readCsvFast path)
+    removeFile path
+    case (result :: Either CsvParseError DataFrame) of
+        Left CsvUnclosedQuote -> return ()
+        Right _ ->
+            assertFailure
+                "readCsvFast should have thrown CsvUnclosedQuote on input with a stray quote"
+
+-- UTF-8 BOM (EF BB BF) must be stripped before the first header is parsed.
+-- Without the strip, `getColumn "name"` fails because the column is keyed
+-- under the three-byte-prefixed "\xEF\xBB\xBFname".
+testUtf8Bom :: Test
+testUtf8Bom = TestLabel "malformed_utf8_bom" $ TestCase $ do
+    df <- D.readCsvFast (fixtureDir <> "utf8_bom.csv")
+    assertEqual "utf8_bom.csv: 2 data rows" 2 (fst (dataframeDimensions df))
+    assertEqual "utf8_bom.csv: 2 columns" 2 (snd (dataframeDimensions df))
+    let names =
+            map fst . L.sortBy (compare `on` snd) . M.toList $ columnIndices df
+    assertEqual "utf8_bom.csv: header names" ["name", "value"] names
+    case getColumn "name" df of
+        Nothing -> assertFailure "utf8_bom.csv: 'name' missing"
+        Just col ->
+            assertEqual
+                "utf8_bom.csv: name values"
+                (DI.fromList @T.Text ["Alice", "Bob"])
+                col
+
+tests :: [Test]
+tests =
+    [ testSimpleFast
+    , testCommaInQuotesFast
+    , testQuotesAndNewlinesFast
+    , testEscapedQuotesFast
+    , testNewlinesFast
+    , testUtf8Fast
+    , testQuotesAndNewlinesFast
+    , testEmptyValuesFast
+    , testJsonDataFast
+    , testCrlfCsv
+    , testCrlfTsv
+    , testHeaderOnly
+    , testTrailingBlankLine
+    , testAllEmptyRow
+    , testSingleCol
+    , testWhitespaceFields
+    , testMissingFields
+    , testExtraFieldsTruncate
+    , testNoTrailingNewline
+    , testEmptyFile
+    , testUtf8Bom
+    , testRfc4180EscapedQuote
+    , testQuotedWhitespacePreserved
+    , testUnclosedQuote
+    , testRaiseOnRagged
+    , testTrimUnquotedOpt
+    , testSchemaPushdown
+    , testProjection
+    ]
diff --git a/tests/Properties/Csv.hs b/tests/Properties/Csv.hs
new file mode 100644
--- /dev/null
+++ b/tests/Properties/Csv.hs
@@ -0,0 +1,238 @@
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE MonoLocalBinds #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
+
+{- | QuickCheck properties for the SIMD CSV parser.
+
+Each property builds a fresh input on disk under @/tmp@, exercises
+'DataFrame.IO.CSV.Fast.fastReadCsv' (or a variant), and asserts a
+property-level invariant.  Temp files are cleaned up by the caller.
+
+The properties deliberately avoid depending on type-inference: every
+test cell starts with a letter so 'parseFromExamples' always lands on
+@Text@, which makes roundtrip comparison unambiguous.
+-}
+module Properties.Csv (tests) where
+
+import Control.Exception (try)
+import qualified Data.List as L
+import qualified Data.Map as M
+import qualified Data.Text as T
+import qualified Data.Text.IO as TIO
+import qualified Data.Vector as V
+
+import Data.Type.Equality (testEquality, (:~:) (Refl))
+import DataFrame.IO.CSV.Fast (CsvParseError (..))
+import qualified DataFrame.IO.CSV.Fast as D
+import DataFrame.Internal.Column (Column (..))
+import DataFrame.Internal.DataFrame (
+    DataFrame (..),
+    columnIndices,
+    columns,
+    dataframeDimensions,
+ )
+
+import System.Directory (removeFile)
+import Test.QuickCheck
+import Test.QuickCheck.Monadic (PropertyM, assert, monadicIO, run)
+import Type.Reflection (typeRep)
+
+-- | A generator for a single CSV cell.  Each cell starts with an ASCII
+-- letter (so type inference always picks 'Text') and may contain a small
+-- menagerie of special characters the parser must escape correctly.
+newtype Cell = Cell {unCell :: T.Text}
+    deriving (Eq, Show)
+
+instance Arbitrary Cell where
+    arbitrary = do
+        leading <- elements "abcdefghij"
+        rest <- listOf (elements "abcdef,\n\"\r ")
+        pure (Cell (T.pack (leading : rest)))
+    shrink (Cell t)
+        | T.null t || T.length t == 1 = []
+        | otherwise = [Cell (T.take (T.length t - 1) t)]
+
+-- | Like 'Cell' but guaranteed to contain neither @\\n@ nor @\\r@.  Used
+-- by properties that diff across line-ending dialects, where embedded
+-- line breaks become part of the data under CRLF transform and break
+-- the otherwise-clean invariant.
+newtype SingleLineCell = SingleLineCell {unSingleLineCell :: T.Text}
+    deriving (Eq, Show)
+
+instance Arbitrary SingleLineCell where
+    arbitrary = do
+        leading <- elements "abcdefghij"
+        rest <- listOf (elements "abcdef, \"")
+        pure (SingleLineCell (T.pack (leading : rest)))
+    shrink (SingleLineCell t)
+        | T.null t || T.length t == 1 = []
+        | otherwise = [SingleLineCell (T.take (T.length t - 1) t)]
+
+-- | A generator for a non-empty list of ASCII column names, unique.
+newtype ColumnNames = ColumnNames {unColumnNames :: [T.Text]}
+    deriving (Eq, Show)
+
+instance Arbitrary ColumnNames where
+    arbitrary = do
+        n <- chooseInt (1, 4)
+        pure (ColumnNames [T.pack ('c' : show i) | i <- [0 .. n - 1]])
+    shrink (ColumnNames ns)
+        | length ns <= 1 = []
+        | otherwise = [ColumnNames (init ns)]
+
+-- | RFC 4180 cell encoding: wrap a field in @"@ and double embedded
+-- quotes when the cell contains the separator, a line break, or a
+-- quote.
+encodeCell :: Char -> T.Text -> T.Text
+encodeCell sep t
+    | T.any needsQuote t = T.concat ["\"", T.replace "\"" "\"\"" t, "\""]
+    | otherwise = t
+  where
+    needsQuote c = c == sep || c == '\n' || c == '\r' || c == '"'
+
+-- | Encode a grid @[[cell]]@ (rows of cells) as an RFC 4180 CSV.  The
+-- first row is written as the header, followed by one line per data
+-- row.  Always uses @\\n@ as the line terminator; callers that need
+-- CRLF can transform the output themselves.
+encodeCsv :: Char -> [T.Text] -> [[T.Text]] -> T.Text
+encodeCsv sep header rows =
+    T.unlines (encodeRow header : map encodeRow rows)
+  where
+    encodeRow = T.intercalate (T.singleton sep) . map (encodeCell sep)
+
+-- | Pull the raw 'Text' values of the named column out of a DataFrame.
+-- Fails the property if the column is missing or is not a 'Text'
+-- column (which would signal a type-inference accident — the cell
+-- generator is designed to avoid that).
+columnAsText :: T.Text -> DataFrame -> Maybe [T.Text]
+columnAsText name df = do
+    idx <- M.lookup name (columnIndices df)
+    col <- columns df V.!? idx
+    case col of
+        BoxedColumn _ (vec :: V.Vector a) ->
+            case testEquality (typeRep @a) (typeRep @T.Text) of
+                Just Refl -> Just (V.toList vec)
+                Nothing -> Nothing
+        UnboxedColumn{} -> Nothing
+
+-- | Run a property in @IO@ against a generated CSV text, cleaning up
+-- the temp file afterwards no matter what.
+withCsvFile :: String -> T.Text -> (FilePath -> IO a) -> IO a
+withCsvFile label body action = do
+    let path = "/tmp/fastcsv_prop_" <> label <> ".csv"
+    TIO.writeFile path body
+    r <- action path
+    removeFile path
+    pure r
+
+--------------------------------------------------------------------------------
+-- Properties
+
+-- | Any CSV parses the same whether or not a UTF-8 BOM (EF BB BF) is
+-- prepended.  Excel / PowerShell CSV exports all come with a BOM.
+prop_bom_invariant :: ColumnNames -> [[Cell]] -> Property
+prop_bom_invariant (ColumnNames header) rows = monadicIO $ do
+    let cellRows = map (map unCell) rows
+        csv = encodeCsv ',' header cellRows
+        csvWithBom = T.pack "\xFEFF" <> csv
+    plain <- run $ withCsvFile "bom_plain" csv D.fastReadCsv
+    withBom <- run $ withCsvFile "bom_with" csvWithBom D.fastReadCsv
+    assert (plain == withBom)
+    assert (dataframeDimensions plain == dataframeDimensions withBom)
+
+-- | A CSV with no embedded newlines inside quoted fields parses the
+-- same whether record separators are @\\n@ or @\\r\\n@.  When quoted
+-- cells themselves contain @\\n@, a CRLF-encoded file contains a literal
+-- @\\r\\n@ inside the quotes (that's data, not line ending), so the
+-- invariant correctly no longer holds; we filter such inputs out.
+prop_crlf_invariant :: ColumnNames -> [[SingleLineCell]] -> Property
+prop_crlf_invariant (ColumnNames header) rows = monadicIO $ do
+    let cellRows = map (map unSingleLineCell) rows
+        lfCsv = encodeCsv ',' header cellRows
+        crlfCsv = T.replace "\n" "\r\n" lfCsv
+    dfLf <- run $ withCsvFile "crlf_lf" lfCsv D.fastReadCsv
+    dfCrlf <- run $ withCsvFile "crlf_crlf" crlfCsv D.fastReadCsv
+    assert (dfLf == dfCrlf)
+
+-- | A DataFrame produced by the RFC 4180 encoder round-trips through
+-- the fast reader: column names survive, row count survives, and every
+-- cell comes back bit-for-bit.  Restricted to ASCII-letter-prefixed
+-- cells so that type inference always stays on 'Text'.
+prop_roundtrip_ascii :: Property
+prop_roundtrip_ascii = forAll (chooseInt (0, 12)) $ \nRows ->
+    forAll (chooseInt (1, 4)) $ \nCols ->
+        let header = [T.pack ('c' : show i) | i <- [0 .. nCols - 1]]
+         in forAll (vectorOf nRows (vectorOf nCols arbitrary)) $
+                \(rawRows :: [[Cell]]) -> monadicIO $ do
+                    let rows = map (map unCell) rawRows
+                        csv = encodeCsv ',' header rows
+                    df <- run $ withCsvFile "roundtrip" csv D.fastReadCsv
+                    assertColumns header rows df
+
+assertColumns :: [T.Text] -> [[T.Text]] -> DataFrame -> PropertyM IO ()
+assertColumns header rows df = do
+    let (gotRows, gotCols) = dataframeDimensions df
+    assert (gotRows == length rows)
+    assert (gotCols == length header)
+    let expected = L.transpose rows
+    mapM_
+        ( \(name, expectedCol) ->
+            case columnAsText name df of
+                Just actual -> assert (actual == expectedCol)
+                Nothing -> assert False
+        )
+        (zip header expected)
+
+-- | A file that ends with an unmatched @"@ must raise 'CsvUnclosedQuote'
+-- under the default policy.  The generator deliberately builds a valid
+-- prefix so the property is checking the error path, not a random
+-- parse failure.
+prop_unclosed_quote_throws :: Property
+prop_unclosed_quote_throws = forAll (listOf1 arbitrary) $ \(cells :: [Cell]) ->
+    monadicIO $ do
+        let plainRow =
+                T.intercalate "," (map (encodeCell ',' . unCell) cells)
+            csv = "v\n" <> plainRow <> ",\"dangling\n"
+        result <- run $ do
+            let path = "/tmp/fastcsv_prop_unclosed.csv"
+            TIO.writeFile path csv
+            r <- try @CsvParseError (D.fastReadCsv path)
+            removeFile path
+            pure r
+        case result of
+            Left CsvUnclosedQuote -> assert True
+            _ -> assert False
+
+-- | Files whose length sits exactly at a SIMD chunk boundary must parse
+-- without an off-by-one.  We generate inputs at the key critical sizes
+-- (64, 127, 128, 129, 192 bytes after the 'v\n' header) by padding a
+-- single column with ASCII letters.
+prop_boundary_sizes :: Property
+prop_boundary_sizes =
+    forAll (elements [64, 127, 128, 129, 192]) $ \targetBytes ->
+        let header = "v\n"
+            bodyLen = targetBytes - T.length header
+            -- Build a "aaaa\nbbbb\n..." body of exactly `bodyLen` bytes
+            -- ending in a newline.
+            rows = take bodyLen $ cycle (map T.singleton "abcdefghij")
+            body = T.intercalate "\n" rows <> "\n"
+            csv = header <> T.take bodyLen body
+         in counterexample (show csv) $ monadicIO $ do
+                df <- run $ withCsvFile "boundary" csv D.fastReadCsv
+                let (rs, cs) = dataframeDimensions df
+                -- Property: reader terminates and returns one column; the
+                -- row count may vary depending on how the padding lines up,
+                -- but the column metadata must be consistent.
+                assert (cs == 1)
+                assert (rs >= 0)
+
+tests :: [Property]
+tests =
+    [ property prop_bom_invariant
+    , property prop_crlf_invariant
+    , property prop_roundtrip_ascii
+    , property prop_unclosed_quote_throws
+    , property prop_boundary_sizes
+    ]
diff --git a/tests/data/unstable_csv/all_empty_row.csv b/tests/data/unstable_csv/all_empty_row.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/all_empty_row.csv
@@ -0,0 +1,2 @@
+a,b,c
+,,
diff --git a/tests/data/unstable_csv/comma_in_quotes.csv b/tests/data/unstable_csv/comma_in_quotes.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/comma_in_quotes.csv
@@ -0,0 +1,3 @@
+first,last,address,city,zip
+John,Doe,"120 any st.","Anytown, WW",08123
+
diff --git a/tests/data/unstable_csv/crlf.csv b/tests/data/unstable_csv/crlf.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/crlf.csv
@@ -0,0 +1,3 @@
+name,city
+Alice,New York
+Bob,Los Angeles
diff --git a/tests/data/unstable_csv/crlf.tsv b/tests/data/unstable_csv/crlf.tsv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/crlf.tsv
@@ -0,0 +1,2 @@
+name	city
+Alice	New York
diff --git a/tests/data/unstable_csv/empty_file.csv b/tests/data/unstable_csv/empty_file.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/empty_file.csv
diff --git a/tests/data/unstable_csv/empty_values.csv b/tests/data/unstable_csv/empty_values.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/empty_values.csv
@@ -0,0 +1,4 @@
+a,b,c
+1,,3
+,2,
+
diff --git a/tests/data/unstable_csv/escaped_quotes.csv b/tests/data/unstable_csv/escaped_quotes.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/escaped_quotes.csv
@@ -0,0 +1,4 @@
+a,b
+1,"ha ""ha"" ha"
+3,"4"
+
diff --git a/tests/data/unstable_csv/extra_fields.csv b/tests/data/unstable_csv/extra_fields.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/extra_fields.csv
@@ -0,0 +1,2 @@
+a,b,c
+1,2,3,EXTRA
diff --git a/tests/data/unstable_csv/header_only.csv b/tests/data/unstable_csv/header_only.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/header_only.csv
@@ -0,0 +1,1 @@
+first,second,third
diff --git a/tests/data/unstable_csv/json_data.csv b/tests/data/unstable_csv/json_data.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/json_data.csv
@@ -0,0 +1,3 @@
+key,val
+1,"{""type"": ""Point"", ""coordinates"": [102.0, 0.5]}"
+
diff --git a/tests/data/unstable_csv/missing_fields.csv b/tests/data/unstable_csv/missing_fields.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/missing_fields.csv
@@ -0,0 +1,3 @@
+a,b,c
+1,2
+X,Y,Z
diff --git a/tests/data/unstable_csv/newlines.csv b/tests/data/unstable_csv/newlines.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/newlines.csv
@@ -0,0 +1,6 @@
+a,b,c
+1,2,3
+"Once upon 
+a time",5,6
+7,8,9
+
diff --git a/tests/data/unstable_csv/no_trailing_newline.csv b/tests/data/unstable_csv/no_trailing_newline.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/no_trailing_newline.csv
@@ -0,0 +1,2 @@
+name,city
+Alice,London
diff --git a/tests/data/unstable_csv/quotes_and_newlines.csv b/tests/data/unstable_csv/quotes_and_newlines.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/quotes_and_newlines.csv
@@ -0,0 +1,6 @@
+a,b
+1,"ha 
+""ha"" 
+ha"
+3,4
+
diff --git a/tests/data/unstable_csv/simple.csv b/tests/data/unstable_csv/simple.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/simple.csv
@@ -0,0 +1,3 @@
+a,b,c
+1,2,3
+
diff --git a/tests/data/unstable_csv/simple.tsv b/tests/data/unstable_csv/simple.tsv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/simple.tsv
@@ -0,0 +1,4 @@
+a	b	c
+1	2	3
+4	5	6
+
diff --git a/tests/data/unstable_csv/single_col.csv b/tests/data/unstable_csv/single_col.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/single_col.csv
@@ -0,0 +1,4 @@
+name
+Alice
+Bob
+Carol
diff --git a/tests/data/unstable_csv/temp_escaped_quotes.csv b/tests/data/unstable_csv/temp_escaped_quotes.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/temp_escaped_quotes.csv
@@ -0,0 +1,3 @@
+a,b
+1,"ha """"ha"""" ha"
+3,4
diff --git a/tests/data/unstable_csv/temp_json_data.csv b/tests/data/unstable_csv/temp_json_data.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/temp_json_data.csv
@@ -0,0 +1,2 @@
+key,val
+1,"{""""type"""": """"Point"""", """"coordinates"""": [102.0, 0.5]}"
diff --git a/tests/data/unstable_csv/temp_quotes_and_newlines.csv b/tests/data/unstable_csv/temp_quotes_and_newlines.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/temp_quotes_and_newlines.csv
@@ -0,0 +1,5 @@
+a,b
+1,"ha 
+""""ha"""" 
+ha"
+3,4
diff --git a/tests/data/unstable_csv/trailing_blank_line.csv b/tests/data/unstable_csv/trailing_blank_line.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/trailing_blank_line.csv
@@ -0,0 +1,3 @@
+a,b,c
+1,2,3
+
diff --git a/tests/data/unstable_csv/utf8.csv b/tests/data/unstable_csv/utf8.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/utf8.csv
@@ -0,0 +1,4 @@
+a,b,c
+1,2,3
+4,5,ʤ
+
diff --git a/tests/data/unstable_csv/utf8_bom.csv b/tests/data/unstable_csv/utf8_bom.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/utf8_bom.csv
@@ -0,0 +1,3 @@
+name,value
+Alice,1
+Bob,2
diff --git a/tests/data/unstable_csv/whitespace_fields.csv b/tests/data/unstable_csv/whitespace_fields.csv
new file mode 100644
--- /dev/null
+++ b/tests/data/unstable_csv/whitespace_fields.csv
@@ -0,0 +1,3 @@
+name,city
+  Alice  ,  New York
+  Bob  ,  Los Angeles