在球拍中将SQL查询处理为惰性流

Question

语言:球拍(带SQL查询代码/指针)库: db、球拍/流、球拍/序列

目标:懒惰地使用Racket中的streams处理sql查询的值。

问题1:如何操作球拍中的SQL查询流对象？(我可以获得流的stream-first值，但不能获得流的其余部分！)

#lang racket/base
(require db
         racket/sequence
         racket/stream)

(define db_sql_local
  (mysql-connect
   #:user "<my-username>"
   #:database "<my-database>"
   #:server "<my-server>"
   #:port <my-port>
   #:password "<my-password>"))

;; PROBLEM 1 HERE
(define test-stream
  (sequence->stream
   (in-query
    chembl_sql_local
    "SELECT * FROM <table-name>"
    #:fetch +inf.0)))

(stream-first test-stream)

;; stream-first of test-stream returns the first-row of the table as a '(#vector). 

如有任何建议或意见，我们将不胜感激--谢谢！

Answer 1

首先，in-query返回的序列不包含向量；序列的每个“元素”都包含多个值，每一列返回一个值。请参阅the Sequence docs中以“序列的各个元素...”开头的段落。关于多值元素。

其次，使用#:fetch +inf.0 (默认行为)意味着在返回序列之前获取所有行。因此，上面的代码并没有什么懒惰之处；您可以改用query-rows并获得一个更容易处理的列表(并且query-rows确实将每一行表示为一个向量)。

最后，使用stream-rest获取流的其余部分。例如：

(require db racket/stream racket/sequence)
(define c (sqlite3-connect #:database 'memory))
(define qseq (in-query c "SELECT 1, 2 UNION SELECT 3, 4" #:fetch 1))
qseq
;; => #<sequence>
(define qstream (sequence->stream qseq))
qstream
;; => #<stream>
(stream-first qstream)
;; => 1 2
(stream-rest qstream)
;; => #<stream>
(stream-first (stream-rest qstream))
;; => 3 4

Answer 2

感谢您的快速回复。#:fetch 1 arg绝对是我想让它变得懒惰的原因。我已经附加了更新的代码，应该可以懒惰地流式sql查询以导出tsv文件。

(define sql_server
  (mysql-connect
   #:user <username>
   #:database <db-name>
   #:server <server>
   #:port <port-num>
   #:password <password>))

(define query-->stream
  (lambda (db-conn query)
    (sequence->stream
     (in-query
      db-conn
      query
      #:fetch 1))))

(define print-table-row-to-tsv
  (lambda (ls port)
    (cond
      ((null? ls)
       (fprintf port "~c" #\newline)
       (void))
      ((sql-null? (car ls))
       (fprintf port "~a~c" "NULL" #\tab)
       (print-table-row-to-tsv (cdr ls) port))
      ((null? (cdr ls))
       (fprintf port "~a" (car ls))
       (print-table-row-to-tsv (cdr ls) port))
      (else
       (fprintf port "~a~c" (car ls) #\tab)
       (print-table-row-to-tsv (cdr ls) port)))))

(define get-table-col-names
  (lambda (db-conn tbl-name)
    (map (lambda (x) (vector-ref x 0))
       (query-rows db-conn (string-append "DESCRIBE " tbl-name)))))

(define export-query-result-to-tsv
  (lambda (db-conn tbl-name query)
    (let* ((tbl-col-names (get-table-col-names db-conn tbl-name))
           (output-file (open-output-file (format "~achembl_~a_table.tsv" (find-system-path 'home-dir) tbl-name) #:exists 'replace))        
           (stream (query-->stream db-conn query)))
      (begin
        (print-table-row-to-tsv tbl-col-names output-file)
        (process-stream-to-tsv stream output-file)
        (close-output-port output-file)))))

(define process-stream-to-tsv
  (lambda (stream port)
    (cond
      ((stream-empty? stream)
       (void))
      (else
       (begin
         (print-table-row-to-tsv (call-with-values (lambda () (stream-first stream)) list) port)
         (process-stream-to-tsv (stream-rest stream) port))))))


(export-query-result-to-tsv sql_server "<table-name>" "SELECT * FROM <table-name>;")