ahmedabu98 commented on code in PR #39426:
URL: https://github.com/apache/beam/pull/39426#discussion_r3661775853
##########
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java:
##########
@@ -174,4 +180,124 @@ static Schema.FieldType convertToBeamFieldType(DataType
deltaType) {
}
}
}
+
+ @AutoValue
+ public abstract static class ReadChanges extends PTransform<PBegin,
PCollection<Row>> {
+ public abstract @Nullable String getTablePath();
+
+ public abstract @Nullable Long getStartVersion();
+
+ public abstract @Nullable String getStartTimestamp();
+
+ public abstract @Nullable Long getEndVersion();
+
+ public abstract @Nullable String getEndTimestamp();
+
+ public abstract @Nullable Map<String, String> getHadoopConfig();
+
+ abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setTablePath(String tablePath);
+
+ abstract Builder setStartVersion(@Nullable Long startVersion);
+
+ abstract Builder setStartTimestamp(@Nullable String startTimestamp);
+
+ abstract Builder setEndVersion(@Nullable Long endVersion);
+
+ abstract Builder setEndTimestamp(@Nullable String endTimestamp);
+
+ abstract Builder setHadoopConfig(@Nullable Map<String, String>
hadoopConfig);
+
+ abstract ReadChanges build();
+ }
+
+ public ReadChanges from(String tablePath) {
+ return toBuilder().setTablePath(tablePath).build();
+ }
+
+ public ReadChanges withStartVersion(long startVersion) {
+ return toBuilder().setStartVersion(startVersion).build();
+ }
+
+ public ReadChanges withStartTimestamp(String startTimestamp) {
+ return toBuilder().setStartTimestamp(startTimestamp).build();
+ }
+
+ public ReadChanges withEndVersion(long endVersion) {
+ return toBuilder().setEndVersion(endVersion).build();
+ }
+
+ public ReadChanges withEndTimestamp(String endTimestamp) {
+ return toBuilder().setEndTimestamp(endTimestamp).build();
+ }
+
+ public ReadChanges withConfig(Map<String, String> config) {
+ return toBuilder().setHadoopConfig(config).build();
+ }
+
+ @Override
+ public PCollection<Row> expand(PBegin input) {
+ String path = getTablePath();
+ if (path == null) {
+ throw new IllegalArgumentException("Table path must be set.");
+ }
+ if (getStartVersion() == null && getStartTimestamp() == null) {
+ throw new IllegalArgumentException("Either startVersion or
startTimestamp must be set.");
+ }
Review Comment:
The IcebergIO default is we read from the beginning for batch and from the
very end for streaming
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
