Add COPY IN support

Implement PostgreSQL's COPY ... FROM STDIN protocol for bulk data
loading. The pull-based CopyInReceiver interface gives callers O(1)
bounded memory: the session calls pg_copy_ready after each
send_copy_data, so the client controls data flow.

Session API: copy_in(sql, receiver), send_copy_data(data),
finish_copy(), abort_copy(reason). CopyInReceiver callbacks:
pg_copy_ready, pg_copy_complete, pg_copy_failed.

The COPY query is sent via the simple query protocol. The server
responds with CopyInResponse, then the client streams CopyData
messages, finishing with CopyDone (success) or CopyFail (abort).
The server responds with CommandComplete + ReadyForQuery on success,
or ErrorResponse + ReadyForQuery on failure.

New query state _CopyInInFlight manages the lifecycle. Data-stream
methods (send_copy_data, finish_copy, abort_copy) use direct type
matching on query_state (following the cancel pattern) rather than
going through the _QueryState interface, since they are only valid
during an active COPY.

Design: #104

Author: SeanTAllen

CLAUDE.md

@@ -30,6 +30,10 @@ Multi-query workflow mixing `SimpleQuery` and `PreparedQuery`. Creates a table,
 
 Asynchronous notifications using PostgreSQL's LISTEN/NOTIFY mechanism. Subscribes to a channel with `LISTEN`, sends a notification with `NOTIFY`, receives it via the `pg_notification` callback on `SessionStatusNotify`, and unsubscribes with `UNLISTEN`. Shows the `Notification` class fields (channel, payload, pid).
 
+## copy-in
+
+Bulk data loading using `COPY ... FROM STDIN`. Creates a table, loads three rows of tab-delimited text data via `Session.copy_in()`, verifies the data with a SELECT, then drops the table. Demonstrates the pull-based `CopyInReceiver` interface: `pg_copy_ready` fires after each `send_copy_data`, and `finish_copy` completes the operation.
+
 ## transaction-status
 
 Transaction status tracking using `pg_transaction_status`. Sends `BEGIN` and `COMMIT` and prints the `TransactionStatus` reported at each step. Shows how `SessionStatusNotify.pg_transaction_status` fires on every `ReadyForQuery` with `TransactionIdle`, `TransactionInBlock`, or `TransactionFailed`.

examples/README.md

