Skip to content

Spark: Refactor SparkReadConf use primitive type for confs with default values #7429

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
aokolnychyi merged 1 commit into from
Apr 27, 2023
Merged

Spark: Refactor SparkReadConf use primitive type for confs with default values #7429

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ public boolean handleTimestampWithoutZone() {
.parse();
}

public Long streamFromTimestamp() {
public long streamFromTimestamp() {
return confParser
.longConf()
.option(SparkReadOptions.STREAM_FROM_TIMESTAMP)
Expand Down
8 changes: 4 additions & 4 deletions spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,14 +69,14 @@ public class SparkMicroBatchStream implements MicroBatchStream {
private final boolean caseSensitive;
private final String expectedSchema;
private final Broadcast<Table> tableBroadcast;
private final Long splitSize;
private final Integer splitLookback;
private final Long splitOpenFileCost;
private final long splitSize;
private final int splitLookback;
private final long splitOpenFileCost;
private final boolean localityPreferred;
private final StreamingOffset initialOffset;
private final boolean skipDelete;
private final boolean skipOverwrite;
private final Long fromTimestamp;
private final long fromTimestamp;

SparkMicroBatchStream(
JavaSparkContext sparkContext,
Expand Down
2 changes: 1 addition & 1 deletion spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ public boolean handleTimestampWithoutZone() {
.parse();
}

public Long streamFromTimestamp() {
public long streamFromTimestamp() {
return confParser
.longConf()
.option(SparkReadOptions.STREAM_FROM_TIMESTAMP)
Expand Down
8 changes: 4 additions & 4 deletions spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,14 +70,14 @@ public class SparkMicroBatchStream implements MicroBatchStream {
private final boolean caseSensitive;
private final String expectedSchema;
private final Broadcast<Table> tableBroadcast;
private final Long splitSize;
private final Integer splitLookback;
private final Long splitOpenFileCost;
private final long splitSize;
private final int splitLookback;
private final long splitOpenFileCost;
private final boolean localityPreferred;
private final StreamingOffset initialOffset;
private final boolean skipDelete;
private final boolean skipOverwrite;
private final Long fromTimestamp;
private final long fromTimestamp;

SparkMicroBatchStream(
JavaSparkContext sparkContext,
Expand Down
6 changes: 3 additions & 3 deletions spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ public boolean handleTimestampWithoutZone() {
.parse();
}

public Long streamFromTimestamp() {
public long streamFromTimestamp() {
return confParser
.longConf()
.option(SparkReadOptions.STREAM_FROM_TIMESTAMP)
Expand All @@ -255,15 +255,15 @@ public Long endTimestamp() {
return confParser.longConf().option(SparkReadOptions.END_TIMESTAMP).parseOptional();
}

public Integer maxFilesPerMicroBatch() {
public int maxFilesPerMicroBatch() {
return confParser
.intConf()
.option(SparkReadOptions.STREAMING_MAX_FILES_PER_MICRO_BATCH)
.defaultValue(Integer.MAX_VALUE)
.parse();
}

public Integer maxRecordsPerMicroBatch() {
public int maxRecordsPerMicroBatch() {
return confParser
.intConf()
.option(SparkReadOptions.STREAMING_MAX_ROWS_PER_MICRO_BATCH)
Expand Down
12 changes: 6 additions & 6 deletions spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,16 +77,16 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
private final boolean caseSensitive;
private final String expectedSchema;
private final Broadcast<Table> tableBroadcast;
private final Long splitSize;
private final Integer splitLookback;
private final Long splitOpenFileCost;
private final long splitSize;
private final int splitLookback;
private final long splitOpenFileCost;
private final boolean localityPreferred;
private final StreamingOffset initialOffset;
private final boolean skipDelete;
private final boolean skipOverwrite;
private final Long fromTimestamp;
private final Integer maxFilesPerMicroBatch;
private final Integer maxRecordsPerMicroBatch;
private final long fromTimestamp;
private final int maxFilesPerMicroBatch;
private final int maxRecordsPerMicroBatch;

SparkMicroBatchStream(
JavaSparkContext sparkContext,
Expand Down
2 changes: 1 addition & 1 deletion spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ public boolean handleTimestampWithoutZone() {
.parse();
}

public Long streamFromTimestamp() {
public long streamFromTimestamp() {
return confParser
.longConf()
.option(SparkReadOptions.STREAM_FROM_TIMESTAMP)
Expand Down
8 changes: 4 additions & 4 deletions spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,14 +72,14 @@ public class SparkMicroBatchStream implements MicroBatchStream {
private final boolean caseSensitive;
private final String expectedSchema;
private final Broadcast<Table> tableBroadcast;
private final Long splitSize;
private final Integer splitLookback;
private final Long splitOpenFileCost;
private final long splitSize;
private final int splitLookback;
private final long splitOpenFileCost;
private final boolean localityPreferred;
private final StreamingOffset initialOffset;
private final boolean skipDelete;
private final boolean skipOverwrite;
private final Long fromTimestamp;
private final long fromTimestamp;

SparkMicroBatchStream(
JavaSparkContext sparkContext,
Expand Down