[core] Generate changelog by copying data when records are insert-only

[yunfengzhou-hub]

Purpose

Linked issue: #3567

• Implements custom copy in local filesystem and OSS
• Adds the optimization to generate changelog by copying data file
• Triggers the optimization when the following conditions are met
  • The filesystem implements copy
  • The changelog producer is INPUT
  • all records received since last flush are insert-only

Tests

API and Format

This PR does not affect API or storage format.

Documentation

This PR does not add a new feature that requires document.

[yunfengzhou-hub]

Hi @JingsongLi could you please take a look at this PR?

[Sxnan]

@yunfengzhou-hub Thanks for the PR. I left some comments below.

[yunfengzhou-hub]

Thanks for the comments @JingsongLi. I have updated the PR according to the comments.

[JingsongLi]

Can we just introduce a new FileIO interface?

interface NativeCopyFileIO extends FileIO {
    void nativeCopy(Path src, Path dst);
}

[yunfengzhou-hub]

According to offline discussion, I benchmarked the throughput of the default implementation introduced in FileIO in Aliyun, and it seems that the default implementation has been enough to bring performance improvement. I did not verify against other filesystems, but theoretically speaking copying should also bring performance improvement to them. Given this result, I'll directly add the copy method to FileIO, instead of introducing the new NativeCopyFileIO interface.

[JingsongLi]

+1

[liming30]

@yunfengzhou-hub hi, when will StoreSinkWrite#withInsertOnly be called? I only see manual calls in the test code, which does not seem to be automatically inferred by the engine.

https://github.com/apache/paimon/blob/master/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWrite.java#L49