@@ -0,0 +1,158 @@
+use "cli"
+use "collections"
+use lori = "lori"
+// in your code this `use` statement would be:
+// use "postgres"
+use "../../postgres"
+
+actor Main
+  new create(env: Env) =>
+    let server_info = ServerInfo(env.vars)
+    let auth = lori.TCPConnectAuth(env.root)
+
+    let client = Client(auth, server_info, env.out)
+
+// This example demonstrates COPY IN for bulk data loading. It creates a table,
+// uses COPY FROM STDIN to load three rows of tab-delimited text data, verifies
+// the data with a SELECT query, then drops the table.
+//
+// The COPY IN protocol uses a pull-based flow: the session calls pg_copy_ready
+// after each send_copy_data, letting the client send the next chunk. When all
+// data is sent, call finish_copy to complete the operation.
+actor Client is (SessionStatusNotify & ResultReceiver & CopyInReceiver)
+  let _session: Session
+  let _out: OutStream
+  var _phase: USize = 0
+  var _rows_sent: USize = 0
+
+  new create(auth: lori.TCPConnectAuth, info: ServerInfo, out: OutStream) =>
+    _out = out
+    _session = Session(
+      ServerConnectInfo(auth, info.host, info.port),
+      DatabaseConnectInfo(info.username, info.password, info.database),
+      this)
+
+  be close() =>
+    _session.close()
+
+  be pg_session_authenticated(session: Session) =>
+    _out.print("Authenticated.")
+    _phase = 0
+    session.execute(
+      SimpleQuery("DROP TABLE IF EXISTS copy_in_example"), this)
+
+  be pg_session_authentication_failed(
+    s: Session,
+    reason: AuthenticationFailureReason)
+  =>
+    _out.print("Failed to authenticate.")
+
+  be pg_copy_ready(session: Session) =>
+    _rows_sent = _rows_sent + 1
+    if _rows_sent <= 3 then
+      // Send one row per callback. Tab-delimited, newline-terminated.
+      let row: Array[U8] val = recover val
+        ("row" + _rows_sent.string() + "\t" + (_rows_sent * 10).string()
+          + "\n").array()
+      end
+      _out.print("  Sending row " + _rows_sent.string() + "...")
+      session.send_copy_data(row)
+    else
+      _out.print("  All rows sent. Finishing COPY...")
+      session.finish_copy()
+    end
+
+  be pg_copy_complete(session: Session, count: USize) =>
+    _out.print("COPY complete: " + count.string() + " rows copied.")
+    // Verify with a SELECT
+    _out.print("Verifying with SELECT...")
+    _session.execute(
+      SimpleQuery("SELECT name, value FROM copy_in_example ORDER BY name"),
+      this)
+
+  be pg_copy_failed(session: Session,
+    failure: (ErrorResponseMessage | ClientQueryError))
+  =>
+    match failure
+    | let e: ErrorResponseMessage =>
+      _out.print("COPY failed: [" + e.severity + "] " + e.code + ": "
+        + e.message)
+    | let e: ClientQueryError =>
+      _out.print("COPY failed: client error")
+    end
+    close()
+
+  be pg_query_result(session: Session, result: Result) =>
+    _phase = _phase + 1
+
+    match _phase
+    | 1 =>
+      // Table dropped (or didn't exist). Create it.
+      _out.print("Creating table...")
+      _session.execute(
+        SimpleQuery(
+          """
+          CREATE TABLE copy_in_example (
+            name VARCHAR(50) NOT NULL,
+            value INT NOT NULL
+          )
+          """),
+        this)
+    | 2 =>
+      // Table created. Start COPY IN.
+      _out.print("Table created. Starting COPY IN...")
+      _session.copy_in(
+        "COPY copy_in_example (name, value) FROM STDIN", this)
+    | 3 =>
+      // SELECT done. Print results and drop table.
+      match result
+      | let r: ResultSet =>
+        _out.print("ResultSet (" + r.rows().size().string() + " rows):")
+        for row in r.rows().values() do
+          _out.write(" ")
+          for field in row.fields.values() do
+            _out.write(" " + field.name + "=")
+            match field.value
+            | let v: String => _out.write(v)
+            | let v: I32 => _out.write(v.string())
+            | None => _out.write("NULL")
+            end
+          end
+          _out.print("")
+        end
+      end
+      _out.print("Dropping table...")
+      _session.execute(
+        SimpleQuery("DROP TABLE copy_in_example"), this)
+    | 4 =>
+      // Table dropped. Done.
+      _out.print("Done.")
+      close()
+    end
+
+  be pg_query_failed(session: Session, query: Query,
+    failure: (ErrorResponseMessage | ClientQueryError))
+  =>
+    match failure
+    | let e: ErrorResponseMessage =>
+      _out.print("Query failed: [" + e.severity + "] " + e.code + ": "
+        + e.message)
+    | let e: ClientQueryError =>
+      _out.print("Query failed: client error")
+    end
+    close()
+
+class val ServerInfo
+  let host: String
+  let port: String
+  let username: String
+  let password: String
+  let database: String
+
+  new val create(vars: (Array[String] val | None)) =>
+    let e = EnvVars(vars)
+    host = try e("POSTGRES_HOST")? else "127.0.0.1" end
+    port = try e("POSTGRES_PORT")? else "5432" end
+    username = try e("POSTGRES_USERNAME")? else "postgres" end
+    password = try e("POSTGRES_PASSWORD")? else "postgres" end
+    database = try e("POSTGRES_DATABASE")? else "postgres" end

examples/copy-in/copy-in-example.pony

@@ -169,6 +169,18 @@ class val _NotificationResponseMessage
     channel = channel'
     payload = payload'
 
+class val _CopyInResponseMessage
+  """
+  Message from the backend indicating it is ready to receive COPY data.
+  Contains the overall format (0=text, 1=binary) and per-column format codes.
+  """
+  let format: U8
+  let column_formats: Array[U8] val
+
+  new val create(format': U8, column_formats': Array[U8] val) =>
+    format = format'
+    column_formats = column_formats'
+
 primitive _PortalSuspendedMessage
   """
   Message from the backend indicating that an Execute command has been

postgres/_backend_messages.pony

@@ -464,6 +464,79 @@ primitive _FrontendMessage
       []
     end
 
+  fun copy_data(data: Array[U8] val): Array[U8] val =>
+    """
+    Build a CopyData message containing a chunk of COPY stream data.
+
+    Format: Byte1('d') Int32(4 + data.size()) Byte[](data)
+    """
+    try
+      recover val
+        let length: U32 = 4 + data.size().u32()
+        let msg_size = (length + 1).usize()
+        let msg: Array[U8] = Array[U8].init(0, msg_size)
+        msg.update_u8(0, 'd')?
+        ifdef bigendian then
+          msg.update_u32(1, length)?
+        else
+          msg.update_u32(1, length.bswap())?
+        end
+        msg.copy_from(data, 0, 5, data.size())
+        msg
+      end
+    else
+      _Unreachable()
+      []
+    end
+
+  fun copy_done(): Array[U8] val =>
+    """
+    Build a CopyDone message signaling successful end of COPY data.
+
+    Format: Byte1('c') Int32(4)
+    """
+    try
+      recover val
+        let msg: Array[U8] = Array[U8].init(0, 5)
+        msg.update_u8(0, 'c')?
+        ifdef bigendian then
+          msg.update_u32(1, U32(4))?
+        else
+          msg.update_u32(1, U32(4).bswap())?
+        end
+        msg
+      end
+    else
+      _Unreachable()
+      []
+    end
+
+  fun copy_fail(reason: String): Array[U8] val =>
+    """
+    Build a CopyFail message aborting the COPY operation.
+
+    Format: Byte1('f') Int32(4 + reason.size() + 1) String(reason)
+    """
+    try
+      recover val
+        let length: U32 = reason.size().u32() + 5
+        let msg_size = (length + 1).usize()
+        let msg: Array[U8] = Array[U8].init(0, msg_size)
+        msg.update_u8(0, 'f')?
+        ifdef bigendian then
+          msg.update_u32(1, length)?
+        else
+          msg.update_u32(1, length.bswap())?
+        end
+        msg.copy_from(reason.array(), 0, 5, reason.size())
+        // space for null left here
+        msg
+      end
+    else
+      _Unreachable()
+      []
+    end
+
   fun terminate(): Array[U8] val =>
     """
     Build a Terminate message. Sent before closing the TCP connection to

