[MYSQL] REDUNDANT行格式的数据解析

原创

大大刺猬

发布于 2024-12-06 15:42:44

7220

导读

mysql的行格式有4种,REDUNDANT,COMPACT,DYNAMIC和COMPRESSED. 最常用的就是DYNAMIC, 也是mysql默认的行格式(很早只有REDUNDANT). 该行格式虽然复杂一点, 但是支持的索引前缀可达3072字节.(REDUNDANT只支持到768字节).

除此之外,未发现明显优势....

这4种格式对比如下(来自官网)

而我们之前写的ibd2sql工具已经支持了DYNAMIC和COMPRESSED的格式. 但还不支持REDUNDANT格式.

那我们今天就又又又来补齐'最后一块拼图'

REDUNDANT格式简介

该行格式比较老, 已经很多人都解析过了的. 网上应该一搜一大堆. 而且和DYNAMIC之类的差别不大, 甚至还简单很多, 所以我们简单介绍下即可.

innodb的行格式都如下结构, 标红的地方就是和DYNAMIC格式的一丢丢区别.

RECORD HEADER

先来看看record header, 比DYNAMIC格式多了1字节. 不再记录行是MIN/MAX/叶子/非叶子了. 多了个记录 var使用1还是2字节表示的, 还多了个本行有多少字段了(含rowid,trxid,rollptr). 下一字段位置变成了绝对位置(所以是无符号的了)

对象	描述
instant_flag	DYNAMIC格式为:是否做过onlineddl
row_version_flag	DYNAMIC格式为:是否做过onlineddl 8.0.29之后
deleted	是否删除了
min_rec	是否是最小
owned	本slot有多少行
heap_no	heap no
n_fields	本行有多少字段(含rowid之类的)
byte1_flag	是否使用1字节来记录长度
next_record	下一字段的绝对位置(无符号)

VARSIZE&NULLABLE

DYNAMIC中是有专门的nullable bitmast的. 而且只记录变长字段,不记录int之类的固定长度的长度. 但REDUNDANT中就通通记录长度, 并且不再根据某个字段是否达到128来决定使用1/2字节了, 而是通通使用1/2字节. 看起来有丢丢浪费, 但毕竟都叫REDUNDANT了, 能不冗余么....

而使用1/2字节是根据是否有超过128字节的字段, 都不再具体判断某字段是否使用1-2字节了, 为啥还是使用128字节呢? 还有 1 bit 哪去了呢?

那就是用来表示是否为空了. 即第1bit表示是否为空. 第1bit是1就表示为空, 否则表示不为空. (为空的时候,也是存在长度的, 也得去读.方便给某字段置为空?)

结构如下:

总结一个字就是:花里胡哨

解析REDUNDANT行格式的脚本

说了这么多, 那我们就来解析解析REDUNDANT格式吧. 为了方便使用, 我们就单独写个脚本来解析. 但有部分存储较为复杂, 单独写的话,会使脚本变得太长. 所以对于那部分数据我们将他置为null. 但不是直接置为null,而是在无法引入ibd2sql的包的时候再将其置为null. 这样我们的脚本健壮性就高一些. 这样的设计其实也比较简单. 比如:

try:
        HAVE_IBD2SQL = True # 部分数据懒得解析了, 直接沿用ibd2sql去实现. 不行的话,就取为null
        from ibd2sql.blob import first_blob
        from ibd2sql.mysql_json import jsonob
        from ibd2sql.innodb_page import page
        from ibd2sql.innodb_page_index import char_decode
        from ibd2sql.COLLATIONS import COLLID_TO_CHAR
        from ibd2sql.innodb_type import innodb_type_isvar
        class MINI_PAGE(page):
                def __init__(self,bdata):
                        super(MINI_PAGE,self).__init__(bdata)
                        self.decimal_data = b''
                def read(self,n):
                        return self.decimal_data
except:
        HAVE_IBD2SQL = False

非常方便的try,except. 对于那些设计得不那么科学的类, 我们只需要修改丢丢代码就能实现, 只解析我们需要的字段类型(decimal).

由于我们是设计的单脚本, 所以就没去考虑online DDL的情况了. 总得有牺牲!

