[MYSQL] python扫描磁盘恢复数据的可行性验证与速度测试

-- 准备测试表和数据
create table db1.t20251128_for_drop(id int primary key auto_increment, name varchar(200));
insert into db1.t20251128_for_drop(name) values('ddcw');
insert into db1.t20251128_for_drop(name) select name from db1.t20251128_for_drop;
insert into db1.t20251128_for_drop(name) select name from db1.t20251128_for_drop;
-- ....

-- 然后干掉它(你就可以跑路了)
drop table db1.t20251128_for_drop;

python3 main.py /data/mysql_3306/mysqldata/mysql.ibd --delete --set table=tables  | grep t20251128_for_drop
python3 main.py /data/mysql_3306/mysqldata/mysql.ibd --delete --set table=indexes | grep 463

python3 scan_drop_table_demo.py --device /dev/vda1 --indexid 254 --parallel 8

python3 main.py 0000000000000254.page.ibd --sdi /data/mysql_3306/mysqldata/db1/t20251128_for_drop_new.ibd --sql --limit 10 --set leafno=0 --set rootno=0

#!/usr/bin/env python3
# write by ddcw @https://github.com/ddcw
# 测试扫描磁获取相关Indexid的page的测试例子,验证可行性和效率

import os
import sys
import stat
import time
import struct
import shutil
import argparse
from multiprocessing import Process

PAGE_SIZE = 16384
def print_error_and_exit(msg,exit_code=1):
	msg += "\n"
	sys.stdout.write(msg)
	sys.exit(exit_code)

def print_info(msg):
	msg += "\n"
	sys.stdout.write(msg)

def format_size(n):
	if n < 1024:
		return f'{n} B'
	elif n < 1024*1024:
		return f'{round(n/1024,2)} KB'
	elif n < 1024*1024*1024:
		return f'{round(n/1024/1024,2)} MB'
	elif n < 1024*1024*1024*1024:
		return f'{round(n/1024/1024/1024,2)} GB'
	elif n < 1024*1024*1024*1024*1024:
		return f'{round(n/1024/1024/1024/1024,2)} TB'
	else:
		return f'{round(n/1024/1024/1024/1024/1024,2)} PB'

def _argparse():
	parser = argparse.ArgumentParser(add_help=True,description="测试扫描磁获取相关Indexid的page的测试例子,验证可行性和效率")
	parser.add_argument('--device',dest="DEVICE_NAME",required=True,help='磁盘设备/文件')
	parser.add_argument('--start',dest="OFFSET_START",type=int,default=0,help='要扫描的磁盘设备/文件的起始地址,默认0')
	parser.add_argument('--end',dest="OFFSET_END",type=int,default=-1,help='要磁盘设备/文件的结束地址,默认全部')
	parser.add_argument('--step',dest="OFFSET_STEP",type=int,default=512,help='扫描步长,默认512字节')
	parser.add_argument('--buffering',dest="BUFFERING",type=int,default=16*1024*1024,help='缓存大小(非open缓存),默认16MB')
	parser.add_argument('--indexid',dest="INDEXID",type=int,required=True,help='要扫描的表的Indexid')
	#parser.add_argument('--tablespaceid',dest="TABLESPACE_ID",help='要扫描的表的tablespace id')
	parser.add_argument('--parallel',dest="PARALLEL",type=int,default=1,help='并发度')
	parser.add_argument('--output',dest="OUTPUT_FILENAME",type=int,default=1,help='输出文件名,默认为indexid.page')
	parser = parser.parse_args()
	return parser

# 初始化屏幕
def init_screen():
	x = os.system('clear')
	columns,lines = shutil.get_terminal_size()
	return columns

