ahmedabu98 commented on code in PR #38837:
URL: https://github.com/apache/beam/pull/38837#discussion_r3589766223
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java:
##########
@@ -462,11 +481,76 @@ void validate(Table table) {
toSnapshotId);
}
+ if (fromSnapshotId != null) {
+ checkArgumentNotNull(
+ table.snapshot(fromSnapshotId),
+ error("configured starting snapshot does not exist: '%s'"),
+ fromSnapshotId);
+ }
+ if (toSnapshotId != null) {
+ checkArgumentNotNull(
+ table.snapshot(toSnapshotId),
+ error("configured end snapshot does not exist: '%s'"),
+ toSnapshotId);
+ }
+ if (fromSnapshotId != null && toSnapshotId != null) {
+ checkArgument(
+ SnapshotUtil.isAncestorOf(table, toSnapshotId, fromSnapshotId),
+ error("fromSnapshot '%s' is not an ancestor of toSnapshot '%s'"),
+ fromSnapshotId,
+ toSnapshotId);
+ }
+
if (getPollInterval() != null) {
checkArgument(
Boolean.TRUE.equals(getStreaming()),
error("'poll_interval_seconds' can only be set when streaming is
true"));
}
+
+ @Nullable String watermarkColumn = getWatermarkColumn();
+ if (watermarkColumn != null) {
+ checkArgument(getUseCdc(), error("'watermark_column' is only supported
in CDC mode"));
+ NestedField field = table.schema().findField(watermarkColumn);
+ checkArgument(
+ field != null, error("'watermark_column' refers to unknown column:
%s"), watermarkColumn);
+ checkArgument(
+ field.isRequired(),
+ error("'watermark_column' needs to be a non-nullable column: %s"),
+ watermarkColumn);
+ checkArgument(
+ field.type().typeId() == TIMESTAMP || field.type().typeId() == LONG,
+ error("'watermark_column' must be a timestamp-typed column, but '%s'
has type %s"),
+ watermarkColumn,
+ field.type().typeId());
+ checkArgumentNotNull(
+ getProjectedSchema().findField(watermarkColumn),
+ "'watermark_column' column should not be dropped.");
+ }
+
+ @Nullable String watermarkColumnTimeUnit = getWatermarkColumnTimeUnit();
+ if (watermarkColumnTimeUnit != null) {
+ checkArgument(
+ table
+ .schema()
+ .findField(
+ checkStateNotNull(
+ watermarkColumn,
+ "watermark_column_time_unit is configured without a
specified watermark_column"))
+ .type()
+ .typeId()
+ == LONG,
+ error("watermark_column_time_unit is only applicable for LONG
columns."));
+ try {
+ TimeUnit.valueOf(watermarkColumnTimeUnit.toUpperCase());
Review Comment:
Done
--
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]
