This is an automated email from the ASF dual-hosted git repository. yhu pushed a commit to branch revert-23969-sqlttypes in repository https://gitbox.apache.org/repos/asf/beam.git
commit e48b329e7a7e868ed38f7e8e0973957d81ed8678 Author: Yi Hu <[email protected]> AuthorDate: Fri Dec 2 08:42:34 2022 -0500 Revert "Support SqlTypes Date and Timestamp (MicrosInstant) in AvroUtils (#23969)" This reverts commit d5d76b974592d45de368ab641647ca5cc4ec12ec. --- .../apache/beam/sdk/schemas/utils/AvroUtils.java | 91 ++-------------------- .../beam/sdk/schemas/utils/AvroUtilsTest.java | 33 -------- 2 files changed, 6 insertions(+), 118 deletions(-) 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 e61dbe505e2..8d01ed0406a 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.HashMap; 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.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; @@ -89,7 +87,6 @@ import org.apache.beam.sdk.transforms.SerializableFunction; 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; @@ -101,42 +98,7 @@ import org.joda.time.Duration; import org.joda.time.Instant; import org.joda.time.ReadableInstant; -/** - * Utils to convert AVRO records to Beam rows. Imposes a mapping between common avro types and Beam - * portable schemas (https://s.apache.org/beam-schemas): - * - * <pre> - * Avro Beam Field Type - * INT <-----> INT32 - * LONG <-----> INT64 - * FLOAT <-----> FLOAT - * DOUBLE <-----> DOUBLE - * BOOLEAN <-----> BOOLEAN - * STRING <-----> STRING - * BYTES <-----> BYTES - * <------ LogicalType(urn="beam:logical_type:var_bytes:v1") - * FIXED <-----> LogicalType(urn="beam:logical_type:fixed_bytes:v1") - * ARRAY <-----> ARRAY - * ENUM <-----> LogicalType(EnumerationType) - * MAP <-----> MAP - * 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 - * </pre> - * - * For SQL CHAR/VARCHAR types, an Avro schema - * - * <pre> - * LogicalType({"type":"string","logicalType":"char","maxLength":MAX_LENGTH}) or - * LogicalType({"type":"string","logicalType":"varchar","maxLength":MAX_LENGTH}) - * </pre> - * - * is used. - */ +/** Utils to convert AVRO records to Beam rows. */ @Experimental(Kind.SCHEMAS) @SuppressWarnings({ "nullness", // TODO(https://github.com/apache/beam/issues/20497) @@ -810,11 +772,9 @@ public class AvroUtils { 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 +887,8 @@ public class AvroUtils { 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 +933,10 @@ public class AvroUtils { 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 +1061,9 @@ 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)) { - // "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 +1137,10 @@ public class AvroUtils { } 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 +1198,6 @@ public class AvroUtils { } } - /** 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 +1247,6 @@ public class AvroUtils { 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 0605aa6fd1c..3087959c1e0 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.nio.ByteBuffer; 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.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; @@ -231,12 +229,6 @@ public class AvroUtilsTest { 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 @@ public class AvroUtilsTest { .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 @@ public class AvroUtilsTest { 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 @@ public class AvroUtilsTest { .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 @@ public class AvroUtilsTest { .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 class AvroUtilsTest { 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);
