Rails原始查询CSV格式，通过控制器返回。

Question

我使用Active Record获取我的故事，然后生成一个CSV，这是Rails Cast中的标准方式。但是我的行数很多，需要花费数分钟的时间。我认为如果我可以让PostgreSQL来进行CSV渲染，那么我就可以节省一些时间。

这是我目前拥有的内容：

query = "COPY stories TO STDOUT WITH CSV HEADER;"
results = ActiveRecord::Base.connection.execute(query);

但是此查询没有结果：

 => #<PG::Result:0x00000006ea0488 @connection=#<PG::Connection:0x00000006c62fb8 @socket_io=nil, @notice_receiver=nil, @notice_processor=nil>> 
2.0.0-p247 :053 > result.count
 => 0 

更好的了解方式：

2.0.0-p247 :059 >   result.to_json
 => "[]" 

我猜测我的控制器会长成这个样子：

format.csv { send_data raw_results }

这适用于普通查询，我只是无法弄清楚SQL语法，以便将CSV结果返回到Rails。

更新:

将CSV导出时间从120000毫秒降至290毫秒

我的模型：

def self.to_csv(story_ids)

    csv  = []
    conn = ActiveRecord::Base.connection.raw_connection
    conn.copy_data("COPY (SELECT * FROM stories WHERE stories.id IN (#{story_ids.join(',')})) TO STDOUT WITH (FORMAT CSV, HEADER TRUE, FORCE_QUOTE *, ESCAPE E'\\\\');") do
      while row = conn.get_copy_data
        csv.push(row)
      end
    end
    csv.join("\r\n")
  end

我的控制器：

send_data Story.to_csv(Story.order(:created_at).pluck(:id))

- penner

---

有没有办法直接从数据库中发送数据，而不必将其保存到“csv”数组中？ - Fernando Fabreti

@FernandoFabreti 听起来像是copy_data函数返回需要合并成一个文件的行。我认为没有任何方法可以在没有某种变量赋值的情况下组合这些行。你可能可以从一开始就使用字符串，并在循环中进行追加。对性能差异感兴趣。 - penner

我不得不将 csv.join("\r\n") 更改为 csv.join("\n")，以便正确生成行。最初它会添加一个额外的换行符。不确定这是否会影响其他非*nix机器... - allthesignals

@penner 对我也非常有效，感谢你的更新！不过有两个问题：

1. 当一行由多个涉及关联的复杂 AR 查询生成时，我们如何生成单个 SQL 查询并在上面的示例中传递它？
2. 虽然它肯定会影响时间性能，但它是否也会影响操作使用的内存？

- Aayush Kothari

@FernandoFabreti 我最终将答案封装成了一个枚举器，然后将其作为参数传递给 self.response_body，就像这里使用的那样（https://coderwall.com/p/kad56a/streaming-large-data-responses-with-rails）。链接中的示例不完整，需要在生成行之前添加 lines << "#{row.length.to_s(16)}\r\n" 以使分块响应正常工作。 - mlt

Answer 1

据我所知，你需要在底层的PostgreSQL数据库连接上使用copy_data方法来实现这一点：

  

- (Object) copy_data(sql)

     

调用序列：

conn.copy_data( sql ) {|sql_result| ... } -> PG::Result

执行复制过程以将数据从或传输至服务器。

此将通过 #exec 命令发出 SQL COPY 命令。如果命令没有错误，则响应是一个 PG::Result 对象，该对象带有 PGRES_COPY_OUT 或 PGRES_COPY_IN 状态代码（取决于指定的复制方向），并将传递给块。然后，应用程序应使用 #put_copy_data 或 #get_copy_data 接收或传输数据行，并在完成后从该块中返回。

甚至还有一个示例：

conn.copy_data "COPY my_table TO STDOUT CSV" do
  while row=conn.get_copy_data
    p row
  end
end

ActiveRecord对原始数据库连接的包装器不知道copy_data是什么，但您可以使用raw_connection来取消包装它：

conn = ActiveRecord::Base.connection.raw_connection
csv  = [ ]
conn.copy_data('copy stories to stdout with csv header') do
  while row = conn.get_copy_data
    csv.push(row)
  end
end

这将会给你一个由CSV字符串组成的数组，存储在csv中（每个数组元素对应一行CSV），你可以使用csv.join("\r\n")来获取最终的CSV数据。

- mu is too short

---

最终不得不使用另一种查询方式，这种方式可以更好地转义数据。conn.copy_data("COPY stories TO STDOUT WITH (FORMAT CSV, HEADER TRUE, FORCE_QUOTE *, ESCAPE E'\\');")。感谢您的帮助！ - penner

Answer 2

这个答案是在 @mu-is-too-short 提供的 答案 基础上构建的，但是不使用临时对象，而是使用 流式处理。

headers['X-Accel-Buffering'] = 'no'
headers["Cache-Control"] = 'no-cache'
headers["Transfer-Encoding"] = 'chunked'
headers['Content-Type'] = 'text/csv; charset=utf-8'
headers['Content-Disposition'] = 'inline; filename="data.csv"'
headers.delete('Content-Length')
sql = "SELECT * FROM stories WHERE stories.id IN (#{story_ids.join(',')})"
self.response_body = Enumerator.new do |chunk|
  conn = ActiveRecord::Base.connection.raw_connection
  conn.copy_data("COPY (#{sql.chomp(';')}) TO STDOUT WITH (FORMAT CSV, HEADER TRUE, RCE_QUOTE *, ESCAPE E'\\\\');") do
    while row = conn.get_copy_data
      chunk << "#{row.length.to_s(16)}\r\n"
      chunk << row
      chunk << "\r\n"
    end
    chunk << "0\r\n\r\n"
  end
end

你也可以使用gz = Zlib::GzipWriter.new(Stream.new(chunk))和gz.write row，其中类似于：

class Stream
  def initialize(block)
    @block = block
  end
  def write(row)
    @block << "#{row.length.to_s(16)}\r\n"
    @block << row
    @block << "\r\n"
  end
end

记得 headers['Content-Encoding'] = 'gzip'。另请参见this gist。

- mlt