AvroSource extends BlockBasedSource<T> { @VisibleForTesting static AvroMetadata readMetadataFromFile(ResourceId fileResource) throws IOException { String codec = null; String schemaString = null; byte[] syncMarker; try (InputStream stream = Channels.newInputStream(FileSystems.open(fileResource))) { BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(stream, null); byte[] magic = new byte[DataFileConstants.MAGIC.length]; decoder.readFixed(magic); if (!Arrays.equals(magic, DataFileConstants.MAGIC)) { throw new IOException("Missing Avro file signature: " + fileResource); } ByteBuffer valueBuffer = ByteBuffer.allocate(512); long numRecords = decoder.readMapStart(); while (numRecords > 0) { for (long recordIndex = 0; recordIndex < numRecords; recordIndex++) { String key = decoder.readString(); valueBuffer = decoder.readBytes(valueBuffer); byte[] bytes = new byte[valueBuffer.remaining()]; valueBuffer.get(bytes); if (key.equals(DataFileConstants.CODEC)) { codec = new String(bytes, StandardCharsets.UTF_8); } else if (key.equals(DataFileConstants.SCHEMA)) { schemaString = new String(bytes, StandardCharsets.UTF_8); } } numRecords = decoder.mapNext(); } if (codec == null) { codec = DataFileConstants.NULL_CODEC; } 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); } private  AvroSource(
      ValueProvider<String> fileNameOrPattern,
      EmptyMatchTreatment emptyMatchTreatment,
      long minBundleSize,
      Mode<T> mode); private  AvroSource(
      Metadata metadata, long minBundleSize, long startOffset, long endOffset, Mode<T> mode); static AvroSource<GenericRecord> from(ValueProvider<String> fileNameOrPattern); static AvroSource<GenericRecord> from(Metadata metadata); static AvroSource<GenericRecord> from(String fileNameOrPattern); AvroSource<T> withEmptyMatchTreatment(EmptyMatchTreatment emptyMatchTreatment); AvroSource<GenericRecord> withSchema(String schema); AvroSource<GenericRecord> withSchema(Schema schema); AvroSource<X> withSchema(Class<X> clazz); AvroSource<X> withParseFn(
      SerializableFunction<GenericRecord, X> parseFn, Coder<X> coder); AvroSource<T> withMinBundleSize(long minBundleSize); AvroSource<T> withDatumReaderFactory(DatumReaderFactory<?> factory); @Override void validate(); @Deprecated BlockBasedSource<T> createForSubrangeOfFile(String fileName, long start, long end); @Override BlockBasedSource<T> createForSubrangeOfFile(Metadata fileMetadata, long start, long end); @Override Coder<T> getOutputCoder();  }
@Test public void testReadMetadataWithCodecs() throws Exception { String[] codecs = { DataFileConstants.NULL_CODEC, DataFileConstants.BZIP2_CODEC, DataFileConstants.DEFLATE_CODEC, DataFileConstants.SNAPPY_CODEC, DataFileConstants.XZ_CODEC }; List<Bird> expected = createRandomRecords(DEFAULT_RECORD_COUNT); for (String codec : codecs) { String filename = generateTestFile( codec, expected, SyncBehavior.SYNC_DEFAULT, 0, AvroCoder.of(Bird.class), codec); Metadata fileMeta = FileSystems.matchSingleFileSpec(filename); AvroMetadata metadata = AvroSource.readMetadataFromFile(fileMeta.resourceId()); assertEquals(codec, metadata.getCodec()); } }
@Test public void testReadSchemaString() throws Exception { List<Bird> expected = createRandomRecords(DEFAULT_RECORD_COUNT); String codec = DataFileConstants.NULL_CODEC; String filename = generateTestFile( codec, expected, SyncBehavior.SYNC_DEFAULT, 0, AvroCoder.of(Bird.class), codec); Metadata fileMeta = FileSystems.matchSingleFileSpec(filename); AvroMetadata metadata = AvroSource.readMetadataFromFile(fileMeta.resourceId()); Schema schema = new Schema.Parser().parse(metadata.getSchemaString()); assertEquals(4, schema.getFields().size()); }