Manually retrying commits can result in partially applied changes

[jasonf20]

Apache Iceberg version

1.4.2 (latest release)

Query engine

Athena

Please describe the bug 🐞

Since the following PR: #6335 FastAppend and subclasses of MergingSnapshotProducer will skip newly added data files during manual commit retries (calling commit again inside a try-catch in user code).

This happens because the cached value is set to an empty list instead of null during cleanUncommittedAppends and then during retry when newDataFilesAsManifests is called the logic is skipped and no data files are returned.

The result can be partially applied changes if the user does manual retries of commits. For example, the following code will produce a rewrite that applies the deletes but does not add the new file:

import java.util
import java.util.UUID

import scala.collection.JavaConverters._

import org.apache.iceberg.aws.glue.GlueCatalog
import org.apache.iceberg.catalog._
import org.apache.iceberg.data.GenericRecord
import org.apache.iceberg.data.parquet.GenericParquetWriter
import org.apache.iceberg.parquet.Parquet
import org.apache.iceberg.types.Types
import org.apache.iceberg.{DataFile, PartitionSpec, Schema, Table, data}

object TestRewriteCommits {

  def main(args: Array[String]): Unit = {
    val catalog = new GlueCatalog()
    catalog.initialize("iceberg", Map.empty[String, String].asJava)

    val schema = new Schema(
      Types.NestedField.required(1, "id", Types.StringType.get()),
    );
    val tableName = "temp4"
    val tableId = TableIdentifier.of("prod_iceberg", tableName)
    val basePath = s"s3://s3-bucket-path/ice/tables/${tableName}/"

     val tableProperties: util.Map[String, String] = Map(
      "format-version" -> "2",
      "commit.retry.num-retries" -> "0" //turn off retries for more control during testing process
    ).asJava
    if (!catalog.tableExists(tableId)) {
      catalog.createTable(tableId, schema, PartitionSpec.unpartitioned(), basePath, tableProperties)
    }

    val table = catalog.loadTable(tableId)

    val addedFiles = (1 to 2).map(i => {
      val file: DataFile = writeFile(basePath, table)
      val append = table.newAppend()
      append.appendFile(file)
      append.commit()
      file
    })

    val transaction = table.newTransaction()
    val rewrite = transaction.newRewrite()
    addedFiles.foreach(rewrite.deleteFile)
    rewrite.addFile(writeFile(basePath, table))
    rewrite.commit()
    try {
      // Make sure this commit fails (I failed it by breaking at glue.updateTable(updateTableRequest.build()); and changing the table from athena.
      transaction.commitTransaction()
    } catch {
      case e: Throwable =>
        // This retry will run successfully but the result will not contain the data file added during the rewrite. 
        transaction.commitTransaction()
    }
  }

  private def writeFile(basePath: String, table: Table) = {
    val writer = Parquet.writeData(
        table.io().newOutputFile(basePath + UUID.randomUUID().toString + ".parquet"))
      .forTable(table)
      .overwrite(true)
      .createWriterFunc(GenericParquetWriter.buildWriter)
      .build[data.Record]()
    writer.write(Iterable(GenericRecord.create(table.schema()).copy("id", "1")).asJava)
    writer.close()
    val file = writer.toDataFile
    file
  }
}

I think fixing this can be done by either setting the cached value to null like it was before or by forbidding calling commit more than once.