聊聊SpinalTap的Transaction

@Value
@RequiredArgsConstructor
public class Transaction {
  private final long timestamp;
  private final long offset;
  private final BinlogFilePos position;
  private final String gtid;

  public Transaction(long timestamp, long offset, BinlogFilePos position) {
    this.timestamp = timestamp;
    this.offset = offset;
    this.position = position;
    this.gtid = null;
  }
}

@Value
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
public class MysqlMutationMetadata extends Mutation.Metadata {
  private final DataSource dataSource;
  private final BinlogFilePos filePos;
  private final Table table;
  private final long serverId;
  private final Transaction beginTransaction;
  private final Transaction lastTransaction;

  /** The leader epoch of the node resource processing the event. */
  private final long leaderEpoch;

  /** The mutation row position in the given binlog event. */
  private final int eventRowPosition;

  public MysqlMutationMetadata(
      DataSource dataSource,
      BinlogFilePos filePos,
      Table table,
      long serverId,
      long id,
      long timestamp,
      Transaction beginTransaction,
      Transaction lastTransaction,
      long leaderEpoch,
      int eventRowPosition) {
    super(id, timestamp);

    this.dataSource = dataSource;
    this.filePos = filePos;
    this.table = table;
    this.serverId = serverId;
    this.beginTransaction = beginTransaction;
    this.lastTransaction = lastTransaction;
    this.leaderEpoch = leaderEpoch;
    this.eventRowPosition = eventRowPosition;
  }
}

@Slf4j
@RequiredArgsConstructor
public abstract class MysqlMutationMapper<R extends BinlogEvent, T extends MysqlMutation>
    implements Mapper<R, List<T>> {
  @NonNull private final DataSource dataSource;
  @NonNull private final TableCache tableCache;
  @NonNull private final AtomicReference<Transaction> beginTransaction;
  @NonNull private final AtomicReference<Transaction> lastTransaction;
  @NonNull private final AtomicLong leaderEpoch;

  public static Mapper<BinlogEvent, List<? extends Mutation<?>>> create(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final SchemaTracker schemaTracker,
      @NonNull final AtomicLong leaderEpoch,
      @NonNull final AtomicReference<Transaction> beginTransaction,
      @NonNull final AtomicReference<Transaction> lastTransaction,
      @NonNull final MysqlSourceMetrics metrics) {
    final AtomicReference<String> gtid = new AtomicReference<>();
    return ClassBasedMapper.<BinlogEvent, List<? extends Mutation<?>>>builder()
        .addMapper(TableMapEvent.class, new TableMapMapper(tableCache))
        .addMapper(GTIDEvent.class, new GTIDMapper(gtid))
        .addMapper(QueryEvent.class, new QueryMapper(beginTransaction, gtid, schemaTracker))
        .addMapper(XidEvent.class, new XidMapper(lastTransaction, gtid, metrics))
        .addMapper(StartEvent.class, new StartMapper(dataSource, tableCache, metrics))
        .addMapper(
            UpdateEvent.class,
            new UpdateMutationMapper(
                dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch))
        .addMapper(
            WriteEvent.class,
            new InsertMutationMapper(
                dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch))
        .addMapper(
            DeleteEvent.class,
            new DeleteMutationMapper(
                dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch))
        .build();
  }

  protected abstract List<T> mapEvent(@NonNull final Table table, @NonNull final R event);

  public List<T> map(@NonNull final R event) {
    Table table = tableCache.get(event.getTableId());

    return mapEvent(table, event);
  }

  MysqlMutationMetadata createMetadata(
      @NonNull final Table table, @NonNull final BinlogEvent event, final int eventPosition) {
    return new MysqlMutationMetadata(
        dataSource,
        event.getBinlogFilePos(),
        table,
        event.getServerId(),
        event.getOffset(),
        event.getTimestamp(),
        beginTransaction.get(),
        lastTransaction.get(),
        leaderEpoch.get(),
        eventPosition);
  }

  static ImmutableMap<String, Column> zip(
      @NonNull final Serializable[] row, @NonNull final Collection<ColumnMetadata> columns) {
    if (row.length != columns.size()) {
      log.error("Row length {} and column length {} don't match", row.length, columns.size());
    }

    final ImmutableMap.Builder<String, Column> builder = ImmutableMap.builder();
    final Iterator<ColumnMetadata> columnIterator = columns.iterator();

    for (int position = 0; position < row.length && columnIterator.hasNext(); position++) {
      final ColumnMetadata col = columnIterator.next();
      builder.put(col.getName(), new Column(col, row[position]));
    }

    return builder.build();
  }
}

