Class: Bricolage::PSQLTask

Inherits:
DataSourceTask show all
Includes:
VacuumLock
Defined in:
lib/bricolage/psqldatasource.rb

Overview

We don’t support dynamodb still now

Defined Under Namespace

Classes: SQLAction

Constant Summary collapse

GRANT_OPTS = 
%w[privilege to]

Constants included from VacuumLock

VacuumLock::DEFAULT_VACUUM_LOCK_FILE, VacuumLock::DEFAULT_VACUUM_LOCK_TIMEOUT

Instance Attribute Summary

Attributes inherited from DataSourceTask

#ds

Instance Method Summary collapse

Methods included from VacuumLock

cleanup_vacuum_lock, create_lockfile_cmd, create_vacuum_lock_file, enable_vacuum_lock?, locking?, psql_serialize_vacuum_begin, psql_serialize_vacuum_end, serialize_vacuum, using, #using_vacuum_lock, vacuum_lock_parameters

Methods inherited from DataSourceTask

#bind, #initialize

Constructor Details

This class inherits a constructor from Bricolage::DataSourceTask

Instance Method Details

#analyze_if(enabled, target = '${dest_table}') ⇒ Object 



285
286
287
 
# File 'lib/bricolage/psqldatasource.rb', line 285

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) ⇒ Object 



319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
 
# File 'lib/bricolage/psqldatasource.rb', line 319

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
  opts.provide_defaults(src_ds)
  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) ⇒ Object 



221
222
223
224
225
 
# File 'lib/bricolage/psqldatasource.rb', line 221

def create_dummy_table(target)
  exec SQLStatement.for_string(
    "create table if not exists #{target} (x int);\n"
  )
end

#drop(target_table) ⇒ Object 



227
228
229
 
# File 'lib/bricolage/psqldatasource.rb', line 227

def drop(target_table)
  exec SQLStatement.for_string("drop table #{target_table} cascade;")
end

#drop_force(target_table) ⇒ Object 



241
242
243
 
# File 'lib/bricolage/psqldatasource.rb', line 241

def drop_force(target_table)
  drop_obj_force('table', target_table)
end

#drop_force_if(enabled) ⇒ Object 



249
250
251
 
# File 'lib/bricolage/psqldatasource.rb', line 249

def drop_force_if(enabled)
  drop_force('${dest_table}') if enabled
end

#drop_if(enabled) ⇒ Object 



231
232
233
 
# File 'lib/bricolage/psqldatasource.rb', line 231

def drop_if(enabled)
  drop '${dest_table}' if enabled
end

#drop_obj_force(type, name) ⇒ Object 



235
236
237
238
239
 
# File 'lib/bricolage/psqldatasource.rb', line 235

def drop_obj_force(type, name)
  exec SQLStatement.for_string(
    "drop #{type} if exists #{name} cascade;\n"
  )
end

#drop_view_force(target_view) ⇒ Object 



245
246
247
 
# File 'lib/bricolage/psqldatasource.rb', line 245

def drop_view_force(target_view)
  drop_obj_force('view', target_view)
end

#drop_view_force_if(enabled) ⇒ Object 



253
254
255
 
# File 'lib/bricolage/psqldatasource.rb', line 253

def drop_view_force_if(enabled)
  drop_view_force('${dest_table}') if enabled
end

#each_statement ⇒ Object 



161
162
163
164
165
 
# File 'lib/bricolage/psqldatasource.rb', line 161

def each_statement
  each_action do |action|
    yield action.statement
  end
end

#exec(stmt) ⇒ Object 



157
158
159
 
# File 'lib/bricolage/psqldatasource.rb', line 157

def exec(stmt)
  add SQLAction.new(stmt)
end

#explain_source ⇒ Object 



200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/bricolage/psqldatasource.rb', line 200

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) ⇒ Object 



336
337
338
339
340
341
342
343
344
345
346
347
 
