HIVE-28047: Iceberg: Major QB Compaction with a single commit

[difin]

What changes were proposed in this pull request?

Improvements to Hive Iceberg QB Major Compaction to perform compaction in one commit instead of two commits as was done till now.

Why are the changes needed?

Existing implementation of compaction creates 2 commits which creates 2 snapshots: first snapshot with all the files deleted and second snapshot with compacted files. If a user queries the table based on snapshot id of the first snapshot, the result would be invalid as no data is present in the table in that snapshot. To avoid this problem this PR is proposed.

Does this PR introduce any user-facing change?

No

Is the change a dependency upgrade?

No

How was this patch tested?

Hive contains 4 query tests for testing Hive Iceberg QB Major Compaction. The outputs of these q-tests were updated as part of this PR.

[zhangbutao]

Can we reuse FilesForCommit results to get the dataFiles and deleteFiles?

[difin]

I don't think so, because FilesForCommit contains only the compacted, new files. They don't contain existing data and delete files.

[zhangbutao]

Since the Major QB Compaction is essentially a IOW operation, why don't we follow the regular IOW flow? Why do we need the extra operation to find the existing data and delete files?
Just reuse the ReplacePartitions api:

hive/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java

Lines 556 to 561 in e08a600

	ReplacePartitions overwrite = transaction.newReplacePartitions();
	results.dataFiles().forEach(overwrite::addFile);
	if (StringUtils.isNotEmpty(branchName)) {
	overwrite.toBranch(HiveUtils.getTableSnapshotRef(branchName));
	}
	overwrite.commit();

[difin]

Because IOW isn't supported on tables that had schema evolution as per https://issues.apache.org/jira/browse/HIVE-26133, but we want to support compaction even for tables that have had schema evolution.

[zhangbutao]

Got it, thx.

It seems that we have tradeoff about the change.
(twice commit) vs (single commit: loop to get and store all existing data and delte files)

I know twice commit will generate two snapshot and maybe get inconsistency of data when users reading table.
But for this change, if the original iceberg table has many many small data and delete files, and we need to loop and store these files in the List, can this behavior would give memory pressure on HS?
I am not sure which way is the best one. ;(

Can other folks give some other thought? @deniskuzZ @ayushtkn @SourabhBadhya

[deniskuzZ]

@difin, do we know if DeleteFiles API is doing the same full table listing when Expressions.alwaysTrue() filter is provided. If not, maybe we could extend OverwriteFiles API to support delete filters?

[deniskuzZ]

In my opinion this approach looks more expensive, maybe we should reach out to iceberg community with the. proposal to extend current API with atomic IOW semantic?

[difin]

In my opinion this approach looks more expensive, maybe we should reach out to iceberg community with the. proposal to extend current API with atomic IOW semantic?

I agree it is a little bit more expensive approach, but it doesn't have the correctness issue like the existing approach and this approach is used in another engines like Amoro, Trino and Impala is also in process of changing to this approach.

[sonarcloud]

Quality Gate passed

Issues
0 New issues

Measures
0 Security Hotspots
No data about Coverage
No data about Duplication

See analysis details on SonarCloud

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.
Feel free to reach out on the dev@hive.apache.org list if the patch is in need of reviews.

[deniskuzZ]

you should be doing

    long startingSnapshotId = table.currentSnapshot().snapshotId();
    
    StructLikeMap<List<List<FileScanTask>>> fileGroupsByPartition =
    planFileGroups(startingSnapshotId);

    CloseableIterable<FileScanTask> fileScanTasks =
        table
            .newScan()
            .useSnapshot(startingSnapshotId)
            .filter(filter)
            .ignoreResiduals()
            .planFiles();

see RewriteDataFilesSparkAction

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.
Feel free to reach out on the dev@hive.apache.org list if the patch is in need of reviews.