docs: private EventData internParse(Pipeline pipeline, Entry entry, RowChange rowChange, RowData rowData) {
          EventData eventData = new EventData();
          eventData.setTableName(entry.getHeader().getTableName());
          eventData.setSchemaName(entry.getHeader().getSchemaName());
          eventData.setEventType(EventType.valueOf(rowChange.getEventType().name()));
          eventData.setExecuteTime(entry.getHeader().getExecuteTime());
          EventType eventType = eventData.getEventType();
          TableInfoHolder tableHolder = null;
  
          if (!StringUtils.equalsIgnoreCase(pipeline.getParameters().getSystemSchema(), eventData.getSchemaName())) {
              boolean useTableTransform = pipeline.getParameters().getUseTableTransform();
              Table table = null;
              DataMediaPair dataMediaPair = ConfigHelper.findDataMediaPairBySourceName(pipeline,
                  eventData.getSchemaName(),
                  eventData.getTableName());
              DataMedia dataMedia = dataMediaPair.getSource();
              eventData.setTableId(dataMedia.getId());
              // 获取目标表
              DataMedia targetDataMedia = dataMediaPair.getTarget();
              if (useTableTransform || dataMedia.getSource().getType().isOracle()) {// oracle需要反查一次meta
                  // 如果设置了需要进行table meta转化，则反查一下table信息
                  // 比如oracle erosa解析时可能使用了非物理主键，需要直接使用，信任erosa的信息
                  DbDialect dbDialect = dbDialectFactory.getDbDialect(pipeline.getId(),
                      (DbMediaSource) dataMedia.getSource());
                  table = dbDialect.findTable(eventData.getSchemaName(), eventData.getTableName());// 查询一下meta信息
                  if (table == null) {
                      logger.warn("find table[{}.{}] is null , may be drop table.",
                          eventData.getSchemaName(),
                          eventData.getTableName());
                  }
                  // 获取一下目标库的拆分字段,设置源表为主键
                  // 首先要求源和目标的库名表名是一致的
                  DataMediaSource targetSource = targetDataMedia.getSource();
                  if (targetSource instanceof DbMediaSource
                      && StringUtils.containsIgnoreCase(((DbMediaSource) targetSource).getUrl(), "drds")) {
                      // 优先判断是否为drds
                      DbDialect targetDbDialect = dbDialectFactory.getDbDialect(pipeline.getId(),
                          (DbMediaSource) targetDataMedia.getSource());
                      if (targetDbDialect.isDRDS()) {
                          String schemaName = buildName(eventData.getSchemaName(),
                              dataMedia.getNamespaceMode(),
                              targetDataMedia.getNamespaceMode());
                          String tableName = buildName(eventData.getSchemaName(),
                              dataMedia.getNameMode(),
                              targetDataMedia.getNameMode());
                          String shardColumns = targetDbDialect.getShardColumns(schemaName, tableName);
                          if (StringUtils.isNotEmpty(shardColumns)) {
                              String columns[] = StringUtils.split(shardColumns, ',');
                              for (String key : columns) {
                                  org.apache.ddlutils.model.Column col = table.findColumn(key, false);
                                  if (col != null) {
                                      col.setPrimaryKey(true);
                                  } else {
                                      logger.warn(String.format("shardColumn %s in table[%s.%s] is not found",
                                          key,
                                          eventData.getSchemaName(),
                                          eventData.getTableName()));
                                  }
                              }
                          }
                      }
                  }
                  tableHolder = new TableInfoHolder(dbDialect, table, useTableTransform);
              }
          }
  
          List<Column> beforeColumns = rowData.getBeforeColumnsList();
          List<Column> afterColumns = rowData.getAfterColumnsList();
          String tableName = eventData.getSchemaName() + "." + eventData.getTableName();
  
          // 判断一下是否需要all columns
          boolean isRowMode = pipeline.getParameters().getSyncMode().isRow(); // 如果是rowMode模式，所有字段都需要标记为updated
          boolean needAllColumns = isRowMode || checkNeedAllColumns(pipeline);
  
          // 变更后的主键
          Map<String, EventColumn> keyColumns = new LinkedHashMap<String, EventColumn>();
          // 变更前的主键
          Map<String, EventColumn> oldKeyColumns = new LinkedHashMap<String, EventColumn>();
          // 有变化的非主键
          Map<String, EventColumn> notKeyColumns = new LinkedHashMap<String, EventColumn>();
  
          if (eventType.isInsert()) {
              for (Column column : afterColumns) {
                  if (isKey(tableHolder, tableName, column)) {
                      keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                  } else {
                      // mysql 有效
                      notKeyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                  }
              }
          } else if (eventType.isDelete()) {
              for (Column column : beforeColumns) {
                  if (isKey(tableHolder, tableName, column)) {
                      keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                  } else {
                      // mysql 有效
                      notKeyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                  }
              }
          } else if (eventType.isUpdate()) {
              // 获取变更前的主键.
              for (Column column : beforeColumns) {
                  if (isKey(tableHolder, tableName, column)) {
                      oldKeyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                      // 同时记录一下new
                      // key,因为mysql5.6之后出现了minimal模式,after里会没有主键信息,需要在before记录中找
                      keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                  } else {
                      if (needAllColumns && entry.getHeader().getSourceType() == CanalEntry.Type.ORACLE) {
                          // 针对行记录同步时，针对oracle记录一下非主键的字段，因为update时针对未变更的字段在aftercolume里没有
                          notKeyColumns.put(column.getName(), copyEventColumn(column, isRowMode, tableHolder));
                      }
                  }
              }
              for (Column column : afterColumns) {
                  if (isKey(tableHolder, tableName, column)) {
                      // 获取变更后的主键
                      keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                  } else if (needAllColumns || entry.getHeader().getSourceType() == CanalEntry.Type.ORACLE
                             || column.getUpdated()) {
                      // 在update操作时，oracle和mysql存放变更的非主键值的方式不同,oracle只有变更的字段;
                      // mysql会把变更前和变更后的字段都发出来，只需要取有变更的字段.
                      // 如果是oracle库，after里一定为对应的变更字段
  
                      boolean isUpdate = true;
                      if (entry.getHeader().getSourceType() == CanalEntry.Type.MYSQL) { // mysql的after里部分数据为未变更,oracle里after里为变更字段
                          isUpdate = column.getUpdated();
                      }
  
                      notKeyColumns.put(column.getName(), copyEventColumn(column, isRowMode || isUpdate, tableHolder));// 如果是rowMode，所有字段都为updated
                  }
              }
  
              if (entry.getHeader().getSourceType() == CanalEntry.Type.ORACLE) { // 针对oracle进行特殊处理
                  checkUpdateKeyColumns(oldKeyColumns, keyColumns);
              }
          }
  
          List<EventColumn> keys = new ArrayList<EventColumn>(keyColumns.values());
          List<EventColumn> oldKeys = new ArrayList<EventColumn>(oldKeyColumns.values());
          List<EventColumn> columns = new ArrayList<EventColumn>(notKeyColumns.values());
  
          Collections.sort(keys, new EventColumnIndexComparable());
          Collections.sort(oldKeys, new EventColumnIndexComparable());
          Collections.sort(columns, new EventColumnIndexComparable());
          if (!keyColumns.isEmpty()) {
              eventData.setKeys(keys);
              if (eventData.getEventType().isUpdate() && !oldKeys.equals(keys)) { // update类型，如果存在主键不同,则记录下old
                                                                                  // keys为变更前的主键
                  eventData.setOldKeys(oldKeys);
              }
              eventData.setColumns(columns);
              // } else if (CanalEntry.Type.MYSQL ==
              // entry.getHeader().getSourceType()) {
              // // 只支持mysql无主键同步
              // if (eventType.isUpdate()) {
              // List<EventColumn> oldColumns = new ArrayList<EventColumn>();
              // List<EventColumn> newColumns = new ArrayList<EventColumn>();
              // for (Column column : beforeColumns) {
              // oldColumns.add(copyEventColumn(column, true, tableHolder));
              // }
              //
              // for (Column column : afterColumns) {
              // newColumns.add(copyEventColumn(column, true, tableHolder));
              // }
              // Collections.sort(oldColumns, new EventColumnIndexComparable());
              // Collections.sort(newColumns, new EventColumnIndexComparable());
              // eventData.setOldKeys(oldColumns);// 做为老主键
              // eventData.setKeys(newColumns);// 做为新主键，需要保证新老主键字段数量一致
              // } else {
              // // 针对无主键，等同为所有都是主键进行处理
              // eventData.setKeys(columns);
              // }
          } else {
              throw new SelectException("this rowdata has no pks , entry: " + entry.toString() + " and rowData: "
                                        + rowData);
          }
  
          return eventData;
      }