Spark: Structured Streaming read limit support follow-up #12260
Conversation
Use the ReadLimit passed in to SparkMicroBatchStream::latestOffset. In addition, fix a bug.
| readLimits[1] = ReadLimit.maxRows(maxFilesPerMicroBatch); | ||
| readLimits[1] = ReadLimit.maxRows(maxRecordsPerMicroBatch); |
There was a problem hiding this comment.
Thank you for catching ! This got missed, as we don't take the Readlimit we get from latestOffset API but rather from the configs which are set in constructor earlier!
| public void testReadStreamOnIcebergTableWithMultipleSnapshots_WithNumberOfFiles_1() | ||
| throws Exception { | ||
| public void testReadStreamWithMaxFiles1() throws Exception { |
There was a problem hiding this comment.
I renamed a few tests to be more concise. The old names were unwieldy and also not conforming to Java style.
| assertThat( | ||
| microBatchCount( | ||
| ImmutableMap.of( | ||
| SparkReadOptions.STREAMING_MAX_FILES_PER_MICRO_BATCH, "1", | ||
| SparkReadOptions.STREAMING_MAX_ROWS_PER_MICRO_BATCH, "2"))) | ||
| .isEqualTo(6); |
There was a problem hiding this comment.
This fails without the fix to SparkMicroBatchStream::getDefaultReadLimit(), as Spark then calls SparkMicroBatchStream::latestOffset(Offset, ReadLimit) with a CompositeReadLimit where one of the ReadLimits is a ReadMaxRows(1).
|
@singhpk234 @jackye1995 @RussellSpitzer this is a small fix; can you please review? |
| for (int i = 0; i < limits.length; i++) { | ||
| ReadLimit limit = limits[i]; | ||
| if (limit instanceof ReadMaxFiles) { | ||
| return ((ReadMaxFiles) limit).maxFiles(); | ||
| } | ||
| } |
There was a problem hiding this comment.
[minor] can we use this ?
| for (int i = 0; i < limits.length; i++) { | |
| ReadLimit limit = limits[i]; | |
| if (limit instanceof ReadMaxFiles) { | |
| return ((ReadMaxFiles) limit).maxFiles(); | |
| } | |
| } | |
| for (ReadLimit limit: limits) { | |
| if (limit instanceof ReadMaxFiles) { | |
| return ((ReadMaxFiles) limit).maxFiles(); | |
| } | |
| } |
| readLimits[1] = ReadLimit.maxRows(maxFilesPerMicroBatch); | ||
| readLimits[1] = ReadLimit.maxRows(maxRecordsPerMicroBatch); |
There was a problem hiding this comment.
Thank you for catching ! This got missed, as we don't take the Readlimit we get from latestOffset API but rather from the configs which are set in constructor earlier!
|
Thanks @singhpk234. |
This fixes the TODO in #4479.
Use the ReadLimit passed in to
SparkMicroBatchStream::latestOffset(Offset, ReadLimit). In testing this, a bug was found inSparkMicroBatchStream::getDefaultReadLimit()and fixed.