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 e61dbe505e22..8a19672f9ea1 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 @@ -34,7 +34,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import javax.annotation.Nonnull; import net.bytebuddy.description.type.TypeDescription.ForLoadedType; @@ -77,7 +76,6 @@ 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; @@ -89,7 +87,6 @@ import org.apache.beam.sdk.transforms.SimpleFunction; import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.values.TypeDescriptor; -import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting; import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.CaseFormat; import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Strings; import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables; @@ -122,9 +119,7 @@ * RECORD <-----> ROW * UNION <-----> LogicalType(OneOfType) * LogicalTypes.Date <-----> LogicalType(DATE) - * <------ LogicalType(urn="beam:logical_type:date:v1") * LogicalTypes.TimestampMillis <-----> DATETIME - * LogicalTypes.TimestampMicros <-----> LogicalType(urn="beam:logical_type:micros_instant:v1") * LogicalTypes.Decimal <-----> DECIMAL * * @@ -810,11 +805,9 @@ private static Schema.FieldType toFieldType(TypeWithNullability type) { if (logicalType instanceof LogicalTypes.Decimal) { fieldType = FieldType.DECIMAL; } else if (logicalType instanceof LogicalTypes.TimestampMillis) { - // TODO(https://github.com/apache/beam/issues/19215) DATETIME primitive will be removed when - // fully migrates to java.time lib from joda-time + // TODO: There is a desire to move Beam schema DATETIME to a micros representation. When + // this is done, this logical type needs to be changed. fieldType = FieldType.DATETIME; - } else if (logicalType instanceof LogicalTypes.TimestampMicros) { - fieldType = FieldType.logicalType(SqlTypes.TIMESTAMP); } else if (logicalType instanceof LogicalTypes.Date) { fieldType = FieldType.DATETIME; } @@ -927,8 +920,8 @@ private static org.apache.avro.Schema getFieldSchema( break; case DATETIME: - // TODO(https://github.com/apache/beam/issues/19215) DATETIME primitive will be removed when - // fully migrates to java.time lib from joda-time + // TODO: There is a desire to move Beam schema DATETIME to a micros representation. When + // this is done, this logical type needs to be changed. baseType = LogicalTypes.timestampMillis().addToSchema(org.apache.avro.Schema.create(Type.LONG)); break; @@ -973,13 +966,10 @@ private static org.apache.avro.Schema getFieldSchema( oneOfType.getOneOfSchema().getFields().stream() .map(x -> getFieldSchema(x.getType(), x.getName(), namespace)) .collect(Collectors.toList())); - } else if ("DATE".equals(identifier) || SqlTypes.DATE.getIdentifier().equals(identifier)) { + } else if ("DATE".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)); - } else if (SqlTypes.TIMESTAMP.getIdentifier().equals(identifier)) { - baseType = - LogicalTypes.timestampMicros().addToSchema(org.apache.avro.Schema.create(Type.LONG)); } else { throw new RuntimeException( "Unhandled logical type " + fieldType.getLogicalType().getIdentifier()); @@ -1104,17 +1094,9 @@ private static org.apache.avro.Schema getFieldSchema( 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)) { - // "TIME" is backed by joda.time.Instant return (int) ((Instant) value).getMillis(); - } else if (SqlTypes.TIMESTAMP.getIdentifier().equals(identifier)) { - // portable SqlTypes.TIMESTAMP is backed by java.time.Instant - return getMicrosFromJavaInstant((java.time.Instant) value); } else { throw new RuntimeException("Unhandled logical type " + identifier); } @@ -1188,19 +1170,10 @@ private static org.apache.avro.Schema getFieldSchema( } else { return convertDateTimeStrict((Long) value, fieldType); } - } else if (logicalType instanceof LogicalTypes.TimestampMicros) { - if (value instanceof java.time.Instant) { - return convertMicroMillisStrict( - getMicrosFromJavaInstant((java.time.Instant) value), fieldType); - } else { - return convertMicroMillisStrict((Long) value, fieldType); - } } else if (logicalType instanceof LogicalTypes.Date) { 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); } @@ -1258,14 +1231,6 @@ private static org.apache.avro.Schema getFieldSchema( } } - /** Helper method to get epoch micros required by Avro TimeStampMicros logical type. */ - @SuppressWarnings("JavaInstantGetSecondsGetNano") - @VisibleForTesting - static long getMicrosFromJavaInstant(java.time.Instant value) { - return TimeUnit.SECONDS.toMicros(value.getEpochSecond()) - + TimeUnit.NANOSECONDS.toMicros(value.getNano()); - } - private static Object convertRecordStrict(GenericRecord record, Schema.FieldType fieldType) { checkTypeName(fieldType.getTypeName(), Schema.TypeName.ROW, "record"); return toBeamRowStrict(record, fieldType.getRowSchema()); @@ -1315,17 +1280,6 @@ private static Object convertDateTimeStrict(Long value, Schema.FieldType fieldTy return new Instant(value); } - private static Object convertMicroMillisStrict(Long value, Schema.FieldType fieldType) { - checkTypeName( - fieldType.getTypeName(), TypeName.LOGICAL_TYPE, SqlTypes.TIMESTAMP.getIdentifier()); - checkArgument( - fieldType.getLogicalType().getIdentifier().equals(SqlTypes.TIMESTAMP.getIdentifier())); - - return java.time.Instant.ofEpochSecond( - TimeUnit.MICROSECONDS.toSeconds(value), - TimeUnit.MICROSECONDS.toNanos(Math.floorMod(value, TimeUnit.SECONDS.toMicros(1)))); - } - private static Object convertFloatStrict(Float value, Schema.FieldType fieldType) { checkTypeName(fieldType.getTypeName(), Schema.TypeName.FLOAT, "float"); return value; 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 0605aa6fd1cc..3087959c1e02 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 @@ -29,7 +29,6 @@ import java.sql.JDBCType; import java.util.List; import java.util.Map; -import java.util.concurrent.TimeUnit; import org.apache.avro.Conversions; import org.apache.avro.LogicalType; import org.apache.avro.LogicalTypes; @@ -47,7 +46,6 @@ 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; @@ -231,12 +229,6 @@ private static org.apache.avro.Schema getAvroSchema() { LogicalTypes.timestampMillis().addToSchema(org.apache.avro.Schema.create(Type.LONG)), "", (Object) null)); - fields.add( - new org.apache.avro.Schema.Field( - "timestampMicros", - LogicalTypes.timestampMicros().addToSchema(org.apache.avro.Schema.create(Type.LONG)), - "", - (Object) null)); fields.add(new org.apache.avro.Schema.Field("row", getAvroSubSchema("row"), "", (Object) null)); fields.add( new org.apache.avro.Schema.Field( @@ -269,7 +261,6 @@ private Schema getBeamSchema() { .addField(Field.of("bytes", FieldType.BYTES)) .addField(Field.of("decimal", FieldType.DECIMAL)) .addField(Field.of("timestampMillis", FieldType.DATETIME)) - .addField(Field.of("timestampMicros", FieldType.logicalType(SqlTypes.TIMESTAMP))) .addField(Field.of("row", FieldType.row(subSchema))) .addField(Field.of("array", FieldType.array(FieldType.row(subSchema)))) .addField(Field.of("map", FieldType.map(FieldType.STRING, FieldType.row(subSchema)))) @@ -279,9 +270,6 @@ private Schema getBeamSchema() { private static final byte[] BYTE_ARRAY = new byte[] {1, 2, 3, 4}; private static final DateTime DATE_TIME = new DateTime().withDate(1979, 3, 14).withTime(1, 2, 3, 4).withZone(DateTimeZone.UTC); - private static final java.time.Instant MICROS_INSTANT = - java.time.Instant.ofEpochMilli(DATE_TIME.getMillis()) - .plusNanos(TimeUnit.MICROSECONDS.toNanos(123)); private static final BigDecimal BIG_DECIMAL = new BigDecimal(3600); private Row getBeamRow() { @@ -296,7 +284,6 @@ private Row getBeamRow() { .addValue(BYTE_ARRAY) .addValue(BIG_DECIMAL) .addValue(DATE_TIME) - .addValue(MICROS_INSTANT) .addValue(subRow) .addValue(ImmutableList.of(subRow, subRow)) .addValue(ImmutableMap.of("k1", subRow, "k2", subRow)) @@ -329,7 +316,6 @@ private static GenericRecord getGenericRecord() { .set("bytes", ByteBuffer.wrap(BYTE_ARRAY)) .set("decimal", encodedDecimal) .set("timestampMillis", DATE_TIME.getMillis()) - .set("timestampMicros", AvroUtils.getMicrosFromJavaInstant(MICROS_INSTANT)) .set("row", getSubGenericRecord("row")) .set("array", ImmutableList.of(getSubGenericRecord("array"), getSubGenericRecord("array"))) .set( @@ -728,25 +714,6 @@ public void testJdbcLogicalDateAndTimeRowDataToGenericRecord() { assertEquals(expectedRecord, AvroUtils.toGenericRecord(rowData, avroSchema)); } - @Test - public void testSqlTypesToGenericRecord() { - // SqlTypes to LogicalTypes.date conversion are 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);