# 获取磁盘设备大小(可能是lv,可能是磁盘,也可能是文件)
def get_size_from_dev(filename):
	f_stat = os.stat(filename)
	file_size = 0
	status = True
	if stat.S_ISREG(f_stat.st_mode): # file
		file_size = f_stat.st_size
	elif stat.S_ISBLK(f_stat.st_mode):
		real_dev = ''
		try:
			real_dev = os.readlink(filename).split('/')[-1] # lv
		except:
			real_dev = os.path.basename(filename) # dev
		with open(f'/sys/class/block/{real_dev}/size') as f:
			sectors = int(f.read().strip())
		try:
			with open(f'/sys/class/block/{real_dev}/queue/hw_sector_size') as f:
				sector_size = int(f.read().strip())
		except:
			sector_size = 512
		file_size = sectors * sector_size
	else:
		status = False
	return status,file_size

# 监控进程,只展示效果,不干活的. (算逑, 不要它了,直接worker输出)
def monitor(q):
	pass

# worker进程,打工仔. (我将给你一个展示的机会!)
def worker(p,filename,start,end,step,indexid,buffering,output_filename,screen_size=0):
	import time
	f = open(filename,'rb')
	fo = open(output_filename,'wb')
	f.seek(start,0)
	buff = b''
	readed_size = 0
	indexid = b'\x00\x00'+struct.pack('>Q',indexid)
	hc = 0
	while end > readed_size:
		start_time = time.time()
		readsize = buffering-len(buff)
		buff += f.read(readsize)
		if len(buff) < 16384:
			break
		readed_size += readsize
		offset = 0
		while True:
			data = buff[offset:offset+16384]
			if len(data) < 16384:
				break
			if data[:4] == data[-8:-4] and data[24:26] == b'E\xbf' and data[64:74] == indexid:
				offset += 16384
				hc += 1
				fo.write(data)
			else:
				offset += step
		buff = buff[offset:]
		end_time = time.time()
		progress = min(100,round(readed_size/end*100,2))
		rate = format_size(readsize/(end_time-start_time)) + '/s'
		progress_bar = "#"*(int(progress)//2)
		progress_bar_wsp = " "*(50-len(progress_bar))
		content = f"[P{str(p+1).zfill(2)}] [{progress_bar}{progress_bar_wsp}] {progress}% {rate} {start}:{start+readed_size} {hc} {' '*5}"
		sys.stdout.write(f"\033[{p+2};0H{content}")
		sys.stdout.flush()
	fo.close()
#	import random
#	for i in range(20):
#		line_num = p+2
#		column = 1
#		content = ''.join([ '#' for _ in range(i) ])
#		sys.stdout.write(f"\033[{line_num};{column}HP[{p}]{content}")
#		sys.stdout.flush()
#		time.sleep(random.random())

def main():
	starttime = time.time()
	parser = _argparse()
	filename = parser.DEVICE_NAME
	parallel = parser.PARALLEL
	if not os.path.exists(filename):
		print_error_and_exit(f'{filename} is not exists')
	screen_size = init_screen()
	status,file_size = get_size_from_dev(filename)
	if not status:
		print_error_and_exit(f'{filename} only support dev/file')
	msg = f"SCAN DEVICE {filename}({format_size(file_size)})"
	msg = " "*((screen_size-len(msg))//2) + msg
	pd = {}
	sys.stdout.write(f'{msg}\n')
	sys.stdout.flush()
	step = parser.OFFSET_STEP
	start = parser.OFFSET_START
	end = parser.OFFSET_END if parser.OFFSET_END > start else file_size
	per_size = (end-start)//parallel//step*step+step
	indexid = parser.INDEXID
	output_filename_pre = '/tmp/' + str(indexid).zfill(16)+'.page'
	for x in range(parallel):
		output_filename = f"{output_filename_pre}{'.'+str(x) if parallel > 1 else ''}"
		pd[x] = Process(target=worker,args=(x,filename,start+x*per_size,per_size,step,indexid,parser.BUFFERING,output_filename,screen_size))
	for x in range(parallel):
		pd[x].start()
	for x in range(parallel):
		pd[x].join()
	stoptime = time.time()
	sys.stdout.write(f"\033[{parallel + 3};{0}HFinish! cost:{round(stoptime-starttime,2)} sec.\n")
	sys.stdout.flush()

if __name__ == '__main__':
	main()