gemini-code-assist[bot] commented on code in PR #36892:
URL: https://github.com/apache/beam/pull/36892#discussion_r2565686676
##########
sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java:
##########
@@ -1387,6 +1410,22 @@ private static Object convertLogicalType(
@Nonnull FieldType fieldType,
@Nonnull GenericData genericData) {
TypeWithNullability type = new TypeWithNullability(avroSchema);
+
+ // TODO: Remove this workaround once Avro is upgraded to 1.12+ where
timestamp-nanos
+ if (TIMESTAMP_NANOS_LOGICAL_TYPE.equals(type.type.getProp("logicalType")))
{
+ if (type.type.getType() == Type.LONG) {
+ Long nanos = (Long) value;
+ // Check if Beam expects Timestamp logical type
+ if (fieldType.getTypeName() == TypeName.LOGICAL_TYPE
+ &&
org.apache.beam.sdk.schemas.logicaltypes.Timestamp.IDENTIFIER.equals(
+ fieldType.getLogicalType().getIdentifier())) {
+ return java.time.Instant.ofEpochSecond(0L, nanos);
Review Comment:
While `Instant.ofEpochSecond(0L, nanos)` works due to the internal
adjustment logic of `Instant`, it's not very explicit and can be confusing for
future maintainers. A more readable and explicit approach would be to calculate
the seconds and nanoseconds-of-second parts separately, which also makes the
handling of negative timestamps clearer.
```java
long seconds = Math.floorDiv(nanos, 1_000_000_000L);
long nanoAdjustment = Math.floorMod(nanos, 1_000_000_000L);
return java.time.Instant.ofEpochSecond(seconds, nanoAdjustment);
```
##########
sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java:
##########
@@ -1186,6 +1196,10 @@ private static org.apache.avro.Schema getFieldSchema(
} else if (SqlTypes.TIMESTAMP.getIdentifier().equals(identifier)) {
baseType =
LogicalTypes.timestampMicros().addToSchema(org.apache.avro.Schema.create(Type.LONG));
+ } else if (Timestamp.IDENTIFIER.equals(identifier)) {
+ baseType =
+ new org.apache.avro.Schema.Parser()
+ .parse("{\"type\": \"long\", \"logicalType\":
\"timestamp-nanos\"}");
Review Comment:
Parsing a JSON string to create an Avro schema can be brittle and less
efficient. It's better to build the schema programmatically. You can create a
`long` schema and then add the `logicalType` property to it.
```java
baseType = org.apache.avro.Schema.create(Type.LONG);
baseType.addProp("logicalType", TIMESTAMP_NANOS_LOGICAL_TYPE);
```
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtils.java:
##########
@@ -161,36 +159,67 @@ static Schema getPrimitiveType(TableFieldSchema schema,
Boolean useAvroLogicalTy
* Formats BigQuery seconds-since-epoch into String matching JSON export.
Thread-safe and
* immutable.
*/
- private static final DateTimeFormatter DATE_AND_SECONDS_FORMATTER =
- DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss").withZoneUTC();
-
- @VisibleForTesting
- static String formatTimestamp(Long timestampMicro) {
- String dateTime = formatDatetime(timestampMicro);
- return dateTime + " UTC";
+ private static final java.time.format.DateTimeFormatter DATE_TIME_FORMATTER =
+ java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
+ .withZone(java.time.ZoneOffset.UTC);
+
+ /** Enum to define the precision of a timestamp since the epoch. */
+ enum TimestampPrecision {
+ MILLISECONDS,
+ MICROSECONDS,
+ NANOSECONDS;
+
+ /** Converts an epoch value of this precision to an Instant. */
+ java.time.Instant toInstant(long epochValue) {
+ switch (this) {
+ case MILLISECONDS:
+ return java.time.Instant.ofEpochMilli(epochValue);
+ case MICROSECONDS:
+ return java.time.Instant.ofEpochSecond(
+ epochValue / 1_000_000L, (epochValue % 1_000_000L) * 1_000L);
+ case NANOSECONDS:
+ return java.time.Instant.ofEpochSecond(
+ epochValue / 1_000_000_000L, epochValue % 1_000_000_000L);
Review Comment:
The use of `/` and `%` operators for negative `epochValue` can be tricky.
While `Instant.ofEpochSecond` adjusts for negative nano parts, using
`Math.floorDiv` and `Math.floorMod` is more explicit and safer for handling
negative timestamps, improving code readability and robustness.
```java
case MICROSECONDS:
{
long seconds = Math.floorDiv(epochValue, 1_000_000L);
long microsOfSecond = Math.floorMod(epochValue, 1_000_000L);
return java.time.Instant.ofEpochSecond(seconds, microsOfSecond *
1_000L);
}
case NANOSECONDS:
{
long seconds = Math.floorDiv(epochValue, 1_000_000_000L);
long nanosOfSecond = Math.floorMod(epochValue, 1_000_000_000L);
return java.time.Instant.ofEpochSecond(seconds, nanosOfSecond);
}
```
--
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]
