Class: Flydata::Mysql::MysqlDumpParser

Inherits:

Object

• Object
• Flydata::Mysql::MysqlDumpParser

Defined in:

lib/flydata/command/sync.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(file_path, option = {}) ⇒ MysqlDumpParser constructor

  A new instance of MysqlDumpParser.

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

Constructor Details

#initialize(file_path, option = {}) ⇒ MysqlDumpParser

Returns a new instance of MysqlDumpParser.

# File 'lib/flydata/command/sync.rb', line 1013

def initialize(file_path, option = {})
  @file_path = file_path
  raise "Dump file does not exist. file_path:#{file_path}" unless File.exist?(file_path)
  @binlog_pos = option[:binlog_pos]
  @option = option
end

Instance Attribute Details

#binlog_pos ⇒ Object

Returns the value of attribute binlog_pos.

# File 'lib/flydata/command/sync.rb', line 1011

def binlog_pos
  @binlog_pos
end

Instance Method Details

#parse(create_table_block, insert_record_block, check_point_block) ⇒ Object

# File 'lib/flydata/command/sync.rb', line 1020

def parse(create_table_block, insert_record_block, check_point_block)
  invalid_file = false
  current_state = State::START
  substate = nil

  state_start = Proc.new do |f|
    line = f.readline.strip
    # -- 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, f.pos, @binlog_pos, current_state)
    end
  end

  current_table = nil
  state_create_table = Proc.new do |f|
    line = f.readline.strip
    # 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 |f|
    line = f.readline.strip
    #  PRIMARY KEY (`id`),
    if line.start_with?(')')
      create_table_block.call(current_table)
      current_state = State::INSERT_RECORD
      check_point_block.call(current_table, f.pos, @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 |f|
    start_pos = f.pos
    line = f.readline.strip
    #  `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
      f.pos = start_pos
      state_create_table_constraints.call(f)
    end
  end

  state_insert_record = Proc.new do |f|
    original_pos = f.pos
    command = f.read(6)
    if command == 'INSERT'
      current_state = State::PARSING_INSERT_RECORD
    else
      f.pos = original_pos
      f.readline
      if command == 'UNLOCK'
        current_state = State::CREATE_TABLE
        check_point_block.call(current_table, f.pos, @binlog_pos, current_state)
      end
    end
  end

  state_parsing_insert_record = Proc.new do |f|
    values_set = InsertParser.new(f).parse
    current_state = State::INSERT_RECORD
    if insert_record_block.call(current_table, values_set)
      check_point_block.call(current_table, f.pos, @binlog_pos, current_state)
    end
  end

  # Start reading file from top
  File.open(@file_path, 'r', encoding: "utf-8") do |f|
    last_saved_pos = 0

    # resume
    if @option[:last_pos]
      f.pos = @option[:last_pos].to_i
      current_state = @option[:state]
      substate = @option[:substate]
      current_table = @option[:mysql_table]
    end

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