[kernel-spark] Implement latestOffset() with rate limiting for dsv2 streaming #5409
Conversation
zikangh
force-pushed
the
stack/latestsnapshot2
branch
from
f602eff to
09ebcbb
Compare
zikangh
changed the title
|
Hi @huan233usc, could you please add @huan233usc, @gengliangwang, @jerrypeng, @tdas to the reviewers list? |
| this.streamingHelper = new StreamingHelper(tablePath, hadoopConf); | ||
|
|
||
| // Initialize snapshot at source init to get table ID, similar to DeltaSource.scala | ||
| this.snapshotAtSourceInit = TableManager.loadSnapshot(tablePath).build(engine); |
There was a problem hiding this comment.
Just call streamingHelper.loadLatestSnapshot(), underlying it will call TableManager.loadSnapshot(tablePath).build(engine)
| @Override | ||
| public Offset latestOffset(Offset startOffset, ReadLimit limit) { | ||
| // For the first batch, initialOffset() should be called before latestOffset(). | ||
| // if startOffset is null: no data is available to read. |
There was a problem hiding this comment.
Will there by a case that startOffset is set to null
There was a problem hiding this comment.
Spark will call initialOffset first to obtain this startOffset for batch 0 -- so it could be null when initialOffset() returns null (i.e. table has no data).
| // TODO(#5318): Check read-incompatible schema changes during stream start | ||
| IndexedFile lastFile = lastFileChange.get(); | ||
| return ScalaUtils.toJavaOptional( | ||
| DeltaSource.buildOffsetFromIndexedFile( |
There was a problem hiding this comment.
DeltaSource.buildOffsetFromIndexedFile seems to always return an offset although it returns Option[Offset].
Can we update the signature for DeltaSource.buildOffsetFromIndexedFile? I just want to minimize the interaction with null if it is not really necessary.
zikangh
force-pushed
the
stack/latestsnapshot2
branch
2 times, most recently
from
797770c to
8298d06
Compare
zikangh
force-pushed
the
stack/latestsnapshot2
branch
from
8298d06 to
df5e0f1
Compare
zikangh
force-pushed
the
stack/latestsnapshot2
branch
2 times, most recently
from
acc4fe8 to
8d106d8
Compare
zikangh
force-pushed
the
stack/latestsnapshot2
branch
from
8d106d8 to
029a596
Compare
2 participants
🥞 Stacked PR
Use this link to review incremental changes.
Which Delta project/connector is this regarding?
Description
We add implementation for
latestOffset(startOffset, limit)andgetDefaultReadLimit()for a completeSupportsAdmissionControlimplementation.Also refactored a few
DeltaSource.scalamethods -- we make them static so we can call them from SparkMicrobatchStream.java.How was this patch tested?
Parameterized tests verifying parity between DSv1 (DeltaSource) and DSv2 (SparkMicroBatchStream).
Does this PR introduce any user-facing changes?
No