Skip to content

Fix IcebergIO rejecting timestamp strings with a space separator #40861

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

Open
mananmangal wants to merge 1 commit into
base: master
Choose a base branch
from
+98 −2
Open

Fix IcebergIO rejecting timestamp strings with a space separator #40861

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
42 changes: 40 additions & 2 deletions sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
*/
package org.apache.beam.sdk.io.iceberg;

import static java.time.format.DateTimeFormatter.ISO_LOCAL_DATE;
import static java.time.format.DateTimeFormatter.ISO_LOCAL_TIME;
import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
Expand All @@ -27,6 +29,10 @@
import java.time.LocalTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.chrono.IsoChronology;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.format.ResolverStyle;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
Expand Down Expand Up @@ -88,6 +94,26 @@ private IcebergUtils() {}
.put(EnumerationType.IDENTIFIER, Types.StringType.get())
.build();

private static final DateTimeFormatter TIMESTAMP_SPACE_FORMATTER =
new DateTimeFormatterBuilder()
.parseCaseInsensitive()
.append(ISO_LOCAL_DATE)
.appendLiteral(' ')
.append(ISO_LOCAL_TIME)
.toFormatter()
.withResolverStyle(ResolverStyle.STRICT)
.withChronology(IsoChronology.INSTANCE);

private static final DateTimeFormatter TIMESTAMPTZ_SPACE_FORMATTER =
new DateTimeFormatterBuilder()
.append(TIMESTAMP_SPACE_FORMATTER)
.parseLenient()
.appendOffsetId()
.parseStrict()
.toFormatter()
.withResolverStyle(ResolverStyle.STRICT)
.withChronology(IsoChronology.INSTANCE);

private static Schema.FieldType icebergTypeToBeamFieldType(
final Type type, @Nullable String updateCompatibilityVersion) {
switch (type.typeId()) {
Expand Down Expand Up @@ -491,7 +517,12 @@ private static Object getIcebergTimestampValue(Object beamValue, boolean shouldA
} else if (beamValue instanceof Long) { // FieldType.INT64
return DateTimeUtil.timestamptzFromMicros((Long) beamValue);
} else if (beamValue instanceof String) { // FieldType.STRING
return OffsetDateTime.parse((String) beamValue).withOffsetSameInstant(ZoneOffset.UTC);
String value = (String) beamValue;
OffsetDateTime offsetDateTime =
hasSpaceSeparator(value)
? OffsetDateTime.parse(value, TIMESTAMPTZ_SPACE_FORMATTER)
: OffsetDateTime.parse(value);
return offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC);
} else {
throw new UnsupportedOperationException(
"Unsupported Beam type for Iceberg timestamp with timezone: " + beamValue.getClass());
Expand All @@ -510,13 +541,20 @@ private static Object getIcebergTimestampValue(Object beamValue, boolean shouldA
} else if (beamValue instanceof Long) { // FieldType.INT64
return DateTimeUtil.timestampFromMicros((Long) beamValue);
} else if (beamValue instanceof String) { // FieldType.STRING
return LocalDateTime.parse((String) beamValue);
String value = (String) beamValue;
return hasSpaceSeparator(value)
? LocalDateTime.parse(value, TIMESTAMP_SPACE_FORMATTER)
: LocalDateTime.parse(value);
} else {
throw new UnsupportedOperationException(
"Unsupported Beam type for Iceberg timestamp with timezone: " + beamValue.getClass());
}
}

private static boolean hasSpaceSeparator(String timestamp) {
return timestamp.length() > 10 && timestamp.charAt(10) == ' ';
}

/** Converts a {@link StructLike} to a Beam {@link Row}. */
public static Row structToRow(Schema schema, StructLike struct) {
checkState(
Expand Down
58 changes: 58 additions & 0 deletions sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeParseException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -149,6 +150,13 @@ private void checkRowValueToRecordValue(
assertThat(record.getField("v"), equalTo(destValue));
}

private Record toIcebergRecord(Type destType, String sourceValue) {
Schema beamSchema = Schema.of(Schema.Field.of("v", Schema.FieldType.STRING));
Row row = Row.withSchema(beamSchema).addValues(sourceValue).build();
return IcebergUtils.beamRowToIcebergRecord(
new org.apache.iceberg.Schema(required(0, "v", destType)), row);
}

@Test
public void testBoolean() {
checkRowValueToRecordValue(Schema.FieldType.BOOLEAN, Types.BooleanType.get(), true);
Expand Down Expand Up @@ -225,6 +233,13 @@ public void testTimestamp() {
LocalDateTime localDateTime = LocalDateTime.of(2024, 10, 8, 13, 18, 20, 53_000_000);
checkRowValueToRecordValue(
Schema.FieldType.STRING, val, Types.TimestampType.withoutZone(), localDateTime);

// Schema.FieldType.STRING, with a space separating date and time
checkRowValueToRecordValue(
Schema.FieldType.STRING,
"2024-10-08 13:18:20.053",
Types.TimestampType.withoutZone(),
localDateTime);
}

@Test
Expand Down Expand Up @@ -268,6 +283,49 @@ public void testTimestampWithZone() {
val,
Types.TimestampType.withZone(),
offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC));

// Schema.FieldType.STRING, with a space separating date and time
checkRowValueToRecordValue(
Schema.FieldType.STRING,
"2024-10-08 13:18:20.053+03:27",
Types.TimestampType.withZone(),
offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC));
checkRowValueToRecordValue(
Schema.FieldType.STRING,
"2024-10-08 09:51:20.053Z",
Types.TimestampType.withZone(),
offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC));
}

@Test
public void testTimestampStringWithInvalidSeparator() {
for (String val :
Arrays.asList(
"2024-10-08 T13:18:20.053", "2024-10-0813:18:20.053", "2024-10-08 13:18:20.053")) {
assertThrows(
DateTimeParseException.class,
() -> toIcebergRecord(Types.TimestampType.withoutZone(), val));
assertThrows(
DateTimeParseException.class,
() -> toIcebergRecord(Types.TimestampType.withZone(), val + "Z"));
}
}

@Test
public void testTimestampStringWithInvalidDate() {
for (String val :
Arrays.asList(
"2023-02-29T13:18:20",
"2023-02-29 13:18:20",
"2024-04-31T13:18:20",
"2024-04-31 13:18:20")) {
assertThrows(
DateTimeParseException.class,
() -> toIcebergRecord(Types.TimestampType.withoutZone(), val));
assertThrows(
DateTimeParseException.class,
() -> toIcebergRecord(Types.TimestampType.withZone(), val + "Z"));
}
}

@Test
Expand Down
Loading