This is an automated email from the ASF dual-hosted git repository.
stankiewicz 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 4fbc6a12113 Fix nullness in extensions/avro (#39976)
4fbc6a12113 is described below
commit 4fbc6a121135e25303edea2b690cd3d6c116a615
Author: Kenneth Knowles <[email protected]>
AuthorDate: Fri Sep 25 06:01:34 2026 -0400
Fix nullness in extensions/avro (#39976)
---
.../beam/sdk/extensions/avro/coders/AvroCoder.java | 45 ++++---
.../sdk/extensions/avro/io/AvroDatumFactory.java | 12 +-
.../apache/beam/sdk/extensions/avro/io/AvroIO.java | 133 ++++++++++++---------
.../extensions/avro/io/AvroSchemaIOProvider.java | 18 ++-
.../beam/sdk/extensions/avro/io/AvroSink.java | 26 ++--
.../beam/sdk/extensions/avro/io/AvroSource.java | 87 +++++++++-----
.../avro/io/ConstantAvroDestination.java | 5 +-
.../avro/io/SerializableAvroCodecFactory.java | 24 ++--
.../avro/schemas/utils/AvroByteBuddyUtils.java | 5 +-
.../extensions/avro/schemas/utils/AvroUtils.java | 37 +++---
.../avro/schemas/utils/AvroUtilsTest.java | 83 ++++++++++++-
11 files changed, 308 insertions(+), 167 deletions(-)
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java
index f8cc1a4074c..f630e741387 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java
@@ -17,6 +17,8 @@
*/
package org.apache.beam.sdk.extensions.avro.coders;
+import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
import com.google.errorprone.annotations.FormatMethod;
import com.google.errorprone.annotations.FormatString;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -109,9 +111,6 @@ import org.checkerframework.checker.nullness.qual.Nullable;
*
* @param <T> the type of elements handled by this coder
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
public class AvroCoder<T> extends CustomCoder<T> {
private static final Cache<AvroCoderCacheKey, AvroCoder<?>> AVRO_CODER_CACHE
=
@@ -137,7 +136,12 @@ public class AvroCoder<T> extends CustomCoder<T> {
* suite for encoding and decoding.
*/
public static <T> AvroCoder<T> specific(Class<T> type) {
- return specific(type, new
SpecificData(type.getClassLoader()).getSchema(type));
+ return specific(type, specificSchemaOf(type));
+ }
+
+ @SuppressWarnings("nullness") // SpecificData tolerates a null class loader
but is unannotated
+ private static Schema specificSchemaOf(Class<?> type) {
+ return new SpecificData(type.getClassLoader()).getSchema(type);
}
/**
@@ -167,7 +171,12 @@ public class AvroCoder<T> extends CustomCoder<T> {
* suite for encoding and decoding.
*/
public static <T> AvroCoder<T> reflect(Class<T> type) {
- return reflect(type, new
ReflectData(type.getClassLoader()).getSchema(type));
+ return reflect(type, reflectSchemaOf(type));
+ }
+
+ @SuppressWarnings("nullness") // ReflectData tolerates a null class loader
but is unannotated
+ private static Schema reflectSchemaOf(Class<?> type) {
+ return new ReflectData(type.getClassLoader()).getSchema(type);
}
/**
@@ -395,10 +404,10 @@ public class AvroCoder<T> extends CustomCoder<T> {
// writer and reader are unused but kept for serialization update
compatibility.
@SuppressWarnings("unused")
- private final EmptyOnDeserializationThreadLocal<DatumWriter<T>> writer =
null;
+ private final @Nullable EmptyOnDeserializationThreadLocal<DatumWriter<T>>
writer = null;
@SuppressWarnings("unused")
- private final EmptyOnDeserializationThreadLocal<DatumReader<T>> reader =
null;
+ private final @Nullable EmptyOnDeserializationThreadLocal<DatumReader<T>>
reader = null;
// datumReader and datumWriter are initialized in the constructor and
// on deserialization (see readObject).
@@ -424,7 +433,8 @@ public class AvroCoder<T> extends CustomCoder<T> {
this.decoder = new EmptyOnDeserializationThreadLocal<>();
this.encoder = new EmptyOnDeserializationThreadLocal<>();
- initializeAvroDatumReaderAndWriter();
+ this.datumReader = datumFactory.apply(schema, schema);
+ this.datumWriter = datumFactory.apply(schema);
}
/** Returns the type this coder encodes/decodes. */
@@ -473,6 +483,11 @@ public class AvroCoder<T> extends CustomCoder<T> {
BinaryDecoder decoderInstance =
DECODER_FACTORY.directBinaryDecoder(inStream, decoder.get());
// Save the potentially-new instance for later.
decoder.set(decoderInstance);
+ return readWithoutReuse(decoderInstance);
+ }
+
+ @SuppressWarnings("nullness") // DatumReader.read accepts a null reuse but
is unannotated
+ private T readWithoutReuse(BinaryDecoder decoderInstance) throws IOException
{
return datumReader.read(null, decoderInstance);
}
@@ -808,10 +823,10 @@ public class AvroCoder<T> extends CustomCoder<T> {
}
private void checkArray(String context, TypeDescriptor<?> type, Schema
schema) {
- TypeDescriptor<?> elementType = null;
+ TypeDescriptor<?> elementType;
if (type.isArray()) {
// The type is an array (with ordering)-> deterministic iff the
element is deterministic.
- elementType = type.getComponentType();
+ elementType = checkNotNull(type.getComponentType());
} else if (isSubtypeOf(type, Collection.class)) {
if (isSubtypeOf(type, List.class, SortedSet.class)) {
// Ordered collection -> deterministic iff the element is
deterministic
@@ -895,16 +910,12 @@ public class AvroCoder<T> extends CustomCoder<T> {
this.datumReader = cachedCoder.get().datumReader;
this.datumWriter = cachedCoder.get().datumWriter;
} else {
- initializeAvroDatumReaderAndWriter();
+ Schema schema = this.schemaSupplier.get();
+ this.datumReader = this.datumFactory.apply(schema, schema);
+ this.datumWriter = this.datumFactory.apply(schema);
}
}
- private void initializeAvroDatumReaderAndWriter() {
- this.datumReader =
- this.datumFactory.apply(this.schemaSupplier.get(),
this.schemaSupplier.get());
- this.datumWriter = this.datumFactory.apply(this.schemaSupplier.get());
- }
-
enum AvroCoderType {
SPECIFIC,
REFLECT;
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java
index 7d2fd43d7ab..1f010fa4a9b 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java
@@ -34,9 +34,6 @@ import
org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
import org.checkerframework.checker.nullness.qual.Nullable;
/** Create {@link DatumReader} and {@link DatumWriter} for given schemas. */
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
public abstract class AvroDatumFactory<T>
implements AvroSource.DatumReaderFactory<T>,
AvroSink.DatumWriterFactory<T> {
@@ -172,14 +169,14 @@ public abstract class AvroDatumFactory<T>
@Override
public DatumReader<T> apply(Schema writer, Schema reader) {
- ReflectData data = new ReflectData(type.getClassLoader());
+ ReflectData data = newReflectData(type);
AvroUtils.addLogicalTypeConversions(data);
return new ReflectDatumReader<>(writer, reader, data);
}
@Override
public DatumWriter<T> apply(Schema writer) {
- ReflectData data = new ReflectData(type.getClassLoader());
+ ReflectData data = newReflectData(type);
AvroUtils.addLogicalTypeConversions(data);
return new ReflectDatumWriter<>(writer, data);
}
@@ -187,5 +184,10 @@ public abstract class AvroDatumFactory<T>
public static <T> ReflectDatumFactory<T> of(Class<T> type) {
return new ReflectDatumFactory<>(type);
}
+
+ @SuppressWarnings("nullness") // ReflectData tolerates a null class loader
but is unannotated
+ private static ReflectData newReflectData(Class<?> type) {
+ return new ReflectData(type.getClassLoader());
+ }
}
}
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java
index 6b23695c21a..33b78abad4f 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java
@@ -19,8 +19,8 @@ package org.apache.beam.sdk.extensions.avro.io;
import static org.apache.beam.sdk.io.FileIO.ReadMatches.DirectoryTreatment;
import static
org.apache.beam.sdk.io.ReadAllViaFileBasedSource.ReadFileRangesFnExceptionHandler;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
-import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
import com.google.auto.value.AutoValue;
import java.io.IOException;
@@ -344,9 +344,6 @@ import org.joda.time.Duration;
* TypedWrite#withBadRecordErrorHandler(ErrorHandler)}. See documentation in
{@link FileIO} for
* details on usage
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
public class AvroIO {
/**
* Reads records of the given type from an Avro file (or multiple Avro files
matching a pattern).
@@ -619,7 +616,7 @@ public class AvroIO {
}
private static <T> PCollection<T> setBeamSchema(
- PCollection<T> pc, Class<T> clazz, @Nullable Schema schema) {
+ PCollection<T> pc, Class<T> clazz, Schema schema) {
return pc.setCoder(AvroUtils.schemaCoder(clazz, schema));
}
@@ -637,9 +634,9 @@ public class AvroIO {
abstract MatchConfiguration getMatchConfiguration();
- abstract @Nullable Class<T> getRecordClass();
+ abstract Class<T> getRecordClass();
- abstract @Nullable Schema getSchema();
+ abstract Schema getSchema();
abstract boolean getInferBeamSchema();
@@ -761,8 +758,7 @@ public class AvroIO {
@Override
@SuppressWarnings("unchecked")
public PCollection<T> expand(PBegin input) {
- checkNotNull(getFilepattern(), "filepattern");
- checkNotNull(getSchema(), "schema");
+ ValueProvider<String> filepattern = checkStateNotNull(getFilepattern(),
"filepattern");
if (getMatchConfiguration().getWatchInterval() == null &&
!getHintMatchesManyFiles()) {
PCollection<T> read =
@@ -770,7 +766,7 @@ public class AvroIO {
"Read",
org.apache.beam.sdk.io.Read.from(
createSource(
- getFilepattern(),
+ filepattern,
getMatchConfiguration().getEmptyMatchTreatment(),
getRecordClass(),
getSchema(),
@@ -784,15 +780,21 @@ public class AvroIO {
(getRecordClass() == GenericRecord.class)
? (ReadFiles<T>) readFilesGenericRecords(getSchema())
: readFiles(getRecordClass());
+ AvroSource.DatumReaderFactory<T> readerFactory = getDatumReaderFactory();
+ if (readerFactory != null) {
+ readFiles = readFiles.withDatumReaderFactory(readerFactory);
+ }
+ Coder<T> coder = getCoder();
+ if (coder != null) {
+ readFiles = readFiles.withCoder(coder);
+ }
return input
- .apply("Create filepattern", Create.ofProvider(getFilepattern(),
StringUtf8Coder.of()))
+ .apply("Create filepattern", Create.ofProvider(filepattern,
StringUtf8Coder.of()))
.apply("Match All",
FileIO.matchAll().withConfiguration(getMatchConfiguration()))
.apply(
"Read Matches",
FileIO.readMatches().withDirectoryTreatment(DirectoryTreatment.PROHIBIT))
- .apply(
- "Via ReadFiles",
-
readFiles.withDatumReaderFactory(getDatumReaderFactory()).withCoder(getCoder()));
+ .apply("Via ReadFiles", readFiles);
}
@Override
@@ -842,9 +844,9 @@ public class AvroIO {
public abstract static class ReadFiles<T>
extends PTransform<PCollection<ReadableFile>, PCollection<T>> {
- abstract @Nullable Class<T> getRecordClass();
+ abstract Class<T> getRecordClass();
- abstract @Nullable Schema getSchema();
+ abstract Schema getSchema();
abstract boolean getUsesReshuffle();
@@ -924,7 +926,6 @@ public class AvroIO {
@Override
public PCollection<T> expand(PCollection<ReadableFile> input) {
- checkNotNull(getSchema(), "schema");
Coder<T> coder =
Optional.ofNullable(getCoder()).orElse(AvroCoder.of(getRecordClass(),
getSchema()));
PCollection<T> read =
@@ -965,9 +966,9 @@ public class AvroIO {
public abstract static class ReadAll<T> extends
PTransform<PCollection<String>, PCollection<T>> {
abstract MatchConfiguration getMatchConfiguration();
- abstract @Nullable Class<T> getRecordClass();
+ abstract Class<T> getRecordClass();
- abstract @Nullable Schema getSchema();
+ abstract Schema getSchema();
abstract long getDesiredBundleSizeBytes();
@@ -1025,7 +1026,6 @@ public class AvroIO {
@Override
public PCollection<T> expand(PCollection<String> input) {
- checkNotNull(getSchema(), "schema");
PCollection<T> read =
input
.apply(FileIO.matchAll().withConfiguration(getMatchConfiguration()))
@@ -1052,13 +1052,13 @@ public class AvroIO {
private final Class<T> recordClass;
private final Supplier<Schema> schemaSupplier;
private final Coder<T> coder;
- private final AvroSource.DatumReaderFactory<T> readerFactory;
+ private final AvroSource.@Nullable DatumReaderFactory<T> readerFactory;
CreateSourceFn(
Class<T> recordClass,
String jsonSchema,
Coder<T> coder,
- AvroSource.DatumReaderFactory<T> readerFactory) {
+ AvroSource.@Nullable DatumReaderFactory<T> readerFactory) {
this.recordClass = recordClass;
this.schemaSupplier =
Suppliers.memoize(
@@ -1158,18 +1158,18 @@ public class AvroIO {
@Override
public PCollection<T> expand(PBegin input) {
- checkNotNull(getFilepattern(), "filepattern");
+ ValueProvider<String> filepattern = checkStateNotNull(getFilepattern(),
"filepattern");
Coder<T> coder = inferCoder(getCoder(), getParseFn(),
input.getPipeline().getCoderRegistry());
if (getMatchConfiguration().getWatchInterval() == null &&
!getHintMatchesManyFiles()) {
return input.apply(
org.apache.beam.sdk.io.Read.from(
- AvroSource.from(getFilepattern()).withParseFn(getParseFn(),
coder)));
+ AvroSource.from(filepattern).withParseFn(getParseFn(),
coder)));
}
// All other cases go through FileIO + ParseFilesGenericRecords.
return input
- .apply("Create filepattern", Create.ofProvider(getFilepattern(),
StringUtf8Coder.of()))
+ .apply("Create filepattern", Create.ofProvider(filepattern,
StringUtf8Coder.of()))
.apply("Match All",
FileIO.matchAll().withConfiguration(getMatchConfiguration()))
.apply(
"Read Matches",
@@ -1377,12 +1377,15 @@ public class AvroIO {
@Override
public PCollection<T> expand(PCollection<String> input) {
+ ParseFiles<T> parseFiles = parseFilesGenericRecords(getParseFn());
+ Coder<T> coder = getCoder();
+ if (coder != null) {
+ parseFiles = parseFiles.withCoder(coder);
+ }
return input
.apply(FileIO.matchAll().withConfiguration(getMatchConfiguration()))
.apply(FileIO.readMatches().withDirectoryTreatment(DirectoryTreatment.PROHIBIT))
- .apply(
- "Parse all via FileBasedSource",
- parseFilesGenericRecords(getParseFn()).withCoder(getCoder()));
+ .apply("Parse all via FileBasedSource", parseFiles);
}
@Override
@@ -1404,6 +1407,13 @@ public class AvroIO {
static final SerializableAvroCodecFactory DEFAULT_SERIALIZABLE_CODEC =
new SerializableAvroCodecFactory(DEFAULT_CODEC);
+ private static final String SCHEMA_REQUIRED =
+ "Unless using DynamicDestinations, .withSchema() is required.";
+ private static final String FORMAT_FUNCTION_REQUIRED =
+ "Unless using DynamicDestinations, .withFormatFunction() is required.";
+ private static final String FILENAME_PREFIX_REQUIRED =
+ "Need to set either the filename prefix or the tempDirectory of a
AvroIO.Write transform.";
+
abstract @Nullable SerializableFunction<UserT, OutputT>
getFormatFunction();
abstract @Nullable ValueProvider<ResourceId> getFilenamePrefix();
@@ -1753,7 +1763,7 @@ public class AvroIO {
if (usedFilenamePolicy == null) {
usedFilenamePolicy =
DefaultFilenamePolicy.fromStandardParameters(
- getFilenamePrefix(),
+ checkStateNotNull(getFilenamePrefix(),
FILENAME_PREFIX_REQUIRED),
getShardTemplate(),
getFilenameSuffix(),
getWindowedWrites());
@@ -1762,10 +1772,10 @@ public class AvroIO {
(DynamicAvroDestinations<UserT, DestinationT, OutputT>)
constantDestinations(
usedFilenamePolicy,
- getSchema(),
+ checkStateNotNull(getSchema(), SCHEMA_REQUIRED),
getMetadata(),
getCodec().getCodec(),
- getFormatFunction(),
+ checkStateNotNull(getFormatFunction(),
FORMAT_FUNCTION_REQUIRED),
getDatumWriterFactory());
}
return dynamicDestinations;
@@ -1774,9 +1784,7 @@ public class AvroIO {
@Override
public WriteFilesResult<DestinationT> expand(PCollection<UserT> input) {
checkArgument(
- getFilenamePrefix() != null || getTempDirectory() != null,
- "Need to set either the filename prefix or the tempDirectory of a
AvroIO.Write "
- + "transform.");
+ getFilenamePrefix() != null || getTempDirectory() != null,
FILENAME_PREFIX_REQUIRED);
if (getFilenamePolicy() != null) {
checkArgument(
getShardTemplate() == null && getFilenameSuffix() == null,
@@ -1789,13 +1797,12 @@ public class AvroIO {
"A format function should not be specified "
+ "with DynamicDestinations. Use
DynamicDestinations.formatRecord instead");
} else {
- checkArgument(
- getSchema() != null, "Unless using DynamicDestinations,
.withSchema() is required.");
+ checkArgument(getSchema() != null, SCHEMA_REQUIRED);
}
ValueProvider<ResourceId> tempDirectory = getTempDirectory();
if (tempDirectory == null) {
- tempDirectory = getFilenamePrefix();
+ tempDirectory = checkStateNotNull(getFilenamePrefix(),
FILENAME_PREFIX_REQUIRED);
}
WriteFiles<UserT, DestinationT, OutputT> write =
WriteFiles.to(
@@ -1813,11 +1820,13 @@ public class AvroIO {
if (getNoSpilling()) {
write = write.withNoSpilling();
}
- if (getMaxNumWritersPerBundle() != null) {
- write = write.withMaxNumWritersPerBundle(getMaxNumWritersPerBundle());
+ Integer maxNumWritersPerBundle = getMaxNumWritersPerBundle();
+ if (maxNumWritersPerBundle != null) {
+ write = write.withMaxNumWritersPerBundle(maxNumWritersPerBundle);
}
- if (getBadRecordErrorHandler() != null) {
- write = write.withBadRecordErrorHandler(getBadRecordErrorHandler());
+ ErrorHandler<BadRecord, ?> badRecordErrorHandler =
getBadRecordErrorHandler();
+ if (badRecordErrorHandler != null) {
+ write = write.withBadRecordErrorHandler(badRecordErrorHandler);
}
return input.apply("Write", write);
}
@@ -2020,13 +2029,13 @@ public class AvroIO {
private static class FormattedDatumWriter<ElementT> implements
DatumWriter<ElementT> {
private Schema root;
- private RecordFormatter<ElementT> formatter;
- private GenericDatumWriter<GenericRecord> writer;
+ private final RecordFormatter<ElementT> formatter;
+ private final GenericDatumWriter<GenericRecord> writer;
public FormattedDatumWriter(Schema schema, RecordFormatter<ElementT>
formatter) {
this.formatter = formatter;
this.writer = new GenericDatumWriter<>(schema);
- setSchema(schema);
+ this.root = schema;
}
@Override
@@ -2101,7 +2110,7 @@ public class AvroIO {
@Deprecated
abstract @Nullable RecordFormatter<ElementT> getRecordFormatter();
- abstract @Nullable String getJsonSchema();
+ abstract String getJsonSchema();
abstract Map<String, Object> getMetadata();
@@ -2151,47 +2160,53 @@ public class AvroIO {
return toBuilder().setDatumWriterFactory(datumWriterFactory).build();
}
- private transient @Nullable Schema schema;
private transient @Nullable DataFileWriter<ElementT> writer;
@Override
public void open(WritableByteChannel channel) throws IOException {
- this.schema = new Schema.Parser().parse(getJsonSchema());
+ Schema schema = new Schema.Parser().parse(getJsonSchema());
+ RecordFormatter<ElementT> recordFormatter = getRecordFormatter();
+ AvroSink.DatumWriterFactory<ElementT> datumWriterFactory =
getDatumWriterFactory();
DatumWriter<ElementT> datumWriter;
- if (getRecordFormatter() != null) {
- datumWriter = new FormattedDatumWriter<>(schema, getRecordFormatter());
- } else if (getDatumWriterFactory() != null) {
- datumWriter = getDatumWriterFactory().apply(schema);
+ if (recordFormatter != null) {
+ datumWriter = new FormattedDatumWriter<>(schema, recordFormatter);
+ } else if (datumWriterFactory != null) {
+ datumWriter = datumWriterFactory.apply(schema);
} else {
datumWriter = new ReflectDatumWriter<>(schema);
}
- writer = new DataFileWriter<>(datumWriter);
- writer.setCodec(getCodec().getCodec());
+ DataFileWriter<ElementT> dataFileWriter = new
DataFileWriter<>(datumWriter);
+ dataFileWriter.setCodec(getCodec().getCodec());
for (Map.Entry<String, Object> entry : getMetadata().entrySet()) {
Object v = entry.getValue();
if (v instanceof String) {
- writer.setMeta(entry.getKey(), (String) v);
+ dataFileWriter.setMeta(entry.getKey(), (String) v);
} else if (v instanceof Long) {
- writer.setMeta(entry.getKey(), (Long) v);
+ dataFileWriter.setMeta(entry.getKey(), (Long) v);
} else if (v instanceof byte[]) {
- writer.setMeta(entry.getKey(), (byte[]) v);
+ dataFileWriter.setMeta(entry.getKey(), (byte[]) v);
} else {
throw new IllegalStateException(
"Metadata value type must be one of String, Long, or byte[].
Found "
+ v.getClass().getSimpleName());
}
}
- writer.create(schema, Channels.newOutputStream(channel));
+ dataFileWriter.create(schema, Channels.newOutputStream(channel));
+ writer = dataFileWriter;
}
@Override
public void write(ElementT element) throws IOException {
- writer.append(element);
+ writer().append(element);
}
@Override
public void flush() throws IOException {
- writer.flush();
+ writer().flush();
+ }
+
+ private DataFileWriter<ElementT> writer() {
+ return checkStateNotNull(writer, "open() has not been called");
}
}
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java
index 08a9f3a2946..85431c4d9fb 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java
@@ -17,6 +17,8 @@
*/
package org.apache.beam.sdk.extensions.avro.io;
+import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
+
import com.google.auto.service.AutoService;
import java.io.Serializable;
import org.apache.avro.generic.GenericRecord;
@@ -45,9 +47,6 @@ import org.joda.time.Duration;
*/
@Internal
@AutoService(SchemaIOProvider.class)
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
public class AvroSchemaIOProvider implements SchemaIOProvider {
/** Returns an id that uniquely represents this IO. */
@Override
@@ -69,8 +68,10 @@ public class AvroSchemaIOProvider implements
SchemaIOProvider {
* resides there, and some IO-specific configuration object.
*/
@Override
- public AvroSchemaIO from(String location, Row configuration, Schema
dataSchema) {
- return new AvroSchemaIO(location, dataSchema, configuration);
+ public AvroSchemaIO from(String location, Row configuration, @Nullable
Schema dataSchema) {
+ // requiresDataSchema() is true, so callers must supply a data schema
+ return new AvroSchemaIO(
+ location, checkArgumentNotNull(dataSchema, "dataSchema is required"),
configuration);
}
@Override
@@ -94,11 +95,8 @@ public class AvroSchemaIOProvider implements
SchemaIOProvider {
private AvroSchemaIO(String location, Schema dataSchema, Row
configuration) {
this.dataSchema = dataSchema;
this.location = location;
- if (configuration.getInt64("writeWindowSizeSeconds") != null) {
- windowSize =
Duration.standardSeconds(configuration.getInt64("writeWindowSizeSeconds"));
- } else {
- windowSize = null;
- }
+ Long windowSizeSeconds =
configuration.getInt64("writeWindowSizeSeconds");
+ windowSize = windowSizeSeconds == null ? null :
Duration.standardSeconds(windowSizeSeconds);
}
@Override
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java
index a6e6353235c..2ec4f429dde 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java
@@ -17,6 +17,8 @@
*/
package org.apache.beam.sdk.extensions.avro.io;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+
import java.io.Serializable;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
@@ -34,9 +36,6 @@ import org.apache.beam.sdk.util.MimeTypes;
import org.checkerframework.checker.nullness.qual.Nullable;
/** A {@link FileBasedSink} for Avro files. */
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
public class AvroSink<UserT, DestinationT, OutputT>
extends FileBasedSink<UserT, DestinationT, OutputT> {
private final Class<OutputT> type;
@@ -121,33 +120,38 @@ public class AvroSink<UserT, DestinationT, OutputT>
.orElse(AvroDatumFactory.of(type))
.apply(schema);
- dataFileWriter = new DataFileWriter<>(datumWriter).setCodec(codec);
+ DataFileWriter<OutputT> writer = new
DataFileWriter<>(datumWriter).setCodec(codec);
for (Map.Entry<String, Object> entry : metadata.entrySet()) {
Object v = entry.getValue();
if (v instanceof String) {
- dataFileWriter.setMeta(entry.getKey(), (String) v);
+ writer.setMeta(entry.getKey(), (String) v);
} else if (v instanceof Long) {
- dataFileWriter.setMeta(entry.getKey(), (Long) v);
+ writer.setMeta(entry.getKey(), (Long) v);
} else if (v instanceof byte[]) {
- dataFileWriter.setMeta(entry.getKey(), (byte[]) v);
+ writer.setMeta(entry.getKey(), (byte[]) v);
} else {
throw new IllegalStateException(
"Metadata value type must be one of String, Long, or byte[].
Found "
+ v.getClass().getSimpleName());
}
}
- dataFileWriter.setSyncInterval(syncInterval);
- dataFileWriter.create(schema, Channels.newOutputStream(channel));
+ writer.setSyncInterval(syncInterval);
+ writer.create(schema, Channels.newOutputStream(channel));
+ dataFileWriter = writer;
}
@Override
public void write(OutputT value) throws Exception {
- dataFileWriter.append(value);
+ dataFileWriter().append(value);
}
@Override
protected void finishWrite() throws Exception {
- dataFileWriter.flush();
+ dataFileWriter().flush();
+ }
+
+ private DataFileWriter<OutputT> dataFileWriter() {
+ return checkStateNotNull(dataFileWriter, "prepareWrite() has not been
called");
}
}
}
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java
index 6d65c800c0a..bd21a0d1412 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java
@@ -18,9 +18,9 @@
package org.apache.beam.sdk.extensions.avro.io;
import static
org.apache.beam.sdk.io.FileBasedSource.Mode.SINGLE_FILE_OR_SUBRANGE;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
-import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
import java.io.IOException;
import java.io.InputStream;
@@ -125,9 +125,6 @@ import org.checkerframework.checker.nullness.qual.Nullable;
*/
// CHECKSTYLE.ON: JavadocStyle
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
public class AvroSource<T> extends BlockBasedSource<T> {
// Default minimum bundle size (chosen as two default-size Avro blocks to
attempt to
// ensure that every source has at least one block of records).
@@ -174,7 +171,8 @@ public class AvroSource<T> extends BlockBasedSource<T> {
@Nullable Coder<T> outputCoder,
@Nullable DatumReaderFactory<?> readerFactory) {
this.type = type;
- this.readerSchemaString = internSchemaString(readerSchemaString);
+ this.readerSchemaString =
+ readerSchemaString == null ? null :
internSchemaString(readerSchemaString);
this.parseFn = parseFn;
this.outputCoder = outputCoder;
this.readerFactory = readerFactory;
@@ -182,21 +180,24 @@ public class AvroSource<T> extends BlockBasedSource<T> {
private void readObject(ObjectInputStream is) throws IOException,
ClassNotFoundException {
is.defaultReadObject();
- readerSchemaString = internSchemaString(readerSchemaString);
+ String schemaString = readerSchemaString;
+ readerSchemaString = schemaString == null ? null :
internSchemaString(schemaString);
}
private Coder<T> getOutputCoder() {
if (parseFn == null && outputCoder == null) {
+ // validate() rejects a null readerSchemaString when there is no parse
fn
+ Schema readerSchema =
+ internOrParseSchemaString(checkStateNotNull(readerSchemaString,
"readerSchemaString"));
if (readerFactory != null && readerFactory instanceof
AvroDatumFactory) {
// create custom AvroCoder from the AvroDatumFactory
- return AvroCoder.of(
- (AvroDatumFactory<T>) readerFactory,
internOrParseSchemaString(readerSchemaString));
+ return AvroCoder.of((AvroDatumFactory<T>) readerFactory,
readerSchema);
} else {
// fallback with default avro coder for the type & schema
- return AvroCoder.of((Class<T>) type,
internOrParseSchemaString(readerSchemaString));
+ return AvroCoder.of((Class<T>) type, readerSchema);
}
} else {
- return outputCoder;
+ return checkStateNotNull(outputCoder, "outputCoder is required when
using a parse fn");
}
}
@@ -476,7 +477,7 @@ public class AvroSource<T> extends BlockBasedSource<T> {
String schemaString = null;
byte[] syncMarker;
try (InputStream stream =
Channels.newInputStream(FileSystems.open(fileResource))) {
- BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(stream, null);
+ BinaryDecoder decoder = newBinaryDecoder(stream);
// The header of an object container file begins with a four-byte magic
number, followed
// by the file metadata (including the schema and codec), encoded as a
map. Finally, the
@@ -519,8 +520,16 @@ public class AvroSource<T> extends BlockBasedSource<T> {
syncMarker = new byte[DataFileConstants.SYNC_SIZE];
decoder.readFixed(syncMarker);
}
- checkState(schemaString != null, "No schema present in Avro file metadata
%s", fileResource);
- return new AvroMetadata(syncMarker, codec, schemaString);
+ return new AvroMetadata(
+ syncMarker,
+ codec,
+ checkStateNotNull(
+ schemaString, "No schema present in Avro file metadata %s",
fileResource));
+ }
+
+ @SuppressWarnings("nullness") // DecoderFactory accepts a null reuse but is
unannotated
+ private static BinaryDecoder newBinaryDecoder(InputStream stream) {
+ return DecoderFactory.get().binaryDecoder(stream, null);
}
// A logical reference cache used to store schemas and schema strings to
allow us to
@@ -583,12 +592,14 @@ public class AvroSource<T> extends BlockBasedSource<T> {
private final Iterator<?> iterator;
- private final SerializableFunction<GenericRecord, T> parseFn;
+ private final @Nullable SerializableFunction<GenericRecord, T> parseFn;
private final long numRecordsInBlock;
AvroBlock(
- Iterator<?> iter, SerializableFunction<GenericRecord, T> parseFn, long
numRecordsInBlock) {
+ Iterator<?> iter,
+ @Nullable SerializableFunction<GenericRecord, T> parseFn,
+ long numRecordsInBlock) {
this.iterator = iter;
this.parseFn = parseFn;
this.numRecordsInBlock = numRecordsInBlock;
@@ -596,7 +607,7 @@ public class AvroSource<T> extends BlockBasedSource<T> {
@Override
public T getCurrentRecord() {
- return currentRecord;
+ return checkStateNotNull(currentRecord, "readNextRecord() has not been
called");
}
@Override
@@ -605,7 +616,7 @@ public class AvroSource<T> extends BlockBasedSource<T> {
return false;
}
- Object record = iterator.next();
+ Object record = checkStateNotNull(iterator.next(), "Avro block contained
a null record");
currentRecord = (parseFn == null) ? ((T) record) :
parseFn.apply((GenericRecord) record);
currentRecordIndex++;
return true;
@@ -699,34 +710,38 @@ public class AvroSource<T> extends BlockBasedSource<T> {
// Postcondition: same as above, but for the new current (formerly next)
block.
@Override
public boolean readNextBlock() {
- if (!dataFileReader.hasNext()) {
+ DataFileReader<?> reader = dataFileReader();
+ if (!reader.hasNext()) {
return false;
}
long headerLength =
- (long) VarInt.getLength(dataFileReader.getBlockCount())
- + VarInt.getLength(dataFileReader.getBlockSize())
+ (long) VarInt.getLength(reader.getBlockCount())
+ + VarInt.getLength(reader.getBlockSize())
+ DataFileConstants.SYNC_SIZE;
currentBlock =
- new AvroBlock<>(
- dataFileReader, getCurrentSource().mode.parseFn,
dataFileReader.getBlockCount());
+ new AvroBlock<>(reader, getCurrentSource().mode.parseFn,
reader.getBlockCount());
// Atomically update both the position and offset of the new block.
synchronized (progressLock) {
- currentBlockOffset = dataFileReader.previousSync();
+ currentBlockOffset = reader.previousSync();
// Total block size includes the header, block content, and trailing
sync marker.
- currentBlockSizeBytes = dataFileReader.getBlockSize() + headerLength;
+ currentBlockSizeBytes = reader.getBlockSize() + headerLength;
}
return true;
}
@Override
- public AvroBlock<T> getCurrentBlock() {
+ public @Nullable AvroBlock<T> getCurrentBlock() {
return currentBlock;
}
+ private DataFileReader<?> dataFileReader() {
+ return checkStateNotNull(dataFileReader, "startReading() has not been
called");
+ }
+
@Override
public long getCurrentBlockOffset() {
synchronized (progressLock) {
@@ -773,23 +788,33 @@ public class AvroSource<T> extends BlockBasedSource<T> {
}
DatumReader<?> reader =
-
Optional.<DatumReaderFactory<?>>ofNullable(this.getCurrentSource().mode.readerFactory)
- .orElse(AvroDatumFactory.of(this.getCurrentSource().mode.type))
- .apply(readerSchema, readerSchema);
+ newDatumReader(
+
Optional.<DatumReaderFactory<?>>ofNullable(this.getCurrentSource().mode.readerFactory)
+
.orElse(AvroDatumFactory.of(this.getCurrentSource().mode.type)),
+ readerSchema);
- dataFileReader = new DataFileReader<>(seekableChannelInput, reader);
+ DataFileReader<?> fileReader = new
DataFileReader<>(seekableChannelInput, reader);
long startOffset = getCurrentSource().getStartOffset();
if (startOffset != 0) {
// the start offset may be in the middle of a sync marker, by
rewinding SYNC_SIZE bytes we
// ensure that we won't miss the block if so.
- dataFileReader.sync(Math.max(0, startOffset -
DataFileConstants.SYNC_SIZE));
+ fileReader.sync(Math.max(0, startOffset -
DataFileConstants.SYNC_SIZE));
}
synchronized (progressLock) {
- currentBlockOffset = dataFileReader.previousSync();
+ currentBlockOffset = fileReader.previousSync();
currentBlockSizeBytes = 0;
}
+ dataFileReader = fileReader;
+ }
+
+ // A null schema tells the factory to derive both the writer and the
reader schema from the
+ // file header. DatumReaderFactory predates nullness annotations and
cannot express that.
+ @SuppressWarnings("nullness")
+ private static <X> DatumReader<X> newDatumReader(
+ DatumReaderFactory<X> factory, @Nullable Schema schema) {
+ return factory.apply(schema, schema);
}
}
}
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java
index 5b3683c349c..d92cef53b34 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java
@@ -32,9 +32,6 @@ import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.BaseEncodin
import org.checkerframework.checker.nullness.qual.Nullable;
/** Always returns a constant {@link FilenamePolicy}, {@link Schema},
metadata, and codec. */
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
class ConstantAvroDestination<UserT, OutputT>
extends DynamicAvroDestinations<UserT, Void, OutputT> {
private static class SchemaFunction implements Serializable,
Function<String, Schema> {
@@ -51,7 +48,7 @@ class ConstantAvroDestination<UserT, OutputT>
private final Map<String, Object> metadata;
private final SerializableAvroCodecFactory codec;
private final SerializableFunction<UserT, OutputT> formatFunction;
- private final AvroSink.DatumWriterFactory<OutputT> datumWriterFactory;
+ private final AvroSink.@Nullable DatumWriterFactory<OutputT>
datumWriterFactory;
private class Metadata implements HasDisplayData {
@Override
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java
index 215d3b4dd5f..fec1b7ed016 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java
@@ -40,9 +40,6 @@ import org.checkerframework.checker.nullness.qual.Nullable;
* A wrapper that allows {@link CodecFactory}s to be serialized using Java's
standard serialization
* mechanisms.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
class SerializableAvroCodecFactory implements Externalizable {
private static final long serialVersionUID = 7445324844109564303L;
private static final List<String> noOptAvroCodecs =
@@ -64,7 +61,7 @@ class SerializableAvroCodecFactory implements Externalizable {
this.codecFactory = codecFactory;
}
- private boolean checkIsSupportedCodec(CodecFactory codecFactory) {
+ private static boolean checkIsSupportedCodec(CodecFactory codecFactory) {
final String codecStr = codecFactory.toString();
return noOptAvroCodecs.contains(codecStr)
|| deflatePattern.matcher(codecStr).matches()
@@ -74,7 +71,7 @@ class SerializableAvroCodecFactory implements Externalizable {
@Override
public void writeExternal(ObjectOutput out) throws IOException {
- out.writeUTF(codecFactory.toString());
+ out.writeUTF(getCodec().toString());
}
@Override
@@ -91,32 +88,37 @@ class SerializableAvroCodecFactory implements
Externalizable {
Matcher deflateMatcher = deflatePattern.matcher(codecStr);
if (deflateMatcher.find()) {
- codecFactory =
CodecFactory.deflateCodec(Integer.parseInt(deflateMatcher.group("level")));
+ codecFactory = CodecFactory.deflateCodec(matchedLevel(deflateMatcher));
return;
}
Matcher xzMatcher = xzPattern.matcher(codecStr);
if (xzMatcher.find()) {
- codecFactory =
CodecFactory.xzCodec(Integer.parseInt(xzMatcher.group("level")));
+ codecFactory = CodecFactory.xzCodec(matchedLevel(xzMatcher));
return;
}
Matcher zstdMatcher = zstdPattern.matcher(codecStr);
if (zstdMatcher.find()) {
- codecFactory =
CodecFactory.zstandardCodec(Integer.parseInt(zstdMatcher.group("level")));
+ codecFactory = CodecFactory.zstandardCodec(matchedLevel(zstdMatcher));
return;
}
throw new IllegalStateException(codecStr + " is not supported");
}
+ /** Reads the {@code level} group of a matcher that has just matched
successfully. */
+ private static int matchedLevel(Matcher matcher) {
+ return Integer.parseInt(checkNotNull(matcher.group("level")));
+ }
+
public CodecFactory getCodec() {
- return codecFactory;
+ return checkNotNull(
+ codecFactory, "Inner CodecFactory is null, please use non default
constructor");
}
@Override
public String toString() {
- checkNotNull(codecFactory, "Inner CodecFactory is null, please use non
default constructor");
- return codecFactory.toString();
+ return getCodec().toString();
}
}
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java
index 0a82663c177..1812a5738c3 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java
@@ -46,10 +46,7 @@ import org.apache.beam.sdk.util.common.ReflectHelpers;
import org.apache.beam.sdk.values.TypeDescriptor;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
-@SuppressWarnings({
- "nullness", // TODO(https://github.com/apache/beam/issues/20497)
- "rawtypes"
-})
+@SuppressWarnings({"rawtypes"})
class AvroByteBuddyUtils {
private static final ByteBuddy BYTE_BUDDY = new ByteBuddy();
diff --git
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
index 853d43ce3e7..302c524722a 100644
---
a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
+++
b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
@@ -35,6 +35,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
@@ -151,13 +152,12 @@ import org.joda.time.ReadableInstant;
*
* is used.
*/
-@SuppressWarnings({
- "nullness", // TODO(https://github.com/apache/beam/issues/20497)
- "rawtypes"
-})
+@SuppressWarnings({"rawtypes"})
public class AvroUtils {
public static final String VERSION_AVRO =
- org.apache.avro.Schema.class.getPackage().getImplementationVersion();
+ Optional.ofNullable(org.apache.avro.Schema.class.getPackage())
+ .map(Package::getImplementationVersion)
+ .orElse("");
private static final ForLoadedType BYTES = new ForLoadedType(byte[].class);
private static final ForLoadedType JAVA_INSTANT = new
ForLoadedType(java.time.Instant.class);
private static final ForLoadedType JAVA_LOCALE_DATE =
@@ -196,6 +196,12 @@ public class AvroUtils {
builder.set(fieldName, castToNonNull(value));
}
+ private static org.apache.avro.Schema.Field newFieldWithoutDefault(
+ String name, org.apache.avro.Schema schema, String doc) {
+ // a null defaultValue means "no default", but the parameter lacks the
@Nullable annotation
+ return new org.apache.avro.Schema.Field(name, schema, doc,
castToNonNull(null));
+ }
+
private static Object createFixed(
@Nullable Object old, byte[] bytes, org.apache.avro.Schema schema) {
// old is tolerated when null, due to an instanceof check
@@ -528,8 +534,8 @@ public class AvroUtils {
public static org.apache.avro.Schema.Field toAvroField(Field field, String
namespace) {
org.apache.avro.Schema fieldSchema =
getFieldSchema(field.getType(), field.getName(), namespace);
- return new org.apache.avro.Schema.Field(
- field.getName(), fieldSchema, field.getDescription(), (Object) null);
+ return NullnessCheckerWorkarounds.newFieldWithoutDefault(
+ field.getName(), fieldSchema, field.getDescription());
}
private AvroUtils() {}
@@ -1437,8 +1443,8 @@ public class AvroUtils {
}
}
- private static Object convertLogicalType(
- @PolyNull Object value,
+ private static @Nullable Object convertLogicalType(
+ @Nonnull Object value,
@Nonnull org.apache.avro.Schema avroSchema,
@Nonnull FieldType fieldType,
@Nonnull GenericData genericData) {
@@ -1447,11 +1453,11 @@ public class AvroUtils {
// 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() == org.apache.avro.Schema.Type.LONG) {
- Long nanos = (Long) value;
+ 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())) {
+ checkNotNull(fieldType.getLogicalType()).getIdentifier())) {
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);
@@ -1473,7 +1479,10 @@ public class AvroUtils {
if (conversion != null) {
convertedType = conversion.getConvertedType();
if (convertedType.isInstance(value)) {
- rawType = Conversions.convertToRawType(value, avroSchema, logicalType,
conversion);
+ // type.type rather than avroSchema: Conversions.convertToRawType
switches on the schema
+ // type and silently returns the value unconverted for a UNION, so a
nullable field would
+ // never get converted.
+ rawType = Conversions.convertToRawType(value, type.type, logicalType,
conversion);
}
}
@@ -1786,8 +1795,8 @@ public class AvroUtils {
Object value,
LogicalType logicalType,
Object rawType,
- Conversion<?> conversion,
- Class<?> convertedType) {
+ @Nullable Conversion<?> conversion,
+ @Nullable Class<?> convertedType) {
String msg =
String.format(
"Value %s of class %s is not a supported type for logical type %s
(%s). "
diff --git
a/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java
b/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java
index d2452452f14..928cdbfe10b 100644
---
a/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java
+++
b/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java
@@ -73,6 +73,7 @@ import org.joda.time.DateTimeFieldType;
import org.joda.time.DateTimeZone;
import org.joda.time.Days;
import org.joda.time.Instant;
+import org.joda.time.LocalDate;
import org.joda.time.LocalTime;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -1112,10 +1113,90 @@ public class AvroUtilsTest {
// Alternatively, a timestamp-millis logical type can have a joda datum.
genericRecord.put("timestampMillis", new
DateTime(genericRecord.get("timestampMillis")));
- row = AvroUtils.toBeamRowStrict(getGenericRecord(), null);
+ row = AvroUtils.toBeamRowStrict(genericRecord, null);
assertEquals(getBeamRow(), row);
}
+ @Test
+ public void testNullableLogicalTypeGenericRecordToBeamRow() {
+ org.apache.avro.Schema decimalSchema =
+ LogicalTypes.decimal(Integer.MAX_VALUE)
+
.addToSchema(org.apache.avro.Schema.create(org.apache.avro.Schema.Type.BYTES));
+ org.apache.avro.Schema avroSchema =
+ org.apache.avro.Schema.createRecord(
+ "topLevelRecord",
+ null,
+ null,
+ false,
+ Lists.newArrayList(
+ new org.apache.avro.Schema.Field(
+ "date",
+ ReflectData.makeNullable(
+ LogicalTypes.date()
+ .addToSchema(
+
org.apache.avro.Schema.create(org.apache.avro.Schema.Type.INT))),
+ "",
+ (Object) null),
+ new org.apache.avro.Schema.Field(
+ "timestampMillis",
+ ReflectData.makeNullable(
+ LogicalTypes.timestampMillis()
+ .addToSchema(
+
org.apache.avro.Schema.create(org.apache.avro.Schema.Type.LONG))),
+ "",
+ (Object) null),
+ new org.apache.avro.Schema.Field(
+ "decimal", ReflectData.makeNullable(decimalSchema), "",
(Object) null)));
+
+ Schema beamSchema =
+ Schema.builder()
+ .addNullableField("date", FieldType.DATETIME)
+ .addNullableField("timestampMillis", FieldType.DATETIME)
+ .addNullableField("decimal", FieldType.DECIMAL)
+ .build();
+ assertEquals(beamSchema, AvroUtils.toBeamSchema(avroSchema));
+
+ // Data written through a GenericData with logical type conversions
registered carries the
+ // converted values (joda LocalDate/DateTime, BigDecimal) rather than the
raw int/long/bytes.
+ GenericRecord converted =
+ new GenericRecordBuilder(avroSchema)
+ .set("date", new LocalDate(1979, 3, 14))
+ .set("timestampMillis", DATE_TIME)
+ .set("decimal", BIG_DECIMAL)
+ .build();
+ assertEquals(
+ Row.withSchema(beamSchema)
+ .addValues(new DateTime(1979, 3, 14, 0, 0, DateTimeZone.UTC),
DATE_TIME, BIG_DECIMAL)
+ .build(),
+ AvroUtils.toBeamRowStrict(converted, beamSchema));
+
+ // The same fields holding their unconverted avro representations.
+ GenericRecord raw =
+ new GenericRecordBuilder(avroSchema)
+ .set("date", (int) java.time.LocalDate.of(1979, 3,
14).toEpochDay())
+ .set("timestampMillis", DATE_TIME.getMillis())
+ .set(
+ "decimal",
+ new Conversions.DecimalConversion()
+ .toBytes(BIG_DECIMAL, decimalSchema,
decimalSchema.getLogicalType()))
+ .build();
+ assertEquals(
+ Row.withSchema(beamSchema)
+ .addValues(new DateTime(1979, 3, 14, 0, 0, DateTimeZone.UTC),
DATE_TIME, BIG_DECIMAL)
+ .build(),
+ AvroUtils.toBeamRowStrict(raw, beamSchema));
+
+ GenericRecord nulls =
+ new GenericRecordBuilder(avroSchema)
+ .set("date", null)
+ .set("timestampMillis", null)
+ .set("decimal", null)
+ .build();
+ assertEquals(
+ Row.withSchema(beamSchema).addValues(null, null, null).build(),
+ AvroUtils.toBeamRowStrict(nulls, beamSchema));
+ }
+
@Test
public void testGenericRecordToRowFunction() {
SerializableUtils.ensureSerializable(AvroUtils.getGenericRecordToRowFunction(Schema.of()));
