class Embulk::Input::Presto
Public Class Methods
build_output_columns(task)
click to toggle source
# File lib/embulk/input/presto.rb, line 41 def self.build_output_columns(task) explain_query = "explain (FORMAT TEXT) " + task["query"] Embulk.logger.debug("SQL: #{explain_query}") explain_result = Connection.get_client(task).run("explain (FORMAT TEXT) " + task["query"]) columns = [] ExplainParser.parse(explain_result).each_with_index do |(name, type), i| columns << Column.new(i, name, TypeConverter.get_type(type)) end columns end
resume(task, columns, count) { |task, columns, count| ... }
click to toggle source
# File lib/embulk/input/presto.rb, line 34 def self.resume(task, columns, count, &control) task_reports = yield(task, columns, count) next_config_diff = {} return next_config_diff end
transaction(config, &control)
click to toggle source
# File lib/embulk/input/presto.rb, line 12 def self.transaction(config, &control) task = { "host" => config.param("host", :string, default: "localhost"), "port" => config.param("port", :integer, default: 8080), "schema" => config.param("schema", :string, default: "default"), "catalog" => config.param("catalog", :string, default: "native"), "query" => config.param("query", :string), "user" => config.param("user", :string, default: "embulk"), "columns" => config.param("columns", :array, default: nil) } columns = if task['columns'] task['columns'].each_with_index.map do |c, i| Column.new(i, c["name"], c["type"].to_sym) end else build_output_columns(task) end resume(task, columns, 1, &control) end
Public Instance Methods
init()
click to toggle source
# File lib/embulk/input/presto.rb, line 53 def init @client = Connection.get_client(task) @query = task["query"] @type_converter = TypeConverter.new Embulk.logger.info "SQL: #{@query}" end
run()
click to toggle source
# File lib/embulk/input/presto.rb, line 61 def run size = 0 @client.query(@query) do |q| q.each_row {|row| converted_values = row.map.with_index { |value,i| @type_converter.convert_value(value, schema[i]) } page_builder.add(converted_values) } size = q.rows.size end page_builder.finish task_report = { size: size } return task_report end