优化逻辑

dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/MappingServiceImpl.java

@@ -19,6 +19,7 @@ import org.dbsyncer.connector.base.ConnectorFactory;
 import org.dbsyncer.manager.ManagerFactory;
 import org.dbsyncer.parser.LogType;
 import org.dbsyncer.parser.ProfileComponent;
+import org.dbsyncer.parser.TableGroupContext;
 import org.dbsyncer.parser.model.ConfigModel;
 import org.dbsyncer.parser.model.Connector;
 import org.dbsyncer.parser.model.Mapping;
@@ -76,6 +77,9 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
     @Resource
     private ConnectorFactory connectorFactory;
 
+    @Resource
+    private TableGroupContext tableGroupContext;
+
     @Override
     public String add(Map<String, String> params) {
         ConfigModel model = mappingChecker.checkAddConfigModel(params);
@@ -152,6 +156,9 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
             profileComponent.removeConfigModel(metaId);
             log(LogType.MetaLog.DELETE, meta);
 
+            // 删除驱动表映射关系
+            tableGroupContext.clear(metaId);
+
             // 删除tableGroup
             List<TableGroup> groupList = profileComponent.getTableGroupAll(id);
             if (!CollectionUtils.isEmpty(groupList)) {

dbsyncer-connector/dbsyncer-connector-mysql/src/main/java/org/dbsyncer/connector/mysql/MySQLConnector.java

@@ -5,6 +5,7 @@ package org.dbsyncer.connector.mysql;
 
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.mysql.cdc.MySQLListener;
+import org.dbsyncer.connector.mysql.schema.MySQLSchemaResolver;
 import org.dbsyncer.connector.mysql.storage.MySQLStorageService;
 import org.dbsyncer.connector.mysql.validator.MySQLConfigValidator;
 import org.dbsyncer.sdk.connector.ConfigValidator;
@@ -15,6 +16,7 @@ import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
 import org.dbsyncer.sdk.listener.Listener;
 import org.dbsyncer.sdk.model.PageSql;
 import org.dbsyncer.sdk.plugin.ReaderContext;
+import org.dbsyncer.sdk.schema.SchemaResolver;
 import org.dbsyncer.sdk.storage.StorageService;
 import org.dbsyncer.sdk.util.PrimaryKeyUtil;
 import org.slf4j.Logger;
@@ -33,12 +35,12 @@ public final class MySQLConnector extends AbstractDatabaseConnector {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
-    private final String TYPE = "MySQL";
     private final MySQLConfigValidator configValidator = new MySQLConfigValidator();
+    private final MySQLSchemaResolver schemaResolver = new MySQLSchemaResolver();
 
     @Override
     public String getConnectorType() {
-        return TYPE;
+        return "MySQL";
     }
 
     @Override
@@ -135,4 +137,8 @@ public final class MySQLConnector extends AbstractDatabaseConnector {
         return true;
     }
 
+    @Override
+    protected SchemaResolver getSchemaResolver() {
+        return schemaResolver;
+    }
 }

dbsyncer-connector/dbsyncer-connector-oracle/src/main/java/org/dbsyncer/connector/oracle/cdc/OracleListener.java

@@ -56,7 +56,6 @@ public class OracleListener extends AbstractDatabaseListener {
     @Override
     public void start() {
         try {
-            // TODO [increment-worker-1184659161326161921] 这里应该单独启动一个线程，线程名要有一定意义，如:binlog-parser-127.0.0.1:3306_123，便于监控排查问题
             final DatabaseConfig config = getConnectorInstance().getConfig();
             String driverClassName = config.getDriverClassName();
             String username = config.getUsername();

dbsyncer-manager/src/main/java/org/dbsyncer/manager/impl/FullPuller.java

@@ -5,14 +5,11 @@ package org.dbsyncer.manager.impl;
 
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.parser.*;
 import org.dbsyncer.sdk.util.PrimaryKeyUtil;
 import org.dbsyncer.manager.AbstractPuller;
-import org.dbsyncer.parser.ParserComponent;
-import org.dbsyncer.parser.ProfileComponent;
 import org.dbsyncer.parser.enums.ParserEnum;
 import org.dbsyncer.parser.event.FullRefreshEvent;
-import org.dbsyncer.parser.LogService;
-import org.dbsyncer.parser.LogType;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.Meta;
 import org.dbsyncer.parser.model.TableGroup;
@@ -53,29 +50,32 @@ public final class FullPuller extends AbstractPuller implements ApplicationListe
     @Resource
     private LogService logService;
 
-    private Map<String, Task> map = new ConcurrentHashMap<>();
+    @Resource
+    private TableGroupContext tableGroupContext;
+
+    private final Map<String, Task> map = new ConcurrentHashMap<>();
 
     @Override
     public void start(Mapping mapping) {
         Thread worker = new Thread(() -> {
             final String metaId = mapping.getMetaId();
-            ExecutorService executor = null;
+            ExecutorService executor = Executors.newFixedThreadPool(mapping.getThreadNum());
             try {
                 List<TableGroup> list = profileComponent.getSortedTableGroupAll(mapping.getId());
                 Assert.notEmpty(list, "映射关系不能为空");
                 logger.info("开始全量同步：{}, {}", metaId, mapping.getName());
-                map.putIfAbsent(metaId, new Task(metaId));
-                executor = Executors.newFixedThreadPool(mapping.getThreadNum());
-                Task task = map.get(metaId);
-                doTask(task, mapping, list, executor);
+                map.computeIfAbsent(metaId, k -> {
+                    tableGroupContext.put(mapping, list);
+                    Task task = new Task(metaId);
+                    doTask(task, mapping, list, executor);
+                    return task;
+                });
             } catch (Exception e) {
                 logger.error(e.getMessage(), e);
                 logService.log(LogType.SystemLog.ERROR, e.getMessage());
             } finally {
                 try {
-                    if (executor != null) {
-                        executor.shutdown();
-                    }
+                    executor.shutdown();
                 } catch (Exception e) {
                     logService.log(LogType.SystemLog.ERROR, e.getMessage());
                 }
@@ -84,7 +84,7 @@ public final class FullPuller extends AbstractPuller implements ApplicationListe
                 logger.info("结束全量同步：{}, {}", metaId, mapping.getName());
             }
         });
-        worker.setName(new StringBuilder("full-worker-").append(mapping.getId()).toString());
+        worker.setName("full-worker-" + mapping.getId());
         worker.setDaemon(false);
         worker.start();
     }

dbsyncer-manager/src/main/java/org/dbsyncer/manager/impl/IncrementPuller.java

@@ -3,13 +3,15 @@
  */
 package org.dbsyncer.manager.impl;
 
-import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.scheduled.ScheduledTaskJob;
+import org.dbsyncer.common.scheduled.ScheduledTaskService;
 import org.dbsyncer.connector.base.ConnectorFactory;
 import org.dbsyncer.manager.AbstractPuller;
 import org.dbsyncer.manager.ManagerException;
 import org.dbsyncer.parser.LogService;
 import org.dbsyncer.parser.LogType;
 import org.dbsyncer.parser.ProfileComponent;
+import org.dbsyncer.parser.TableGroupContext;
 import org.dbsyncer.parser.consumer.ParserConsumer;
 import org.dbsyncer.parser.event.RefreshOffsetEvent;
 import org.dbsyncer.parser.flush.impl.BufferActuatorRouter;
@@ -21,8 +23,6 @@ import org.dbsyncer.sdk.listener.AbstractListener;
 import org.dbsyncer.sdk.listener.AbstractQuartzListener;
 import org.dbsyncer.sdk.listener.Listener;
 import org.dbsyncer.sdk.model.*;
-import org.dbsyncer.common.scheduled.ScheduledTaskJob;
-import org.dbsyncer.common.scheduled.ScheduledTaskService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationListener;
@@ -32,11 +32,7 @@ import org.springframework.util.Assert;
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
@@ -67,7 +63,10 @@ public final class IncrementPuller extends AbstractPuller implements Application
     @Resource
     private LogService logService;
 
-    private Map<String, Listener> map = new ConcurrentHashMap<>();
+    @Resource
+    private TableGroupContext tableGroupContext;
+
+    private final Map<String, Listener> map = new ConcurrentHashMap<>();
 
     @PostConstruct
     private void init() {
@@ -92,8 +91,12 @@ public final class IncrementPuller extends AbstractPuller implements Application
                 meta.setBeginTime(now);
                 meta.setEndTime(now);
                 profileComponent.editConfigModel(meta);
-                map.putIfAbsent(metaId, getListener(mapping, connector, list, meta));
-                map.get(metaId).start();
+                map.computeIfAbsent(metaId, k-> {
+                    tableGroupContext.put(mapping, list);
+                    Listener listener = getListener(mapping, connector, list, meta);
+                    listener.start();
+                    return listener;
+                });
             } catch (Exception e) {
                 close(metaId);
                 logService.log(LogType.TableGroupLog.INCREMENT_FAILED, e.getMessage());

dbsyncer-parser/src/main/java/org/dbsyncer/parser/TableGroupContext.java

@@ -0,0 +1,23 @@
+/**
+ * DBSyncer Copyright 2020-2025 All Rights Reserved.
+ */
+package org.dbsyncer.parser;
+
+import org.dbsyncer.parser.model.Mapping;
+import org.dbsyncer.parser.model.TableGroup;
+
+import java.util.List;
+
+/**
+ * @Author 穿云
+ * @Version 1.0.0
+ * @Date 2025-01-16 23:48
+ */
+public interface TableGroupContext {
+
+    void put(Mapping mapping, List<TableGroup> tableGroups);
+
+    List<TableGroup> getTableGroups(Mapping mapping, String tableName);
+
+    void clear(String metaId);
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/consumer/ParserConsumer.java

@@ -33,7 +33,7 @@ public final class ParserConsumer implements Watcher {
         this.logService = logService;
         this.metaId = metaId;
         // 注册到路由服务中
-        tableGroups.forEach(t -> bufferActuatorRouter.bind(metaId, t.getSourceTable().getName()));
+        bufferActuatorRouter.bind(metaId, tableGroups);
     }
 
     @Override

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/BufferActuatorRouter.java

@@ -4,9 +4,10 @@
 package org.dbsyncer.parser.flush.impl;
 
 import org.dbsyncer.common.config.TableGroupBufferConfig;
-import org.dbsyncer.sdk.listener.ChangedEvent;
 import org.dbsyncer.parser.flush.BufferActuator;
+import org.dbsyncer.parser.model.TableGroup;
 import org.dbsyncer.parser.model.WriterRequest;
+import org.dbsyncer.sdk.listener.ChangedEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
@@ -14,6 +15,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
@@ -45,39 +47,49 @@ public final class BufferActuatorRouter implements DisposableBean {
     private final Map<String, Map<String, TableGroupBufferActuator>> router = new ConcurrentHashMap<>();
 
     public void execute(String metaId, ChangedEvent event) {
-        if (router.containsKey(metaId) && router.get(metaId).containsKey(event.getSourceTableName())) {
-            router.get(metaId).get(event.getSourceTableName()).offer(new WriterRequest(event));
-            return;
+        if (router.containsKey(metaId)) {
+            router.computeIfPresent(metaId, (k, processor) -> {
+                processor.computeIfPresent(event.getSourceTableName(), (x, actuator) -> {
+                    actuator.offer(new WriterRequest(event));
+                    return actuator;
+                });
+                return processor;
+            });
         }
         generalBufferActuator.offer(new WriterRequest(event));
     }
 
-    public void bind(String metaId, String tableName) {
-        router.computeIfAbsent(metaId, k -> new ConcurrentHashMap<>());
-
-        // TODO 暂定执行器上限，待替换为LRU模型
-        if (router.get(metaId).size() >= tableGroupBufferConfig.getMaxBufferActuatorSize()) {
-            return;
-        }
-
-        router.get(metaId).computeIfAbsent(tableName, k -> {
-            TableGroupBufferActuator newBufferActuator = null;
-            try {
-                newBufferActuator = (TableGroupBufferActuator) tableGroupBufferActuator.clone();
-                newBufferActuator.setTableName(tableName);
-                newBufferActuator.buildConfig();
-            } catch (CloneNotSupportedException ex) {
-                logger.error(ex.getMessage(), ex);
+    public void bind(String metaId, List<TableGroup> tableGroups) {
+        router.computeIfAbsent(metaId, k -> {
+            Map<String, TableGroupBufferActuator> processor = new ConcurrentHashMap<>();
+            for (TableGroup tableGroup : tableGroups) {
+                // 超过执行器上限
+                if (processor.size() >= tableGroupBufferConfig.getMaxBufferActuatorSize()) {
+                    break;
+                }
+                final String tableName = tableGroup.getSourceTable().getName();
+                processor.computeIfAbsent(tableName, name -> {
+                    TableGroupBufferActuator newBufferActuator = null;
+                    try {
+                        newBufferActuator = (TableGroupBufferActuator) tableGroupBufferActuator.clone();
+                        newBufferActuator.setTableName(name);
+                        newBufferActuator.buildConfig();
+                    } catch (CloneNotSupportedException ex) {
+                        logger.error(ex.getMessage(), ex);
+                    }
+                    return newBufferActuator;
+                });
             }
-            return newBufferActuator;
+            return processor;
         });
     }
 
     public void unbind(String metaId) {
-        if (router.containsKey(metaId)) {
-            router.get(metaId).values().forEach(TableGroupBufferActuator::stop);
-            router.remove(metaId);
-        }
+        router.computeIfPresent(metaId, (k, processor) -> {
+            processor.values().forEach(TableGroupBufferActuator::stop);
+            return processor;
+        });
+        router.remove(metaId);
     }
 
     @Override

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/GeneralBufferActuator.java

@@ -12,13 +12,13 @@ import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.base.ConnectorFactory;
 import org.dbsyncer.parser.ParserComponent;
 import org.dbsyncer.parser.ProfileComponent;
+import org.dbsyncer.parser.TableGroupContext;
 import org.dbsyncer.parser.ddl.DDLParser;
 import org.dbsyncer.parser.event.RefreshOffsetEvent;
 import org.dbsyncer.parser.flush.AbstractBufferActuator;
 import org.dbsyncer.parser.model.*;
 import org.dbsyncer.parser.strategy.FlushStrategy;
 import org.dbsyncer.parser.util.ConvertUtil;
-import org.dbsyncer.parser.util.PickerUtil;
 import org.dbsyncer.plugin.PluginFactory;
 import org.dbsyncer.plugin.enums.ProcessEnum;
 import org.dbsyncer.plugin.impl.IncrementPluginContext;
@@ -80,6 +80,9 @@ public class GeneralBufferActuator extends AbstractBufferActuator<WriterRequest,
     @Resource
     private DDLParser ddlParser;
 
+    @Resource
+    private TableGroupContext tableGroupContext;
+
     @PostConstruct
     public void init() {
         setConfig(generalBufferConfig);
@@ -117,89 +120,81 @@ public class GeneralBufferActuator extends AbstractBufferActuator<WriterRequest,
 
     @Override
     public void pull(WriterResponse response) {
-        // TODO add cache
-        List<TableGroup> groupAll = profileComponent.getTableGroupAll(response.getChangedOffset().getMetaId());
-        if (!CollectionUtils.isEmpty(groupAll)) {
-            groupAll.forEach(tableGroup -> {
-                if (StringUtil.equals(tableGroup.getSourceTable().getName(), response.getTableName())) {
-                    distributeTableGroup(response, tableGroup);
-                }
-            });
+        Meta meta = profileComponent.getMeta(response.getChangedOffset().getMetaId());
+        if (meta == null) {
+            return;
         }
-    }
-
-    private void distributeTableGroup(WriterResponse response, TableGroup tableGroup) {
-        // 0、获取配置信息
-        final Mapping mapping = profileComponent.getMapping(tableGroup.getMappingId());
-        final TableGroup group = PickerUtil.mergeTableGroupConfig(mapping, tableGroup);
+        final Mapping mapping = profileComponent.getMapping(meta.getMappingId());
+        List<TableGroup> tableGroups = tableGroupContext.getTableGroups(mapping, response.getTableName());
 
         // 1、ddl解析
         if (ChangedEventTypeEnum.isDDL(response.getTypeEnum())) {
-            parseDDl(response, mapping, group);
+            tableGroups.forEach(tableGroup -> parseDDl(response, mapping, tableGroup));
             return;
         }
 
-        // 2、映射字段
-        final Picker picker = new Picker(group.getFieldMapping());
+        // 2、dml解析
+        tableGroups.forEach(tableGroup -> distributeTableGroup(response, mapping, tableGroup));
+    }
+
+    @Override
+    protected void offerFailed(BlockingQueue<WriterRequest> queue, WriterRequest request) {
+        throw new QueueOverflowException("缓存队列已满");
+    }
+
+    @Override
+    protected void meter(TimeRegistry timeRegistry, long count) {
+        // 统计执行器同步效率TPS
+        timeRegistry.meter(TimeRegistry.GENERAL_BUFFER_ACTUATOR_TPS).add(count);
+    }
+
+    @Override
+    public Executor getExecutor() {
+        return generalExecutor;
+    }
+
+    private void distributeTableGroup(WriterResponse response, Mapping mapping, TableGroup tableGroup) {
+        // 1、映射字段
+        final Picker picker = new Picker(tableGroup.getFieldMapping());
         List<Map> sourceDataList = new ArrayList<>();
         List<Map> targetDataList = picker.pickTargetData(response.getDataList(), sourceDataList);
 
-        // 3、参数转换
-        ConvertUtil.convert(group.getConvert(), targetDataList);
+        // 2、参数转换
+        ConvertUtil.convert(tableGroup.getConvert(), targetDataList);
 
-        // 4、插件转换
+        // 3、插件转换
         final IncrementPluginContext context = new IncrementPluginContext();
         context.setSourceConnectorInstance(connectorFactory.connect(getConnectorConfig(mapping.getSourceConnectorId())));
         context.setTargetConnectorInstance(connectorFactory.connect(getConnectorConfig(mapping.getTargetConnectorId())));
-        context.setSourceTableName(group.getSourceTable().getName());
-        context.setTargetTableName(group.getTargetTable().getName());
+        context.setSourceTableName(tableGroup.getSourceTable().getName());
+        context.setTargetTableName(tableGroup.getTargetTable().getName());
         context.setEvent(response.getEvent());
         context.setTargetFields(picker.getTargetFields());
-        context.setCommand(group.getCommand());
+        context.setCommand(tableGroup.getCommand());
         context.setBatchSize(generalBufferConfig.getBufferWriterCount());
         context.setSourceList(sourceDataList);
         context.setTargetList(targetDataList);
-        context.setPluginExtInfo(group.getPluginExtInfo());
+        context.setPluginExtInfo(tableGroup.getPluginExtInfo());
         context.setForceUpdate(mapping.isForceUpdate());
-        pluginFactory.process(group.getPlugin(), context, ProcessEnum.CONVERT);
+        pluginFactory.process(tableGroup.getPlugin(), context, ProcessEnum.CONVERT);
 
-        // 5、批量执行同步
+        // 4、批量执行同步
         Result result = parserComponent.writeBatch(context, getExecutor());
 
-        // 6.发布刷新增量点事件
+        // 5.发布刷新增量点事件
         applicationContext.publishEvent(new RefreshOffsetEvent(applicationContext, response.getChangedOffset()));
 
-        // 7、持久化同步结果
+        // 6、持久化同步结果
         result.setTableGroupId(tableGroup.getId());
         result.setTargetTableGroupName(context.getTargetTableName());
         flushStrategy.flushIncrementData(mapping.getMetaId(), result, response.getEvent());
 
-        // 8、执行批量处理后的
-        pluginFactory.process(group.getPlugin(), context, ProcessEnum.AFTER);
-    }
-
-    @Override
-    protected void offerFailed(BlockingQueue<WriterRequest> queue, WriterRequest request) {
-        throw new QueueOverflowException("缓存队列已满");
-    }
-
-    @Override
-    protected void meter(TimeRegistry timeRegistry, long count) {
-        // 统计执行器同步效率TPS
-        timeRegistry.meter(TimeRegistry.GENERAL_BUFFER_ACTUATOR_TPS).add(count);
-    }
-
-    @Override
-    public Executor getExecutor() {
-        return generalExecutor;
+        // 7、执行批量处理后的
+        pluginFactory.process(tableGroup.getPlugin(), context, ProcessEnum.AFTER);
     }
 
     /**
      * 解析DDL
-     *
-     * @param response
-     * @param mapping
-     * @param tableGroup
      */
     private void parseDDl(WriterResponse response, Mapping mapping, TableGroup tableGroup) {
         try {
@@ -248,9 +243,6 @@ public class GeneralBufferActuator extends AbstractBufferActuator<WriterRequest,
 
     /**
      * 获取连接器配置
-     *
-     * @param connectorId
-     * @return
      */
     private ConnectorConfig getConnectorConfig(String connectorId) {
         Assert.hasText(connectorId, "Connector id can not be empty.");

dbsyncer-parser/src/main/java/org/dbsyncer/parser/impl/ProfileComponentImpl.java

@@ -149,7 +149,7 @@ public class ProfileComponentImpl implements ProfileComponent {
     @Override
     public List<TableGroup> getTableGroupAll(String mappingId) {
         TableGroup tableGroup = new TableGroup().setMappingId(mappingId);
-        return operationTemplate.queryAll(new QueryConfig(tableGroup, GroupStrategyEnum.TABLE));
+        return operationTemplate.queryAll(new QueryConfig<>(tableGroup, GroupStrategyEnum.TABLE));
     }
 
     @Override

dbsyncer-parser/src/main/java/org/dbsyncer/parser/impl/TableGroupContextImpl.java

@@ -0,0 +1,69 @@
+/**
+ * DBSyncer Copyright 2020-2025 All Rights Reserved.
+ */
+package org.dbsyncer.parser.impl;
+
+import org.dbsyncer.parser.TableGroupContext;
+import org.dbsyncer.parser.model.Mapping;
+import org.dbsyncer.parser.model.TableGroup;
+import org.dbsyncer.parser.util.PickerUtil;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @Author 穿云
+ * @Version 1.0.0
+ * @Date 2025-01-16 23:34
+ */
+@Component
+public final class TableGroupContextImpl implements TableGroupContext {
+
+    /**
+     * 驱动表映射关系
+     */
+    private final Map<String, InnerMapping> tableGroupMap = new ConcurrentHashMap<>();
+
+    private final List<TableGroup> EMPTY_LIST = new ArrayList<>();
+
+    @Override
+    public void put(Mapping mapping, List<TableGroup> tableGroups) {
+        tableGroupMap.computeIfAbsent(mapping.getMetaId(), k -> {
+            InnerMapping innerMap = new InnerMapping();
+            tableGroups.forEach(tableGroup -> {
+                String sourceTableName = tableGroup.getSourceTable().getName();
+                innerMap.add(sourceTableName, PickerUtil.mergeTableGroupConfig(mapping, tableGroup));
+            });
+            return innerMap;
+        });
+    }
+
+    @Override
+    public List<TableGroup> getTableGroups(Mapping mapping, String tableName) {
+        InnerMapping innerMapping = tableGroupMap.get(mapping.getMetaId());
+        if (innerMapping != null) {
+            return innerMapping.get(tableName);
+        }
+        return EMPTY_LIST;
+    }
+
+    @Override
+    public void clear(String metaId) {
+        tableGroupMap.remove(metaId);
+    }
+
+    static final class InnerMapping {
+        Map<String, List<TableGroup>> mapping = new ConcurrentHashMap<>();
+
+        public List<TableGroup> get(String tableName) {
+            return mapping.get(tableName);
+        }
+
+        public void add(String tableName, TableGroup tableGroup) {
+            mapping.computeIfAbsent(tableName, k -> new ArrayList<>()).add(tableGroup);
+        }
+    }
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/model/Picker.java

@@ -49,7 +49,7 @@ public class Picker {
                 source = new HashMap<>();
                 target = new HashMap<>();
                 row = rows.get(i);
-                for (int j = 0; j < row.size(); i++) {
+                for (int j = 0; j < row.size(); j++) {
                     source.put(sFields.get(j).getName(), row.get(j));
                 }
                 exchange(sFieldSize, tFieldSize, sourceFields, targetFields, source, target);

dbsyncer-parser/src/main/java/org/dbsyncer/parser/model/QueryConfig.java

@@ -7,7 +7,7 @@ import org.dbsyncer.parser.enums.GroupStrategyEnum;
 
 public class QueryConfig<T> {
 
-    private ConfigModel configModel;
+    private final ConfigModel configModel;
 
     private GroupStrategyEnum groupStrategyEnum = GroupStrategyEnum.DEFAULT;
 

dbsyncer-sdk/src/main/java/org/dbsyncer/sdk/enums/ChangedEventTypeEnum.java

@@ -17,7 +17,7 @@ public enum ChangedEventTypeEnum {
      */
     SCAN,
     /**
-     * 表行变更，比如mysql, 会获取变动行所有列的值
+     * 表行变更
      */
     ROW;