class InsertMutationMapper extends MysqlMutationMapper<WriteEvent, MysqlInsertMutation> {
  InsertMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference<Transaction> beginTransaction,
      @NonNull final AtomicReference<Transaction> lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

  @Override
  protected List<MysqlInsertMutation> mapEvent(
      @NonNull final Table table, @NonNull final WriteEvent event) {
    final List<Serializable[]> rows = event.getRows();
    final List<MysqlInsertMutation> mutations = new ArrayList<>();
    final Collection<ColumnMetadata> cols = table.getColumns().values();

    for (int position = 0; position < rows.size(); position++) {
      mutations.add(
          new MysqlInsertMutation(
              createMetadata(table, event, position),
              new Row(table, zip(rows.get(position), cols))));
    }

    return mutations;
  }
}

final class UpdateMutationMapper extends MysqlMutationMapper<UpdateEvent, MysqlMutation> {
  UpdateMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference<Transaction> beginTransaction,
      @NonNull final AtomicReference<Transaction> lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

  @Override
  protected List<MysqlMutation> mapEvent(
      @NonNull final Table table, @NonNull final UpdateEvent event) {
    final List<MysqlMutation> mutations = Lists.newArrayList();
    final Collection<ColumnMetadata> cols = table.getColumns().values();
    final List<Map.Entry<Serializable[], Serializable[]>> rows = event.getRows();

    for (int position = 0; position < rows.size(); position++) {
      MysqlMutationMetadata metadata = createMetadata(table, event, position);

      final Row previousRow = new Row(table, zip(rows.get(position).getKey(), cols));
      final Row newRow = new Row(table, zip(rows.get(position).getValue(), cols));

      // If PK value has changed, then delete before image and insert new image
      // to retain invariant that a mutation captures changes to a single PK
      if (table.getPrimaryKey().isPresent()
          && !previousRow.getPrimaryKeyValue().equals(newRow.getPrimaryKeyValue())) {
        mutations.add(new MysqlDeleteMutation(metadata, previousRow));
        mutations.add(new MysqlInsertMutation(metadata, newRow));
      } else {
        mutations.add(new MysqlUpdateMutation(metadata, previousRow, newRow));
      }
    }

    return mutations;
  }
}

final class DeleteMutationMapper extends MysqlMutationMapper<DeleteEvent, MysqlDeleteMutation> {
  DeleteMutationMapper(
      @NonNull final DataSource dataSource,
      @NonNull final TableCache tableCache,
      @NonNull final AtomicReference<Transaction> beginTransaction,
      @NonNull final AtomicReference<Transaction> lastTransaction,
      @NonNull final AtomicLong leaderEpoch) {
    super(dataSource, tableCache, beginTransaction, lastTransaction, leaderEpoch);
  }

  @Override
  protected List<MysqlDeleteMutation> mapEvent(
      @NonNull final Table table, @NonNull final DeleteEvent event) {
    final Collection<ColumnMetadata> cols = table.getColumns().values();
    final List<MysqlDeleteMutation> mutations = new ArrayList<>();
    final List<Serializable[]> rows = event.getRows();

    for (int position = 0; position < rows.size(); position++) {
      mutations.add(
          new MysqlDeleteMutation(
              createMetadata(table, event, position),
              new Row(table, zip(rows.get(position), cols))));
    }

    return mutations;
  }
}