source: project/release/4/postgresql/trunk/postgresql.scm @ 21512

;;; Bindings to the PostgreSQL C library
;;
;; Copyright (C) 2008-2010 Peter Bex
;; Copyright (C) 2004 Johannes Grødem <johs@copyleft.no>
;; Redistribution and use in source and binary forms, with or without
;; modification, is permitted.
;;
;; THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS
;; OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
;; WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
;; ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE
;; LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
;; CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
;; OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
;; BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
;; LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
;; (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
;; USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
;; DAMAGE.

(module postgresql
 (type-parsers update-type-parsers! default-type-parsers
  char-parser bool-parser bytea-parser numeric-parser
  make-array-parser make-composite-parser
  scheme-value->db-value type-unparsers update-type-unparsers!
  default-type-unparsers bool-unparser vector-unparser list-unparser
  
  connect reset-connection disconnect connection?
  
  query query* with-transaction in-transaction?
  
  result? clear-result! row-count column-count
  column-index column-name column-names column-format
  column-type column-type-modifier table-oid table-column-index
  value-at row-values row-alist column-values affected-rows inserted-oid

  invalid-oid
  
  escape-string escape-bytea unescape-bytea

  put-copy-data put-copy-end get-copy-data
  
  row-fold row-fold* row-fold-right row-fold-right*
  row-for-each row-for-each* row-map row-map*
  column-fold column-fold* column-fold-right column-fold-right*
  column-for-each column-for-each* column-map column-map*
  copy-query-fold copy-query*-fold copy-query-fold-right copy-query*-fold-right
  copy-query-for-each copy-query*-for-each copy-query-map copy-query*-map
  call-with-output-copy-query call-with-output-copy-query*
  with-output-to-copy-query with-output-to-copy-query*)

(import chicken scheme foreign)

(require-extension srfi-1 srfi-4 srfi-13 srfi-18 srfi-69
                   extras data-structures ports sql-null)

(foreign-declare "#include <libpq-fe.h>")

(define-foreign-type pg-polling-status (enum "PostgresPollingStatusType"))
(define-foreign-variable PGRES_POLLING_FAILED pg-polling-status)
(define-foreign-variable PGRES_POLLING_READING pg-polling-status)
(define-foreign-variable PGRES_POLLING_WRITING pg-polling-status)
(define-foreign-variable PGRES_POLLING_OK pg-polling-status)

(define-foreign-type pg-exec-status (enum "ExecStatusType"))
(define-foreign-variable PGRES_EMPTY_QUERY pg-exec-status)
(define-foreign-variable PGRES_COMMAND_OK pg-exec-status)
(define-foreign-variable PGRES_TUPLES_OK pg-exec-status)
(define-foreign-variable PGRES_COPY_OUT pg-exec-status)
(define-foreign-variable PGRES_COPY_IN pg-exec-status)
(define-foreign-variable PGRES_BAD_RESPONSE pg-exec-status)
(define-foreign-variable PGRES_NONFATAL_ERROR pg-exec-status)
(define-foreign-variable PGRES_FATAL_ERROR pg-exec-status)

;(define-foreign-type pgconn* (c-pointer "PGconn"))
(define-foreign-type pgconn* c-pointer)

(define PQconnectStart (foreign-lambda pgconn* PQconnectStart (const c-string)))
(define PQconnectPoll (foreign-lambda pg-polling-status PQconnectPoll pgconn*))
(define PQresetStart (foreign-lambda bool PQresetStart pgconn*))
(define PQresetPoll (foreign-lambda pg-polling-status PQresetPoll pgconn*))
(define PQfinish (foreign-lambda void PQfinish pgconn*))
(define PQstatus (foreign-lambda (enum "ConnStatusType") PQstatus (const pgconn*)))
(define PQerrorMessage (foreign-lambda c-string PQerrorMessage (const pgconn*)))

;(define-foreign-type oid "Oid")
(define-foreign-type oid unsigned-int)

(define invalid-oid (foreign-value "InvalidOid" oid))

(define PQisBusy (foreign-lambda bool PQisBusy pgconn*))
(define PQconsumeInput (foreign-lambda bool PQconsumeInput pgconn*))

(define-foreign-type pgresult* (c-pointer "PGresult"))

(define PQgetResult (foreign-lambda pgresult* PQgetResult pgconn*))
(define PQresultStatus (foreign-lambda pg-exec-status PQresultStatus (const pgresult*)))
(define PQresultErrorMessage (foreign-lambda c-string PQresultErrorMessage (const pgresult*)))
(define PQresultErrorField (foreign-lambda c-string PQresultErrorField (const pgresult*) int))

(define PQclear (foreign-lambda void PQclear pgresult*))
(define PQntuples (foreign-lambda int PQntuples (const pgresult*)))
(define PQnfields (foreign-lambda int PQnfields (const pgresult*)))
(define PQfname (foreign-lambda c-string PQfname (const pgresult*) int))
(define PQfnumber (foreign-lambda int PQfnumber (const pgresult*) (const c-string)))
(define PQftable (foreign-lambda oid PQftable (const pgresult*) int))
(define PQftablecol (foreign-lambda int PQftablecol (const pgresult*) int))
(define PQfformat (foreign-lambda int PQfformat (const pgresult*) int))
(define PQftype (foreign-lambda oid PQftype (const pgresult*) int))
(define PQfmod (foreign-lambda int PQfmod (const pgresult*) int))
(define PQgetisnull (foreign-lambda bool PQgetisnull (const pgresult*) int int))
(define PQcmdTuples (foreign-lambda nonnull-c-string PQcmdTuples pgresult*))
(define PQoidValue (foreign-lambda oid PQoidValue pgresult*))

(define PQputCopyData (foreign-lambda int PQputCopyData pgconn* blob int))
(define PQputCopyEnd (foreign-lambda int PQputCopyEnd pgconn* (const c-string)))

;; TODO: Create a real callback system?
(foreign-declare "static void nullNoticeReceiver(void *arg, const PGresult *res){ }")

(define-syntax define-foreign-int
  (er-macro-transformer
   (lambda (e r c)
     ;; cannot rename define-foreign-variable; it's a really special form
    `(define-foreign-variable ,(cadr e) int ,(conc "(int) " (cadr e))))))

(define-foreign-int PG_DIAG_SEVERITY)
(define-foreign-int PG_DIAG_SQLSTATE)
(define-foreign-int PG_DIAG_MESSAGE_PRIMARY)
(define-foreign-int PG_DIAG_MESSAGE_DETAIL)
(define-foreign-int PG_DIAG_MESSAGE_HINT)
(define-foreign-int PG_DIAG_STATEMENT_POSITION)
(define-foreign-int PG_DIAG_CONTEXT)
(define-foreign-int PG_DIAG_SOURCE_FILE)
(define-foreign-int PG_DIAG_SOURCE_LINE)
(define-foreign-int PG_DIAG_SOURCE_FUNCTION)

;; Helper procedure for lists (TODO: use ANY instead of IN with an array?)
(define (in-list len)
  (string-intersperse
   (list-tabulate len (lambda (p) (conc "$" (add1 p)))) ","))

(define (postgresql-error loc message . args)
  (signal (make-pg-condition loc message args: args)))

(define (make-pg-condition loc message #!key (args '()) severity
                           error-class error-code message-detail
                           message-hint statement-position context
                           source-file source-line
                           source-function)
  (make-composite-condition
    (make-property-condition
     'exn 'location loc 'message message 'arguments args)
    (make-property-condition
     'postgresql 'severity severity 'error-class error-class
     'error-code error-code 'message-detail message-detail
     'message-hint message-hint 'statement-position statement-position
     'context context 'source-file source-file 'source-line source-line
     ;; Might break not-terribly-old versions of postgresql
     ;;'internal-position internal-position 'internal-query internal-query
     'source-function source-function)))

;;;;;;;;;;;;;;;;;;;;;;;;
;;;; Type parsers
;;;;;;;;;;;;;;;;;;;;;;;;

(define (char-parser str) (string-ref str 0))

(define (bool-parser str) (string=? str "t"))

(define (numeric-parser str)
  (or (string->number str)
      (postgresql-error 'numeric-parser "Unable to parse number" str)))

(define (bytea-parser str)
  (blob->u8vector/shared (string->blob (unescape-bytea str))))

;; Here be dragons
(define (make-array-parser element-parser #!optional (delim #\,))
  (define (parse str)
    (if (string-ci=? "NULL" str)
        (sql-null)
        (element-parser str)))
  (lambda (str)
    (let loop ((chars (let ignore-bounds ((chars (string->list str)))
                        (if (char=? (car chars) #\{)
                            chars
                            (ignore-bounds (cdr chars)))))
               (result (list)))
      (if (null? chars)
          (car result) ; Should contain only one vector
          (case (car chars)
            ((#\{) (receive (value rest-chars)
                     (loop (cdr chars) (list))
                     (loop rest-chars (cons value result))))
            ((#\}) (values (list->vector (reverse! result)) (cdr chars)))
            ((#\") (let consume-string ((chars (cdr chars))
                                        (consumed (list)))
                     (case (car chars)
                       ((#\\) (consume-string ; Don't interpret, just add it
                               (cddr chars) (cons (cadr chars) consumed)))
                       ((#\") (loop (cdr chars)
                                    (cons (element-parser
                                           (reverse-list->string consumed))
                                          result)))
                       (else (consume-string (cdr chars)
                                             (cons (car chars) consumed))))))
            ((#\tab #\newline #\space) (loop (cdr chars) result))
            (else
             (if (char=? (car chars) delim)
                 (loop (cdr chars) result)
                 (let consume-string ((chars chars)
                                      (consumed (list)))
                   (cond
                    ((char=? (car chars) delim)
                     (loop (cdr chars)
                           (cons (parse (reverse-list->string consumed))
                                 result)))
                    ((or (char=? (car chars) #\})
                         (char=? (car chars) #\}))
                     (loop chars
                           (cons (parse (reverse-list->string consumed))
                                 result)))
                    (else (consume-string (cdr chars)
                                          (cons (car chars) consumed))))))))))))

(define (make-composite-parser element-parsers)
  (define (parse str element-parser)
    (if (string=? "" str)
        (sql-null)
        (element-parser str)))
  (lambda (str)
    (let loop ((chars (cdr (string->list (string-trim str)))) ; skip leading (
               (maybe-null? #f)
               (result (list))
               (parsers element-parsers))
      (case (car chars)
        ((#\)) (reverse! (if maybe-null?
                             (cons (sql-null) result)
                             result)))
        ((#\") (let consume-string ((chars (cdr chars))
                                    (consumed (list)))
                 (case (car chars)
                   ((#\\) (consume-string ; Don't interpret, just add it
                           (cddr chars) (cons (cadr chars) consumed)))
                   ((#\") (if (char=? #\" (cadr chars)) ; double escapes
                              (consume-string (cddr chars)
                                              (cons #\" consumed))
                              (let skip-spaces ((chars (cdr chars)))
                                (case (car chars)
                                  ((#\space #\newline #\tab)
                                   (skip-spaces (cdr chars)))
                                  ((#\,)
                                   (loop (cdr chars)
                                         #t
                                         (cons ((car parsers)
                                                (reverse-list->string consumed))
                                               result)
                                         (cdr parsers)))
                                  ((#\)) (loop chars
                                               #f
                                               (cons ((car parsers)
                                                      (reverse-list->string consumed))
                                                     result)
                                               (cdr parsers)))
                                  (else
                                   (postgresql-error 'make-composite-parser
                                                     "Bogus trailing characters"
                                                     str))))))
                   (else (consume-string (cdr chars)
                                         (cons (car chars) consumed))))))
        (else (let consume-string ((chars chars)
                                   (consumed (list)))
                (case (car chars)
                  ((#\,) (loop (cdr chars)
                               #t
                               (cons (parse (reverse-list->string consumed)
                                            (car parsers))
                                     result)
                               (cdr parsers)))
                  ;; Nothing should precede this one
                  ((#\)) (loop chars
                               #f
                               (cons (parse (reverse-list->string consumed)
                                            (car parsers))
                                     result)
                               (cdr parsers)))
                  (else (consume-string (cdr chars)
                                        (cons (car chars) consumed))))))))))

;; Array parsers and composite parsers are automatically cached when such
;; a value is requested.
(define default-type-parsers
  (make-parameter
   `(("text" . ,identity)
     ("bytea" . ,bytea-parser)
     ("char" . ,char-parser)
     ("bpchar" . ,identity)
     ("bool" . ,bool-parser)
     ("int8" . ,numeric-parser)
     ("int4" . ,numeric-parser)
     ("int2" . ,numeric-parser)
     ("float4" . ,numeric-parser)
     ("float8" . ,numeric-parser)
     ("numeric" . ,numeric-parser)
     ("oid" . ,numeric-parser)
     ;; Nasty hack, or clever hack? :)
     ("record" . ,(make-composite-parser (circular-list identity))))))

;;;;;;;;;;;;;;;;;;;;;;;
;;;; Type unparsers
;;;;;;;;;;;;;;;;;;;;;;;

(define (scheme-value->db-value conn value)
  (cond ((find (lambda (parse?)
                 ((car parse?) value))
               (pg-connection-type-unparsers conn)) =>
               (lambda (parse)
                 ((cdr parse) conn value)))
        (else value)))

(define (bool-unparser conn b)
  (if b "TRUE" "FALSE"))

(define (vector-unparser conn v)
  (let loop ((result (list))
             (pos 0)
             (len (vector-length v)))
    (if (= pos len)
        (string-append "{" (string-intersperse (reverse! result) ",") "}")
        (let* ((value (vector-ref v pos))
               (unparsed-value (scheme-value->db-value conn value))
               (serialized (cond
                            ((sql-null? unparsed-value) "NULL")
                            ((not (string? unparsed-value))
                             (postgresql-error
                              'vector-unparser
                              (sprintf "Param value is not string: ~S"
                                       unparsed-value)))
                            ((vector? value) unparsed-value) ;; don't quote!
                            (else
                             (sprintf "\"~A\""
                                      (string-translate*
                                       unparsed-value
                                       '(("\\" . "\\\\") ("\"" . "\\\""))))))))
          (loop (cons serialized result) (add1 pos) len)))))

(define (list-unparser conn l)
  (let loop ((result (list))
             (items l))
    (if (null? items)
        (string-append "(" (string-intersperse (reverse! result) ",") ")")
        (let* ((unparsed-value (scheme-value->db-value conn (car items)))
               (serialized (cond
                            ((sql-null? unparsed-value) "")
                            ((not (string? unparsed-value))
                             (postgresql-error
                              'list-unparser
                              (sprintf "Param value is not string: ~S"
                                       unparsed-value)))
                            (else
                             (sprintf "\"~A\""
                                      (string-translate*
                                       unparsed-value
                                       '(("\\" . "\\\\") ("\"" . "\\\""))))))))
          (loop (cons serialized result) (cdr items))))))

(define default-type-unparsers
  (make-parameter
   `((,string? . ,(lambda (conn s) s))
     (,u8vector? . ,(lambda (conn v) (u8vector->blob/shared v)))
     (,char? . ,(lambda (conn c) (string c)))
     (,boolean? . ,bool-unparser)
     (,number? . ,(lambda (conn n) (number->string n)))
     (,vector? . ,vector-unparser)
     (,pair? . ,list-unparser))))

;; Retrieve type-oids from PostgreSQL:
(define (update-type-parsers! conn #!optional new-type-parsers)
  (let ((type-parsers (or new-type-parsers (pg-connection-type-parsers conn)))
        (ht (make-hash-table))
        (result '()))
    (pg-connection-oid-parsers-set! conn ht)
    (pg-connection-type-parsers-set! conn type-parsers)
    (unless (null? type-parsers)   ; empty IN () clause is not allowed
      (row-for-each*
       (lambda (oid typname)
         (and-let* ((procedure (assoc typname type-parsers)))
           (hash-table-set! ht (string->number oid) (cdr procedure))))
       (query* conn (sprintf
                     "SELECT oid, typname FROM pg_type WHERE typname IN (~A)"
                     (in-list (length type-parsers)))
        (map car type-parsers) raw: #t)))))

(define (update-type-unparsers! conn new-type-unparsers)
  (pg-connection-type-unparsers-set! conn new-type-unparsers))

;;;;;;;;;;;;;;;;;;;;
;;;; Connections
;;;;;;;;;;;;;;;;;;;;

(define-record
  pg-connection ptr
  type-parsers oid-parsers type-unparsers
  transaction-depth)
(define connection? pg-connection?)
(define type-parsers pg-connection-type-parsers)
(define type-unparsers pg-connection-type-unparsers)

(define (pgsql-connection->fd conn)
  ((foreign-lambda int PQsocket pgconn*) (pg-connection-ptr conn)))

(define (wait-for-connection! conn poll-function)
  (let ((conn-fd (pgsql-connection->fd conn))
        (conn-ptr (pg-connection-ptr conn)))
    (let loop ((result (poll-function conn-ptr)))
      (cond ((= result PGRES_POLLING_OK) (void))
            ((= result PGRES_POLLING_FAILED)
             (let ((error-message (PQerrorMessage conn-ptr)))
               (disconnect conn)
               (postgresql-error 'connect
                                 (conc "Polling Postgres database failed. "
                                       error-message) conn)))
            ((member result (list PGRES_POLLING_WRITING PGRES_POLLING_READING))
             (thread-wait-for-i/o! conn-fd (if (= PGRES_POLLING_READING result)
                                               #:input
                                               #:output))
             (loop (poll-function conn-ptr)))
            (else
             (postgresql-error 'connect (conc "Unknown status code!") conn))))))

(define (alist->connection-spec alist)
  (string-join
   (map (lambda (subspec)
          (sprintf "~A='~A'"
                   (car subspec) ;; this had better not contain [ =\']
                   (string-translate* (->string (cdr subspec))
                                      '(("\\" . "\\\\") ("'" . "\\'")))))
        alist)))

(define (connect connection-spec
                 #!optional
                 (type-parsers (default-type-parsers))
                 (type-unparsers (default-type-unparsers)))
  (let* ((connection-spec (if (string? connection-spec)
                              connection-spec
                              (alist->connection-spec connection-spec)))
         (conn-ptr (PQconnectStart connection-spec)))
    (cond
     ((not conn-ptr)
      (postgresql-error 'connect
                        "Unable to allocate a Postgres connection structure."
                        connection-spec))
     ((= (foreign-value "CONNECTION_BAD" int) (PQstatus conn-ptr))
      (let ((error-message (PQerrorMessage conn-ptr)))
        (PQfinish conn-ptr)
        (postgresql-error 'connect
                          (conc "Connection to Postgres database failed: "
                                error-message)
                          connection-spec)))
     (else
      (let ((conn (make-pg-connection conn-ptr type-parsers
                                      (make-hash-table) type-unparsers 0)))
        ;; We don't want libpq to piss in our stderr stream
        ((foreign-lambda* void ((pgconn* conn))
          "PQsetNoticeReceiver(conn, nullNoticeReceiver, NULL);") conn-ptr)
        (wait-for-connection! conn PQconnectPoll)
        (set-finalizer! conn disconnect)
        ;; Retrieve type-information from PostgreSQL metadata for use by
        ;; the various value-parsers.
        (update-type-parsers! conn)
        conn)))))

(define (reset-connection connection)
  (let ((conn-ptr (pg-connection-ptr connection)))
    (if (PQresetStart conn-ptr) ;; Update oid-parsers?
        (wait-for-connection! connection PQresetPoll)
        (let ((error-message (PQerrorMessage conn-ptr)))
          (disconnect connection)
          (postgresql-error 'reset-connection
                            (conc "Reset of connection failed " error-message)
                            connection)))))

(define (disconnect connection)
  (and-let* ((conn-ptr (pg-connection-ptr connection)))
    (pg-connection-ptr-set! connection #f)
    (pg-connection-type-parsers-set! connection #f)
    (pg-connection-oid-parsers-set! connection #f)
    (PQfinish conn-ptr))
  (void))

;;;;;;;;;;;;;;;
;;;; Results
;;;;;;;;;;;;;;;

(define-record pg-result ptr value-parsers)
(define result? pg-result?)

(define (clear-result! result)
  (and-let* ((result-ptr (pg-result-ptr result)))
    (pg-result-ptr-set! result #f)
    (PQclear result-ptr)))

(define (row-count result)
  (PQntuples (pg-result-ptr result)))

(define (column-count result)
  (PQnfields (pg-result-ptr result)))

;; Helper procedures for bounds checking; so we can distinguish between
;; out of bounds and nonexistant columns, and signal it.
(define (check-column-index! result index location)
  (when (>= index (column-count result))
    (postgresql-error
     location (sprintf "Result column ~A out of bounds" index) result index)))

(define (check-row-index! result index location)
  (when (>= index (row-count result))
    (postgresql-error
     location (sprintf "Result row ~A out of bounds" index) result index)))

(define (column-name result index)
  (check-column-index! result index 'column-name)
  (string->symbol (PQfname (pg-result-ptr result) index)))

(define (column-names result)
  (let ((ptr (pg-result-ptr result)))
   (let loop ((names '())
              (column (column-count result)))
     (if (= column 0)
         names
         (loop (cons (string->symbol (PQfname ptr (sub1 column))) names)
               (sub1 column))))))

(define (column-index result name)
  (let ((idx (PQfnumber (pg-result-ptr result) (symbol->string name))))
    (and (>= idx 0) idx)))

(define (table-oid result index)
  (check-column-index! result index 'table-oid)
  (let ((oid (PQftable (pg-result-ptr result) index)))
    (and (not (= oid invalid-oid)) oid)))

;; Fixes the off-by-1 unexpectedness in libpq/the protocol to make it more
;; consistent with the rest of Scheme.  However, this is inconsistent with
;; almost all other PostgreSQL interfaces...
(define (table-column-index result index)
  (check-column-index! result index 'table-column-index)
  (let ((idx (PQftablecol (pg-result-ptr result) index)))
    (and (> idx 0) (sub1 idx))))

(define format-table
  '((0 . text) (1 . binary)))

(define (format->symbol format)
  (or (alist-ref format format-table eq?)
      (postgresql-error 'format->symbol "Unknown format" format)))

(define (symbol->format symbol)
  (or (and-let* ((res (rassoc symbol format-table eq?)))
        (car res))
      (postgresql-error 'format->symbol "Unknown format" symbol)))

(define (column-format result index)
  (check-column-index! result index 'column-format)
  (format->symbol (PQfformat (pg-result-ptr result) index)))

(define (column-type result index)
  (check-column-index! result index 'column-type)
  (PQftype (pg-result-ptr result) index))

;; This is really not super-useful as it requires intimate knowledge
;; about the internal implementations of types in PostgreSQL.
(define (column-type-modifier result index)
  (check-column-index! result index 'column-type)
  (let ((mod (PQfmod (pg-result-ptr result) index)))
    (and (>= mod 0) mod)))

;; Unchecked version, for speed
(define (value-at* result column row #!key raw)
  (if (PQgetisnull (pg-result-ptr result) row column)
      (sql-null)
      (let ((value ((foreign-safe-lambda*
                     scheme-object ((c-pointer res) (int row) (int col))
                     "C_word fin, *str; char *val; int len;"
                     "len = PQgetlength(res, row, col);"
                     "str = C_alloc(C_bytestowords(len + sizeof(C_header)));"
                     "val = PQgetvalue(res, row, col);"
                     "fin = C_string(&str, len, val);"
                     "if (PQfformat(res, col) == 1) /* binary? */"
                     "        C_string_to_bytevector(fin);"
                     "C_return(fin);")
                    (pg-result-ptr result) row column)))
        (if (or raw (blob? value))
            value
            ((vector-ref (pg-result-value-parsers result) column) value)))))

(define (value-at result #!optional (column 0) (row 0) #!key raw)
  (check-row-index! result row 'value)
  (check-column-index! result column 'value)
  (value-at* result column row raw: raw))

(define (row-values result #!optional (row 0) #!key raw)
  (check-row-index! result row 'row)
  (let loop ((list '())
             (column (column-count result)))
    (if (= column 0)
        list
        (loop (cons (value-at* result (sub1 column) row raw: raw) list)
              (sub1 column)))))

(define (column-values result #!optional (column 0) #!key raw)
  (check-column-index! result column 'column)
  (let loop ((list '())
             (row (row-count result)))
    (if (= row 0)
        list
        (loop (cons (value-at* result column (sub1 row) raw: raw) list)
              (sub1 row)))))

;; (define (row-alist result #!optional (row 0) #!key raw)
;;   (map cons (column-names result) (row-values result row raw: raw)))
(define (row-alist result #!optional (row 0) #!key raw)
  (check-row-index! result row 'row-alist)
  (let loop ((alist '())
             (column (column-count result)))
    (if (= column 0)
        alist
        (loop (cons (cons (string->symbol
                           (PQfname (pg-result-ptr result) (sub1 column)))
                          (value-at* result (sub1 column) row raw: raw)) alist)
              (sub1 column)))))

;;; TODO: Do we want/need PQnparams and PQparamtype bindings?

(define (affected-rows result)
  (string->number (PQcmdTuples (pg-result-ptr result))))

(define (inserted-oid result)
  (let ((oid (PQoidValue (pg-result-ptr result))))
    (and (not (= oid invalid-oid)) oid)))


;;;;;;;;;;;;;;;;;;;;;;;;
;;;; Query procedures
;;;;;;;;;;;;;;;;;;;;;;;;

;; Buffer all available input, yielding if nothing is available:
(define (buffer-available-input! conn)
  (let ((conn-ptr (pg-connection-ptr conn))
        (conn-fd (pgsql-connection->fd conn)))
    (let loop ()
      (if (PQconsumeInput conn-ptr)
          (when (PQisBusy conn-ptr)
            (thread-wait-for-i/o! conn-fd #:input)
            (loop))
          (postgresql-error 'buffer-available-input!
                            (conc "Error reading reply from server. "
                                  (PQerrorMessage conn-ptr))
                            conn)))))

;; Here be more dragons
(define (resolve-unknown-types! conn oids)
  (unless (null? oids)
    (let* ((parsers (pg-connection-oid-parsers conn))
           (q (conc "SELECT t.oid, t.typtype, t.typelem, t.typdelim, "
                    "       t.typbasetype, t.typcategory, a.attrelid, a.atttypid "
                    "FROM pg_type t "
                    "     LEFT JOIN pg_attribute a "
                    "     ON t.typrelid = a.attrelid AND a.attnum > 0 "
                    "WHERE t.oid IN (~A)  "
                    "ORDER BY COALESCE(t.typrelid,-1) ASC, a.attnum ASC"))
           (result (query* conn (sprintf q (in-list (length oids)))
                           (map number->string oids) raw: #t))
           (count (row-count result)))
      (let dissect-types ((unknown-oids (list))
                          (pos 0)
                          (domains (list))
                          (arrays (list))
                          (classes (list))
                          (last-class 0))
        (cond
         ((>= pos count)     ; Done scanning rows?
          ;; Keep going until all oids are resolved
          (resolve-unknown-types! conn unknown-oids)
          ;; Postprocessing step: resolve all nested types
          (for-each (lambda (d)
                      (and-let* ((p (hash-table-ref/default parsers (cdr d) #f)))
                        (hash-table-set! parsers (car d) p)))
                    domains)
          (for-each (lambda (a)
                      (and-let* ((p (hash-table-ref/default parsers (cddr a) #f)))
                        (hash-table-set! parsers (car a)
                                         (make-array-parser p (cadr a)))))
                    arrays)
          (for-each
           (lambda (c)
             (and-let* ((p-list
                         (fold
                          (lambda (att l)
                            (and-let* ((l)
                                       (p (hash-table-ref/default parsers att #f)))
                              (cons p l)))
                          '()
                          (cdr c))))
               (hash-table-set! parsers (car c)
                                (make-composite-parser p-list))))
           classes))
         ((not (string=? (value-at* result 4 pos) "0")) ; Domain type?
          (let* ((basetype-oid (string->number (value-at* result 4 pos)))
                 (parser (hash-table-ref/default parsers basetype-oid #f))
                 (oid (string->number (value-at* result 0 pos))))
            (dissect-types (if parser
                               unknown-oids
                               (cons basetype-oid unknown-oids))
                           (add1 pos) (cons (cons oid basetype-oid) domains)
                           arrays classes last-class)))
         ((string=? (value-at* result 5 pos) "A") ; Array value?
          (let* ((elem (string->number (value-at* result 2 pos)))
                 (delim (string-ref (value-at* result 3 pos) 0))
                 (parser (hash-table-ref/default parsers elem #f))
                 (oid (string->number (value-at* result 0 pos))))
            (dissect-types (if parser
                               unknown-oids
                               (cons elem unknown-oids))
                           (add1 pos) domains
                           (cons (cons oid (cons delim elem)) arrays)
                           classes last-class)))
         ((string=? (value-at* result 1 pos) "c") ; Class? (i.e., table or type)
          (let* ((classid (string->number (value-at* result 6 pos)))
                 (attrid (string->number (value-at* result 7 pos)))
                 (parser (hash-table-ref/default parsers attrid #f)))
            (dissect-types (if parser
                               unknown-oids
                               (cons attrid unknown-oids))
                           (add1 pos) domains arrays
                           ;; Keep oid at the front of the list, insert this
                           ;; attr after it, before the other attrs, if any.
                           (if (= last-class classid)
                               (cons (cons (caar classes)
                                           (cons attrid (cdar classes)))
                                     (cdr classes))
                               (cons (cons (string->number
                                            (value-at* result 0 pos))
                                           (list attrid)) classes))
                           classid)))
         (else
          (dissect-types unknown-oids (add1 pos)
                         domains arrays classes last-class)))))))

(define (make-value-parsers conn pqresult #!key raw)
  (let* ((nfields (PQnfields pqresult))
         (parsers (make-vector nfields))
         (ht (pg-connection-oid-parsers conn)))
    (let loop ((col 0)
               (unknowns (list)))
      (if (= col nfields)
          (begin
            (resolve-unknown-types! conn (map cdr unknowns))
            (for-each (lambda (unknown)
                        (let* ((col (car unknown))
                               (oid (cdr unknown))
                               (parser (hash-table-ref/default ht oid identity)))
                          (vector-set! parsers col parser)))
                      unknowns)
            parsers)
          (let* ((oid (PQftype pqresult col))
                 (parser (if raw identity (hash-table-ref/default ht oid #f))))
            (vector-set! parsers col parser)
            (loop (add1 col) (if parser
                                 unknowns
                                 (cons (cons col oid) unknowns))))))))

;; Collect the result pointers from the last query.
;;
;; A pgresult represents an entire resultset and is always read into memory
;; all at once.
(define (get-last-result conn #!key raw)
  (buffer-available-input! conn)
  (let* ((conn-ptr (pg-connection-ptr conn))
         ;; Read out all remaining results (including the current one).
         ;; TODO: Is this really needed? libpq does it (in pqExecFinish),
         ;; but ostensibly only to concatenate the error messages for
         ;; each query.  OTOH, maybe we want to do that, too.
         (clean-results! (lambda (result)
                           (let loop ((result result))
                             (when result
                               (PQclear result)
                               (loop (PQgetResult conn-ptr))))))
         (result (PQgetResult conn-ptr))
         (status (PQresultStatus result)))
    (cond
     ((not result) (postgresql-error
                    'get-last-result
                    "Internal error! No result object available from server"
                    conn))
     ((member status (list PGRES_BAD_RESPONSE PGRES_FATAL_ERROR
                           PGRES_NONFATAL_ERROR))
      (let* ((get-error-field (lambda (d) (PQresultErrorField result d)))
             (sqlstate (get-error-field PG_DIAG_SQLSTATE))
             (maybe-severity (get-error-field PG_DIAG_SEVERITY))
             (maybe-statement-position
              (get-error-field PG_DIAG_STATEMENT_POSITION))
             (condition
              (make-pg-condition
               'get-last-result
               (PQresultErrorMessage result)
               severity:           (and maybe-severity
                                        (string->symbol
                                         (string-downcase maybe-severity)))
               error-class:        (and sqlstate (string-take sqlstate 2))
               error-code:         sqlstate
               message-detail:     (get-error-field PG_DIAG_MESSAGE_DETAIL)
               message-hint:       (get-error-field PG_DIAG_MESSAGE_HINT)
               statement-position: (and maybe-statement-position
                                        (string->number
                                         maybe-statement-position))
               context:            (get-error-field PG_DIAG_CONTEXT)
               source-file:        (get-error-field PG_DIAG_SOURCE_FILE)
               source-line:        (get-error-field PG_DIAG_SOURCE_LINE)
               source-function:    (get-error-field PG_DIAG_SOURCE_FUNCTION))))
        (clean-results! result)
        (signal condition)))
     ((member status (list PGRES_COPY_OUT PGRES_COPY_IN))
      ;; These are weird; A COPY puts the connection in "copy mode".
      ;; As long as it's in "copy mode", pqgetresult will return the
      ;; same result every time you call it, so don't try to call
      ;; clean-results!
      (let ((result-obj (make-pg-result result #f)))
        (set-finalizer! result-obj clear-result!)
        result-obj))
     ((member status (list PGRES_EMPTY_QUERY PGRES_COMMAND_OK
                           PGRES_TUPLES_OK))
      (let ((result-obj (make-pg-result result #f)))
        (set-finalizer! result-obj clear-result!)
        (let ((trailing-result (PQgetResult conn-ptr)))
          (when trailing-result
            (clean-results! trailing-result)
            (postgresql-error 'get-last-result
                              (conc "Internal error! Unexpected extra "
                                    "results after first query result")
                              conn)))
        (pg-result-value-parsers-set!
         result-obj (make-value-parsers conn result raw: raw))
        result-obj))
     (else (postgresql-error 'get-last-result
                             (conc "Internal error! Unknown status code: "
                                   status)
                             conn)))))

(define (query conn query . params)
  (query* conn query params))

(define (query* conn query #!optional (params '()) #!key (format 'text) raw)
  (let* ((params ;; Check all params and ensure they are proper pairs
          (map ;; See if this can be moved into C
           (lambda (p)
             (let ((obj (if raw p (scheme-value->db-value conn p))))
               (when (and (not (string? obj))
                          (not (blob? obj))
                          (not (sql-null? obj)))
                 (postgresql-error
                  'query*
                  (sprintf "Param value is not string, sql-null or blob: ~S" p)
                  conn query params format))
               (if (sql-null? obj) #f obj))) params))
         (send-query
          (foreign-lambda*
           bool ((pgconn* conn) (nonnull-c-string query)
                 (int num) (scheme-object params) (int resfmt))
           "int res = 0, i = 0, *lens = NULL;"
           "char **vals = NULL;"
           "int *fmts = NULL;"
           "C_word obj, cons;"
           "if (num > 0) {"
           "    vals = C_malloc(num * sizeof(char *));"
           "    lens = C_malloc(num * sizeof(int));"
           "    fmts = C_malloc(num * sizeof(int));"
           "}"
           "for (i=0,cons=params; i < num; ++i,cons=C_u_i_cdr(cons)) {"
           "    obj = C_u_i_car(cons);"
           "    if (obj == C_SCHEME_FALSE) {"
           "        fmts[i] = 0; /* don't care */"
           "        lens[i] = 0;"
           "        vals[i] = NULL;"
           "    } else if (C_header_bits(obj) == C_BYTEVECTOR_TYPE) {"
           "        fmts[i] = 1; /* binary */"
           "        lens[i] = C_header_size(obj);"
           "        vals[i] = C_c_string(obj);"
           "    } else {"
           "        /* text needs to be copied; it expects ASCIIZ */"
           "        fmts[i] = 0; /* text */"
           "        lens[i] = C_header_size(obj);"
           "        vals[i] = malloc(lens[i] + 1);"
           "        memcpy(vals[i], C_c_string(obj), lens[i]);"
           "        vals[i][lens[i]] = '\\0';"
           "    }"
           "}"
           "res = PQsendQueryParams((PGconn *)conn, query, num, NULL,"
           "                        (const char * const *)vals, lens, fmts, resfmt);"
           "for (i=0,cons=params; i < num; ++i,cons=C_u_i_cdr(cons)) {"
           "    obj = C_u_i_car(cons);"
           "    if (!C_immediatep(obj) && C_header_bits(obj) == C_STRING_TYPE)"
           "        free(vals[i]); /* Clear copied strings only */"
           "}"
           "if (num > 0) {"
           "    free(fmts);"
           "    free(lens);"
           "    free(vals);"
           "}"
           "C_return(res);")))
   (if (send-query (pg-connection-ptr conn) query
                   (length params) params (symbol->format format))
       (get-last-result conn raw: raw)
       (postgresql-error 'query*
                         (conc "Unable to send query to server. "
                               (PQerrorMessage (pg-connection-ptr conn)))
                         conn query params format))))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;;; Transaction management
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(define (with-transaction conn thunk)
  (let* ((old-depth (pg-connection-transaction-depth conn))
         (rollback!
          (lambda ()
            (if (= old-depth 0)
                (query conn "ROLLBACK")
                ;; We do not *need* to give savepoints unique names,
                ;; but it aids debugging and we know the depth anyway.
                (query conn (conc "ROLLBACK TO SAVEPOINT s_" old-depth)))))
         (commit!
          (lambda ()
            (if (= old-depth 0)
                (query conn "COMMIT")
                (query conn (conc "RELEASE SAVEPOINT s_" old-depth))))))
    (if (= old-depth 0)
        (query conn "BEGIN")
        (query conn (conc "SAVEPOINT s_" old-depth)))
    (pg-connection-transaction-depth-set! conn (add1 old-depth))
    ;; TODO: Add a warning mechanism (using dynamic-wind) for when the
    ;; user tries to jump into/out of transactions with continuations?
    (handle-exceptions exn
      (begin
        (pg-connection-transaction-depth-set! conn old-depth)
        (rollback!)
        (raise exn))
      (let ((res (thunk)))
        (pg-connection-transaction-depth-set! conn old-depth)
        (if res (commit!) (rollback!))
        res))))

(define (in-transaction? conn)
  (> (pg-connection-transaction-depth conn) 0))

;;;;;;;;;;;;;;;;;;;;
;;;; COPY support
;;;;;;;;;;;;;;;;;;;;

(define (put-copy-data conn data)
  (let* ((data (cond
                ((blob? data) data)
                ((string? data) (string->blob data))
                ((u8vector? data) (u8vector->blob/shared data))
                (else (postgresql-error
                       'put-copy-data "Expected a blob, string or u8vector"
                       conn data))))
         (len (blob-size data))
         (conn-ptr (pg-connection-ptr conn))
         (conn-fd (pgsql-connection->fd conn)))
    (let loop ((res (PQputCopyData conn-ptr data len)))
      (cond
       ((= res 0)
        (thread-wait-for-i/o! conn-fd #:output)
        (loop (PQputCopyData conn-ptr data len)))
       ((= res 1) (void))
       ((= res -1) (postgresql-error 'put-copy-data
                                     (conc "Error putting COPY data. "
                                           (PQerrorMessage conn-ptr))
                                     conn))
       (else (postgresql-error
              'put-copy-data
              (conc "Internal error! Unexpected return value: " res)
              conn))))))

(define (put-copy-end conn #!optional error-message)
  (let ((conn-ptr (pg-connection-ptr conn))
        (conn-fd (pgsql-connection->fd conn)))
   (let loop ((res (PQputCopyEnd conn-ptr error-message)))
     (cond
      ((= res 0)
       (thread-wait-for-i/o! conn-fd #:output)
       (loop (PQputCopyEnd conn-ptr error-message)))
      ((= res 1) (get-last-result conn))
      ((= res -1) (postgresql-error 'put-copy-end
                                    (conc "Error ending put COPY data. "
                                          (PQerrorMessage conn-ptr))
                                    conn error-message))
      (else
       (postgresql-error 'put-copy-end
                         (conc "Internal error! Unexpected return value: " res)
                         conn))))))

(define (get-copy-data conn #!key (format 'text))
  (let* ((conn-ptr (pg-connection-ptr conn))
         (conn-fd (pgsql-connection->fd conn)))
    (let loop ()
      (let-location ((res int))
        (let ((data ((foreign-safe-lambda*
                      scheme-object ((pgconn* conn) ((c-pointer int) res)
                                     (int format))
                      "C_word fin = C_SCHEME_FALSE, *str; char *buf; "
                      "*res = PQgetCopyData(conn, &buf, 1); "
                      "if (buf != NULL) { /* res is length */ "
                      "  str = C_alloc(C_bytestowords(*res + sizeof(C_header))); "
                      "  fin = C_string(&str, *res, buf); "
                      "  if (format == 1) "
                      "    C_string_to_bytevector(fin);"
                      "  PQfreemem(buf); "
                      "} "
                      "C_return(fin);")
                     conn-ptr (location res) (symbol->format format))))
          (cond
           ((> res 0) data)
           ((= res 0) (thread-wait-for-i/o! conn-fd #:input) (loop))
           ((= res -1) (get-last-result conn))
           ((= res -2) (postgresql-error 'put-copy-data
                                         (conc "Error putting COPY data. "
                                               (PQerrorMessage conn-ptr))
                                         conn))
           (else (postgresql-error
                  'get-copy-data
                  (conc "Internal error! Unexpected return value: " res)
                  conn))))))))

;;;;;;;;;;;;;;;;;;;;;;
;;;; Value escaping
;;;;;;;;;;;;;;;;;;;;;;

(define (escape-string conn str)
  (define %escape-string-conn
    ;; This could be more efficient by copying straight into a Scheme object.
    ;; Now it's being copied by PQescapeStringConn, and Chicken copies it again.
    ;; This can allocate up to twice as much memory than the string actually
    ;; uses; in extreme cases this could be a problem.
    (foreign-lambda* c-string* ((pgconn* conn) (c-string from) (int flen))
                     "int err = 0; char *to;"
                     "to = malloc(sizeof(char) * (flen * 2 + 1));"
                     "PQescapeStringConn((PGconn *)conn, to, from, (size_t)flen, &err);"
                     "if (err) {"
                     "        free(to);"
                     "        C_return(NULL);"
                     "}"
                     "C_return(to);"))
  (or (%escape-string-conn (pg-connection-ptr conn) str (string-length str))
      (postgresql-error 'escape-string
                        (conc "String escaping failed. "
                              (PQerrorMessage conn)) conn str)))

(define (escape-bytea conn str)
  (define %escape-bytea-conn
    ;; This must copy because libpq returns a malloced ptr...
    (foreign-safe-lambda* scheme-object ((pgconn* conn)
                                         (scheme-pointer from)
                                         (int flen))
                     "size_t tolen=0; C_word res, *fin; unsigned char *esc;"
                     "esc = PQescapeByteaConn((PGconn *)conn, from, (size_t)flen, &tolen);"
                     "if (esc == NULL)"
                     "    C_return(C_SCHEME_FALSE);"
                     "fin = C_alloc(C_bytestowords(tolen + sizeof(C_header)));"
                     "/* tolen includes the resulting NUL byte */"
                     "res = C_string(&fin, tolen - 1, (char *)esc);"
                     "PQfreemem(esc);"
                     "C_return(res);"))
  (or (%escape-bytea-conn (pg-connection-ptr conn) str (string-length str))
      (postgresql-error 'escape-bytea
                        (conc "Byte array escaping failed. "
                              (PQerrorMessage conn)) conn str)))

(define (unescape-bytea str)
  (define %unescape-bytea
    ;; This must copy because libpq returns a malloced ptr...
    (foreign-safe-lambda* scheme-object (((const unsigned-c-string*) from))
                     "size_t tolen=0; C_word res, *fin; unsigned char *unesc;"
                     "unesc = PQunescapeBytea(from, &tolen);"
                     "if (unesc == NULL)"
                     "    C_return(C_SCHEME_FALSE);"
                     "fin = C_alloc(C_bytestowords(tolen + sizeof(C_header)));"
                     "res = C_string(&fin, tolen, (char *)unesc);"
                     "PQfreemem(unesc);"
                     "C_return(res);"
                     ))
  (or (%unescape-bytea str)
      (postgresql-error 'unescape-bytea
                        "Byte array unescaping failed (out of memory?)" str)))


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;;; High-level interface
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(define (make-result-fold item-count extract-item)
  (lambda (kons knil result)
   (let ((items (item-count result)))
     (let loop ((seed knil)
                (item 0))
       (if (= item items)
           seed
           (loop (kons (extract-item result item) seed) (add1 item)))))))

(define row-fold (make-result-fold row-count row-values))
(define (row-fold* kons knil result)
  (row-fold (lambda (values seed)
              (apply kons (append values (list seed)))) knil result))

(define column-fold (make-result-fold column-count column-values))
(define (column-fold* kons knil result)
  (column-fold (lambda (values seed)
                 (apply kons (append values (list seed)))) knil result))


(define (make-result-fold-right item-count extract-item)
  (lambda (kons knil result)
    (let loop ((seed knil)
               (item (item-count result)))
      (if (= item 0)
          seed
          (loop (kons (extract-item result (sub1 item)) seed) (sub1 item))))))

(define row-fold-right (make-result-fold-right row-count row-values))
(define (row-fold-right* kons knil result)
  (row-fold-right (lambda (values seed)
                    (apply kons (append values (list seed)))) knil result))

(define column-fold-right (make-result-fold-right column-count column-values))
(define (column-fold-right* kons knil result)
  (column-fold-right (lambda (values seed)
                       (apply kons (append values (list seed)))) knil result))


(define (row-for-each proc result)
  (row-fold (lambda (values seed) (proc values)) #f result)
  (void))
(define (row-for-each* proc result)
  (row-fold (lambda (values seed) (apply proc values)) #f result)
  (void))

(define (column-for-each proc result)
  (column-fold (lambda (values seed) (proc values)) #f result)
  (void))
(define (column-for-each* proc result)
  (column-fold (lambda (values seed) (apply proc values)) #f result)
  (void))

;; Like regular Scheme map, the order in which the procedure is applied is
;; undefined.  We make good use of that by traversing the resultset from
;; the end back to the beginning, thereby avoiding a reverse! on the result.
(define (row-map proc res)
  (row-fold-right (lambda (row lst) (cons (proc row) lst)) '() res))
(define (row-map* proc res)
  (row-fold-right (lambda (row lst) (cons (apply proc row) lst)) '() res))
(define (column-map proc res)
  (column-fold-right (lambda (col lst) (cons (proc col) lst)) '() res))
(define (column-map* proc res)
  (column-fold-right (lambda (col lst) (cons (apply proc col) lst)) '() res))

(define (result-format result)
  (if (and result ((foreign-lambda bool PQbinaryTuples pgresult*)
                   (pg-result-ptr result)))
     'binary 'text))

(define (copy-query*-fold kons knil conn query
                          #!optional (params '()) #!key (format 'text) raw)
  (let* ((result (query* conn query params format: format raw: raw))
         (data-format (result-format result)))
    (handle-exceptions exn
      (let cleanup () (if (result? (get-copy-data conn)) (raise exn) (cleanup)))
      (let loop ((data (get-copy-data conn format: data-format))
                 (seed knil))
        (if (result? data)
            seed
            ;; Explicit ordering; data could be _very_ big, allow one to be GCed
            (let ((next (kons data seed)))
              (loop (get-copy-data conn format: data-format) next)))))))

(define (copy-query-fold kons knil conn query . params)
  (copy-query*-fold kons knil conn query params))


;; This is slow and memory-intensive if data is big. Provided for completeness
(define (copy-query*-fold-right kons knil conn query
                                #!optional (params '()) #!key (format 'text) raw)
  (let* ((result (query* conn query params format: format raw: raw))
         (data-format (result-format result)))
    (handle-exceptions exn
      (let cleanup () (if (result? (get-copy-data conn)) (raise exn) (cleanup)))
      (let loop ((data (get-copy-data conn format: data-format)))
        (if (result? data)
            knil
            (kons data (loop (get-copy-data conn format: data-format))))))))

(define (copy-query-fold-right kons knil conn query . params)
  (copy-query*-fold-right kons knil conn query params))


(define (copy-query*-map proc conn query
                         #!optional (params '()) #!key (format 'text) raw)
  (reverse! (copy-query*-fold (lambda (data seed) (cons (proc data) seed))
                              '() conn query params format: format raw: raw)))

(define (copy-query-map proc conn query . params)
  (copy-query*-map proc conn query params))


(define (copy-query*-for-each proc conn query
                              #!optional (params '()) #!key (format 'text) raw)
  (copy-query*-fold (lambda (data seed) (proc data))
                    #f conn query params format: format raw: raw)
  (void))

(define (copy-query-for-each proc conn query . params)
  (copy-query*-for-each proc conn query params))

;; A bit of a weird name but consistent
(define (call-with-output-copy-query*
         proc conn query #!optional (params '()) #!key (format 'text) raw)
  (query* conn query params format: format raw: raw)
  (let* ((closed? #f)
         (output-port (make-output-port
                       (lambda (data) (put-copy-data conn data))
                       (lambda () (put-copy-end conn) (set! closed? #t)))))
    (handle-exceptions exn
      (if closed?
          (raise exn)
          (handle-exceptions _
            (raise exn)
            ;; Previously written data will be discarded to guarantee atomicity
            (put-copy-end conn "Chicken PostgreSQL egg -- forcing error")))
      (call-with-values (lambda () (proc output-port))
        (lambda args
          (unless closed? (put-copy-end conn))
          (apply values args))))))

(define (call-with-output-copy-query proc conn query . params)
  (call-with-output-copy-query* proc conn query params))

(define (with-output-to-copy-query*
         thunk conn query #!optional (params '()) #!key (format 'text) raw)
  (call-with-output-copy-query* (lambda (x) (with-output-to-port x thunk))
                                conn query params format: format raw: raw))

(define (with-output-to-copy-query thunk conn query . params)
  (with-output-to-copy-query* thunk conn query params))

)