Skip to content

compatiable ror bitmap rdb type #1079

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jackjeyis wants to merge 7 commits into
base: master
Choose a base branch
from
Open

compatiable ror bitmap rdb type #1079

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
1406f24
compatiable ror bitmap rdb type
Jul 16, 2026
04a3858
rdb type add version
Jul 20, 2026
beb548b
add conflict type into version map
Jul 20, 2026
758d1ad
support 8.2 new command
Jul 20, 2026
e807c09
support multi split
Jul 21, 2026
61e3910
RedisOpType add fields attr
Jul 22, 2026
22abb62
fix hashtag check & subkey parser merge general fields logic
Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 27 additions & 1 deletion redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/redis/operation/RedisOpType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,15 +40,24 @@ public enum RedisOpType {
LREM(false, 4),
LSET(false, 4),
LTRIM(false, 4),
BLMPOP(true,-5),
LMPOP(true,-4),

// Hash single
HDEL(false, -3),
HINCRBY(false, 4),
HINCRBYFLOAT(false, 4),
HMSET(false, -4),
HSET(false, -4),
HSETEX(false, -4),
HSETNX(false, 4),
HSETEX(false, true,-6),
HEXPIREAT(false,true,-6),
HPEXPIREAT(false,true,-6),
HEXPIRE(false,true,-6),
HPEXPIRE(false,true,-6),
HGETDEL(false,true,-5),
HPERSIST(false,true,-5),
HGETEX(false,true,-5),

// Set single
SADD(false, -3),
Expand All @@ -62,13 +71,18 @@ public enum RedisOpType {
ZREMRANGEBYLEX(false, 4),
ZREMRANGEBYRANK(false, 4),
ZREMRANGEBYSCORE(false, 4),
ZMPOP(true,-4),
BZMPOP(true,-5),

// Stream single
XADD(false, -5),
XDEL(false, -3),
XSETID(false, 3),
XGROUP(false, -2),
XCLAIM(false, -6),
XDELEX(false,-6),
XACKDEL(false,-7),


// TTL single
EXPIRE(false, 3),
Expand All @@ -83,6 +97,8 @@ public enum RedisOpType {

// Bit single
SETBIT(false, 4),
BITCOUNT(false, 4),


// String multi
DEL(true, -2),
Expand Down Expand Up @@ -154,6 +170,8 @@ public enum RedisOpType {

private RedisOpCrdtTransfer transfer;

private boolean fields;

private static final Map<String, RedisOpType> NAME_CACHE = new HashMap<>();
static {
for (RedisOpType op : values()) {
Expand All @@ -175,6 +193,12 @@ public enum RedisOpType {
this(multiKey, arity, swallow, null);
}

RedisOpType(boolean multiKey, boolean fields, int artiy) {
this.supportMultiKey = multiKey;
this.fields = fields;
this.arity = artiy;
}

RedisOpType(boolean multiKey, int arity, RedisOpCrdtTransfer transfer) {
this(multiKey, arity, false, transfer);
}
Expand All @@ -198,6 +222,8 @@ public boolean isSwallow() {
return swallow;
}

public boolean isFields() {return fields;}

public Pair<RedisOpType, byte[][]> transfer(RedisOpType redisOpType, byte[][] args) {
if (null == transfer) {
return Pair.of(redisOpType, args);
Expand Down
2 changes: 1 addition & 1 deletion ...s-core/src/main/java/com/ctrip/xpipe/redis/core/redis/operation/op/RedisOpItemParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ private RedisOpItemParser() {
public static RedisOpItem parse(RedisOpParser redisOpParser, Object[] payload) {
RedisOpItem redisOpItem = new RedisOpItem();
try {
RedisOp redisOp = redisOpParser.parse(payload);
RedisOp redisOp = redisOpParser.parse(payload);
redisOpItem.setRedisOpType(redisOp.getOpType());
redisOpItem.setGtid(redisOp.getOpGtid());
redisOpItem.setDbId(redisOp.getDbId());
Expand Down
4 changes: 4 additions & 0 deletions ...src/main/java/com/ctrip/xpipe/redis/core/redis/operation/parser/RedisOpMultiKeysEnum.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@ public enum RedisOpMultiKeysEnum {
MSETNX(RedisOpType.MSETNX, 1, 2),
DEL(RedisOpType.DEL, 1, 1),
UNLINK(RedisOpType.UNLINK, 1, 1),
LMPOP(RedisOpType.LMPOP, 2, 1),
BLMPOP(RedisOpType.BLMPOP, 3, 1),
ZMPOP(RedisOpType.ZMPOP, 2, 1),
BZMPOP(RedisOpType.BZMPOP, 3, 1),

//crdt,
CRDT_DEL_REG(RedisOpType.CRDT_DEL_REG, 1, 1),
Expand Down
20 changes: 19 additions & 1 deletion ...c/main/java/com/ctrip/xpipe/redis/core/redis/operation/parser/RedisOpMultiKeysParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,10 @@
import com.ctrip.xpipe.redis.core.redis.operation.RedisOpType;
import com.ctrip.xpipe.redis.core.redis.operation.op.RedisOpMultiKVs;
import com.ctrip.xpipe.tuple.Pair;
import org.checkerframework.checker.units.qual.A;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
Expand All @@ -17,6 +19,12 @@
*/
public class RedisOpMultiKeysParser extends AbstractRedisOpParser implements RedisOpParser {

private static final byte[] MIN_BYTES = "MIN".getBytes();
private static final byte[] MIN_BYTES_LOWER = "min".getBytes();

private static final byte[] MAX_BYTES = "MAX".getBytes();
private static final byte[] MAX_BYTES_LOWER = "max".getBytes();

private RedisOpType redisOpType;
private Integer keyStartIndex;
private Integer kvNum;
Expand All @@ -37,7 +45,17 @@ public RedisOp parse(byte[][] args) {
List<Pair<RedisKey, byte[]>> kvs = new ArrayList<>();

int i = keyStartIndex;
while (i < args.length) {
int argLen = args.length;

@LanternLee LanternLee Jul 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

扫一下8.0注册了getkeyrequests_proc的命令，如果都符合可以用这种写法；否则可以参考下Redis的写法，对于特殊的key获取定义方法注册到命令定义枚举

if(keyStartIndex>1){
int numKeys;
try {
numKeys = Integer.parseInt(new String(args[keyStartIndex-1]));
} catch (NumberFormatException e) {
throw new IllegalArgumentException("Invalid numkeys for " + redisOpType);
}
argLen = i+numKeys;
}
while (i < argLen) {
RedisKey key = new RedisKey(args[i++]);
kvs.add(kvNum == 1 ? Pair.of(key, null) : Pair.of(key, args[i++]));
}
Expand Down
9 changes: 8 additions & 1 deletion ...c/main/java/com/ctrip/xpipe/redis/core/redis/operation/parser/RedisOpWithSubKeysEnum.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,14 @@ public enum RedisOpWithSubKeysEnum {
HINCRBYFLOAT(RedisOpType.HINCRBYFLOAT, 1, 2,false),
HMSET(RedisOpType.HMSET, 1, 2,false),
HSET(RedisOpType.HSET, 1, 2,false),
HSETEX(RedisOpType.HSET, 1, 2,false),
HSETEX(RedisOpType.HSETEX, 1, 2,false),
HGETEX(RedisOpType.HGETEX, 1, 1,false),
HGETDEL(RedisOpType.HGETDEL, 1, 1,false),
HEXPIRE(RedisOpType.HEXPIRE, 1, 1,false),
HEXPIREAT(RedisOpType.HEXPIREAT, 1, 1,false),
HPEXPIRE(RedisOpType.HPEXPIRE, 1, 1,false),
HPEXPIREAT(RedisOpType.HPEXPIREAT, 1, 1,false),
HPERSIST(RedisOpType.HPERSIST, 1, 1,false),
HSETNX(RedisOpType.HSETNX, 1, 2,false),

ZADD(RedisOpType.ZADD, 1, 2,true),
Expand Down
79 changes: 65 additions & 14 deletions ...main/java/com/ctrip/xpipe/redis/core/redis/operation/parser/RedisOpWithSubKeysParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,11 @@
import com.ctrip.xpipe.redis.core.redis.operation.RedisOp;
import com.ctrip.xpipe.redis.core.redis.operation.RedisOpParser;
import com.ctrip.xpipe.redis.core.redis.operation.RedisOpType;
import com.ctrip.xpipe.redis.core.redis.operation.op.RedisOpMultiKVs;
import com.ctrip.xpipe.redis.core.redis.operation.op.RedisOpMultiSubKey;
import com.ctrip.xpipe.tuple.Pair;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.nio.ByteBuffer;
import java.util.*;

/**
* @author tb
Expand All @@ -24,12 +22,28 @@ public class RedisOpWithSubKeysParser extends AbstractRedisOpParser implements R
private Integer kvNum;
private boolean kvReverse;

private byte[] NX_BYTES = new byte[]{'N','X'};
private byte[] XX_BYTES = new byte[]{'X','X'};
private byte[] GT_BYTES = new byte[]{'G','T'};
private byte[] LT_BYTES = new byte[]{'L','T'};
private byte[] CH_BYTES = new byte[]{'C','H'};
private byte[] INCR_BYTES = new byte[]{'I','N','C','R'};
private static final Set<ByteBuffer> NON_KEY_COMMANDS = new HashSet<>();
private static final Set<ByteBuffer> NON_KEY_COMMANDS_WITH_ARGS = new HashSet<>();

private static final byte[] FIELDS_BYTES = "FIELDS".getBytes();
private static final byte[] FIELDS_BYTES_LOWER = "fields".getBytes();

static {
// 无参数关键字
String[] noArgCmds = {
"NX", "nx", "XX", "xx", "GT", "gt", "LT", "lt", "CH", "ch",
"INCR", "incr", "KEEPTTL", "keepttl", "FNX", "fnx", "FXX", "fxx", "PERSIST", "persist"
};
for (String cmd : noArgCmds) {
NON_KEY_COMMANDS.add(ByteBuffer.wrap(cmd.getBytes()));
}
// 带一个参数的关键字
String[] oneArgCmds = { "EX", "ex", "PX", "px", "EXAT", "exat", "PXAT", "pxat" };
for (String cmd : oneArgCmds) {
NON_KEY_COMMANDS_WITH_ARGS.add(ByteBuffer.wrap(cmd.getBytes()));
NON_KEY_COMMANDS.add(ByteBuffer.wrap(cmd.getBytes())); // 也在无参数集合中，因为需要先识别
}
}

public RedisOpWithSubKeysParser(RedisOpType redisOpType, int keyStartIndex, int kvNum,boolean kvReverse) {
this.redisOpType = redisOpType;
Expand All @@ -43,6 +57,13 @@ public RedisOp parse(byte[][] args) {
Pair<RedisOpType, byte[][]> pair = redisOpType.transfer(redisOpType, args);
args = pair.getValue();

if(redisOpType.isFields()) {
return parseFieldsCommand(redisOpType,args);
}
return parseGeneralCommands(args, pair);
}

private RedisOpMultiSubKey parseGeneralCommands(byte[][] args, Pair<RedisOpType, byte[][]> pair) {
// 计算容量 - 主键 + 子键数量
int subKeyCount = (args.length - keyStartIndex-1) / kvNum;
int capacity = 1 + subKeyCount;
Expand Down Expand Up @@ -76,7 +97,7 @@ public RedisOp parse(byte[][] args) {
subKeys.add(subKey);
}

return new RedisOpMultiSubKey(pair.getKey(), args, key,subKeys);
return new RedisOpMultiSubKey(pair.getKey(), args, key, subKeys);
}

@Override
Expand All @@ -85,8 +106,38 @@ public int getOrder() {
}

private boolean nonKey(byte[] args){
return Arrays.equals(NX_BYTES,args) || Arrays.equals(XX_BYTES,args)
|| Arrays.equals(GT_BYTES,args) || Arrays.equals(LT_BYTES,args)
|| Arrays.equals(CH_BYTES,args) || Arrays.equals(INCR_BYTES,args);
return NON_KEY_COMMANDS.contains(ByteBuffer.wrap(args));
}

private RedisOp parseFieldsCommand(RedisOpType opType, byte[][] args) {

@LanternLee LanternLee Jul 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

和parseGeneralCommands的关系？

int idx = keyStartIndex;
RedisKey key = new RedisKey(args[idx++]);

// 跳过 FIELDS 之前的所有可选标志
while (idx < args.length) {
ByteBuffer tokenBuf = ByteBuffer.wrap(args[idx]);
if (NON_KEY_COMMANDS.contains(tokenBuf)) {
idx++;
if (NON_KEY_COMMANDS_WITH_ARGS.contains(tokenBuf) && idx < args.length) {
idx++; // 跳过标志的参数
}
} else if (Arrays.equals(args[idx], FIELDS_BYTES) || Arrays.equals(args[idx], FIELDS_BYTES_LOWER)) {
idx++;
break;
} else {
idx++;
}
}

if (idx >= args.length) throw new IllegalArgumentException("Missing numfields");
int numFields = Integer.parseInt(new String(args[idx++]));
List<RedisKey> subKeys = new ArrayList<>(numFields);

for (int i = 0; i < numFields; i++) {
if (idx >= args.length) throw new IllegalArgumentException("Incomplete field list");
subKeys.add(new RedisKey(args[idx]));
idx += kvNum;
}
return new RedisOpMultiSubKey(opType, args, key, subKeys);
}
}
3 changes: 2 additions & 1 deletion redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/redis/rdb/RdbConstant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,8 @@ private RdbConstant() {
public static final short REDIS_RDB_TYPE_HASH_ZIPLIST = 13;
public static final short REDIS_RDB_TYPE_LIST_QUICKLIST = 14;
public static final short REDIS_RDB_TYPE_STREAM_LISTPACKS = 15;
// public static final short REDIS_RDB_TYPE_BITMAP = 16;
public static final short REDIS_RDB_TYPE_BITMAP = 192;
public static final short REDIS_RDB_TYPE_BITMAP_9 = 16;
public static final short REDIS_RDB_TYPE_CRDT = 200;

// Redis 8.x 新增操作码 (rdb.h)
Expand Down
26 changes: 24 additions & 2 deletions redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/redis/rdb/RdbParseContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,8 @@ enum RdbType {
HASH_ZIPLIST(RdbConstant.REDIS_RDB_TYPE_HASH_ZIPLIST, false, RdbHashZipListParser::new),
LIST_QUICKLIST(RdbConstant.REDIS_RDB_TYPE_LIST_QUICKLIST, false, RdbQuickListParser::new),
STREAM_LISTPACKS(RdbConstant.REDIS_RDB_TYPE_STREAM_LISTPACKS, false, RdbStreamListpacksParser::new),
// BITMAP(RdbConstant.REDIS_RDB_TYPE_BITMAP, false, RdbBitmapParser::new),
BITMAP(RdbConstant.REDIS_RDB_TYPE_BITMAP, false, RdbBitmapParser::new),
BITMAP_9(9,RdbConstant.REDIS_RDB_TYPE_BITMAP_9, false, RdbBitmapParser::new),
CRDT(RdbConstant.REDIS_RDB_TYPE_CRDT, false, DefaultRdbCrdtParser::new),
// MODULE_AUX(RdbConstant.REDIS_RDB_OP_CODE_MODULE_AUX),
IDLE(RdbConstant.REDIS_RDB_OP_CODE_IDLE, true, RdbIdleParser::new),
Expand Down Expand Up @@ -132,15 +133,22 @@ enum RdbType {
STREAM_LISTPACKS_3(RdbConstant.REDIS_RDB_TYPE_STREAM_LISTPACKS_3, false, RdbStreamListpacks3Parser::new),
STREAM_LISTPACKS_4(RdbConstant.REDIS_RDB_TYPE_STREAM_LISTPACKS_4, false, RdbStreamListpacks4Parser::new);

private int version;
private short code;

private boolean rdbOp;

private Function<RdbParseContext, RdbParser> parserConstructor;

private static final Map<Short, RdbType> types = new HashMap<>();
private static final Map<Integer,Map<Short, RdbType>> versionTypes = new HashMap<>();

RdbType(short code, boolean rdbOp, Function<RdbParseContext, RdbParser> parserConstructor) {
this(0,code,rdbOp,parserConstructor);
}

RdbType(int version,short code, boolean rdbOp, Function<RdbParseContext, RdbParser> parserConstructor) {
this.version = version;
this.code = code;
this.rdbOp = rdbOp;
this.parserConstructor = parserConstructor;
Expand All @@ -163,14 +171,28 @@ public RdbParser<?> makeParser(RdbParseContext parserManager) {

static {
for (RdbType rdbType : values()) {
types.put(rdbType.code, rdbType);
if(rdbType.version != 0) {
Map<Short, RdbType> versionMap = versionTypes.computeIfAbsent(rdbType.version, (version) -> new HashMap<>());
versionMap.put(rdbType.code, rdbType);
}else {
types.put(rdbType.code, rdbType);
}
}
}

public static RdbType findByCode(short code) {
return types.get(code);
}

public static RdbType findByCode(int version,short code) {
Map<Short,RdbType> versionMap = versionTypes.getOrDefault(version,types);
RdbType rdbType = versionMap.get(code);
if(rdbType == null){
return findByCode(code);
}
return rdbType;
}

}

String cSet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
Expand Down
2 changes: 1 addition & 1 deletion ...edis-core/src/main/java/com/ctrip/xpipe/redis/core/redis/rdb/parser/DefaultRdbParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ public Void read(ByteBuf byteBuf) {

case READ_TYPE:
short type = byteBuf.readUnsignedByte();
RdbParseContext.RdbType newType = RdbParseContext.RdbType.findByCode(type);
RdbParseContext.RdbType newType = RdbParseContext.RdbType.findByCode(rdbVersion,type);
if (currentType == RdbParseContext.RdbType.AUX && newType != RdbParseContext.RdbType.AUX) {
auxFinished.set(true);
notifyAuxEnd(rdbParseContext.getAllAux());
Expand Down
5 changes: 5 additions & 0 deletions ...redis-core/src/main/java/com/ctrip/xpipe/redis/core/redis/rdb/parser/RdbBitmapParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,11 @@ private void propagateCmdIfNeed(byte[] value) {
RedisOpType.SET,
new byte[][]{RedisOpType.SET.name().getBytes(), context.getKey().get(), value},
context.getKey(), value));

notifyRedisOp(new RedisOpSingleKey(
RedisOpType.BITCOUNT,
new byte[][]{RedisOpType.BITCOUNT.name().getBytes(), context.getKey().get()},
context.getKey(), value));
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion ...s-core/src/main/java/com/ctrip/xpipe/redis/core/redis/rdb/parser/RdbQuickList2Parser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,8 +63,8 @@ public Integer read(ByteBuf byteBuf) {
break;
case READ_DATA:
byte[] data = rdbStringParser.read(byteBuf);
rdbStringParser.reset();
if (data != null) {
rdbStringParser.reset();
readCnt++;

if (readCnt >= len.getLenValue()) {
Expand Down
Loading
Loading