class Charrington::AlterPostgresTable

Constants

AlterFailed
Error

Attributes

column_types[RW]
columns[R]
connection[R]
event[R]
schema[R]
table_name[R]

Public Class Methods

new(connection, event, schema, table_name, columns) click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 16
def initialize(connection, event, schema, table_name, columns)
  @connection = connection
  @event = event.to_hash
  @table_name = table_name
  @schema = schema
  @columns = columns
  @column_types = []
end

Public Instance Methods

call() click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 25
def call
  set_column_types
  alter_table
  true
rescue => e
  raise AlterFailed, e.message
ensure
  @column_types.clear if @column_types.is_a? Array
end

Private Instance Methods

alter_table() click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 37
def alter_table
  execute("ALTER TABLE IF EXISTS #{schema}#{table_name} #{columns_fragment}")
end
columns_fragment() click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 41
def columns_fragment
  column_types.map do |column|
    "ADD COLUMN IF NOT EXISTS #{column}"
  end.join(",")
end
current_table_columns() click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 69
def current_table_columns
  sql = "SELECT * FROM #{schema}#{table_name} LIMIT 1;"
  stmt, rs = executeQuery(prep_sql(sql))
  meta_data = rs.getMetaData()
  column_count = meta_data.getColumnCount()
  (1..column_count).map {|i| meta_data.getColumnName(i) }
ensure
  stmt.close unless stmt.nil?
end
execute(sql) click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 79
def execute(sql)
  stmt = connection.prepareStatement(prep_sql(sql))
  stmt.execute()
rescue Java::OrgPostgresqlUtil::PSQLException => e
  self.logger.error "PSQLException: #{e.message}"
ensure
  stmt.close unless stmt.nil?
end
executeQuery(sql) click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 88
def executeQuery(sql)
  stmt = connection.createStatement()
  # only close the statement if something goes wrong
  # otherwise, the caller is responsible for closing the
  # statement when they are doen with the result set
  return stmt, stmt.executeQuery(prep_sql(sql))
rescue Java::OrgPostgresqlUtil::PSQLException => e
  puts "PSQLException: #{e.message}"
  self.logger.info "PSQLException: #{e.message}"
  stmt.close unless stmt.nil?
  # @logger.error("#{e.message}")
rescue => e
  puts "Unknown exception: #{e.message}"
  self.logger.info "Unknown exception: #{e.message}"
  stmt.close unless stmt.nil?
end
prep_sql(sql) click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 105
def prep_sql(sql)
  sql.gsub(/\s+/, " ").strip
end
set_column_types() click to toggle source
# File lib/logstash/outputs/charrington/alter_postgres_table.rb, line 47
def set_column_types
  (columns - current_table_columns).each_with_index do |key, idx|

    case event[key]
    when Time, LogStash::Timestamp
      column_types << "#{key} TIMESTAMP"
    when Date
      column_types << "#{key} DATE"
    when Integer
      column_types << "#{key} BIGINT"
    when BigDecimal
      column_types << "#{key} DECIMAL"
    when Float
      column_types << "#{key} DOUBLE PRECISION"
    when true, false
      column_types << "#{key} BOOLEAN"
    else
      column_types << "#{key} VARCHAR"
    end
  end
end