解析逻辑

本次我们直接遍历整个数据文件, 如果PAGE符合要求, 就解析.

要求为:

是INDEX_PAGE (b'E\xbf')
是叶子节点, PAGE_LEVEL是0
是主键的索引ID

这样我们就只解析我们要的叶子页了, 避免了坏块的问题. 之前是顺着叶子节点去解析,遇到坏块就不知道往哪走了.

那我们去哪获取主键索引的ID呢?

当然是解析主键页的70:74位置的值了呗? 那主键页在哪呢?

主键页是inode page的第1对segment(第0对是sdi page). 或者sdi page中的第0条索引中的se_private_data里面的root值也可以. 我们这里选择了后者, 所以埋下了一个坑(5.7场景的sdi是来自8.0的, 所以就对不上咯.). 没关系, 后面遇到了再修....

然后就是读取KEY/ROWID, TRXID,ROLLPTR,剩余字段了.

演示

直接开演.

注: json,decimal,和溢出页这三目前只有在存在ibd2sql包的时候才可使用,其它数据类型目前均支持.

-- 创建测试表
CREATE TABLE `t20241206_test_redundant` (
  `id` int NOT NULL DEFAULT '0',
  `int_col` int DEFAULT NULL,
  `tinyint_col` tinyint DEFAULT '1',
  `smallint_col` smallint DEFAULT NULL,
  `mediumint_col` mediumint DEFAULT NULL,
  `bigint_col` bigint DEFAULT NULL,
  `float_col` float DEFAULT NULL,
  `double_col` double DEFAULT NULL,
  `decimal_col` decimal(10,2) DEFAULT NULL,
  `date_col` date DEFAULT NULL,
  `datetime_col` datetime DEFAULT NULL,
  `timestamp_col` timestamp NULL DEFAULT NULL,
  `time_col` time DEFAULT NULL,
  `year_col` year DEFAULT NULL,
  `char_col` char(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
  `varchar_col` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
  `binary_col` binary(10) DEFAULT NULL,
  `varbinary_col` varbinary(20) DEFAULT NULL,
  `bit_col` bit(4) DEFAULT NULL,
  `enum_col` enum('A','B','C') CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
  `set_col` set('X','Y','Z') CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
  `josn_type` json DEFAULT NULL
) ENGINE=InnoDB ROW_FORMAT=REDUNDANT;

-- 插入测试数据
INSERT INTO t20241206_test_redundant values(1,929866,-82,4981,6013,-7147,-2930.0,1155.0,-3196.00,'1994-03-25','2017-11-05 20:58:46','2021-04-26 20:36:14','04:13:07',2017,'MYuXdlBNNYRjzXrcYmvDYertYOIvvZlQfwcxegVPBZQqSSTAzeSkNFuTdD','RlyqLNGtWmqEUNoUukYKSOqRlPGtgliBKqxtooqZYzHPPHTdgQBlEGhBMaOqJheooN',0x31390000000000000000,0x3737,0x01,'A','Y','{"aa": "c", "bb": {"dd": 1}}');

然后我们使用ibd2sql_mini_for_redundant.py来解析(脚本见文末)

python3 ibd2sql_mini_for_redundant.py /data/mysql_3314/mysqldata/db1/t20241206_test_redundant.ibd

看起来数据没毛病呢, 然后我们可以再插入数据库并校验.

# 创建新表
mysql -h127.0.0.1 -P3314 -p123456 -e 'create table db2.t20241206_test_redundant as select * from db1.t20241206_test_redundant where 1>2'
# 修改行格式为redundant
mysql -h127.0.0.1 -P3314 -p123456 -e 'alter table db2.t20241206_test_redundant row_format=redundant'
# 解析数据并插入新表
python3 ibd2sql_mini_for_redundant.py /data/mysql_3314/mysqldata/db1/t20241206_test_redundant.ibd | mysql -h127.0.0.1 -P3314 -p123456 db2
# 校验数据是否一致
mysql -h127.0.0.1 -P3314 -p123456 -e 'checksum table db1.t20241206_test_redundant,db2.t20241206_test_redundant'

数据确实没问题. 证明我们确实解析对了.

5.7或者分区表的话, 使用方法就是多加个8.0里面的同样的表(为了获取表结构的)

python3 ibd2sql_mini_for_redundant.py /data/mysql_3308/mysqldata/db1/t20241206_test_redundant.ibd /data/mysql_3314/mysqldata/db2/t20241206_test_redundant.ibd

第一个参数是: 要解析的5.7的ibd文件

第二个参数是: 对应的8.0里面的ibd文件(在8.0里面建个一样的空表即可)

注: 这个版本的时候, 我把结尾的空格也保留了.对于char来说就会很丑, 不过无所谓了拉.

总结

innodb的行格式总体长得差不多, 都是细微的差距. 而compress的话,只是压缩了一下, 其它的就没得区别了.

参考:

https://dev.mysql.com/doc/refman/8.0/en/innodb-row-format.html

https://github.com/ddcw/ibd2sql

附源码

单独使用如下脚本的话, 不支持decimal之类的复杂字段类型.

或者github上下载

#!/usr/bin/env python
# write by ddcw @https://github.com/ddcw
# 解析row format=REDUNDANT的脚本, 先试试, 后面2.0的时候再合并.
# 本脚本可以单独使用, 对于sdi之类的都单独解析, 所以不再拼接DDL了. 也不支持compress,加密之类的,页不支持溢出页,json,decimal之类的. 就一个Mini脚本,要啥自行车..
# 功能: 解析8.0中的ibd文件(ROW_FORMAT=REDUNDANT)
# 用法:
#  python3  ibd2sql_mini_for_redundant.py  /PATH/xxxx.ibd

# 本脚本采用强制解析, 即每页都直接解析, 不做校验, 只要是PAGE_LEVEL=0的通通解析

import sys
import os
import struct
import json
import zlib
import base64
import time

try:
	HAVE_IBD2SQL = True # 部分数据懒得解析了, 直接沿用ibd2sql去实现. 不行的话,就取为null
	from ibd2sql.blob import first_blob
	from ibd2sql.mysql_json import jsonob
	from ibd2sql.innodb_page import page
	from ibd2sql.innodb_page_index import char_decode
	from ibd2sql.COLLATIONS import COLLID_TO_CHAR
	from ibd2sql.innodb_type import innodb_type_isvar
	class MINI_PAGE(page):
		def __init__(self,bdata):
			super(MINI_PAGE,self).__init__(bdata)
			self.decimal_data = b''
		def read(self,n):
			return self.decimal_data
except:
	HAVE_IBD2SQL = False

ARGV = sys.argv
def USAGE():
	print(f"\nUSAGE:\n\tpython3 {ARGV[0]} /PATH/xxxx.ibd")

WITH_COLNAME = False # 带上字段名
class IBDREADER(object):
	def __init__(self,filename):
		self.filename = filename
		self.f = open(self.filename,'rb')
		self.PAGE_SIZE = 16384
		self.current_page = 0
	def read(self,n=0):
		if n > 0:
			self.current_page = n
		self.f.seek(self.PAGE_SIZE*self.current_page,0)
		data = self.f.read(self.PAGE_SIZE)
		self.current_page += 1
		return data
	def close(self,):
		self.f.close()

def BDATA2INTBD(bdata,):
	n = len(bdata)
	sn = ">"+str(n)+"B"
	tdata = struct.unpack(sn,bdata)
	rdata = 0
	for x in range(n):
		rdata += tdata[x]<<((n-x-1)*8)
	return rdata
	

def TOINT4(bdata,unsigned):
	data = BDATA2INTBD(bdata)
	rdata = 0
	if unsigned:
		rdata = data
	else:
		if data&(2**(len(bdata)*8-1)):
			rdata = data&(2**(len(bdata)*8-1)-1)
		else:
			rdata = -((2**(len(bdata)*8-1))-data) if data > 0 else  data
	return rdata

def INNODB_TIMESPLIT(data,n,rule1):
	# data是int数字,  n是多少bit位, rule1是时间
	# rule1 : [[0,1], [1,11], [12,18] [start,count]   ]
	rdata1 = []
	for start,count in rule1:
		t = ( data & ((2**(n-start)-1)-(2**(n-start-count)-1)) )>>(n-start-count)
		rdata1.append(t)
	return rdata1

class ROWREAD(object):
	def __init__(self,bdata):
		self.bdata = bdata
		self.offset = 101
		self._offset = 101
		self._last_offset = 0

	def read(self,n):
		data = self.bdata[self.offset:self.offset+n]
		self.offset += n
		return data

	def readreverse(self,n):
		data = self.bdata[self._offset-n:self._offset]
		self._offset -= n
		return data

	def read_record_header(self,):
		bdata = self.readreverse(6)
		bb = struct.unpack('>6B',bdata)
		return {
			"instant_flag": True if bb[0]&128 else False,
			"row_version_flag": True if bb[0]&64 else False,
			"deleted": True if bb[0]&32 else False,
			"min_rec": True if bb[0]&16 else False,
			"owned": bb[0]&15,
			"heap_no": 0,
			"n_fields": (((bb[2]<<8)+bb[3])&(2**11-1))>>1,
			"byte1_flag": True if bb[3]&1 else False,
			"next_record":  (bb[4]<<8) + bb[5],
		}

	def read_trx_rollptr(self,):
		trx = self.read(6)
		rollptr = self.read(7)
		return {'trx':trx,'rollptr':rollptr}

	def read_filed(self,colid,dd,rh,f):
		datasize,isnull = self.read_nullandsize(rh)
		extradata = b''
		if datasize > 16384 and HAVE_IBD2SQL:
			bdata = self.read(768)
			SPACE_ID,PAGENO,BLOB_HEADER,REAL_SIZE = struct.unpack('>3LQ',self.read(20))
			extradata = first_blob(f,PAGENO)
		elif datasize > 16384:
			bdata = self.read(768)
		else:
			bdata = self.read(datasize)
		bdata += extradata
		rdata = None
		col = dd['columns'][colid]
		if isnull:
			return rdata
		else:
			if col['type'] in [2,3,4,9,10]: # int
				rdata = TOINT4(bdata,col['is_unsigned'])
			elif col['type'] == 5: # float
				rdata = struct.unpack('f',bdata)[0]
			elif col['type'] == 6: # double
				rdata = struct.unpack('d',bdata)[0]
			elif col['type'] == 14: # year
				rdata = BDATA2INTBD(bdata)+1900
			elif col['type'] == 22: # enum
				rdata = col['elements_dict'][BDATA2INTBD(bdata)]
				rdata = repr(rdata)
			elif col['type'] == 23: # set # 可以包含多个
				_tdata = BDATA2INTBD(bdata)
				_sn = 0
				_sdata = ''
				for x in col['elements_dict']:
					if 1<<_sn & _tdata:
						 _sdata += col['elements_dict'][x] + ","
					_sn += 1
				rdata = repr(_sdata[:-1])
			elif col['type'] == 31: # json
				if HAVE_IBD2SQL and extradata == b'':
					_tdata = jsonob(bdata[1:],struct.unpack('<B',bdata[:1])[0]).init()
					rdata = repr(json.dumps(_tdata))
				else:
					rdata = None
				
			elif col['type'] == 15: # date
				signed,year,month,day = INNODB_TIMESPLIT(
					BDATA2INTBD(bdata[:3]),
					24,
					[[0,1], [1,14], [15,4], [19,5]])
				jingdu = BDATA2INTBD(bdata[3:]) if datasize > 3 else ''
				rdata = f"{'-' if signed == 0 else ''}{year}-{str(month).zfill(2)}-{str(day).zfill(2)}{'.'+str(jingdu) if datasize > 3 else ''}"
				rdata = repr(rdata)
			elif col['type'] == 18: # timestamp
				rdata = repr(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(BDATA2INTBD(bdata))))
			elif col['type'] == 19: # datetime
				signed,year_month,day,hour,minute,second = INNODB_TIMESPLIT(
					BDATA2INTBD(bdata[:5]),
					40,
					[[0,1], [1,17], [18,5], [23,5], [28,6], [34,6]]
					)
				jingdu = BDATA2INTBD(bdata[5:]) if datasize > 5 else ''
				year = int(year_month/13)
				month = int(year_month%13)
				rdata = f"{'-' if signed == 0 else ''}{year}-{str(month).zfill(2)}-{str(day).zfill(2)} {str(hour).zfill(2)}:{str(minute).zfill(2)}:{str(second).zfill(2)}{'.'+str(jingdu) if datasize > 5 else ''}"
				rdata = repr(rdata)
			elif col['type'] == 20: # time
				signed,hour,minute,second = INNODB_TIMESPLIT(
					BDATA2INTBD(bdata[:3]),
					24,
					[[0,1], [1,11], [12,6], [18,6]]
				)
				jingdu = BDATA2INTBD(bdata[3:]) if datasize > 3 else ''
				rdata = f"{'-' if signed == 0 else ''}{str(hour).zfill(2)}:{str(minute).zfill(2)}:{str(second).zfill(2)}{'.'+str(jingdu) if datasize > 3 else ''}"
				rdata = repr(rdata)
			elif col['type'] == 21: # decimal #TODO
				if HAVE_IBD2SQL:
					aa = MINI_PAGE(b'\x00'*16384)
					aa.decimal_data = bdata
					rdata = aa.read_innodb_decimal(len(bdata),col['extra'])
				else:
					rdata = None
			elif col['type'] == 29 and col['column_type_utf8'].startswith('binary'): # binary
				rdata = hex(BDATA2INTBD(bdata))
			elif col['type'] == 16 and col['column_type_utf8'].startswith('varbinary'): # varbinary
				rdata = hex(BDATA2INTBD(bdata))
			elif col['type'] in [29,16]: # varchar,char
				if HAVE_IBD2SQL:
					#rdata = repr(char_decode(bdata,col).rstrip())
					rdata = repr(char_decode(bdata,col))
				else:
					try:
						#rdata = repr((bdata).decode().rstrip())
						rdata = repr((bdata).decode())
					except:
						rdata = '0x'+(bdata).hex()
			else:
				rdata = '0x'+bdata.hex()
		return rdata

	def read_nullandsize(self,rh):
		nmask = 128 if rh['byte1_flag'] else 32768
		data = 1
		if rh['byte1_flag']:
			data = struct.unpack('>B',self.readreverse(1))[0]
		else:
			data = struct.unpack('>H',self.readreverse(2))[0]
		isnull = True if nmask&data else False
		value = data&(nmask-1) if isnull else data
		_t = value
		value = value - self._last_offset
		#print('SIZE',value,'NULLABLE:',isnull)
		self._last_offset = _t
		return value,isnull

