docs: private void doDdl(DbLoadContext context, List<EventData> eventDatas) {
          for (final EventData data : eventDatas) {
              DataMedia dataMedia = ConfigHelper.findDataMedia(context.getPipeline(), data.getTableId());
              final DbDialect dbDialect = dbDialectFactory.getDbDialect(context.getIdentity().getPipelineId(),
                  (DbMediaSource) dataMedia.getSource());
              Boolean skipDdlException = context.getPipeline().getParameters().getSkipDdlException();
              try {
                  Boolean result = dbDialect.getJdbcTemplate().execute(new StatementCallback<Boolean>() {
  
                      public Boolean doInStatement(Statement stmt) throws SQLException, DataAccessException {
                          Boolean result = true;
                          if (dbDialect instanceof MysqlDialect && StringUtils.isNotEmpty(data.getDdlSchemaName())) {
                              // 如果mysql，执行ddl时，切换到在源库执行的schema上
                              // result &= stmt.execute("use " +
                              // data.getDdlSchemaName());
  
                              // 解决当数据库名称为关键字如"Order"的时候,会报错,无法同步
                              result &= stmt.execute("use `" + data.getDdlSchemaName() + "`");
                          }
                          result &= stmt.execute(data.getSql());
                          return result;
                      }
                  });
                  if (result) {
                      context.getProcessedDatas().add(data); // 记录为成功处理的sql
                  } else {
                      context.getFailedDatas().add(data);
                  }
  
              } catch (Throwable e) {
                  if (skipDdlException) {
                      // do skip
                      logger.warn("skip exception for ddl : {} , caused by {}", data, ExceptionUtils.getFullStackTrace(e));
                  } else {
                      throw new LoadException(e);
                  }
              }
  
          }
      }