class Bricolage::PSQLTask

We don't support dynamodb still now

Constants

GRANT_OPTS

Public Instance Methods

analyze_if(enabled, target = '${dest_table}') click to toggle source
# File lib/bricolage/psqldatasource.rb, line 293
def analyze_if(enabled, target = '${dest_table}')
  exec SQLStatement.for_string("analyze #{target};") if enabled
end
copy_statement(src_ds, src_path, dest_table, format, jsonpath, opts) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 327
def copy_statement(src_ds, src_path, dest_table, format, jsonpath, opts)
  unless src_ds.redshift_loader_source?
    raise ParameterError, "input data source does not support redshift as bulk loading source: #{src_ds.name}"
  end
  buf = StringIO.new
  buf.puts "copy #{dest_table}"
  buf.puts "from '#{src_ds.url(src_path)}'"
  buf.puts "credentials '#{src_ds.credential_string}'"
  buf.puts format_option(format, src_ds, jsonpath)
  opts.each do |opt|
    buf.puts opt.to_s
  end
  buf.puts ';'
  buf.string
end
create_dummy_table(target) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 229
def create_dummy_table(target)
  exec SQLStatement.for_string(
    "create table if not exists #{target} (x int);\n"
  )
end
drop(target_table) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 235
def drop(target_table)
  exec SQLStatement.for_string("drop table #{target_table} cascade;")
end
drop_force(target_table) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 249
def drop_force(target_table)
  drop_obj_force('table', target_table)
end
drop_force_if(enabled) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 257
def drop_force_if(enabled)
  drop_force('${dest_table}') if enabled
end
drop_if(enabled) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 239
def drop_if(enabled)
  drop '${dest_table}' if enabled
end
drop_obj_force(type, name) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 243
def drop_obj_force(type, name)
  exec SQLStatement.for_string(
    "drop #{type} if exists #{name} cascade;\n"
  )
end
drop_view_force(target_view) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 253
def drop_view_force(target_view)
  drop_obj_force('view', target_view)
end
drop_view_force_if(enabled) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 261
def drop_view_force_if(enabled)
  drop_view_force('${dest_table}') if enabled
end
each_statement() { |statement| ... } click to toggle source
# File lib/bricolage/psqldatasource.rb, line 169
def each_statement
  each_action do |action|
    yield action.statement
  end
end
exec(stmt) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 165
def exec(stmt)
  add SQLAction.new(stmt)
end
explain_source() click to toggle source
# File lib/bricolage/psqldatasource.rb, line 208
def explain_source
  buf = StringIO.new
  each_statement do |stmt|
    buf.puts
    buf.puts "-- #{stmt.location}" if stmt.location
    if support_explain?(stmt.kind)
      buf.puts "explain #{stmt.stripped_source}"
    else
      buf.puts "/* #{stmt.stripped_source} */"
    end
  end
  buf.string
end
format_option(fmt, src_ds, jsonpath) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 343
def format_option(fmt, src_ds, jsonpath)
  case fmt
  when 'tsv'
    %q(delimiter '\t')
  when 'csv'
    %q(delimiter ',')
  when 'json'
    "json '#{json_param(jsonpath)}'"
  else
    raise ParameterError, "unsupported format: #{fmt}"
  end
end
format_query(query) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 396
def format_query(query)
  query.gsub(/^--.*/, '').strip.gsub(/[ \t]*\n[ \t]*/, ' ').gsub("'", "\\\\'")
end
grant(privilege:, on:, to:) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 297
def grant(privilege:, on:, to:)
  exec SQLStatement.for_string("grant #{privilege} on #{on} to #{to};")
