2025-12-07 RubyのTCPソケットでMySQLにクエリを発行する

前回 Ruby の TCPソケット通信で MySQL の認証を通すやり方を書いたので、その続き。
認証を通せれば、Rubyから SQL を発行して実行できるようになるのでやり方をまとめる。

MySQL には各種処理を実行するためのコマンドというのがあり、SQL の実行は COM_QUERY というコマンドになる。

主要なコマンド一覧

コマンド	値	渡すもの
COM_QUIT	0x01	なし
COM_INIT_DB	0x02	database名
COM_QUERY	0x03	SQLクエリ文字列
COM_PING	0x0E	なし
COM_BINLOG_DUMP	0x12	binlog位置情報
COM_REGISTER_SLAVE	0x15	スレーブ情報
COM_STMT_PREPARE	0x16	SQLクエリ
COM_STMT_EXECUTE	0x17	statement_id + パラメータ
COM_STMT_CLOSE	0x19	statement_id

コマンド値は 8ビット符号なし整数でパケット送信する。

例えば、接続確認用の PING コマンドの実行と結果の確認はこんな感じで取れる。

socket = TCPSocket.new(host, port)
socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)

sequence_id = 0

def read_packet
  header = socket.read(4)

  packet_length = header[0].unpack1('C') | (header[1].unpack1('C') << 8) | (header[2].unpack1('C') << 16)
  sequence_id = header[3]unpack1('C')
  payload = socket.read(packet_length)

  { packet_length:, payload: }
end

def send_packet(payload)
  sequence_id += 1

  packet_length = payload.length
  header = [packet_length].pack('V')[0..2] + [sequence_id].pack('C')
  socket.write(header + payload)
end

payload = [0x0E].pack('C') # PINGコマンド
send_packet(payload)

response = read_packet(payload)
if response[:payload][0].unpack1('C') == 0x00 # 成功
  puts 'PONG'
else
  puts 'Oops'
end

COM_QUERY コマンドを実行する

SQL を実行させるための COM_QUERY コマンドのペイロードはこれでOK

payload = [0x03].pack('C') + sql.encode('UTF-8')

クエリ実行のパケットを送信すると、MySQL でクエリが実行されて結果をパケット受信できるようになる。
実行結果は OK Packet / ERR Packet / Result Set の3パターンあって、構造が違うのでそれぞれパースする必要がある。

どのパターンかはパケットのペイロードの 1 バイト目を見ればわかるようになっている。

0x00 -> OK Packet
0xFF -> Err Packet
0x01~0xFA -> Result Set
0xFB -> Local INFILE Request

LOCAL INFILE Request は特殊なので今回は未対応。

OK Packet の場合

結果セットを返さないクエリが成功したときがこれ。
メジャーなところだと、INSERT / UPDATE / DELETE文を実行したときこれが返る。

パケットの構造はこんな感じ。

サイズ	フィールド
1 byte integer	header	0x00が入る
length-encoded integer	affected_rows
length-encoded integer	last_insert_id
2 bytes integer	status_flags	サーバ状態を表すフラグ
2 bytes integer	warnings
N bytes string	info	NULL終端

length-encoded integer は最初の 1 バイトで長さが決まる整数。
最初の 1 バイトごとにこんな感じで長さと値を取る。

0x00 から 0xFA（250）
- 最初の 1 バイトがそのまま値になる
0xFC
- 後続の 2 バイトが値
0xFD
- 後続の 3 バイトが値
0xFE
- 後続の 8 バイトが値
0xFB
- NULL

def length_encoded_integer(payload, offset)
  first_byte = payload[offset].unpack1('C')

  case first_byte
  when 0..250
    { value: first_byte, bytes_read: 1 }
  when 0xFC
    value = payload[(offset + 1)..(offset + 2)].unpack1('s<')
    { value: value, bytes_read: 3 }
  when 0xFD
    value = payload[(offset + 1)..(offset + 3)].unpack1('V') & 0xFFFFFF
    { value: value, bytes_read: 4 }
  when 0xFE
    value = payload[(offset + 1)..(offset + 8)].unpack1('Q<')
    { value: value, bytes_read: 9 }
  else # Included 0xFB
    { value: nil, bytes_read: 1 }
  end
end

ERR Packet の場合

文字通りクエリ実行がエラーになった時がこれ。

パケットの構造はこんな感じ。

サイズ	フィールド
1 byte integer	header	0xFFが入る
2 bytes integer	error_code
1 byte integer	sql_state_marker	"#"固定
5 bytes string	sql_state
N bytes string	error_message	NULL終端

