Skip to content
Merged
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@
private Map<String, String> renameColumns = new HashMap<String, String>();
private boolean hashRowId = false;

private static final Long DATETIME_POSITIVE_INFINITY = 9223372036825200000L;
private static final Long DATETIME_NEGATIVE_INFINITY = -9223372036832400000L;

private FormatDatastreamRecordToJson() {}

public static FormatDatastreamRecordToJson create() {
Expand Down Expand Up @@ -505,12 +508,19 @@
(ByteBuffer) element.get(fieldName), fieldSchema, fieldSchema.getLogicalType());
jsonObject.put(fieldName, bigDecimal.toPlainString());
} else if (fieldSchema.getLogicalType() instanceof LogicalTypes.TimeMicros) {
Long nanoseconds = (Long) element.get(fieldName) * TimeUnit.MICROSECONDS.toNanos(1);
Duration duration =
Duration.ofSeconds(
TimeUnit.NANOSECONDS.toSeconds(nanoseconds),
nanoseconds % TimeUnit.SECONDS.toNanos(1));
jsonObject.put(fieldName, duration.toString());
Long microseconds = (Long) element.get(fieldName);

Check warning on line 511 in v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java

View check run for this annotation

Codecov / codecov/patch

v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java#L511

Added line #L511 was not covered by tests
if (microseconds == DATETIME_POSITIVE_INFINITY) {
jsonObject.put(fieldName, "infinity");

Check warning on line 513 in v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java

View check run for this annotation

Codecov / codecov/patch

v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java#L513

Added line #L513 was not covered by tests
} else if (microseconds == DATETIME_NEGATIVE_INFINITY) {
jsonObject.put(fieldName, "-infinity");

Check warning on line 515 in v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java

View check run for this annotation

Codecov / codecov/patch

v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java#L515

Added line #L515 was not covered by tests
} else {
Long nanoseconds = microseconds * TimeUnit.MICROSECONDS.toNanos(1);
Duration duration =
Duration.ofSeconds(
TimeUnit.NANOSECONDS.toSeconds(nanoseconds),
nanoseconds % TimeUnit.SECONDS.toNanos(1));
jsonObject.put(fieldName, duration.toString());

Check warning on line 522 in v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java

View check run for this annotation

Codecov / codecov/patch

v2/datastream-common/src/main/java/com/google/cloud/teleport/v2/datastream/transforms/FormatDatastreamRecordToJson.java#L517-L522

Added lines #L517 - L522 were not covered by tests
}
} else if (fieldSchema.getLogicalType() instanceof LogicalTypes.TimeMillis) {
Duration duration = Duration.ofMillis(((Long) element.get(fieldName)));
jsonObject.put(fieldName, duration.toString());
Expand Down
Loading