end
grant_if(opts, target) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 303
def grant_if(opts, target)
  return unless opts
  return if opts.empty?
  unknown_keys = opts.keys - GRANT_OPTS
  raise ParameterError, "unknown grant options: #{unknown_keys.inspect}" unless unknown_keys.empty?
  missing_keys = GRANT_OPTS - opts.keys
  raise ParameterError, %Q(missing grant options: #{missing_keys.inspect}) unless missing_keys.empty?
  args = {on: target}
  opts.each do |k, v|
    args[k.intern] = v
  end
  grant(**args)
end
json_param(jsonpath) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 356
def json_param(jsonpath)
  case jsonpath
  when nil
    'auto'
  when %r{\As3://}
    jsonpath
  else
    src_ds.url(jsonpath)
  end
end
load(src_ds, src_path, dest_table, format, jsonpath, opts) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 323
def load(src_ds, src_path, dest_table, format, jsonpath, opts)
  exec SQLStatement.for_string(copy_statement(src_ds, src_path, dest_table, format, jsonpath, opts))
end
rename_table(src, dest) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 269
def rename_table(src, dest)
  exec SQLStatement.for_string("alter table #{src} rename to #{dest};")
end
run() click to toggle source

override

# File lib/bricolage/psqldatasource.rb, line 198
def run
  VacuumLock.using {
    @ds.execute source
  }
end
run_explain() click to toggle source
# File lib/bricolage/psqldatasource.rb, line 204
def run_explain
  @ds.execute explain_source
end
serialize_vacuum() { || ... } click to toggle source
# File lib/bricolage/psqldatasource.rb, line 287
def serialize_vacuum   # override
  exec SQLStatement.for_string psql_serialize_vacuum_begin
  yield
  exec SQLStatement.for_string psql_serialize_vacuum_end
end
source() click to toggle source

override

# File lib/bricolage/psqldatasource.rb, line 186
def source
  buf = StringIO.new
  buf.puts '\timing on'
  each_statement do |stmt|
    buf.puts
    buf.puts "/* #{stmt.location} */" if stmt.location
    buf.puts stmt.stripped_source
  end
  buf.string
end
support_explain?(statement_kind) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 222
def support_explain?(statement_kind)
  case statement_kind
  when 'select', 'insert', 'update', 'delete' then true
  else false
  end
end
transaction() { || ... } click to toggle source
# File lib/bricolage/psqldatasource.rb, line 317
def transaction
  exec SQLStatement.for_string('begin transaction;')
  yield
  exec SQLStatement.for_string('commit;')
end
truncate_if(enabled, target = '${dest_table}') click to toggle source
# File lib/bricolage/psqldatasource.rb, line 265
def truncate_if(enabled, target = '${dest_table}')
  exec SQLStatement.for_string("truncate #{target};") if enabled
end
unload(stmt, dest_ds, dest_path, format, opts) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 367
def unload(stmt, dest_ds, dest_path, format, opts)
  exec unload_statement(stmt, dest_ds, dest_path, format, opts)
end
unload_format_option(format, ds) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 385
def unload_format_option(format, ds)
  case format
  when 'tsv'
    %q(delimiter '\t')
  when 'csv'
    %q(delimiter ',')
  else
    raise ParameterError, "unsupported format: #{fmt}"
  end
end
unload_statement(stmt, dest_ds, dest_path, format, opts) click to toggle source
# File lib/bricolage/psqldatasource.rb, line 371
def unload_statement(stmt, dest_ds, dest_path, format, opts)
  buf = StringIO.new
  buf.puts "unload ('#{format_query(stmt.stripped_raw_content)}')"
  buf.puts "to '#{dest_ds.url(dest_path)}'"
  buf.puts "credentials '#{dest_ds.credential_string}'"
  buf.puts unload_format_option(format, dest_ds)
  opts.each do |opt|
    buf.puts opt.to_s
  end
  buf.puts ';'
  res = StringResource.new(buf.string, stmt.location)
  SQLStatement.new(res, stmt.declarations)
end
vacuum_if(enable_vacuum, enable_vacuum_sort, target = '${dest_table}') click to toggle source
# File lib/bricolage/psqldatasource.rb, line 273
def vacuum_if(enable_vacuum, enable_vacuum_sort, target = '${dest_table}')
  if enable_vacuum
    serialize_vacuum {
      exec SQLStatement.for_string("vacuum #{target};")
    }
  elsif enable_vacuum_sort
    serialize_vacuum {
      exec SQLStatement.for_string("vacuum sort only #{target};")
    }
  end
end