Phase 11: smart-LOB BLOB/CLOB write via SQ_FILE / filetoblob

Mirrors Phase 10's read implementation in the opposite direction —
extends the SQ_FILE (98) handler with optype 2 (read-from-client)
support. Users register bytes in cursor.virtual_files; the server's
filetoblob('path', 'client') call streams them up via SQ_FILE_READ
(106) chunks. Same architectural pivot as Phase 10 — avoids the
heavy SQ_FPROUTINE+SQ_LODATA stack.

Wire protocol (per IfxSqli.receiveSQFILE case 2 line 5103+):
* Server sends [short SQ_FILE=98][short optype=2][short bufSize]
  [int readAmount][short SQ_EOT]
* Client responds [short 106][int totalAmount] then chunks
  [short 106][short chunkSize][padded data]... terminated by SQ_EOT

API:
* Low-level: cur.virtual_files['/sentinel'] = data, then SQL with
  filetoblob('/sentinel', 'client')
* High-level: cur.write_blob_column(sql, blob_data, params, clob=False)
  — substitutes BLOB_PLACEHOLDER token in the SQL with filetoblob()
  (or filetoclob for CLOB columns) and registers the bytes
  automatically. Cleans up virtual_files after the call.

The BLOB_PLACEHOLDER design was chosen over magic ?-binding because:
* bytes already maps to BYTE type (legacy in-row blobs) for ?-params
* Method on BlobLocator doesn't work for inserts (no locator yet)
* PLACEHOLDER is unmistakable at the call site

Closes the smart-LOB loop in pure Python — Phase 9's tests and
Phase 10's read fixtures previously used JDBC to seed test data.
Phase 11 eliminated that dependency: tests/test_smart_lob.py and
tests/test_smart_lob_read.py now self-seed via write_blob_column.

Bonus: integration test runtime 5.78s → 2.78s (no more per-fixture
JVM spawns). Project goal "pure Python, no native deps" now true
for the test suite too.

Tests: 9 integration tests in test_smart_lob_write.py covering
* BLOB short, multichunk (51KB), empty, binary-safe (256 values)
* BLOB UPDATE
* BLOB multi-row INSERTs
* CLOB via filetoclob
* validation (rejects SQL without BLOB_PLACEHOLDER)
* virtual_files cleanup

Total: 64 unit + 126 integration = 190 tests.

docs/DECISION_LOG.md

@@ -798,6 +798,91 @@ Total project tests: **64 unit + 117 integration = 181 tests**.
 
 ---
 
