Adds DeltasharingSource

[linzhou-db]

Adds DeltasharingSource, similar to DeltaSource, supports streaming addfiles,

• local cache: sortedFetchedFiles
• 30 seconds interval is added for getTableVersion
• ignoreChanges/ignoreDeletes/maxFilesPerTrigger/maxBytesPerTrigger are supported

Not in this PR, will be supported in a separate PR

• readChangeFeed/readChangeData
• read compatibility check on schema
• SupportsTriggerAvailableNow
• startingTimestamp

[chakankardb]

Some initial comments.

[chakankardb]

Question:
Why do we load the snapshot from prev version?
Do we want to use its metadata ?

[linzhou-db]

I don't think it's anything related to the metadata. The schema of the latest table version at the time the stream query starts is used for constructing the dataframe.

I copied this logic from DeltaSource, I guess:
It's more of a product behavior decision, when getBatch is called and startOffsetOption is not set, and endOffset is not a startingVersion, it means: 1) no startingVersion is provided, 2) the stream query is not starting from a checkpoint 3) it's the first getBatch after the stream query started, 4) endOffset is from a previous latestOffset call. 5) and endOffset is NOT startingVersion. 6) and given 1) in L410, some version is the startingVersion.

Now, it needs to decide where the starting version is, then I guess it's a prod decision to use (endOffset.tableVersion - 1), or maybe we should use the version we used in L410?

@zsxwing Do you have an idea on this? Thanks!

[chakankardb]

Consider adding to the comment that we do this to maintain same behavior as delta.

[linzhou-db]

added the comment, will merge this first and still confirm with Ryan later.

[zsxwing]

This hack is for the following case:

When we start a streaming query, we load the latest version X - 1. However, when latestOffset is called, we get an offset F in version X (new data got written in between)

Now getBatch is called with (startOffsetOption=None, end=the above offset). Since this is the first batch, we need to return the table data in version X - 1 + the data in version X from offset 0 to offset F (Note: we should not return data after offset F). So we use (offset=-1, version = X-1, isStartingVersion=true) to load the table data in version X - 1.

[chakankardb]

@linzhou-db Consider adding Ryan's explanation as a comment.

[linzhou-db]

Thanks! I just want to clarify and confirm:
This only happens to the first latestOffset and getBatch call of a stream query without startingVersion, and the Offset returned by latestOffset and used as end in getBatch can only be 1 version away from the startingVersion returned here, because it is loading snapshot at startingVersion.

I'll add the comment in another PR.

[chakankardb]

Consider adding to the comment that we do this to maintain same behavior as delta.