def ROW2SQL(row,columns,tablename):
	sql = f"INSERT INTO {tablename}"
	colname = ""
	value = ""
	for col in columns:
		if col['name'] in ['DB_ROLL_PTR','DB_TRX_ID','DB_ROW_ID']:
			continue
		colname = f"{colname}{col['name']},"
		value = f"{value}{row[col['name']] if row[col['name']] is not None else 'null'},"
	colname = "("+colname[:-1]+")"
	print(f"{sql}{colname[:-1] if WITH_COLNAME else ''} values({value[:-1]});")

if __name__ == "__main__":
	if len(ARGV) not in [2,3]:
		USAGE()
	for x in ARGV[1:]:
		if str(x).upper().find('-H') >= 0:
			USAGE()

	for filename in ARGV[1:]:
		if not os.path.exists(filename):
			print(f"{filename}不存在!")
			USAGE()
	filename = ARGV[1]
	# 开整
	f = IBDREADER(filename)
	if len(ARGV) == 3: # 分区表和5.7的情况
		_f = IBDREADER(ARGV[2])
		fsp_bdata = _f.read()
	else:
		fsp_bdata = f.read()
	if fsp_bdata[24:26] != b'\x00\x08': 
		print(f"这文件{filename}不是ibd文件!")
		sys.exit(2)
	SDI_PAGE_NO = struct.unpack('>I',fsp_bdata[38+112+40*256+115:][4:8])[0]
	if fsp_bdata[10390:10390+115] != b'\x00'*115:
		print(f"这文件{filename}加密了, 暂不支持")
	if len(ARGV) == 3:
		bdata = _f.read(SDI_PAGE_NO)
	else:
		bdata = f.read(SDI_PAGE_NO) # 暂不支持general tablespace
	offset = struct.unpack('>h',bdata[97:99])[0]+99 +12 +6 +7 
	dunzip_len,dzip_len = struct.unpack('>LL',bdata[offset:offset+8])
	unzbdata = zlib.decompress(bdata[offset+8:offset+8+dzip_len])
	dic_info = json.loads(unzbdata.decode())
	dd = dic_info['dd_object']
	if dd['row_format'] != 4:
		print('只支持redundant格式')
		sys.exit(4)
	_n = -1
	for col in dd['columns']:
		_n += 1
		if col['type'] in [22,23]: # enum,set
			_xx = {}
			for i in col['elements']:
				_xx[i['index']] = base64.b64decode(i['name']).decode()
			dd['columns'][_n]['elements_dict'] = _xx
		if HAVE_IBD2SQL:
			dd['columns'][_n]['character_set'] = COLLID_TO_CHAR[col['collation_id']][0]
			ct,isvar,size,isbig,elements_dict,varsize,extra = innodb_type_isvar(col)
			dd['columns'][_n]['extra'] = extra
	TABLE_NAME = dd['name']
	f.current_page = 2
	ROWID = True if dd['indexes'][0]['elements'][0]['length'] == 4294967295 else False
	PRIMARRY_KEY = []
	for x in dd['indexes'][0]['elements']:
		if x['length'] == 4294967295:
			continue
		MAX_LEN = len(base64.b64decode(dd['columns'][x['ordinal_position']]['default_value']))
		isprefix = True if MAX_LEN > x['length'] else False
		PRIMARRY_KEY.append([x['ordinal_position'],isprefix])
	# 获取主键的INDEXID, inode的第一个page就是,但太麻烦了, 所以我们从PK ROOT PAGE获取. 但是对于5.7的话,就可能得人工指定了.
	rootpagebdata = f.read(int(dict([ x.split('=') for x in  dd['indexes'][0]['se_private_data'].split(';')[:-1] ])['root']))
	INDEXID = struct.unpack('>L',rootpagebdata[70:74])[0]
	INDEXID = struct.pack('>L',INDEXID) # 转为二进制,方便比较, 上面是为了获取.
	for _pageno in range(int(os.stat(filename).st_size/16384)-2):
		bdata = f.read(_pageno)
		if bdata[24:26] != b'E\xbf' or bdata[64:66] != b'\x00\x00' or bdata[70:74] != INDEXID:
			continue # b'E\xbf': 17855,idxpage   b'\x00\x00' page_level:0叶子节点
		rr = ROWREAD(bdata)
		isfirst = True
		while True:
			rh = rr.read_record_header()
			if rh['next_record'] == 0:
				break
			if isfirst:
				rr.offset = rh['next_record']
				rr._offset = rh['next_record']
				isfirst = False
				continue
			row = {}
			for k,isprefix in PRIMARRY_KEY:
				_tdata = rr.read_filed(k,dd,rh,f.f)
				if not isprefix:
					row[dd['columns'][k]['name']] = _tdata
			if len(PRIMARRY_KEY) == 0:
				rr.read_nullandsize(rh) # rowid
				rowid = rr.read(6)
			rr.read_nullandsize(rh) # trx
			rr.read_nullandsize(rh) # rollptr
			trxrollptr = rr.read_trx_rollptr()
			_idx = -1
			for col in dd['columns']:
				_idx += 1
				if col['name'] in row or (col['name'] in ['DB_ROW_ID','DB_TRX_ID','DB_ROLL_PTR']):
					continue
				try:
					row[col['name']] = rr.read_filed(_idx,dd,rh,f.f)
				except Exception as e:
					print(e,f.current_page,rr.offset,row,rh) # 大概率是坏块或者索引不匹配的问题
					sys.exit(2)
			rr.offset = rh['next_record']
			rr._offset = rh['next_record']
			rr._last_offset = 0
			ROW2SQL(row,dd['columns'],TABLE_NAME)
		#break # 先1页试试水
	f.close()

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

mysql

#ibd2sql