Skip to content

Commit

Permalink
address comments
Browse files Browse the repository at this point in the history
  • Loading branch information
@reuvenlax
reuvenlax committed Apr 29, 2022
1 parent 8d56420 commit 8195ce6
Show file tree
Hide file tree
Showing 2 changed files with 99 additions and 10 deletions.
18 changes: 12 additions & 6 deletions ...platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -400,24 +400,30 @@ static Object singularFieldToProtoValue(
org.joda.time.Instant.EPOCH.toDateTime().toLocalDate(),
(org.joda.time.LocalDate) value)
.getDays();
} else if (value instanceof Integer || value instanceof Long) {
return ((Number) value).intValue();
}
break;
case "NUMERIC":
if (value instanceof String) {
return BigDecimalByteStringEncoder.encodeToNumericByteString(new BigDecimal((String)value));
return BigDecimalByteStringEncoder.encodeToNumericByteString(
new BigDecimal((String) value));
} else if (value instanceof BigDecimal) {
return BigDecimalByteStringEncoder.encodeToNumericByteString(((BigDecimal) value));
} else if (value instanceof Double || value instanceof Float) {
return BigDecimalByteStringEncoder.encodeToNumericByteString(BigDecimal.valueOf(((Number) value).doubleValue()));
return BigDecimalByteStringEncoder.encodeToNumericByteString(
BigDecimal.valueOf(((Number) value).doubleValue()));
}
break;
case "BIGNUMERIC":
if (value instanceof String) {
return BigDecimalByteStringEncoder.encodeToBigNumericByteString(new BigDecimal((String)value));
return BigDecimalByteStringEncoder.encodeToBigNumericByteString(
new BigDecimal((String) value));
} else if (value instanceof BigDecimal) {
return BigDecimalByteStringEncoder.encodeToBigNumericByteString(((BigDecimal) value));
} else if (value instanceof Double || value instanceof Float) {
return BigDecimalByteStringEncoder.encodeToBigNumericByteString(BigDecimal.valueOf(((Number) value).doubleValue()));
return BigDecimalByteStringEncoder.encodeToBigNumericByteString(
BigDecimal.valueOf(((Number) value).doubleValue()));
}
break;
case "DATETIME":
Expand All @@ -426,9 +432,9 @@ static Object singularFieldToProtoValue(
} else if (value instanceof Number) {
return ((Number) value).longValue();
} else if (value instanceof LocalDateTime) {
return CivilTimeEncoder.encodePacked64DatetimeMicros((LocalDateTime)value);
return CivilTimeEncoder.encodePacked64DatetimeMicros((LocalDateTime) value);
} else if (value instanceof org.joda.time.LocalDateTime) {
return CivilTimeEncoder.encodePacked64DatetimeMicros((org.joda.time.LocalDateTime)value);
return CivilTimeEncoder.encodePacked64DatetimeMicros((org.joda.time.LocalDateTime) value);
}
break;
case "TIME":
Expand Down
91 changes: 87 additions & 4 deletions ...atform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProtoIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ public class TableRowToStorageApiProtoIT {
.set("floatValue", 2.817f)
.set("boolValue", true)
.set("booleanValue", true)
.set("timestampValue", org.joda.time.Instant.parse("1970-01-01T00:00:00.000043Z"))
.set("timestampValue", org.joda.time.Instant.parse("1970-01-01T00:00:00.0043Z"))
.set("timeValue", org.joda.time.LocalTime.parse("00:52:07.123456"))
.set("datetimeValue", org.joda.time.LocalDateTime.parse("2019-08-16T00:52:07.123456"))
.set("dateValue", org.joda.time.LocalDate.parse("2019-08-16"))
Expand All @@ -147,6 +147,42 @@ public class TableRowToStorageApiProtoIT {
.set("bigNumericValue", "23334.4")
.set("arrayValue", Arrays.asList("hello", "goodbye"));

private static final TableRow BASE_TABLE_ROW_NUM_TIME =
new TableRow()
.set("stringValue", "string")
.set("bytesValue", "string".getBytes(StandardCharsets.UTF_8))
.set("int64Value", 42)
.set("intValue", 43)
.set("float64Value", 2.8168f)
.set("floatValue", 2.817f)
.set("boolValue", true)
.set("booleanValue", true)
.set("timestampValue", 43)
.set("timeValue", 3497124416L)
.set("datetimeValue", 142111881387172416L)
.set("dateValue", 18124)
.set("numericValue", new BigDecimal("23.4"))
.set("bigNumericValue", "23334.4")
.set("arrayValue", ImmutableList.of("hello", "goodbye"));

private static final TableRow BASE_TABLE_ROW_FLOATS =
new TableRow()
.set("stringValue", "string")
.set("bytesValue", "string".getBytes(StandardCharsets.UTF_8))
.set("int64Value", 42)
.set("intValue", 43)
.set("float64Value", 2.8168f)
.set("floatValue", 2.817f)
.set("boolValue", true)
.set("booleanValue", true)
.set("timestampValue", 43)
.set("timeValue", 3497124416L)
.set("datetimeValue", 142111881387172416D)
.set("dateValue", 18124)
.set("numericValue", 23.4)
.set("bigNumericValue", "23334.4")
.set("arrayValue", ImmutableList.of("hello", "goodbye"));

private static final TableRow BASE_TABLE_ROW_NULL =
new TableRow()
// .set("stringValue", null) // do not set stringValue, this should work
Expand Down Expand Up @@ -195,14 +231,52 @@ public class TableRowToStorageApiProtoIT {
.set("floatValue", 2.817)
.set("boolValue", true)
.set("booleanValue", true)
.set("timestampValue", "0.0")
.set("timestampValue", "0.004")
.set("timeValue", "00:52:07.123000")
.set("datetimeValue", "2019-08-16T00:52:07.123000")
.set("dateValue", "2019-08-16")
.set("numericValue", "23.4")
.set("bigNumericValue", "23334.4")
.set("arrayValue", ImmutableList.of("hello", "goodbye"));

private static final TableRow BASE_TABLE_ROW_NUM_EXPECTED =
new TableRow()
.set("stringValue", "string")
.set(
"bytesValue", BaseEncoding.base64().encode("string".getBytes(StandardCharsets.UTF_8)))
.set("int64Value", "42")
.set("intValue", "43")
.set("float64Value", 2.8168)
.set("floatValue", 2.817)
.set("boolValue", true)
.set("booleanValue", true)
.set("timestampValue", "4.3E-5")
.set("timeValue", "00:52:07.123456")
.set("datetimeValue", "2019-08-16T00:52:07.123456")
.set("dateValue", "2019-08-16")
.set("numericValue", "23.4")
.set("bigNumericValue", "23334.4")
.set("arrayValue", ImmutableList.of("hello", "goodbye"));

private static final TableRow BASE_TABLE_ROW_FLOATS_EXPECTED =
new TableRow()
.set("stringValue", "string")
.set(
"bytesValue", BaseEncoding.base64().encode("string".getBytes(StandardCharsets.UTF_8)))
.set("int64Value", "42")
.set("intValue", "43")
.set("float64Value", 2.8168)
.set("floatValue", 2.817)
.set("boolValue", true)
.set("booleanValue", true)
.set("timestampValue", "4.3E-5")
.set("timeValue", "00:52:07.123456")
.set("datetimeValue", "2019-08-16T00:52:07.123456")
.set("dateValue", "2019-08-16")
.set("numericValue", "23.4")
.set("bigNumericValue", "23334.4")
.set("arrayValue", ImmutableList.of("hello", "goodbye"));

// only nonnull values are returned, null in arrayValue should be converted to empty list
private static final TableRow BASE_TABLE_ROW_NULL_EXPECTED =
new TableRow().set("arrayValue", ImmutableList.of());
Expand Down Expand Up @@ -265,7 +339,11 @@ public void testNestedRichTypesAndNull() throws IOException, InterruptedExceptio
.set(
"nestedValue2",
Arrays.asList(
BASE_TABLE_ROW_JAVA_TIME, BASE_TABLE_ROW_JODA_TIME, BASE_TABLE_ROW_NULL))
BASE_TABLE_ROW_JAVA_TIME,
BASE_TABLE_ROW_JODA_TIME,
BASE_TABLE_ROW_NUM_TIME,
BASE_TABLE_ROW_FLOATS,
BASE_TABLE_ROW_NULL))
.set("nestedValue3", null);

runPipeline(tableSpec, Collections.singleton(tableRow));
Expand All @@ -275,9 +353,14 @@ public void testNestedRichTypesAndNull() throws IOException, InterruptedExceptio

assertEquals(1, actualTableRows.size());
assertEquals(BASE_TABLE_ROW_EXPECTED, actualTableRows.get(0).get("nestedValue1"));
LOG.info("ACTUAL " + actualTableRows.get(0).get("nestedValue2"));
assertEquals(
ImmutableList.of(
BASE_TABLE_ROW_EXPECTED, BASE_TABLE_ROW_JODA_EXPECTED, BASE_TABLE_ROW_NULL_EXPECTED),
BASE_TABLE_ROW_EXPECTED,
BASE_TABLE_ROW_JODA_EXPECTED,
BASE_TABLE_ROW_NUM_EXPECTED,
BASE_TABLE_ROW_FLOATS_EXPECTED,
BASE_TABLE_ROW_NULL_EXPECTED),
actualTableRows.get(0).get("nestedValue2"));
assertNull(actualTableRows.get(0).get("nestedValue3"));
}
Expand Down

0 comments on commit 8195ce6

Please sign in to comment.