Core: Support incremental compute for partition stats

[ajantha-bhat]

If the previous stats file exist, no need to compute the stats from the scratch.

Identify the latest snapshot for which partition stats file exist. Read the previous stats, incrementally compute the stats for new snapshots, merge the stats and write them to the new file.

[ajantha-bhat]

Some engines want to synchronously write the partition stats. (Similar to how Trino synchronously write the puffin files during insert). Reading all the manifests in a table can be avoided to compute partition stats if the we compute the stats incrementally and merge it with previous stats.

@aokolnychyi, @pvary, @deniskuzZ, @gaborkaszab : Let me know what you guys think.

[ajantha-bhat]

since the new unified tuple is used for reading the old stats file. It automatically handled the schema evolution.

[ajantha-bhat]

I have added 1.9.0 milestone for this PR as it is a small change (excluding refactoring) and we still have some time for 1.9.0 sue to open issues in milestone.

[pvary]

Don't we want this as a default predicate?

manifestFile -> !manifestFile.hasExistingFiles()

[deniskuzZ]

we could add it as default filter:

if (fromSnapshot != null) {
  manifestFilePredicate =
      manifestFile -> snapshotIdsRange.contains(manifestFile.snapshotId())
}
List<ManifestFile> manifests =
  currentSnapshot.allManifests(table.io()).stream()
      .filter(manifestFilePredicate)
      .filter(manifestFile -> !manifestFile.hasExistingFiles())
      .collect(Collectors.toList());

[ajantha-bhat]

good point.

While computing incremental, I observed that it may become duplicate counts. So, I added.
I do have some gaps, I need to understand fully when and all we mark manifest entry as existing.
Is there any scenario exist to consider "existing" entries or just "added" is enough?

There is another check down below, that considers both added and existing (added long back).

iceberg/core/src/main/java/org/apache/iceberg/PartitionStatsUtil.java

Line 103 in d54d81e

if (entry.isLive()) {

I will update the code to just keep added entry and also add a testcase of rewrite data files to ensure stats are same after the rewrite.

[ajantha-bhat]

Also, looks like ManifestFile can have both added and existing entries together? So, Instead of filtering here. I will keep filtering just at the entries level down below in collectStatsForManifest

[deniskuzZ]

what if we have compaction and expire snapshots? new manifests would have the EXISTING entries?

[pvary]

What do we do with the stats of the removed files?

Lets say:

• S1 adds data
• Execute the stats collection
• S2 adds more data
• S3 compacts data from S1, and S2 - This removes files created by S1, and S2 and creates new files
• Execute incremental/normal stats collection

What happens with the stats in this case?

[deniskuzZ]

compaction doesn't remove the data. if we expire S1 and S2 we don't have prev snapshots/stats and start fresh (i.e. full compute)

[pvary]

If we don't expire data, could we detect that S3 is only a compaction commit, and the stats don't need to be changed?

What if S3 instead is a MoW commit? Can we detect the changes and calculate stats incrementally?

[ajantha-bhat]

1. Compaction will have snapshot operation as REPLACE and we can reuse the old stats for that scenario. But need to write the new stats file with same data to handle clean GC of snapshot files.

Compaction will be tested end to end while adding the spark procedure.

2. About the live (existing + added),

For full compute, old manifest files will be marked as deleted and entries will be reused as existing in the manifest files + may have additional added entry. So, for full compute need to consider both existing and added.

For incremental compute, old stats file has some entires which are now existing. So, should consider the existing entires.

This all leads to the next question, what happens when manifest is deleted. That case we just update the snapshot entry (last modified) and not decrement the stats. Hence, we should skip it for incremental compute again.

All these logic present in collectStatsForManifest and existing testcases (full compute and incremental) covers it as it uses mergeAppend which produces manifest mix of added and existing entires.

[ajantha-bhat]

We didn't need decrement stats for full compute because we were discarding the deleted manifests. Only considering live manifests.

Now, I am not really sure for compaction, the current code will work. We may need decrement stats just for incremental compute. I will test compaction scenario tomorrow and handle this.

[deniskuzZ]

LGTM +1

[pvary]

Could this throw an error instead?

[deniskuzZ]

why? That is to handle the case when no stats files existed before, and we need to execute a full computation.
We enter here when computing stats for the first time.

[pvary]

If I understand correctly, the user requested incremental stat compute, but with wrong parameters. In this case we could either "correct" the mistake or throw an error.

The question is, how frequent is the problem, and how easy is to detect from the user side

[deniskuzZ]

what do you mean by wrong parameters? non existing snapshotId?

[pvary]

what do you mean by wrong parameters? non existing snapshotId?

Exactly

[ajantha-bhat]

throwing error now and added the testcase

[deniskuzZ]

i don't agree with that design, see #12629 (comment)

[ajantha-bhat]

@deniskuzZ, @pvary: Thanks guys for the review. I have addressed all the comments. You can take a fresh look again tomorrow :D (after some break :D)

[deniskuzZ]

I don't think it's user-friendly + recompute flag loses it's purpose (you can call directly computeAndWriteStats).
Now every client needs to implement either the same prev stats file check or do the try-catch.

try {
   computeAndWriteStatsFileIncremental()
} catch (RuntimeException e) {
 if (e.getMessage().equals("bla-bla")) {
   computeAndWriteStats
 }
}

I would expect from computeAndWriteStatsFileIncremental do what's needed instead throwing Previous stats not found exception.

Non-existent snapshotId is a diff situation. We should validate if snapshot == null and throw Snapshot doesn't exist

[ajantha-bhat]

recompute flag loses it's purpose

there is no recompute flag exposed to the user. The private method (incrementalComputeAndMerge) which is throwing this exception is also always computing incrementally.

I would expect from computeAndWriteStatsFileIncremental do what's needed instead throwing Previous stats not found exception.

computeAndWriteStatsFileIncremental says incremental compute. Forcefully recomputing when there is an error is not a good idea as the method's responsibility is just to try incremental compute?

Maybe I can expose another method called computeAndWriteStatsWithFallback(), which will internally calls it?

public void computeAndWriteStatsIncrementalWithFallback() {
   try {
       computeAndWriteStatsFileIncremental();
   } catch (RuntimeException e) {
       if ("bla-bla".equals(e.getMessage())) {
           computeAndWriteStats(); // Fallback in case of a specific error
       } else {
           throw e; // Re-throw unexpected errors
       }
   }
}

[deniskuzZ]

I liked how you did it initially. Please disregard the recompute flag comment, it has nothing to do with the incremental workflow.

Think about what changes are needed on the client side. I was planning just to replace the existing call to the incremental one unless it's ANALYZE TABLE (force recompute).

What are the use-cases we would benefit from the prev stats file missing exception?

[ajantha-bhat]

Lets see what @pvary thinks.