+## 2026-05-04 — Phase 11: smart-LOB BLOB/CLOB write via SQ_FILE / filetoblob
+
+**Status**: active
+**Decision**: Implemented BLOB and CLOB *write* using the same `SQ_FILE` (98) protocol pivot as Phase 10 — the symmetric counterpart in the opposite direction. Same pattern: leverage a server-side SQL function (`filetoblob`/`filetoclob`) that orchestrates the byte transfer, with our driver acting as a remote filesystem.
+
+### What ships
+
+Two new pieces:
+
+1. **`SQ_FILE` optype 2 (read-from-client)**: extended the Phase 10 handler. When the server says "open file X for reading, send me chunks", we look up registered bytes in `cursor.virtual_files[X]` and stream them as `SQ_FILE_READ` (106) chunks. The wire format mirrors optype 3 (write-to-client) but reversed.
+
+2. **`cursor.write_blob_column(sql, blob_data, params, *, clob=False)`**: high-level helper. Takes a SQL statement with a `BLOB_PLACEHOLDER` token, replaces it with `filetoblob('<sentinel>', 'client')` (or `filetoclob` for CLOB), registers the bytes under the sentinel, runs the statement. The server reads the bytes via the SQ_FILE protocol mid-statement.
+
+### Wire protocol — SQ_FILE optype 2 in detail
+
+Server sends: `[short SQ_FILE=98][short optype=2][short bufSize][int readAmount][short SQ_EOT]`
+
+We respond with:
+- `[short SQ_FILE_READ=106][int actualAmount]` — the total we'll send
+- For each chunk: `[short SQ_FILE_READ=106][short chunkSize][padded data]`
+- Final `[short SQ_EOT]` (per JDBC's `flip()`)
+
+The server's `bufSize` is the per-chunk cap; we honor it. `readAmount=-1` means "send everything".
+
+### High-level API design
+
+The `BLOB_PLACEHOLDER` token approach was chosen over alternatives:
+- **`?`-style binding**: would conflict with normal parameter substitution and require introspecting parameter types from DESCRIBE
+- **Method on `BlobLocator`**: works for read (Phase 10's deferred design) but not write — there's no locator before the row exists
+- **Implicit bytes-detection in `execute()`**: too magical; `bytes` already maps to BYTE type for legacy in-row blobs
+
+`BLOB_PLACEHOLDER` is unmistakable, doesn't conflict with anything, and makes the code obvious at the call site:
+
+```python
+cur.write_blob_column(
+    "INSERT INTO photos VALUES (?, BLOB_PLACEHOLDER)",
+    jpeg_bytes, (42,),
+)
+```
+
+### Closing the loop: pure Python end-to-end
+
+Phase 9's tests needed JDBC to seed BLOB rows. Phase 10's read tests still needed JDBC for fixtures. **Phase 11 eliminated that dependency entirely** — both `tests/test_smart_lob.py` and `tests/test_smart_lob_read.py` now use our own `write_blob_column` for fixture setup. The full smart-LOB read+write loop is **pure Python, no JVM needed**.
+
+Bonus: integration test runtime dropped from 5.78s → 2.78s because we're no longer spawning Java per fixture. The Phase 0 project goal — "pure Python Informix driver, no native deps" — was already met for the protocol implementation, but Phase 11 finally made it true for the test suite as well.
+
+### Test coverage
+
+9 integration tests in `tests/test_smart_lob_write.py`:
+- BLOB short payload round-trip (single chunk)
+- BLOB 51200 bytes (multi-chunk)
+- BLOB empty bytes
+- BLOB binary-safe (all 256 byte values)
+- BLOB UPDATE
+- BLOB multi-row INSERTs
+- CLOB round-trip (`clob=True` routes through `filetoclob`)
+- `write_blob_column` validation (rejects SQL without `BLOB_PLACEHOLDER`)
+- `virtual_files` cleanup after call
+
+Total project tests: **64 unit + 126 integration = 190 tests**.
+
+### Type matrix complete (for the common types)
+
+| Type | Decode | Encode |
+|------|--------|--------|
+| INT/FLOAT/DECIMAL/etc. | ✓ | ✓ |
+| CHAR/VARCHAR/LVARCHAR/etc. | ✓ | ✓ |
+| BOOL/DATE/DATETIME/INTERVAL | ✓ | ✓ |
+| BYTE/TEXT (legacy in-row blobs) | ✓ | ✓ |
+| **BLOB/CLOB (smart-LOBs)** | **✓ via lotofile** | **✓ via filetoblob** |
+| ROW, COLLECTION | — | — |
+
+Smart-LOBs went from "research-only" (Phase 9) to "fully working in pure Python" (Phase 11) in three phases. The architectural insight that made it tractable: **lean on server-side SQL functions, not client-side RPC**. The fast-path `SQ_FPROUTINE`/`SQ_LODATA` stack would have been ~3-4x the work.
+
+### What's still NOT done
+
+- ROW types (composite UDTs)
+- COLLECTION types (SET, LIST, MULTISET)
+- Async layer (`informix_db.aio`)
+- TLS/SSL
+- Connection pooling
+- SQL fast-path RPC (`SQ_FPROUTINE`/`SQ_LODATA`) — not needed for any common operation we've found, but would be needed for direct stored-procedure invocation with UDT params
+
+---
+
 ## (template — copy below this line for new entries)
 
 ```

src/informix_db/cursors.py

@@ -102,6 +102,15 @@ class Cursor:
         self.blob_files: dict[str, bytes] = {}
         self._sqfile_current_name: str | None = None
         self._sqfile_current_buf: bytearray | None = None
+        # Phase 11: smart-LOB write via ``filetoblob(path, 'client', ...)``.
+        # The server tells us "open file X for reading, send me chunks".
+        # Users register the bytes they want uploaded keyed by filename
+        # in ``virtual_files``; our SQ_FILE handler streams them on
+        # request. After the INSERT completes, the registry can be
+        # cleared (or kept for batched uploads).
+        self.virtual_files: dict[str, bytes] = {}
+        self._sqfile_read_source: bytes | None = None
+        self._sqfile_read_offset: int = 0
 
     # -- PEP 249 attributes ------------------------------------------------
 
@@ -259,6 +268,103 @@ class Cursor:
             new_rows.append(tuple(row_list))
         self._rows = new_rows
 
+    def _lookup_virtual_file(self, fname: str) -> bytes | None:
+        """Look up bytes registered for a given filename.
+
+        Tries exact match first, then falls back to prefix match. The
+        server occasionally rewrites paths (e.g., adds a unique suffix
+        for ``lotofile`` output, but seemingly NOT for ``filetoblob``
+        input — kept conservative for safety).
+        """
+        if fname in self.virtual_files:
+            return self.virtual_files[fname]
+        # Try prefix match (e.g., user registered '/sentinel' and server
+        # opens '/sentinel.SUFFIX').
+        for k, v in self.virtual_files.items():
+            if fname.startswith(k):
+                return v
+        return None
+
+    def _send_sqfile_read_response(self, payload: bytes, buf_size: int) -> None:
+        """Send the SQ_FILE optype=2 response: client→server file read.
+
+        Per ``IfxSqli.receiveSQFILE`` case 2 (line 5103+):
+        ``[short SQ_FILE_READ=106][int total][short 106][short chunkLen]
+        [padded data]...[short SQ_EOT]`` — ``buf_size`` is the chunk cap.
+        """
+        # Cap chunk size to a sane maximum if server requested 0 (defensive)
+        if buf_size <= 0:
+            buf_size = 32_000
+        out = bytearray()
+        # Header: tag + total amount being sent
+        out.extend(struct.pack("!hi", MessageType.SQ_FILE_READ, len(payload)))
+        # Chunks
+        offset = 0
+        while offset < len(payload):
+            chunk = payload[offset : offset + buf_size]
+            out.extend(struct.pack("!hh", MessageType.SQ_FILE_READ, len(chunk)))
+            out.extend(chunk)
+            if len(chunk) & 1:
+                out.append(0)  # writePadded pad
+            offset += len(chunk)
+        # Final SQ_EOT (per JDBC's flip())
+        out.extend(struct.pack("!h", MessageType.SQ_EOT))
+        self._conn._send_pdu(bytes(out))
+
+    def write_blob_column(
+        self,
+        sql: str,
+        blob_data: bytes,
+        params: tuple = (),
+        *,
+        sentinel: str = "/tmp/_informix_db_blob_in",
+        clob: bool = False,
+    ) -> None:
+        """Insert/update a smart-LOB BLOB or CLOB column with the given bytes.
+
+        Wraps the user's SQL by replacing a ``BLOB_PLACEHOLDER`` token
+        with ``filetoblob('<sentinel>', 'client')`` (or ``filetoclob``
+        when ``clob=True``). Registers the bytes in ``virtual_files``
+        keyed by ``<sentinel>`` so the SQ_FILE protocol's read-from-
+        client path streams them up.
+
+        Phase 11 implementation — uses the SQ_FILE optype 2 protocol
+        instead of the heavier ``ifx_lo_create`` + ``SQ_LODATA`` stack.
+
+        Example::
+
+            cur.write_blob_column(
+                "INSERT INTO photos VALUES (?, BLOB_PLACEHOLDER)",
+                jpeg_bytes,
+                (42,),
+            )
+            # CLOB column:
+            cur.write_blob_column(
+                "INSERT INTO docs VALUES (?, BLOB_PLACEHOLDER)",
+                "Lorem ipsum...".encode("iso-8859-1"),
+                (1,),
+                clob=True,
+            )
+
+        The ``BLOB_PLACEHOLDER`` token in the SQL must appear exactly
+        where the BLOB/CLOB-typed value belongs (typically as a
+        ``VALUES`` item or a ``SET col = ...`` RHS).
+        """
+        if "BLOB_PLACEHOLDER" not in sql:
+            raise ProgrammingError(
+                "write_blob_column SQL must include a BLOB_PLACEHOLDER token "
+                "where the BLOB/CLOB value goes"
+            )
+        fn = "filetoclob" if clob else "filetoblob"
+        substitution = f"{fn}('{sentinel}', 'client')"
+        rewritten = sql.replace("BLOB_PLACEHOLDER", substitution)
+        self.virtual_files[sentinel] = blob_data
+        try:
+            self.execute(rewritten, params)
+        finally:
+            # Clean up to avoid leaking bytes into a future call
+            self.virtual_files.pop(sentinel, None)
+
     def read_blob_column(
         self,
         select_blob_sql: str,
@@ -358,10 +464,35 @@ class Cursor:
                 )
             self._sqfile_current_name = fname
             self._sqfile_current_buf = bytearray()
+            # If the user pre-registered bytes for this filename (or any
+            # close-enough match — server may add a suffix), prepare the
+            # read source for an upcoming optype=2.
+            self._sqfile_read_source = self._lookup_virtual_file(fname)
+            self._sqfile_read_offset = 0
             # Acknowledge with bare SQ_EOT (mirrors JDBC's flip() flush)
             self._conn._send_pdu(
                 struct.pack("!h", MessageType.SQ_EOT)
             )
+        elif optype == 2:  # read from client (filetoblob path)
+            buf_size = reader.read_short() & 0xFFFF
+            read_amount = reader.read_int()  # signed int; -1 = read all
+            tail = reader.read_short()
+            if tail != MessageType.SQ_EOT:
+                raise DatabaseError(
+                    f"SQ_FILE read-from-client: expected SQ_EOT, got 0x{tail:04x}"
+                )
+            if self._sqfile_read_source is None:
+                # No virtual file registered — server expects a real file
+                # but we're in-memory only. Send a zero-byte response.
+                self._send_sqfile_read_response(b"", buf_size)
+            else:
+                start = self._sqfile_read_offset
+                if read_amount < 0:
+                    payload = self._sqfile_read_source[start:]
+                else:
+                    payload = self._sqfile_read_source[start : start + read_amount]
+                self._sqfile_read_offset += len(payload)
+                self._send_sqfile_read_response(payload, buf_size)
         elif optype == 3:  # write to client
             total = 0
             while True:
@@ -881,6 +1012,9 @@ class Cursor:
                 reader.read_short()  # xcEvent
                 reader.read_short()  # xcNewLevel
                 reader.read_short()  # xcOldLevel
+            elif tag == 98:  # SQ_FILE — server orchestrates file transfer
+                # mid-DML (e.g., INSERT ... filetoblob('X', 'client'))
+                self._handle_sq_file(reader)
             elif tag == MessageType.SQ_ERR:
                 self._raise_sq_err(reader)
             else:

tests/test_smart_lob.py

@@ -19,11 +19,7 @@ from __future__ import annotations
 
 import contextlib
 import dataclasses
-import os
-import shutil
-import subprocess
 from collections.abc import Iterator
-from pathlib import Path
 
 import pytest
 
@@ -47,80 +43,30 @@ def _connect(params: ConnParams) -> informix_db.Connection:
     )
 
 
-def _java_available() -> bool:
-    """JDBC reference client requires java + the IfxJdbc jar."""
-    if not shutil.which("java"):
-        return False
-    return Path("build/ifxjdbc.jar").exists() and Path("build/tests").exists()
-
-
 @pytest.fixture
 def blob_table_with_data(
     logged_db_params: ConnParams,
 ) -> Iterator[str]:
-    """Create a BLOB table and seed it via the JDBC reference client.
+    """Create a BLOB table and seed it via Phase 11's write_blob_column.
 
-    Smart-LOB writes require the SQ_FPROUTINE + SQ_LODATA protocols
+    Originally (pre-Phase-11) this fixture used a JDBC helper to seed
-    that our driver doesn't implement yet (Phase 10). We use the
+    the row because our driver couldn't write smart-LOBs. Phase 11
-    JDBC reference client (``RefBlob``) to seed test data.
+    eliminated that dependency — pure Python end-to-end.
     """
-    if not _java_available():
-        pytest.skip(
-            "JDBC reference client unavailable (need java + build/ifxjdbc.jar)"
-        )
-
     table = "t_blob_test"
-    # Drop if exists
     with _connect(logged_db_params) as conn:
         cur = conn.cursor()
         with contextlib.suppress(Exception):
             cur.execute(f"DROP TABLE {table}")
-
+        try:
-    # Use Java helper to populate (compile RefBlob inline if needed)
+            cur.execute(f"CREATE TABLE {table} (id INT, data BLOB)")
-    helper_dir = Path("build/tests/reference")
+        except informix_db.Error as e:
-    helper_dir.mkdir(parents=True, exist_ok=True)
+            pytest.skip(f"sbspace unavailable ({e!r})")
-    helper_src = Path("tests/reference/RefBlobTest.java")
+        cur.write_blob_column(
-    if not helper_src.exists():
+            f"INSERT INTO {table} VALUES (?, BLOB_PLACEHOLDER)",
-        helper_src.write_text(
+            b"hello smart-lob bytes",
-            'package tests.reference;\n'
+            (1,),
-            'import java.sql.*;\n'
-            'public class RefBlobTest {\n'
-            '    public static void main(String[] args) throws Exception {\n'
-            '        String table = args[0], payload = args[1];\n'
-            '        Class.forName("com.informix.jdbc.IfxDriver");\n'
-            '        try (Connection c = DriverManager.getConnection(\n'
-            '             "jdbc:informix-sqli://127.0.0.1:9088/testdb:INFORMIXSERVER=informix",\n'
-            '             "informix", "in4mix")) {\n'
-            '            c.setAutoCommit(true);\n'
-            '            try (Statement s = c.createStatement()) {\n'
-            '                s.execute("CREATE TABLE " + table + " (id INT, data BLOB)");\n'
-            '            }\n'
-            '            try (PreparedStatement ps = c.prepareStatement(\n'
-            '                    "INSERT INTO " + table + " VALUES (?, ?)")) {\n'
-            '                ps.setInt(1, 1);\n'
-            '                ps.setBytes(2, payload.getBytes());\n'
-            '                ps.executeUpdate();\n'
-            '            }\n'
-            '        }\n'
-            '    }\n'
-            '}\n'
         )
-    subprocess.run(
-        [
-            "javac", "-cp", "build/ifxjdbc.jar",
-            "-d", "build/", str(helper_src),
-        ],
-        check=True, capture_output=True,
-    )
-    subprocess.run(
-        [
-            "java", "-cp", "build/ifxjdbc.jar:build/",
-            "tests.reference.RefBlobTest", table, "hello smart-lob bytes",
-        ],
-        check=True, capture_output=True,
-        env={**os.environ, "IFX_DATABASE": "testdb"},
-    )
-
     try:
         yield table
     finally:

tests/test_smart_lob_read.py

@@ -11,18 +11,15 @@ Three APIs are exposed:
     ``cursor.blob_files[filename]`` for the low-level path.
   - ``cursor.read_blob_column(sql, params)`` convenience wrapper.
 
-Test data is seeded via the JDBC reference client because writing
+After Phase 11, smart-LOB writes also work in pure Python — so test
-smart-LOBs from our driver still requires the deferred SQ_FPROUTINE
+data is now seeded via our own driver instead of the JDBC reference
-+ SQ_LODATA stack.
+client. The full read+write loop is end-to-end pure Python.
 """
 
 from __future__ import annotations
 
 import contextlib
-import shutil
-import subprocess
 from collections.abc import Iterator
-from pathlib import Path
 
 import pytest
 
@@ -46,69 +43,24 @@ def _connect(params: ConnParams) -> informix_db.Connection:
     )
 
 
-def _java_available() -> bool:
-    return (
-        shutil.which("java") is not None
-        and Path("build/ifxjdbc.jar").exists()
-    )
-
-
-def _seed_blob(table: str, payload: bytes) -> None:
-    """Use JDBC to seed a BLOB row (since smart-LOB write needs Phase 11)."""
-    helper_dir = Path("build/tests/reference")
-    helper_dir.mkdir(parents=True, exist_ok=True)
-    src_path = Path("/tmp/p10/tests/reference/SeedBlob.java")
-    src_path.parent.mkdir(parents=True, exist_ok=True)
-    src_path.write_text(
-        'package tests.reference;\n'
-        'import java.sql.*;\n'
-        'import java.io.*;\n'
-        'import java.util.Base64;\n'
-        'public class SeedBlob {\n'
-        '    public static void main(String[] args) throws Exception {\n'
-        '        String table = args[0];\n'
-        '        byte[] payload = Base64.getDecoder().decode(args[1]);\n'
-        '        Class.forName("com.informix.jdbc.IfxDriver");\n'
-        '        try (Connection c = DriverManager.getConnection(\n'
-        '             "jdbc:informix-sqli://127.0.0.1:9088/testdb:INFORMIXSERVER=informix",\n'
-        '             "informix", "in4mix")) {\n'
-        '            c.setAutoCommit(true);\n'
-        '            try (Statement s = c.createStatement()) {\n'
-        '                try { s.execute("DROP TABLE " + table); } catch (Exception e) {}\n'
-        '                s.execute("CREATE TABLE " + table + " (id INT, data BLOB)");\n'
-        '            }\n'
-        '            try (PreparedStatement ps = c.prepareStatement(\n'
-        '                    "INSERT INTO " + table + " VALUES (1, ?)")) {\n'
-        '                ps.setBytes(1, payload);\n'
-        '                ps.executeUpdate();\n'
-        '            }\n'
-        '        }\n'
-        '    }\n'
-        '}\n'
-    )
-    subprocess.run(
-        ["javac", "-cp", "build/ifxjdbc.jar", "-d", "build/", str(src_path)],
-        check=True, capture_output=True,
-    )
-    import base64
-    subprocess.run(
-        [
-            "java", "-cp", "build/ifxjdbc.jar:build/",
-            "tests.reference.SeedBlob",
-            table, base64.b64encode(payload).decode(),
-        ],
-        check=True, capture_output=True,
-    )
-
-
 @pytest.fixture
 def small_blob(logged_db_params: ConnParams) -> Iterator[str]:
-    """A BLOB table seeded with a small payload."""
+    """A BLOB table seeded (via Phase 11 write) with a small payload."""
-    if not _java_available():
-        pytest.skip("JDBC reference client unavailable")
     table = "p10_small"
     payload = b"hello phase 10 lotofile"
-    _seed_blob(table, payload)
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute(f"DROP TABLE {table}")
+        try:
+            cur.execute(f"CREATE TABLE {table} (id INT, data BLOB)")
+        except informix_db.Error as e:
+            pytest.skip(f"sbspace unavailable ({e!r})")
+        cur.write_blob_column(
+            f"INSERT INTO {table} VALUES (?, BLOB_PLACEHOLDER)",
+            payload,
+            (1,),
+        )
     try:
         yield table
     finally:
@@ -121,12 +73,22 @@ def small_blob(logged_db_params: ConnParams) -> Iterator[str]:
 @pytest.fixture
 def big_blob(logged_db_params: ConnParams) -> Iterator[tuple[str, bytes]]:
     """A BLOB table seeded with a multi-chunk (>1KB) payload."""
-    if not _java_available():
-        pytest.skip("JDBC reference client unavailable")
     table = "p10_big"
     # 30000 bytes — spans many SQ_FILE_WRITE chunks
     payload = (b"X" * 10_000) + (b"Y" * 10_000) + (b"Z" * 10_000)
-    _seed_blob(table, payload)
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute(f"DROP TABLE {table}")
+        try:
+            cur.execute(f"CREATE TABLE {table} (id INT, data BLOB)")
+        except informix_db.Error as e:
+            pytest.skip(f"sbspace unavailable ({e!r})")
+        cur.write_blob_column(
+            f"INSERT INTO {table} VALUES (?, BLOB_PLACEHOLDER)",
+            payload,
+            (1,),
+        )
     try:
         yield table, payload
     finally:

tests/test_smart_lob_write.py

@@ -0,0 +1,262 @@
+"""Phase 11 integration tests — smart-LOB BLOB/CLOB write via SQ_FILE / filetoblob.
+
+Phase 10 implemented BLOB *read* by leveraging ``lotofile(...)`` and
+intercepting the resulting ``SQ_FILE`` (98) protocol. Phase 11 mirrors
+that pattern in the *write* direction: the user calls
+``filetoblob('/sentinel', 'client')`` (or ``filetoclob``) with bytes
+pre-registered in ``cursor.virtual_files``. The server's read-from-
+client SQ_FILE optype=2 messages drive our handler to stream the
+registered bytes up.
+
+The high-level API is ``cursor.write_blob_column(sql, blob_data, params)``
+which uses a ``BLOB_PLACEHOLDER`` token in the SQL.
+
+This is the symmetric counterpart of Phase 10's ``read_blob_column``
+and the missing piece that makes the smart-LOB read+write loop
+complete entirely in pure Python — no JDBC needed for fixture seeding.
+"""
+
+from __future__ import annotations
+
+import contextlib
+from collections.abc import Iterator
+
+import pytest
+
+import informix_db
+from tests.conftest import ConnParams
+
+pytestmark = pytest.mark.integration
+
+
+def _connect(params: ConnParams) -> informix_db.Connection:
+    return informix_db.connect(
+        host=params.host,
+        port=params.port,
+        user=params.user,
+        password=params.password,
+        database=params.database,
+        server=params.server,
+        connect_timeout=10.0,
+        read_timeout=10.0,
+        autocommit=True,
+    )
+
+
+@pytest.fixture
+def blob_table(logged_db_params: ConnParams) -> Iterator[str]:
+    """A fresh BLOB table per test, dropped on teardown."""
+    table = "t_p11_blob"
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute(f"DROP TABLE {table}")
+        try:
+            cur.execute(f"CREATE TABLE {table} (id INT, data BLOB)")
+        except informix_db.Error as e:
+            pytest.skip(f"sbspace unavailable ({e!r})")
+    try:
+        yield table
+    finally:
+        with _connect(logged_db_params) as conn:
+            cur = conn.cursor()
+            with contextlib.suppress(Exception):
+                cur.execute(f"DROP TABLE {table}")
+
+
+@pytest.fixture
+def clob_table(logged_db_params: ConnParams) -> Iterator[str]:
+    """A fresh CLOB table per test."""
+    table = "t_p11_clob"
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute(f"DROP TABLE {table}")
+        try:
+            cur.execute(f"CREATE TABLE {table} (id INT, txt CLOB)")
+        except informix_db.Error as e:
+            pytest.skip(f"sbspace unavailable ({e!r})")
+    try:
+        yield table
+    finally:
+        with _connect(logged_db_params) as conn:
+            cur = conn.cursor()
+            with contextlib.suppress(Exception):
+                cur.execute(f"DROP TABLE {table}")
+
+
+# -------- BLOB write+read round-trip --------
+
+
+def test_write_blob_round_trip_short(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """Short payload — single SQ_FILE_READ chunk."""
+    payload = b"hello phase 11 blob write"
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        cur.write_blob_column(
+            f"INSERT INTO {blob_table} VALUES (?, BLOB_PLACEHOLDER)",
+            payload,
+            (1,),
+        )
+        got = cur.read_blob_column(
+            f"SELECT data FROM {blob_table} WHERE id = ?", (1,)
+        )
+        assert got == payload
+
+
+def test_write_blob_round_trip_multichunk(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """50KB payload — spans many SQ_FILE_READ chunks (32KB cap each)."""
+    payload = bytes(range(256)) * 200  # 51200 bytes
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        cur.write_blob_column(
+            f"INSERT INTO {blob_table} VALUES (?, BLOB_PLACEHOLDER)",
+            payload,
+            (1,),
+        )
+        got = cur.read_blob_column(
+            f"SELECT data FROM {blob_table} WHERE id = ?", (1,)
+        )
+        assert got == payload
+        assert len(got) == 51200
+
+
+def test_write_blob_empty(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """Empty bytes round-trip cleanly."""
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        cur.write_blob_column(
+            f"INSERT INTO {blob_table} VALUES (?, BLOB_PLACEHOLDER)",
+            b"",
+            (1,),
+        )
+        got = cur.read_blob_column(
+            f"SELECT data FROM {blob_table} WHERE id = ?", (1,)
+        )
+        assert got == b""
+
+
+def test_write_blob_binary_safe(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """All-byte-values payload — no encoding artifacts."""
+    payload = bytes(range(256)) * 4  # 1024 bytes covering all values
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        cur.write_blob_column(
+            f"INSERT INTO {blob_table} VALUES (?, BLOB_PLACEHOLDER)",
+            payload,
+            (1,),
+        )
+        got = cur.read_blob_column(
+            f"SELECT data FROM {blob_table} WHERE id = ?", (1,)
+        )
+        assert got == payload
+
+
+def test_write_blob_update(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """UPDATE with BLOB column replaces the prior value."""
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        cur.write_blob_column(
+            f"INSERT INTO {blob_table} VALUES (?, BLOB_PLACEHOLDER)",
+            b"original",
+            (1,),
+        )
+        cur.write_blob_column(
+            f"UPDATE {blob_table} SET data = BLOB_PLACEHOLDER WHERE id = ?",
+            b"replacement",
+            (1,),
+        )
+        got = cur.read_blob_column(
+            f"SELECT data FROM {blob_table} WHERE id = ?", (1,)
+        )
+        assert got == b"replacement"
+
+
+def test_write_blob_multiple_rows(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """Distinct INSERTs round-trip independently."""
+    rows = [
+        (1, b"first row"),
+        (2, b"second row blob"),
+        (3, b"third"),
+    ]
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        for rid, payload in rows:
+            cur.write_blob_column(
+                f"INSERT INTO {blob_table} VALUES (?, BLOB_PLACEHOLDER)",
+                payload,
+                (rid,),
+            )
+        for rid, expected in rows:
+            got = cur.read_blob_column(
+                f"SELECT data FROM {blob_table} WHERE id = ?", (rid,)
+            )
+            assert got == expected
+
+
+# -------- CLOB --------
+
+
+def test_write_clob_round_trip(
+    logged_db_params: ConnParams, clob_table: str
+) -> None:
+    """``clob=True`` routes through ``filetoclob`` (not ``filetoblob``)."""
+    text = "Lorem ipsum dolor sit amet, café résumé".encode("iso-8859-1")
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        cur.write_blob_column(
+            f"INSERT INTO {clob_table} VALUES (?, BLOB_PLACEHOLDER)",
+            text,
+            (1,),
+            clob=True,
+        )
+        got = cur.read_blob_column(
+            f"SELECT txt FROM {clob_table} WHERE id = ?", (1,)
+        )
+        assert got == text
+
+
+# -------- Helper validation --------
+
+
+def test_write_blob_column_requires_placeholder(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """SQL without ``BLOB_PLACEHOLDER`` is rejected."""
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        with pytest.raises(
+            informix_db.ProgrammingError, match="BLOB_PLACEHOLDER"
+        ):
+            cur.write_blob_column(
+                f"INSERT INTO {blob_table} VALUES (1, NULL)",
+                b"data",
+                (),
+            )
+
+
+def test_virtual_files_cleared_after_call(
+    logged_db_params: ConnParams, blob_table: str
+) -> None:
+    """``virtual_files`` doesn't leak the registered bytes between calls."""
+    with _connect(logged_db_params) as conn:
+        cur = conn.cursor()
+        cur.write_blob_column(
+            f"INSERT INTO {blob_table} VALUES (?, BLOB_PLACEHOLDER)",
+            b"some data",
+            (1,),
+        )
+        # The default sentinel should have been removed
+        assert "/tmp/_informix_db_blob_in" not in cur.virtual_files