# File 'lib/bricolage/psqldatasource.rb', line 336

def format_option(fmt, src_ds, jsonpath)
  case fmt
  when 'tsv'
    %q(delimiter '\t')
  when 'csv'
    %q(delimiter ',')
  when 'json'
    jsonpath ? "json \'#{src_ds.url(jsonpath)}\'" : %q(json 'auto')
  else
    raise ParameterError, "unsupported format: #{fmt}"
  end
end

#format_query(query) ⇒ Object 



378
379
380
 
# File 'lib/bricolage/psqldatasource.rb', line 378

def format_query(query)
  query.gsub(/^--.*/, '').strip.gsub(/[ \t]*\n[ \t]*/, ' ').gsub("'", "\\\\'")
end

#grant(privilege:, on:, to:) ⇒ Object 



289
290
291
 
# File 'lib/bricolage/psqldatasource.rb', line 289

def grant(privilege:, on:, to:)
  exec SQLStatement.for_string("grant #{privilege} on #{on} to #{to};")
end

#grant_if(opts, target) ⇒ Object

Raises:



295
296
297
298
299
300
301
302
303
304
305
306
307
 
# File 'lib/bricolage/psqldatasource.rb', line 295

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

#load(src_ds, src_path, dest_table, format, jsonpath, opts) ⇒ Object 



315
316
317
 
# File 'lib/bricolage/psqldatasource.rb', line 315

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) ⇒ Object 



261
262
263
 
# File 'lib/bricolage/psqldatasource.rb', line 261

def rename_table(src, dest)
  exec SQLStatement.for_string("alter table #{src} rename to #{dest};")
end

#run ⇒ Object

override



190
191
192
193
194
 
# File 'lib/bricolage/psqldatasource.rb', line 190

def run
  VacuumLock.using {
    @ds.execute source
  }
end

#run_explain ⇒ Object 



196
197
198
 
# File 'lib/bricolage/psqldatasource.rb', line 196

def run_explain
  @ds.execute explain_source
end

#serialize_vacuum ⇒ Object

override



279
280
281
282
283
 
# File 'lib/bricolage/psqldatasource.rb', line 279

def serialize_vacuum   # override
  exec SQLStatement.for_string psql_serialize_vacuum_begin
  yield
  exec SQLStatement.for_string psql_serialize_vacuum_end
end

#source ⇒ Object

override



178
179
180
181
182
183
184
185
186
187
 
# File 'lib/bricolage/psqldatasource.rb', line 178

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) ⇒ Boolean

Returns:

  • (Boolean) 


214
215
216
217
218
219
 
# File 'lib/bricolage/psqldatasource.rb', line 214

def support_explain?(statement_kind)
  case statement_kind
  when 'select', 'insert', 'update', 'delete' then true
  else false
  end
end

#transaction ⇒ Object 



309
310
311
312
313
 
# File 'lib/bricolage/psqldatasource.rb', line 309

def transaction
  exec SQLStatement.for_string('begin transaction;')
  yield
  exec SQLStatement.for_string('commit;')
end

#truncate_if(enabled, target = '${dest_table}') ⇒ Object 



257
258
259
 
# File 'lib/bricolage/psqldatasource.rb', line 257

def truncate_if(enabled, target = '${dest_table}')
  exec SQLStatement.for_string("truncate #{target};") if enabled
end

#unload(stmt, dest_ds, dest_path, format, opts) ⇒ Object 



349
350
351
 
# File 'lib/bricolage/psqldatasource.rb', line 349

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) ⇒ Object 



367
368
369
370
371
372
373
374
375
376
 
# File 'lib/bricolage/psqldatasource.rb', line 367

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) ⇒ Object 



353
354
355
356
357
358
359
360
361
362
363
364
365
 
# File 'lib/bricolage/psqldatasource.rb', line 353

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}') ⇒ Object 



265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/bricolage/psqldatasource.rb', line 265

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