使用 TiKV 读改写 TiDB 数据

作者：TiDB 社区干货传送门

2024-02-09
北京
本文字数：7135 字
阅读完需：约 23 分钟

作者：表渣渣渣原文来源：https://tidb.net/blog/d9f0295f

一切开始的原因

由于数据开发的需要，我一度尝试将 tidb 的使用范围更大话，同时目前大数据开发中，内存当做堆料，对于公司的开支也会与很大压力，那么就我就尝试将 tikv 当做 kafka 和 redis 使用，本文章中将讲述开发的过程以及衍生品；

row_id 是什么

也许你和我在尝试使用 tikv 的时候会感觉网上的资料好像都是不太对劲的样子，比如：

https://tikv.github.io/client-java/examples/txnkv.html

我们看看这块代码：

import java.util.Arrays;import java.util.List;
import org.tikv.common.BytePairWrapper;import org.tikv.common.ByteWrapper;import org.tikv.common.TiConfiguration;import org.tikv.common.TiSession;import org.tikv.common.util.BackOffer;import org.tikv.common.util.ConcreteBackOffer;import org.tikv.kvproto.Kvrpcpb.KvPair;import org.tikv.shade.com.google.protobuf.ByteString;import org.tikv.txn.KVClient;import org.tikv.txn.TwoPhaseCommitter;
public class App {    public static void main(String[] args) throws Exception {        TiConfiguration conf = TiConfiguration.createDefault("127.0.0.1:2379");        try (TiSession session = TiSession.create(conf)) {            // two-phrase write            long startTS = session.getTimestamp().getVersion();            try (TwoPhaseCommitter twoPhaseCommitter = new TwoPhaseCommitter(session, startTS)) {                BackOffer backOffer = ConcreteBackOffer.newCustomBackOff(1000);                byte[] primaryKey = "key1".getBytes("UTF-8");                byte[] key2 = "key2".getBytes("UTF-8");
                // first phrase: prewrite                twoPhaseCommitter.prewritePrimaryKey(backOffer, primaryKey, "val1".getBytes("UTF-8"));                List<BytePairWrapper> pairs = Arrays                        .asList(new BytePairWrapper(key2, "val2".getBytes("UTF-8")));                twoPhaseCommitter.prewriteSecondaryKeys(primaryKey, pairs.iterator(), 1000);
                // second phrase: commit                long commitTS = session.getTimestamp().getVersion();                twoPhaseCommitter.commitPrimaryKey(backOffer, primaryKey, commitTS);                List<ByteWrapper> keys = Arrays.asList(new ByteWrapper(key2));                twoPhaseCommitter.commitSecondaryKeys(keys.iterator(), commitTS, 1000);            }
            try (KVClient kvClient = session.createKVClient()) {                long version = session.getTimestamp().getVersion();                ByteString key1 = ByteString.copyFromUtf8("key1");                ByteString key2 = ByteString.copyFromUtf8("key2");
                // get value of a single key                ByteString val = kvClient.get(key1, version);                System.out.println(val);
                // get value of multiple keys                BackOffer backOffer = ConcreteBackOffer.newCustomBackOff(1000);                List<KvPair> kvPairs = kvClient.batchGet(backOffer, Arrays.asList(key1, key2), version);                System.out.println(kvPairs);
                // get value of a range of keys                kvPairs = kvClient.scan(key1, ByteString.copyFromUtf8("key3"), version);                System.out.println(kvPairs);            }        }    }}

复制代码

tikv 存储的时候 key 是什么东西呢? 在使用 scan api 后读取的数据全部都是 encode 的数据，参考网上 flink-tidb-cdc 源码，只有 value 的 decode , key 呢？

首先我们打印一下 key 长得样子

key = <ByteString@1b78595f size=19 contents="t\200\000\000\000\000\000\000\212_r\200\000\000\000\000\001N1">

复制代码

看结果好像应该是有个是 table_id，一个是 row_id；

我们继续找资料：

官方文档中 region 切分有提及：

Split Table Region

表中行数据的 key 由 table_id 和 row_id 编码组成，格式如下：

t[table_id]_r[row_id]

复制代码

例如，当 table_id 是 22，row_id 是 11 时：

t22_r11

复制代码

同一表中行数据的 table_id 是一样的，但 row_id 肯定不一样，所以可以根据 row_id 来切分 Region。

TIDB_DECODE_KEY

TIDB_DECODE_KEY 函数用于将 TiDB 编码的键输入解码为包含 _tidb_rowid 和 table_id 的 JSON 结构。你可以在一些系统表和日志输出中找到 TiDB 的编码键。

语法图

语法图代码

TableStmtTIDB_DECODE_KEY(STR) 示例

以下示例中，表 t1 有一个隐藏的 rowid，该 rowid 由 TiDB 生成。语句中使用了 TIDB_DECODE_KEY 函数。结果显示，隐藏的 rowid 被解码后并输出，这是典型的非聚簇主键结果。

SELECT START_KEY, TIDB_DECODE_KEY(START_KEY) FROM information_schema.tikv_region_status WHERE table_name='t1' AND REGION_ID=2\G*************************** 1. row ***************************                 START_KEY: 7480000000000000FF3B5F728000000000FF1DE3F10000000000FATIDB_DECODE_KEY(START_KEY): {"_tidb_rowid":1958897,"table_id":"59"}1 row in set (0.00 sec)

复制代码

但是至此我们还是无法明白 row_id 到底是怎么生成的，受什么控制：

终于我在一篇 tidb 公众号文章（参考资料里）里面找到答案

文中这样写道：


至此我们已经聊完了如何将 Table 映射到 KV 上面，这里再举个简单的例子，便于大家理解，还是以上面的表结构为例。假设表中有 3 行数据：
1, "TiDB", "SQL Layer", 10
2, "TiKV", "KV Engine", 20
3, "PD", "Manager", 30
那么首先每行数据都会映射为一个 Key-Value pair，注意这个表有一个 Int 类型的 Primary Key，所以 RowID 的值即为这个 Primary Key 的值。假设这个表的 Table ID 为 10，其 Row 的数据为：
t_r_10_1  --> ["TiDB", "SQL Layer", 10]
t_r_10_2 --> ["TiKV", "KV Engine", 20]
t_r_10_3 --> ["PD", "Manager", 30]

复制代码

至此，我们可以得到一个结论，key 包含两个信息，一个是 table_id，另外一个是 row_id，且当主键为 INT 类型时，主键就是 row_id；

able_id 可以通过如下获得；

TiTableInfo table = session.getCatalog().getTable("ods", "ods_consult_order_event");                long id = table.getId();

复制代码

写一个生成 key 的方法

我们再扒一扒 tikv Client 源码：

    protected static final byte[] TBL_PREFIX = new byte[]{116};// 这个其实就是t     private static final byte[] REC_PREFIX_SEP = new byte[]{95, 114}; // 这个其实就是_r
private static byte[] encode(long tableId, long handle) {        CodecDataOutput cdo = new CodecDataOutput();        encodePrefix(cdo, tableId);        IntegerCodec.writeLong(cdo, handle);        return cdo.toBytes();    }


    private static void encodePrefix(CodecDataOutput cdo, long tableId) {        cdo.write(TBL_PREFIX);        IntegerCodec.writeLong(cdo, tableId);        cdo.write(REC_PREFIX_SEP);    }
        public static void writeLong(CodecDataOutput cdo, long lVal) {            cdo.writeLong(flipSignBit(lVal));        }
        private static long flipSignBit(long v) {            return v ^ -9223372036854775808L;        }


    /**     * Writes a <code>long</code> to the underlying output stream as eight     * bytes, high byte first. In no exception is thrown, the counter     * <code>written</code> is incremented by <code>8</code>.     *     * @param      v   a <code>long</code> to be written.     * @exception  IOException  if an I/O error occurs.     * @see        java.io.FilterOutputStream#out     */    public final void writeLong(long v) throws IOException {        writeBuffer[0] = (byte)(v >>> 56);        writeBuffer[1] = (byte)(v >>> 48);        writeBuffer[2] = (byte)(v >>> 40);        writeBuffer[3] = (byte)(v >>> 32);        writeBuffer[4] = (byte)(v >>> 24);        writeBuffer[5] = (byte)(v >>> 16);        writeBuffer[6] = (byte)(v >>>  8);        writeBuffer[7] = (byte)(v >>>  0);        out.write(writeBuffer, 0, 8);        incCount(8);    }

复制代码

通过以上我们可以写一个通过 table_id，row_id 获取 key 的方法：

package util;
/** * @Maintainer 蜡笔老舅 * @Email  * @CreateDate 2023/3/17 * @Version 1.0 * @Comment */import java.io.Serializable;import org.tikv.common.codec.CodecDataInput;import org.tikv.common.codec.CodecDataOutput;import org.tikv.common.codec.Codec.IntegerCodec;import org.tikv.common.exception.TiClientInternalException;import org.tikv.common.exception.TiExpressionException;import org.tikv.common.key.Key;import org.tikv.common.key.TypedKey;
import static org.tikv.common.key.RowKey.toRowKey;
public class TidbRowKeyUtil extends Key implements Serializable {    private static final byte[] REC_PREFIX_SEP = new byte[]{95, 114};    private final long tableId;    private final long handle;    private final boolean maxHandleFlag;
    private TidbRowKeyUtil(long tableId, long handle) {        super(encode(tableId, handle));        this.tableId = tableId;        this.handle = handle;        this.maxHandleFlag = false;    }
    public static byte[] encode(long tableId, long handle) {        CodecDataOutput cdo = new CodecDataOutput();        encodePrefix(cdo, tableId);        encodeRow(cdo,handle);        return cdo.toBytes();    }


    private static void encodePrefix(CodecDataOutput cdo, long tableId) {        // 增加 /t table_id 标识        cdo.write(TBL_PREFIX);        cdo.writeLong(tableId ^ Long.MIN_VALUE);        // 写入一个 /r row_id 标识        cdo.write(REC_PREFIX_SEP);    }
    private static void encodeRow(CodecDataOutput cdo, long tableId) {        // 对row_id 进行编码        cdo.writeLong(tableId ^ Long.MIN_VALUE);    }

}

复制代码

然后我们验证：

package tidb;
import com.alibaba.fastjson.JSONObject;import org.tikv.common.TiConfiguration;import org.tikv.common.TiSession;import org.tikv.common.key.RowKey;import org.tikv.common.meta.TiColumnInfo;import org.tikv.common.meta.TiTableInfo;import org.tikv.kvproto.Kvrpcpb;import org.tikv.shade.com.google.protobuf.ByteString;import org.tikv.txn.KVClient;
import java.text.SimpleDateFormat;import java.util.Arrays;import java.util.Date;import java.util.List;
import static org.tikv.common.codec.TableCodec.decodeObjects;
/** * @Maintainer 蜡笔老舅 * @Email  * @CreateDate 2023/3/17 * @Version 1.0 * @Comment */public class KVClientMatchTest implements Runnable{    public static void main(String[] args) {        //3.创建实现类的对象        KVClientMatchTest mThread = new KVClientMatchTest();
        //4.将此对象作为参数传递到Thread类的构造器中，创建Thread类的对象        Thread t1 = new Thread(mThread);        t1.setName("线程1");        t1.start();////        Thread t2 = new Thread(mThread);//        t2.setName("线程2");//        t2.start();////        Thread t3 = new Thread(mThread);//        t3.setName("线程3");//        t3.start();////        Thread t4 = new Thread(mThread);//        t4.setName("线程4");//        t4.start();////        Thread t5 = new Thread(mThread);//        t5.setName("线程5");//        t5.start();    }
    private static void getKey(TiSession session, KVClient kvClient, Integer i) {        long version = session.getTimestamp().getVersion();        System.out.println("version = " + version);        System.out.println("System.currentTimeMillis() = " + System.currentTimeMillis());        TiTableInfo table = session.getCatalog().getTable("库名", "表名");
        byte[] encode = TidbRowKeyUtil.encode(table.getId(), i);        ByteString key = ByteString.copyFrom(encode);        ByteString bytes = kvClient.get(key, version);
        System.out.println("bytes.size() = " + bytes.size());
        if (bytes.size()!=0) {            Kvrpcpb.KvPair.Builder builder = Kvrpcpb.KvPair.newBuilder();            builder.setKey(key);            builder.setValue(bytes);            Kvrpcpb.KvPair build = builder.build();            JSONObject dataJSON = getDataJSON(build, table.getColumns(), table);            System.out.println("i = " + i);            System.out.println("dataJSON = " + dataJSON);        }    }    private static JSONObject getDataJSON(Kvrpcpb.KvPair record, List<TiColumnInfo> columns, TiTableInfo tiTableInfo) {        JSONObject jsonObject = new JSONObject();        Object[] tikvValues =                decodeObjects(                        record.getValue().toByteArray(),                        RowKey.decode(record.getKey().toByteArray()).getHandle(),                        tiTableInfo);        List<Object> objects = Arrays.asList(tikvValues);        for (int i = 0; i < tikvValues.length; i++) {            String type = columns.get(i).getType().getName();            jsonObject.put(columns.get(i).getName(),getDataObject(objects.get(i),type));        }        return jsonObject;    }    protected static Object getDataObject(Object o, String type) {
        if (o!=null) {            switch(type.toLowerCase()) {                case "timestamp": case "datetime@asia/shanghai":                    return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(o);                case "time":                    return new SimpleDateFormat("HH:mm:ss").format(new Date(Long.parseLong(o.toString())));                default:                    return o;            }        }        return o;    }
    @Override    public void run() {        Long startTs=System.currentTimeMillis();        TiConfiguration conf = TiConfiguration.createDefault(Constant.TIDB_PD_ADDR_TEST);        TiSession session = TiSession.create(conf);        KVClient kvClient = session.createKVClient();        for (int j = 1; j <= 10000; j++) {            getKey(session,kvClient,j);        }
        kvClient.close();        try {            session.close();        } catch (Exception e) {            e.printStackTrace();        }        Long endTs=System.currentTimeMillis();        long l = endTs - startTs;        System.out.println("l = " + l);    }}