聊聊SpinalTap的Transaction

序

本文主要研究一下 SpinalTap 的 Transaction

Transaction

SpinalTap/spinaltap-model/src/main/java/com/airbnb/spinaltap/mysql/Transaction.java

@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;
  }
}

• Transaction 定义了 timestamp、offset、position、gtid 属性

MysqlMutationMetadata

SpinalTap/spinaltap-model/src/main/java/com/airbnb/spinaltap/mysql/mutation/MysqlMutationMetadata.java

@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;
  }
}

• MysqlMutationMetadata 定义了 dataSource、filePos、table、serverId、beginTransaction、lastTransaction、leaderEpoch、eventRowPosition 属性

MysqlMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/MysqlMutationMapper.java

@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();}
}

• MysqlMutationMapper 定义了 dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch 属性；它提供了 createMetadata 方法，它接收 table、event、eventPosition 参数返回新建的 MysqlMutationMetadata

InsertMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/InsertMutationMapper.java

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;
  }
}

• InsertMutationMapper 继承了 MysqlMutationMapper，其构造器要求输入 dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch 参数

UpdateMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/UpdateMutationMapper.java

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;
  }
}

• UpdateMutationMapper 继承了 MysqlMutationMapper，其构造器要求输入 dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch 参数

DeleteMutationMapper

SpinalTap/spinaltap-mysql/src/main/java/com/airbnb/spinaltap/mysql/event/mapper/DeleteMutationMapper.java

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;
  }
}

• DeleteMutationMapper 继承了 MysqlMutationMapper，其构造器要求输入 dataSource、tableCache、beginTransaction、lastTransaction、leaderEpoch 参数

小结

Transaction 定义了 timestamp、offset、position、gtid 属性；MysqlMutationMetadata 定义了 dataSource、filePos、table、serverId、beginTransaction、lastTransaction、leaderEpoch、eventRowPosition 属性

doc

• Transaction