postgres/_frontend_message.pony

@@ -37,5 +37,8 @@ primitive _MessageType
   fun parameter_status(): U8 =>
     'S'
 
+  fun copy_in_response(): U8 =>
+    'G'
+
   fun row_description(): U8 =>
     'T'

postgres/_message_type.pony

@@ -55,6 +55,8 @@ primitive _ResponseMessageParser
           s.state.on_backend_key_data(s, msg)
         | let msg: _NotificationResponseMessage =>
           s.state.on_notification(s, msg)
+        | let msg: _CopyInResponseMessage =>
+          s.state.on_copy_in_response(s, msg)
         | _SkippedMessage =>
           // Known async message (ParameterStatus, NoticeResponse) —
           // intentionally not routed.

postgres/_response_message_parser.pony

@@ -23,6 +23,7 @@ type _ResponseParserResult is
   | _NoDataMessage
   | _ParameterDescriptionMessage
   | _NotificationResponseMessage
+  | _CopyInResponseMessage
   | _PortalSuspendedMessage
   | _SkippedMessage
   | _UnsupportedMessage
@@ -218,6 +219,12 @@ primitive _ResponseParser
       // and parse the notification payload in an isolated reader
       let notification_payload = buffer.block(payload_size)?
       return _notification_response(consume notification_payload)?
+    | _MessageType.copy_in_response() =>
+      // Slide past the header...
+      buffer.skip(5)?
+      // and parse the CopyInResponse payload in an isolated reader
+      let copy_payload = buffer.block(payload_size)?
+      return _copy_in_response(consume copy_payload)?
     else
       buffer.skip(message_size)?
       return _UnsupportedMessage
@@ -383,6 +390,23 @@ primitive _ResponseParser
     let payload' = String.from_array(consume payload_bytes)
     _NotificationResponseMessage(pid, channel, payload')
 
+  fun _copy_in_response(payload: Array[U8] val)
+    : _CopyInResponseMessage ?
+  =>
+    """
+    Parse a CopyInResponse message.
+    """
+    let reader: Reader = Reader.>append(payload)
+    let format = reader.u8()?
+    let num_cols = reader.u16_be()?.usize()
+    let col_fmts: Array[U8] iso = recover iso Array[U8](num_cols) end
+
+    for i in Range(0, num_cols) do
+      col_fmts.push(reader.u16_be()?.u8())
+    end
+
+    _CopyInResponseMessage(format, consume col_fmts)
+
   fun _parameter_description(payload: Array[U8] val)
     : _ParameterDescriptionMessage ?
   =>

postgres/_response_parser.pony

@@ -137,6 +137,17 @@ actor \nodoc\ Main is TestList
     test(_TestNotificationDelivery)
     test(_TestNotificationDuringDataRows)
     test(_TestListenNotify)
+    test(_TestResponseParserCopyInResponseMessage)
+    test(_TestFrontendMessageCopyData)
+    test(_TestFrontendMessageCopyDone)
+    test(_TestFrontendMessageCopyFail)
+    test(_TestCopyInSuccess)
+    test(_TestCopyInAbort)
+    test(_TestCopyInServerError)
+    test(_TestCopyInShutdownDrainsCopyQueue)
+    test(_TestCopyInAfterSessionClosed)
+    test(_TestCopyInInsert)
+    test(_TestCopyInAbortRollback)
 
 class \nodoc\ iso _TestAuthenticate is UnitTest
   """