You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2022/12/22 03:13:47 UTC
[beam] branch master updated: Support SqlTypes Date in AvroUtils (#24756)
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new b757ad035b9 Support SqlTypes Date in AvroUtils (#24756)
b757ad035b9 is described below
commit b757ad035b9f30d6e9ee8a77ffee1683e74712c9
Author: Yi Hu <ya...@google.com>
AuthorDate: Wed Dec 21 22:13:40 2022 -0500
Support SqlTypes Date in AvroUtils (#24756)
---
.../org/apache/beam/sdk/schemas/utils/AvroUtils.java | 10 +++++++++-
.../apache/beam/sdk/schemas/utils/AvroUtilsTest.java | 20 ++++++++++++++++++++
2 files changed, 29 insertions(+), 1 deletion(-)
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java
index 8a19672f9ea..8aad6348fba 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java
@@ -76,6 +76,7 @@ import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes;
import org.apache.beam.sdk.schemas.logicaltypes.FixedString;
import org.apache.beam.sdk.schemas.logicaltypes.OneOfType;
+import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
import org.apache.beam.sdk.schemas.logicaltypes.VariableBytes;
import org.apache.beam.sdk.schemas.logicaltypes.VariableString;
import org.apache.beam.sdk.schemas.utils.ByteBuddyUtils.ConvertType;
@@ -119,6 +120,7 @@ import org.joda.time.ReadableInstant;
* RECORD <-----> ROW
* UNION <-----> LogicalType(OneOfType)
* LogicalTypes.Date <-----> LogicalType(DATE)
+ * <------ LogicalType(urn="beam:logical_type:date:v1")
* LogicalTypes.TimestampMillis <-----> DATETIME
* LogicalTypes.Decimal <-----> DECIMAL
* </pre>
@@ -966,7 +968,7 @@ public class AvroUtils {
oneOfType.getOneOfSchema().getFields().stream()
.map(x -> getFieldSchema(x.getType(), x.getName(), namespace))
.collect(Collectors.toList()));
- } else if ("DATE".equals(identifier)) {
+ } else if ("DATE".equals(identifier) || SqlTypes.DATE.getIdentifier().equals(identifier)) {
baseType = LogicalTypes.date().addToSchema(org.apache.avro.Schema.create(Type.INT));
} else if ("TIME".equals(identifier)) {
baseType = LogicalTypes.timeMillis().addToSchema(org.apache.avro.Schema.create(Type.INT));
@@ -1094,7 +1096,11 @@ public class AvroUtils {
oneOfValue.getValue());
}
} else if ("DATE".equals(identifier)) {
+ // "Date" is backed by joda.time.Instant
return Days.daysBetween(Instant.EPOCH, (Instant) value).getDays();
+ } else if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
+ // portable SqlTypes.DATE is backed by java.time.LocalDate
+ return ((java.time.LocalDate) value).toEpochDay();
} else if ("TIME".equals(identifier)) {
return (int) ((Instant) value).getMillis();
} else {
@@ -1174,6 +1180,8 @@ public class AvroUtils {
if (value instanceof ReadableInstant) {
int epochDays = Days.daysBetween(Instant.EPOCH, (ReadableInstant) value).getDays();
return convertDateStrict(epochDays, fieldType);
+ } else if (value instanceof java.time.LocalDate) {
+ return convertDateStrict((int) ((java.time.LocalDate) value).toEpochDay(), fieldType);
} else {
return convertDateStrict((Integer) value, fieldType);
}
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/AvroUtilsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/AvroUtilsTest.java
index 3087959c1e0..5c115aa89bc 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/AvroUtilsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/AvroUtilsTest.java
@@ -46,6 +46,7 @@ import org.apache.beam.sdk.schemas.Schema.Field;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
import org.apache.beam.sdk.schemas.logicaltypes.OneOfType;
+import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
import org.apache.beam.sdk.schemas.utils.AvroGenerators.RecordSchemaGenerator;
import org.apache.beam.sdk.schemas.utils.AvroUtils.TypeWithNullability;
import org.apache.beam.sdk.testing.CoderProperties;
@@ -714,6 +715,25 @@ public class AvroUtilsTest {
assertEquals(expectedRecord, AvroUtils.toGenericRecord(rowData, avroSchema));
}
+ @Test
+ public void testSqlTypesToGenericRecord() {
+ // SqlTypes to LogicalTypes.date conversion is one direction
+ java.time.LocalDate localDate = java.time.LocalDate.of(1979, 3, 14);
+
+ Schema beamSchema =
+ Schema.builder()
+ .addField(Field.of("local_date", FieldType.logicalType(SqlTypes.DATE)))
+ .build();
+
+ Row rowData = Row.withSchema(beamSchema).addValue(localDate).build();
+
+ org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(beamSchema);
+ GenericRecord expectedRecord =
+ new GenericRecordBuilder(avroSchema).set("local_date", localDate.toEpochDay()).build();
+
+ assertEquals(expectedRecord, AvroUtils.toGenericRecord(rowData, avroSchema));
+ }
+
@Test
public void testBeamRowToGenericRecord() {
GenericRecord genericRecord = AvroUtils.toGenericRecord(getBeamRow(), null);