Extend RewriteManifests with a way to add/delete manifests

[aokolnychyi]

This PR extends RewriteManifests with a way to directly delete/add manifests, which enables us to rewrite manifests using an external process.

[aokolnychyi]

@bryanck could you take a look?

[bryanck]

I feel this new functionality might be better suited to a separate class, something like ReplaceManifests? The current rewrite functionality and add/delete don't share much in common. If we did want to share some code, maybe create a base class.

[rdblue]

This shouldn't pass summaryBuilder because the files are not actually added or deleted. The set of files should be the same.

Side note: should we check that the number of data files in deleted manifests matches the number of data files in added manifests before committing?

[aokolnychyi]

Similarly to BaseRewriteManifests, our utility reads only live entries via liveEntries in FilteredManifests and produces staged manifests that contain entries with status EXISTING. ManifestWriter$copyManifest is changed to work for arbitrary use cases. In this particular one, it won't report any files as added or removed because all manifest entries have one status: EXISTING (validated in addManifest).

[rdblue]

In this context, manifests that are copied are allowed to have any entries. The validation for the copy in AppendFiles that all entries were ADDED was because the entries are for files that were being appended to the dataset. But here, any manifest could be rewritten, so any status is allowed.

[aokolnychyi]

I followed the logic in BaseRewriteManifests and MergingSnapshotProducer: rewriting manifests produces a new snapshot where files that were present before are added with status EXISTING while keeping their original snapshot id.

By restricting manifest entries to EXISTING, we validate that RewriteManifests is not used to add/delete files.

[rdblue]

That sounds good to me, but I don't think that the logic in the copy does this. We just need to restrict entries to live entries and convert ADDED entries to EXISTING.

[aokolnychyi]

Left as is according to this comment.

[aokolnychyi]

@bryanck I also thought about having a separate class but I was a bit worried about adding a new method with a similar purpose to the Table API as it already contains quite some logic. The change did feel too big to me as we still reuse a couple of methods.

[aokolnychyi]

@bryanck @rdblue thanks for the initial review! There are a couple of questions I want to clarify before updating the PR. Let me know what you think.

[aokolnychyi]

@rdblue I am not sure changing the status of entries from ADDED to EXISTING while copying manifests would give us any benefits. Instead, I would leave that to the utilities that produce staged manifests. Later, we want to get rid of copying manifests at all.

Here is how we read metadata per file:

    val manifestPathDS = spark.sparkContext.parallelize(manifestPaths, numPartitions.toInt).toDS()
    manifestPathDS.flatMap { path =>
      val reader = ManifestReader.read(io.newInputFile(path), specLookup)
      val spec = reader.spec
      try {
        val filteredManifest = reader.select(ImmutableList.of("*"))
        filteredManifest.liveEntries.asScala.map { entry =>
          val file = entry.file.copy()
          val snapshotId = entry.snapshotId
          (spec.specId, snapshotId, strategy.clusterBy(spec, file), file)
        }
      } finally {
        reader.close()
      }
    }

While writing staged manifets, we call writer.existing(file, snapshotId) to have correct entry status.

[aokolnychyi]

@bryanck @rdblue I've updated the PR and put the logic into a separate class as it seems reasonable to me. I still assume utilities will convert manifest entry statuses from ADDED to EXISTING themselves as it will allow us to avoid copying manifests in the future.

[aokolnychyi]

@rdblue @bryanck Ready for another review round.

[rdblue]

I think this comment and method name were useful to help understand what's happening here. We at least need to document that this function is responsible for adding manifests that were added since the snapshot that was originally rewritten. I can see how using "keep" in the method name makes sense since it adds manifests to keptManifests.

[rdblue]

What about keepNewActiveManifests?

[aokolnychyi]

The reason I modified this part is that the comment is no longer always true after this change. For example, when we use only add/delete manifests, there will be no rewrite and might be no retry, but we will still call keepActiveManifests to keep all active manifests.

[rdblue]

Got it, I think that works.

[rdblue]

Should this clean up added manifest files?

I think that is a responsibility of the caller because the caller may be able to reuse some of them if the commit fails because a deleted manifest is no longer in the table. This is similar to a write operation, where if Iceberg can't commit the write, it doesn't delete the data files that were added in the failed operation. Instead, the writer has abort functions that are called outside of the snapshot commit: https://github.com/apache/incubator-iceberg/blob/master/spark/src/main/java/org/apache/iceberg/spark/source/Writer.java#L183

[aokolnychyi]

This cleans up copies of the added manifests, not the original manifests that were added. I believe the caller doesn't have access to the copied manifests.

[rdblue]

Ah, you're right. Thanks for the explanation!

[rdblue]

@aokolnychyi, there is just one problem with cleanUncommitted. Everything else is minor, though would be nice to fix the nits.

[rdblue]

@aokolnychyi, looks like there's a test failure, but I think this is ready to merge when that's fixed.

[aokolnychyi]

@rdblue, thanks for the review. I modified an error message and forgot to update a test. Updated the PR now.

[aokolnychyi]

Seems like some Python tests failed this time.

[rdblue]

Yeah, looks like python is having trouble. We'll look into it, but it shouldn't block this commit.