我们知道，Iceberg基于Snapshot（快照机制）实现了乐观无锁地数据并发读写能力（MVCC，Multi Versions Concurrency Controll），默认提供了快照级别的事务隔离，因此可以至少避免脏读的问题。
SQL-92 标准定义了如下 4 种隔离级别：
幻读，是对于并发事务的INSERT/DELETE操作，会导致连续读取相同区间的数据记录时，数据条数不同。
不可重复读，是对于并发事务的UPDATE操作，会导致连续读取相同数据记录的数据发生了变化。
基于快照实现方案是解决事务并的方式之一，可以达到可重复读、串行，但会导致另外一个数据更新丢失问题，即写偏斜问题：
事务A开始，读取字段X，更新字段Y，事务结束。
事务B开始，读取字段Y，更新字段X，事务结束。
上面两个并发事务A/B，会基于同一个快照读取分别读取了X、Y的值，同时分别更新了Y、X，完成自己的事务，由于这两个事务在写的时候不会产生数据更新的重叠问题，因此都可以成功，产生的结果就是X、Y的值被互换了。
但如果事务是串行的话，不论是A事务先执行还是B事务，最终的结果则是X的值=Y的值。
实际上，基于快照的读又分为两种情况，产生的结果也不相同：
基于旧快照： 由于旧的快照数据不可能再发生变化，因此可以避免一切并发导致的读问题。
基于当前快照： 由于最新快照会被写事务更新，因此不可避免地会导致幻读、不可重复读的问题。
下面的测试演示了一个事务上的多次更新，每一次从对象创建的快照更新对象的commit，都会产生一个新的事务ID，但这些新产生的Snapshot都归属于创建事务时绑定的起始TableMetadata对象。
一个表会唯一绑定一个最新的TableMetadata实例，而此TableMetadata则包含子所有的历史Snapshots。
对于非事务的快照更新操作，一次commit就会产生一个新的Snapshot，同时也会产生一个新的TableMetadata。
而对于事务的更新，由于一个事务期间可能会基于起始TableMetadata的当前快照，执行一次或多次commit，因此这个TableMetadata包含所有的新生产的Snapshots。
public void testMultipleOperationTransaction() {
Assert.assertEquals("Table should be on version 0", 0, (int) version());
// 先向表中追加一个文件，FILE_C
table.newAppend().appendFile(FILE_C).commit();
List<HistoryEntry> initialHistory = table.history();
// 获取当前的Metadata
TableMetadata base = readMetadata();
// 创建一个事务实例
Transaction txn = table.newTransaction();
// 验证当前表的Metadata是的版本号为1
Assert.assertSame(
"Base metadata should not change when commit is created", base, readMetadata());
Assert.assertEquals("Table should be on version 1 after txn create", 1, (int) version());
// 从事务对象，创建一个AppendFiles的实例，即MergeAppend的对象
// 并追加两个新文件FILE_A、FILE_B
txn.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
// 前面的commit()方法，并不会更新表的TaleMetadata
Assert.assertSame(
"Base metadata should not change when commit is created", base, readMetadata());
Assert.assertEquals("Table should be on version 1 after txn create", 1, (int) version());
// 获取表的追加文件后的最新快照信息
Snapshot appendSnapshot = txn.table().currentSnapshot();
// 通过事务对象创建一个文件删除器，即StreamingDelete的实例
// 并删除FILE_A
txn.newDelete().deleteFile(FILE_A).commit();
// 获取表的删除文件后的最新快照信息
Snapshot deleteSnapshot = txn.table().currentSnapshot();
// 但表对应的最新TableMetadata实例依然没有变化
Assert.assertSame(
"Base metadata should not change when an append is committed", base, readMetadata());
Assert.assertEquals("Table should be on version 1 after append", 1, (int) version());
// 提交事务
txn.commitTransaction();
// TableMetadata发生了变化，版本号更新到了2
Assert.assertEquals("Table should be on version 2 after commit", 2, (int) version());
// 由于前面通过非事务操作提交一了一次数据文件，又通过事务提交了一次，因此会产生2个Manifest Files
Assert.assertEquals(
"Table should have two manifest after commit",
2, readMetadata().currentSnapshot().allManifests(table.io()).size());
// 由于最后一次的提交操作是delete file，因此产生的最后一个Snapshot就是deleteSnapshot
Assert.assertEquals(
"Table snapshot should be the delete snapshot",
deleteSnapshot,
readMetadata().currentSnapshot());
validateManifestEntries(
readMetadata().currentSnapshot().allManifests(table.io()).get(0),
ids(deleteSnapshot.snapshotId(), appendSnapshot.snapshotId()),
files(FILE_A, FILE_B),
statuses(Status.DELETED, Status.EXISTING));
// 表最新的TableMetadata应该包含3个历史Snapshots，一个是通过非事务提交产生的；
// 另外两个是在事务期间提交产生的。
Assert.assertEquals(
"Table should have a snapshot for each operation", 3, readMetadata().snapshots().size());
validateManifestEntries(
readMetadata().snapshots().get(1).allManifests(table.io()).get(0),
ids(appendSnapshot.snapshotId(), appendSnapshot.snapshotId()),
files(FILE_A, FILE_B),
statuses(Status.ADDED, Status.ADDED));
// 验证所有的历史提交
Assertions.assertThat(table.history()).containsAll(initialHistory);
}
支持创建、替换以及通用的事务创建。
public final class Transactions {
private Transactions() {}
public static Transaction createOrReplaceTableTransaction(
String tableName, TableOperations ops, TableMetadata start) {
return new BaseTransaction(tableName, ops, TransactionType.CREATE_OR_REPLACE_TABLE, start);
}
public static Transaction replaceTableTransaction(
String tableName, TableOperations ops, TableMetadata start) {
return new BaseTransaction(tableName, ops, TransactionType.REPLACE_TABLE, start);
}
public static Transaction replaceTableTransaction(
String tableName, TableOperations ops, TableMetadata start, MetricsReporter reporter) {
return new BaseTransaction(tableName, ops, TransactionType.REPLACE_TABLE, start, reporter);
}
public static Transaction createTableTransaction(
String tableName, TableOperations ops, TableMetadata start) {
Preconditions.checkArgument(
ops.current() == null, "Cannot start create table transaction: table already exists");
return new BaseTransaction(tableName, ops, TransactionType.CREATE_TABLE, start);
}
public static Transaction createTableTransaction(
String tableName, TableOperations ops, TableMetadata start, MetricsReporter reporter) {
Preconditions.checkArgument(
ops.current() == null, "Cannot start create table transaction: table already exists");
return new BaseTransaction(tableName, ops, TransactionType.CREATE_TABLE, start, reporter);
}
public static Transaction newTransaction(String tableName, TableOperations ops) {
return new BaseTransaction(tableName, ops, TransactionType.SIMPLE, ops.refresh());
}
public static Transaction newTransaction(
String tableName, TableOperations ops, MetricsReporter reporter) {
return new BaseTransaction(tableName, ops, TransactionType.SIMPLE, ops.refresh(), reporter);
}
}
public class BaseTransaction implements Transaction {
private final String tableName;
// 保存实际的表操作集合对象
private final TableOperations ops;
private final TransactionTable transactionTable;
private final TableOperations transactionOps;
// 保存所有的从事务创建出来的SnapshotUpdate的实现类的对象
private final List<PendingUpdate> updates;
// 保存事物期间删除的所有文件路径
private final Set<String> deletedFiles =
Sets.newHashSet(); // keep track of files deleted in the most recent commit
private final Consumer<String> enqueueDelete = deletedFiles::add;
private TransactionType type;
private TableMetadata base;
private TableMetadata current;
private boolean hasLastOpCommitted;
private final MetricsReporter reporter;
BaseTransaction(
String tableName,
TableOperations ops,
TransactionType type,
TableMetadata start,
MetricsReporter reporter) {
this.tableName = tableName;
this.ops = ops;
// 创建一个事务类开的表对象，提供表相关的事务方法
this.transactionTable = new TransactionTable();
// 保存事务开始时的TableMetadata
this.current = start;
// 事务表上的操作集，可以支持
this.transactionOps = new TransactionTableOperations();
this.updates = Lists.newArrayList();
// 保存最新的TableMetadata，有可能与start不同，但版本号大于等待start
this.base = ops.current();
this.type = type;
this.hasLastOpCommitted = true;
this.reporter = reporter;
}
/**
* 在BaseTransaction实例构造时，会实例化一个此类的实例，它是对原始TableOperations的封装，例如HiveTableOperations，
* 以支持从此事务创建的具体的操作的提交行为。
*/
public class TransactionTableOperations implements TableOperations {
private TableOperations tempOps = ops.temp(current);
@Override
public TableMetadata current() {
return current;
}
@Override
public TableMetadata refresh() {
return current;
}
@Override
@SuppressWarnings("ConsistentOverrides")
public void commit(TableMetadata underlyingBase, TableMetadata metadata) {
// underlyingBase指向的是最新的TableMetadata，
// metata指向的是事务创建时绑定的起始TableMetadata
if (underlyingBase != current) {
// 如果最新的TableMetadata发生了改变，那么就需要触发更新当前事务所绑定的TableMetadata到最新版本，以保证不会将事务期间产生的变更更新到旧的Metadata上
// trigger a refresh and retry
throw new CommitFailedException("Table metadata refresh is required");
}
// 当事务引出的SnapshotUpdate对象commit时，所有变更提交到的TableMeatadata实例与事务绑定的最新的实例相同，
// 则更新当前事务的各种引用，并不会真正提交变更到表
BaseTransaction.this.current = metadata;
this.tempOps = ops.temp(metadata);
BaseTransaction.this.hasLastOpCommitted = true;
}
@Override
public FileIO io() {
return tempOps.io();
}
@Override
public EncryptionManager encryption() {
return tempOps.encryption();
}
@Override
public String metadataFileLocation(String fileName) {
return tempOps.metadataFileLocation(fileName);
}
@Override
public LocationProvider locationProvider() {
return tempOps.locationProvider();
}
@Override
public long newSnapshotId() {
return tempOps.newSnapshotId();
}
}
}
对应于前文，事务测试章节，提到的如下代码行
txn.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
public class BaseTransaction implements Transaction {
@Override
public AppendFiles newAppend() {
checkLastOperationCommitted("AppendFiles");
// 创建一个新的AppendFiles实例，注意到这里为此实例绑定的TableOperations是带事务的，
//
AppendFiles append = new MergeAppend(tableName, transactionOps).reportWith(reporter);
// 为append添加一个回调函数，当删除文件成功后，就将文件路径记录到当前事务维护的队列中
append.deleteWith(enqueueDelete);
// 添加 从事务新创建的SnapshotUpdate类型的对象 到缓存队列
updates.add(append);
return append;
}
}
所有继承自SnapshotProducer的实例类，可以实现自己的验证逻辑，其接口定义如下：
abstract class SnapshotProducer<ThisT> implements SnapshotUpdate<ThisT> {
/**
* Validate the current metadata.
*
* <p>Child operations can override this to add custom validation.
*
* @param currentMetadata current table metadata to validate
* @param snapshot ending snapshot on the lineage which is being validated
*/
protected void validate(TableMetadata currentMetadata, Snapshot snapshot) {}
}
例如对于BaseRewriteFiles的更新，它的实现如下：
@Override
protected void validate(TableMetadata base, Snapshot parent) {
if (replacedDataFiles.size() > 0) {
// if there are replaced data files, there cannot be any new row-level deletes for those data
// files
validateNoNewDeletesForDataFiles(base, startingSnapshotId, replacedDataFiles, parent);
}
}
MergeAppend类没有覆盖父类的commit()方法，故调用的实际上是SnapshotProducer::commit()方法，它与非事务模式下的Snapshot生成及更新过各相同，只是在将新的Snapshot应用到TableMetadata时，会将apply(TableMetadata base, Snapshot snapshot)方法路由到父类MergingSnapshotProducer，具体的代码见下：
class MergeAppend extends MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
/**
* 将新生产的Snapshot应用到目标TableMetadata。
*/
@Override
public List<ManifestFile> apply(TableMetadata base, Snapshot snapshot) {
// filter any existing manifests
// 过滤出可以从当前Snapshot可以访问的Manifest Files
List<ManifestFile> filtered =
filterManager.filterManifests(
SnapshotUtil.schemaFor(base, targetBranch()),
snapshot != null ? snapshot.dataManifests(ops.io()) : null);
// 获取最小的数据文件的序列号，由于TableMetadata记录的lastSequenceNumber并不一定与ManifestFile记录的所有合法文件的最小序列号相同，
// 故在清理垃圾文件时需要找到最小的序列号
long minDataSequenceNumber =
filtered.stream()
.map(ManifestFile::minSequenceNumber)
.filter(
seq ->
seq
!= ManifestWriter
.UNASSIGNED_SEQ) // filter out unassigned in rewritten manifests
.reduce(base.lastSequenceNumber(), Math::min);
// 删除所有序列号小于minDataSequenceNumber的Delete files，这些文件后续也不会被用到
deleteFilterManager.dropDeleteFilesOlderThan(minDataSequenceNumber);
List<ManifestFile> filteredDeletes =
deleteFilterManager.filterManifests(
SnapshotUtil.schemaFor(base, targetBranch()),
snapshot != null ? snapshot.deleteManifests(ops.io()) : null);
// only keep manifests that have live data files or that were written by this commit
Predicate<ManifestFile> shouldKeep =
manifest ->
manifest.hasAddedFiles()
|| manifest.hasExistingFiles()
|| manifest.snapshotId() == snapshotId();
Iterable<ManifestFile> unmergedManifests =
Iterables.filter(Iterables.concat(prepareNewManifests(), filtered), shouldKeep);
Iterable<ManifestFile> unmergedDeleteManifests =
Iterables.filter(Iterables.concat(prepareDeleteManifests(), filteredDeletes), shouldKeep);
// update the snapshot summary
summaryBuilder.clear();
summaryBuilder.merge(addedFilesSummary);
summaryBuilder.merge(appendedManifestsSummary);
summaryBuilder.merge(filterManager.buildSummary(filtered));
summaryBuilder.merge(deleteFilterManager.buildSummary(filteredDeletes));
// 将所有当前Snapshot产生的数据文件或删除文件，与历史的Manifest Files合并到一起，这些文件就是当前表可以访问的活数据。
List<ManifestFile> manifests = Lists.newArrayList();
Iterables.addAll(manifests, mergeManager.mergeManifests(unmergedManifests));
Iterables.addAll(manifests, deleteMergeManager.mergeManifests(unmergedDeleteManifests));
return manifests;
}
}
经过上一小节对事务绑定的TableMetadata的更新之后，就可以真正地执行提交动作，尝试通过TableOperations对象将最新的TableMetadata实例更新到目标表。
但由于我们现在是对整个事务流程中的某一个操作进行提交，因此不应该直接将这些变更到底层的表，即不应该调用底层的HiveTableOperations对象的commit()方法。还记得在前面提到在类BaseTransaction中定义了TransactionTableOperations吧，实际上这个在事务期间，都是通过这个Operations实例来完成TableMetadata的提交的，其具体的方法定义如下：
public void commit(TableMetadata underlyingBase, TableMetadata metadata) {
// underlyingBase指向的是最新的TableMetadata，
// metata指向的是事务创建时绑定的起始TableMetadata
if (underlyingBase != current) {
// 如果最新的TableMetadata发生了改变，那么就需要触发更新当前事务所绑定的TableMetadata到最新版本，以保证不会将事务期间产生的变更更新到旧的Metadata上
// trigger a refresh and retry
throw new CommitFailedException("Table metadata refresh is required");
}
// 当事务引出的SnapshotUpdate对象commit时，所有变更提交到的TableMeatadata实例与事务绑定的最新的实例相同，
// 则更新当前事务的各种引用，并不会真正提交变更到表
BaseTransaction.this.current = metadata;
this.tempOps = ops.temp(metadata);
BaseTransaction.this.hasLastOpCommitted = true;
}
通过上面的commit()方法可以知道，不同于HiveTableOperations中的实现那样，这里仅仅是更新了当前事务的成员变量，以便后续的更新操作能够继续应用到此TableMetadata实例上。
@Override
public void commitTransaction() {
Preconditions.checkState(
hasLastOpCommitted, "Cannot commit transaction: last operation has not committed");
switch (type) {
case CREATE_TABLE:
commitCreateTransaction();
break;
case REPLACE_TABLE:
commitReplaceTransaction(false);
break;
case CREATE_OR_REPLACE_TABLE:
commitReplaceTransaction(true);
break;
case SIMPLE:
commitSimpleTransaction();
break;
}
}
private void commitSimpleTransaction() {
// if there were no changes, don't try to commit
// 在讲解AppendFiles的提交流程时，有提到它的commit()方法，实际上是更新当前事务的current引用，保存了更新的TableMetadata实例
// 因此如果这里base == current，说明在之前没有调用于commit方法，因此不会执行事务的提交
if (base == current) {
return;
}
Set<Long> startingSnapshots =
base.snapshots().stream().map(Snapshot::snapshotId).collect(Collectors.toSet());
try {
Tasks.foreach(ops)
.retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT))
.exponentialBackoff(
base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
2.0 /* exponential */)
.onlyRetryOn(CommitFailedException.class)
.run(
underlyingOps -> {
// 调用BaseTransaction::newAppend()方法后生成一个MergeAppend实例，同时这个实例会被添加到缓存队列中，
// 这里就是遍历这个缓存队列，尝试再次对这个更新操作进行提交，即二次提交
// 至于为什么要再一次提交，见后面的解析
applyUpdates(underlyingOps);
// 当所有的更新都提交完成时，就可以真正地执行底层表上的更新操作，更新最新的TableMetadata
underlyingOps.commit(base, current);
});
} catch (CommitStateUnknownException e) {
throw e;
} catch (PendingUpdateFailedException e) {
cleanUpOnCommitFailure();
throw e.wrapped();
} catch (RuntimeException e) {
cleanUpOnCommitFailure();
throw e;
}
// the commit succeeded
try {
// clean up the data files that were deleted by each operation. first, get the list of
// committed manifests to ensure that no committed manifest is deleted.
// A manifest could be deleted in one successful operation commit, but reused in another
// successful commit of that operation if the whole transaction is retried.
Set<Long> newSnapshots = Sets.newHashSet();
for (Snapshot snapshot : current.snapshots()) {
if (!startingSnapshots.contains(snapshot.snapshotId())) {
newSnapshots.add(snapshot.snapshotId());
}
}
Set<String> committedFiles = committedFiles(ops, newSnapshots);
if (committedFiles != null) {
// delete all of the files that were deleted in the most recent set of operation commits
Tasks.foreach(deletedFiles)
.suppressFailureWhenFinished()
.onFailure((file, exc) -> LOG.warn("Failed to delete uncommitted file: {}", file, exc))
.run(
path -> {
if (!committedFiles.contains(path)) {
ops.io().deleteFile(path);
}
});
} else {
LOG.warn("Failed to load metadata for a committed snapshot, skipping clean-up");
}
} catch (RuntimeException e) {
LOG.warn("Failed to load committed metadata, skipping clean-up", e);
}
}
private void applyUpdates(TableOperations underlyingOps) {
// base保存的是在创建事务实例时，绑定的最新的TableMetadata
if (base != underlyingOps.refresh()) {
// 由于commit事务之前，表的Metadata很可能发生了变化，因此就需要我们将所有的更新操作修正到最新的实例上
// use refreshed the metadata
this.base = underlyingOps.current();
this.current = underlyingOps.current();
for (PendingUpdate update : updates) {
// re-commit each update in the chain to apply it and update current
try {
update.commit();
} catch (CommitFailedException e) {
// Cannot pass even with retry due to conflicting metadata changes. So, break the
// retry-loop.
throw new PendingUpdateFailedException(e);
}
}
}
}
Iceberg基于乐观锁的实现，以达到无锁提交TableMetadata的目的，因此在更新TableMetadata时，需要总是尝试获取表锁以便能够顺利更新表数据，例如对于Hive表保存Metadata时，则需要通过HiveTableOperations::doCommit(...)方法执行上锁、更新、释放锁的流程。
table.refresh();