Raptor read and write with delta delete functionality

[kewang1024]

== RELEASE NOTES ==

Raptor Changes
* Add the ability to read and write with delta deletes.

[kewang1024]

The second PR from #13248

[jessesleeping]

Reviewed "Enhance RaptorSplit and enable delta read".

As reviewed before, the general direction and main logic LGTM. Comments on minor changes.

[jessesleeping]

Reviewed "Add delta delete functionality on worker". Generally looks good. Comments are around code refactoring.

[jessesleeping]

Compared to DeltaShardRewriter, InplaceShardRewriter seems to be more coupled with OrcStorageManager as it's originally part of OrcStorageManager. I suggest we can keep InplaceShardRewriter an embedded class of OrcStorageManageror an anonymous class.

[kewang1024]

@highker what do you think of this suggestion?

[highker]

Some high-level comments

[kewang1024]

All known requests for change have been addressed. Put up a temp commit for refactor of OrcPageSource. Will merge it after review.

[highker]

rebase?

[jessesleeping]

Reviewed "Commit protocol for rewrite delete, delta delete". LGTM with minor comments.

[jessesleeping]

Reviewed " Export statistics of delta delete". LGTM with minor refactoring suggestions.

[highker]

OrcPageSource should not have fundamental surgery.

[highker]

unused

[highker]

This is not the right design. This file should remain (almost) unchanged. The only critical change to be made is on recordReader. That should be replaced by another page source (which inherits ConnectorPageSource) to represent the source data. Let me know if you want to chat in person.

[kewang1024]

Can we chat in person today?

[highker]

more comments

[highker]

early comments; still reviewing

[highker]

If this is only used by compaction, shall we move this with the compaction PR? Also, if this interface is only for delta delete, why do we need flag tableSupportsDeltaDelete? I think that can be removed.

[kewang1024]

If this is only used by compaction, shall we move this with the compaction PR? Also, if this interface is only for delta delete, why do we need flag tableSupportsDeltaDelete? I think that can be removed.

Actually this is a legacy problem.
We write this replaceShardUuids as a function to completely replace the old replaceShardUuids, that's why we need tableSupportsDeltaDelete in this new function.
But during the process, we found ourselves insecure of touching the old logical path, so we restore the old replaceShardUuids and keep these two together.
And we build tests around this new method and everything.
We will remove the old method once we feel the feature is stable. I've already left comments above
// TODO: Will merge these new function with old function once new feature is stable

[highker]

a try is needed to avoid file descriptor leak. Check other use cases.

[kewang1024]

a try is needed to avoid file descriptor leak. Check other use cases.

I think as long as we close OrcDataSource dataSource, we wouldn't have file descriptor leak.

So in above, we use
try (OrcDataSource dataSource = openShard(fileSystem, deltaShardUuid.get(), defaultReaderAttributes))
which will close the datasource once it encounters some exceptions

[highker]

You will miss closing all the stream readers. So use try to make sure it is closed properly.

[kewang1024]

You will miss closing all the stream readers. So use try to make sure it is closed properly.

But I do see from other examples that they used OrcBatchRecordReader with OrcDataSource dataSource as well, and they only close dataSource
eg1:
OrcStorageManager
public ConnectorPageSource getPageSource
it uses try and catch and it only closes the dataSourcecloseQuietly(dataSource);

eg2:
OrcBatchPageSourceFactory
createOrcPageSource
orcDataSource.close();

[highker]

Because readers will be closed by ConnectorPageSource. In your case, no one can close the readers.

[highker]

more comments

[highker]

You will miss closing all the stream readers. So use try to make sure it is closed properly.

[highker]

Please implement try for OrcBatchRecordReader recordReader = reader.createBatchRecordReader(. Otherwise LGTM

[highker]

lgtm