Result Set の場合

SELECT みたいな結果行レコードが返るクエリがこれ。

こいつは複雑で、複数のパケットで構成される。

カラム数パケット
- この後何回カラム定義パケットを取ればいいかを伝える
カラム定義パケット
EOFパケット
- カラム定義パケットが終わったことを伝える
行データパケット
EOFパケット
- 行データパケットが終わったことを伝える

カラム数パケット

サイズ	フィールド
1 byte integer	column_count	カラムの数が入る

カラム定義パケット

サイズ	フィールド
length-encoded string	catalog	"def" というダミー値固定
length-encoded string	schema	データベース名
length-encoded string	table	テーブル名（エイリアス）
length-encoded string	org_table	物理テーブル名
length-encoded string	name	カラム名（エイリアス）
length-encoded string	org_name	物理カラム名
length-encoded integer	length_of_fixed_fields	"0x0c" 固定
2 bytes integer	charset
4 bytes integer	column_length	カラムの最大長
1 byte integer	column_type	データ型
2 bytes integer	flags	カラムの属性を表すビットフラグ
1 byte integer	decimals	小数点以下の桁数
2 bytes integer	filter	"0x00 0x00" 固定

length-encoded string は length-encoded integer + length-encoded integer が返す値の長さの文字列。
こんな感じで取る。

def length_encoded_string(payload, offset)
  length_info = length_encoded_integer(payload, offset)
  return { value: '', bytes_read: length_info[:bytes_read] } if length_info[:value].nil?

  string_start = length_info[:bytes_read]
  string_end = string_start + length_info[:value] - 1
  value = payload[string_start..string_end] || ''

  { value: value, bytes_read: length_info[:bytes_read] + length_info[:value] }
end

EOF パケット

サイズ	フィールド
1 byte integer	eof	"0xFE" 固定

行データパケット
カラム数分 length-encoded string が繰り返され、取れる値がカラムの値になる。
ただし、1 バイト目が 0xFB の場合は値が NULL になっているので最初に NULL チェックする。

これらをまとめると、Result Set パケットを処理するコードはこんな感じになる。

# カラム定義パケットのパース
def parse_column_definition(payload)
  offset = 0

  catalog = length_encoded_string(payload, offset)
  offset += catalog[:bytes_read]

  schema = length_encoded_string(payload, offset)
  offset += schema[:bytes_read]

  table = length_encoded_string(payload, offset)
  offset += table[:bytes_read]

  org_table = length_encoded_string(payload, offset)
  offset += org_table[:bytes_read]

  name = length_encoded_string(payload, offset)
  offset += name[:bytes_read]

  org_name = length_encoded_string(payload, offset)
  offset += org_name[:bytes_read]

  # length of fixed-length fields (1 byte)
  offset += 1

  charset = payload[offset..(offset + 1)].unpack1('v')
  offset += 2

  column_length = payload[offset..(offset + 3)].unpack1('V')
  offset += 4

  column_type = payload[offset].unpack1('C')

  {
    schema: schema[:value],
    table: table[:value],
    org_table: org_table[:value],
    name: name[:value],
    org_name: org_name[:value],
    charset: charset,
    column_length: column_length,
    column_type: column_type
  }
end

# 行データパケットのパース
def parse_row_data(payload, columns)
  first_byte = payload[0].unpack1('C')
  row = {}
  offset = 0

  columns.each do |column|
    column_name_key = column[:name].downcase.to_sym

    if offset >= payload.length
      row[column_name_key] = nil
      next
    end

    if first_byte == 0xFB
      # NULL value
      row[column_name_key] = nil
      offset += 1
    else
      # row data (length-encoded string)
      value = length_encoded_string(payload, offset)
      row[column_name_key] = value[:value]
      offset += value[:bytes_read]
    end
  end

  row
end

# カラム数パケット、カラム定義パケットを処理
columns = []
column_count = length_encoded_integer(payload, 0)[:value].to_i
column_count.times do
  column_packet = read_packet
  column_info = parse_column_definition(column_packet[:payload])
  columns << column_info
end

# EOF パケットを読み進める
connection.read_packet

# 行データパケットを処理
rows = []
loop do
  row_packet = read_packet

  # EOF パケットになったら終了
  if row_packet[:payload][0].unpack1('C') == 0xFE
    break
  end

  rows << parse_row_data(row_packet[:payload], columns)
end

puts rows

waka.dev

2025-12-07 RubyのTCPソケットでMySQLにクエリを発行する

COM_QUERY コマンドを実行する

OK Packet の場合

ERR Packet の場合

Result Set の場合