Class: Flydata::Parser::Mysql::MysqlDumpParser

Inherits:

Object

• Object
• Flydata::Parser::Mysql::MysqlDumpParser

Defined in:

lib/flydata/parser/mysql/dump_parser.rb

Defined Under Namespace

Modules: State Classes: InsertParser

Instance Attribute Summary

• #binlog_pos ⇒ Object

  Returns the value of attribute binlog_pos.

Instance Method Summary

• #initialize(option = {}) ⇒ MysqlDumpParser constructor

  A new instance of MysqlDumpParser.

• #parse(dump_io, create_table_block, insert_record_block, check_point_block) ⇒ Object

Constructor Details

#initialize(option = {}) ⇒ MysqlDumpParser

Returns a new instance of MysqlDumpParser.

# File 'lib/flydata/parser/mysql/dump_parser.rb', line 261

def initialize(option = {})
  @binlog_pos = option[:binlog_pos]
  @option = option
end

Instance Attribute Details

#binlog_pos ⇒ Object

Returns the value of attribute binlog_pos.

# File 'lib/flydata/parser/mysql/dump_parser.rb', line 259

def binlog_pos
  @binlog_pos
end

Instance Method Details

#parse(dump_io, create_table_block, insert_record_block, check_point_block) ⇒ Object

# File 'lib/flydata/parser/mysql/dump_parser.rb', line 266

def parse(dump_io, create_table_block, insert_record_block, check_point_block)
  unless dump_io.kind_of?(IO)
    raise ArgumentError.new("Invalid argument. The first parameter must be io.")
  end

  invalid_file = false
  current_state = State::START
  substate = nil
  buffered_line = nil
  bytesize = 0

  readline_proc = Proc.new do
    line = nil
    if buffered_line
      line = buffered_line
      buffered_line = nil
    else
      rawline = dump_io.readline
      bytesize += rawline.bytesize
      line = rawline.strip
    end
    line
  end

  state_start = Proc.new do
    line = readline_proc.call

    # -- CHANGE MASTER TO MASTER_LOG_FILE='mysql-bin.000002', MASTER_LOG_POS=120;
    m = /^\-\- CHANGE MASTER TO MASTER_LOG_FILE='(?<binfile>[^']+)', MASTER_LOG_POS=(?<pos>\d+)/.match(line)
    if m
      @binlog_pos = {binfile: m[:binfile], pos: m[:pos].to_i}
      current_state = State::CREATE_TABLE
      check_point_block.call(nil, dump_io.pos, bytesize, @binlog_pos, current_state)
    end
  end

  current_table = nil
  state_create_table = Proc.new do
    line = readline_proc.call

    # CREATE TABLE `active_admin_comments` (
    m = /^CREATE TABLE `(?<table_name>[^`]+)`/.match(line)
    if m
      current_table = MysqlTable.new(m[:table_name])
      current_state = State::CREATE_TABLE_COLUMNS
    end
  end

  state_create_table_constraints = Proc.new do
    line = readline_proc.call

    #  PRIMARY KEY (`id`),
    if line.start_with?(')')
      create_table_block.call(current_table)
      current_state = State::INSERT_RECORD
      check_point_block.call(current_table, dump_io.pos, bytesize, @binlog_pos, current_state)
    elsif m = /^PRIMARY KEY \((?<primary_keys>[^\)]+)\)/.match(line)
      current_table.primary_keys = m[:primary_keys].split(',').collect do |pk_str|
        pk_str[1..-2]
      end
    end
  end

  state_create_table_columns = Proc.new do
    start_pos = dump_io.pos
    line = readline_proc.call

    #  `author_type` varchar(255) COLLATE utf8_unicode_ci DEFAULT NULL,
    if line.start_with?("\`")
      column = {}

      # parse column line
      line = line[0..-2] if line.end_with?(',')
      items = line.split
      column[:column_name] = items.shift[1..-2]
      column[:format_type_str] = format_type_str = items.shift
      pos = format_type_str.index('(')
      if pos
        ft = column[:format_type] = format_type_str[0..pos-1]
        if ft == 'decimal'
          precision, scale = format_type_str[pos+1..-2].split(',').collect{|v| v.to_i}
          column[:decimal_precision] = precision
          column[:decimal_scale] = scale
        else
          column[:format_size] = format_type_str[pos+1..-2].to_i
        end
      else
        column[:format_type] = format_type_str
      end
      while (item = items.shift) do
        case item
        when 'DEFAULT'
          value = items.shift
          value = value.start_with?('\'') ? value[1..-2] : value
          value = nil if value == 'NULL'
          column[:default] = value
        when 'NOT'
          if items[1] == 'NULL'
            items.shift
            column[:not_null] = true
          end
        when 'unsigned'
          column[:unsigned] = true
        else
          #ignore other options
        end
      end

      current_table.add_column(column)
    else
      current_state = State::CREATE_TABLE_CONSTRAINTS
      buffered_line = line
      state_create_table_constraints.call
    end
  end

  state_insert_record = Proc.new do
    line = readline_proc.call

    if line.start_with?('INSERT')
      buffered_line = line
      current_state = State::PARSING_INSERT_RECORD
    elsif line.start_with?('UNLOCK')
      current_state = State::CREATE_TABLE
      check_point_block.call(current_table, dump_io.pos, bytesize, @binlog_pos, current_state)
    end
  end

  state_parsing_insert_record = Proc.new do
    line = readline_proc.call

    values_set = InsertParser.new.parse(line)
    current_state = State::INSERT_RECORD

    if insert_record_block.call(current_table, values_set)
      check_point_block.call(current_table, dump_io.pos, bytesize, @binlog_pos, current_state)
    end
  end

  # Start reading file from top
  begin
    # resume(only when using dump file)
    if @option[:last_pos] && (@option[:last_pos].to_i != -1)
      dump_io.pos = @option[:last_pos].to_i
      current_state = @option[:state]
      substate = @option[:substate]
      current_table = @option[:mysql_table]
      bytesize = dump_io.pos
    end

    until dump_io.eof? do
      case current_state
      when State::START
        state_start.call
      when State::CREATE_TABLE
        state_create_table.call
      when State::CREATE_TABLE_COLUMNS
        state_create_table_columns.call
      when State::CREATE_TABLE_CONSTRAINTS
        state_create_table_constraints.call
      when State::INSERT_RECORD
        state_insert_record.call
      when State::PARSING_INSERT_RECORD
        state_parsing_insert_record.call
      end
    end
  end
  @binlog_pos
end