lang/php/lib/DataFile/AvroDataIOReader.php - avro - Git at Google

 <?php

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     https://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 namespace Apache\Avro\DataFile;

 use Apache\Avro\AvroException;
 use Apache\Avro\AvroIO;
 use Apache\Avro\Datum\AvroIOBinaryDecoder;
 use Apache\Avro\Datum\AvroIODatumReader;
 use Apache\Avro\IO\AvroIOException;
 use Apache\Avro\IO\AvroStringIO;
 use Apache\Avro\Schema\AvroSchema;

 /**
  * Reads Avro data from an AvroIO source using an AvroSchema.
  */
 class AvroDataIOReader
 {
     public string $sync_marker;
     /**
      * @var array<string, mixed> object container metadata
      */
     public array $metadata;

     private AvroIOBinaryDecoder $decoder;
     /**
      * @var int count of items in block
      */
     private int $blockCount;

     /**
      * @var string compression codec
      */
     private string $codec;

     /**
      * @param AvroIO $io source from which to read
      * @param AvroIODatumReader $datumReader reader that understands
      *                                        the data schema
      * @throws AvroDataIOException if $io is not an instance of AvroIO
      *                             or the codec specified in the header
      *                             is not supported
      * @uses readHeader()
      */
     public function __construct(
         private AvroIO $io,
         private AvroIODatumReader $datumReader
     ) {
         $this->decoder = new AvroIOBinaryDecoder($this->io);
         $this->readHeader();

         $codec = $this->metadata[AvroDataIO::METADATA_CODEC_ATTR] ?? null;
         if ($codec && !AvroDataIO::isValidCodec($codec)) {
             throw new AvroDataIOException(sprintf('Unknown codec: %s', $codec));
         }
         $this->codec = $codec;

         $this->blockCount = 0;
         // FIXME: Seems unsanitary to set writers_schema here.
         // Can't constructor take it as an argument?
         $this->datumReader->setWritersSchema(
             AvroSchema::parse($this->metadata[AvroDataIO::METADATA_SCHEMA_ATTR])
         );
     }

     /**
      * @throws AvroException
      * @throws AvroIOException
      * @return list<mixed> of data from object container.
      * @internal Would be nice to implement data() as an iterator, I think
      */
     public function data(): array
     {
         $data = [];
         $decoder = $this->decoder;
         while (true) {
             if (0 == $this->blockCount) {
                 if ($this->isEof()) {
                     break;
                 }

                 if ($this->skipSync()) {
                     /** @phpstan-ignore if.alwaysFalse */
                     if ($this->isEof()) {
                         break;
                     }
                 }

                 $length = $this->readBlockHeader();
                 if (AvroDataIO::DEFLATE_CODEC === $this->codec) {
                     $compressed = $decoder->read($length);
                     $datum = $this->gzUncompress($compressed);
                     $decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
                 } elseif (AvroDataIO::ZSTANDARD_CODEC === $this->codec) {
                     $compressed = $decoder->read($length);
                     $datum = $this->zstdUncompress($compressed);
                     $decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
                 } elseif (AvroDataIO::SNAPPY_CODEC === $this->codec) {
                     $compressed = $decoder->read($length);
                     $datum = $this->snappyUncompress($compressed);
                     $decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
                 } elseif (AvroDataIO::BZIP2_CODEC === $this->codec) {
                     $compressed = $decoder->read($length);
                     $datum = $this->bzUncompress($compressed);
                     $decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
                 }
             }
             $data[] = $this->datumReader->read($decoder);
             --$this->blockCount;
         }

         return $data;
     }

     /**
      * Closes this writer (and its AvroIO object.)
      * @uses AvroIO::close()
      */
     public function close(): bool
     {
         return $this->io->close();
     }

     /**
      * Reads header of object container
      * @throws AvroDataIOException if the file is not an Avro data file.
      */
     private function readHeader(): void
     {
         $this->seek(0, AvroIO::SEEK_SET);

         $magic = $this->read(AvroDataIO::magicSize());

         if (strlen((string) $magic) < AvroDataIO::magicSize()) {
             throw new AvroDataIOException(
                 'Not an Avro data file: shorter than the Avro magic block'
             );
         }

         if (AvroDataIO::magic() != $magic) {
             throw new AvroDataIOException(
                 sprintf(
                     'Not an Avro data file: %s does not match %s',
                     $magic,
                     AvroDataIO::magic()
                 )
             );
         }

         $this->metadata = $this->datumReader->readData(
             AvroDataIO::metadataSchema(),
             AvroDataIO::metadataSchema(),
             $this->decoder
         );
         $this->sync_marker = $this->read(AvroDataIO::SYNC_SIZE);
     }

     /**
      * @uses AvroIO::seek()
      */
     private function seek(int $offset, int $whence): bool
     {
         return $this->io->seek($offset, $whence);
     }

     /**
      * @uses AvroIO::read()
      */
     private function read(int $len): string
     {
         return $this->io->read($len);
     }

     /**
      * @uses AvroIO::isEof()
      */
     private function isEof(): bool
     {
         return $this->io->isEof();
     }

     private function skipSync(): bool
     {
         $proposedSyncMarker = $this->read(AvroDataIO::SYNC_SIZE);
         if ($proposedSyncMarker !== $this->sync_marker) {
             $this->seek(-AvroDataIO::SYNC_SIZE, AvroIO::SEEK_CUR);

             return false;
         }

         return true;
     }

     /**
      * Reads the block header (which includes the count of items in the block
      * and the length in bytes of the block)
      * @return int|string length in bytes of the block. It returns a string if AvroGMP is enabled.
      */
     private function readBlockHeader(): string|int
     {
         $this->blockCount = $this->decoder->readLong();

         return $this->decoder->readLong();
     }

     /**
      * @throws AvroException
      */
     private function gzUncompress(string $compressed): string
     {
         $datum = gzinflate($compressed);

         if (false === $datum) {
             throw new AvroException('gzip uncompression failed.');
         }

         return $datum;
     }

     /**
      * @throws AvroException
      */
     private function zstdUncompress(string $compressed): string
     {
         if (!extension_loaded('zstd')) {
             throw new AvroException('Please install ext-zstd to use zstandard compression.');
         }
         $datum = zstd_uncompress($compressed);

         if (false === $datum) {
             throw new AvroException('zstd uncompression failed.');
         }

         return $datum;
     }

     /**
      * @throws AvroException
      */
     private function bzUncompress(string $compressed): string
     {
         if (!extension_loaded('bz2')) {
             throw new AvroException('Please install ext-bz2 to use bzip2 compression.');
         }
         $datum = bzdecompress($compressed);

         if (!is_string($datum)) {
             throw new AvroException('bz2 uncompression failed.');
         }

         return $datum;
     }

     /**
      * @throws AvroException
      */
     private function snappyUncompress(string $compressed): string
     {
         if (!extension_loaded('snappy')) {
             throw new AvroException('Please install ext-snappy to use snappy compression.');
         }
         $crc32 = unpack('N', substr((string) $compressed, -4))[1];
         $datum = snappy_uncompress(substr((string) $compressed, 0, -4));

         if (false === $datum) {
             throw new AvroException('snappy uncompression failed.');
         }

         if ($crc32 !== crc32($datum)) {
             throw new AvroException('snappy uncompression failed - crc32 mismatch.');
         }

         return $datum;
     }
 }
	<?php

	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* https://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	namespace Apache\Avro\DataFile;

	use Apache\Avro\AvroException;
	use Apache\Avro\AvroIO;
	use Apache\Avro\Datum\AvroIOBinaryDecoder;
	use Apache\Avro\Datum\AvroIODatumReader;
	use Apache\Avro\IO\AvroIOException;
	use Apache\Avro\IO\AvroStringIO;
	use Apache\Avro\Schema\AvroSchema;

	/**
	* Reads Avro data from an AvroIO source using an AvroSchema.
	*/
	class AvroDataIOReader
	{
	public string $sync_marker;
	/**
	* @var array<string, mixed> object container metadata
	*/
	public array $metadata;

	private AvroIOBinaryDecoder $decoder;
	/**
	* @var int count of items in block
	*/
	private int $blockCount;

	/**
	* @var string compression codec
	*/
	private string $codec;

	/**
	* @param AvroIO $io source from which to read
	* @param AvroIODatumReader $datumReader reader that understands
	* the data schema
	* @throws AvroDataIOException if $io is not an instance of AvroIO
	* or the codec specified in the header
	* is not supported
	* @uses readHeader()
	*/
	public function __construct(
	private AvroIO $io,
	private AvroIODatumReader $datumReader
	) {
	$this->decoder = new AvroIOBinaryDecoder($this->io);
	$this->readHeader();

	$codec = $this->metadata[AvroDataIO::METADATA_CODEC_ATTR] ?? null;
	if ($codec && !AvroDataIO::isValidCodec($codec)) {
	throw new AvroDataIOException(sprintf('Unknown codec: %s', $codec));
	}
	$this->codec = $codec;

	$this->blockCount = 0;
	// FIXME: Seems unsanitary to set writers_schema here.
	// Can't constructor take it as an argument?
	$this->datumReader->setWritersSchema(
	AvroSchema::parse($this->metadata[AvroDataIO::METADATA_SCHEMA_ATTR])
	);
	}

	/**
	* @throws AvroException
	* @throws AvroIOException
	* @return list<mixed> of data from object container.
	* @internal Would be nice to implement data() as an iterator, I think
	*/
	public function data(): array
	{
	$data = [];
	$decoder = $this->decoder;
	while (true) {
	if (0 == $this->blockCount) {
	if ($this->isEof()) {
	break;
	}

	if ($this->skipSync()) {
	/** @phpstan-ignore if.alwaysFalse */
	if ($this->isEof()) {
	break;
	}
	}

	$length = $this->readBlockHeader();
	if (AvroDataIO::DEFLATE_CODEC === $this->codec) {
	$compressed = $decoder->read($length);
	$datum = $this->gzUncompress($compressed);
	$decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
	} elseif (AvroDataIO::ZSTANDARD_CODEC === $this->codec) {
	$compressed = $decoder->read($length);
	$datum = $this->zstdUncompress($compressed);
	$decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
	} elseif (AvroDataIO::SNAPPY_CODEC === $this->codec) {
	$compressed = $decoder->read($length);
	$datum = $this->snappyUncompress($compressed);
	$decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
	} elseif (AvroDataIO::BZIP2_CODEC === $this->codec) {
	$compressed = $decoder->read($length);
	$datum = $this->bzUncompress($compressed);
	$decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
	}
	}
	$data[] = $this->datumReader->read($decoder);
	--$this->blockCount;
	}

	return $data;
	}

	/**
	* Closes this writer (and its AvroIO object.)
	* @uses AvroIO::close()
	*/
	public function close(): bool
	{
	return $this->io->close();
	}

	/**
	* Reads header of object container
	* @throws AvroDataIOException if the file is not an Avro data file.
	*/
	private function readHeader(): void
	{
	$this->seek(0, AvroIO::SEEK_SET);

	$magic = $this->read(AvroDataIO::magicSize());

	if (strlen((string) $magic) < AvroDataIO::magicSize()) {
	throw new AvroDataIOException(
	'Not an Avro data file: shorter than the Avro magic block'
	);
	}

	if (AvroDataIO::magic() != $magic) {
	throw new AvroDataIOException(
	sprintf(
	'Not an Avro data file: %s does not match %s',
	$magic,
	AvroDataIO::magic()
	)
	);
	}

	$this->metadata = $this->datumReader->readData(
	AvroDataIO::metadataSchema(),
	AvroDataIO::metadataSchema(),
	$this->decoder
	);
	$this->sync_marker = $this->read(AvroDataIO::SYNC_SIZE);
	}

	/**
	* @uses AvroIO::seek()
	*/
	private function seek(int $offset, int $whence): bool
	{
	return $this->io->seek($offset, $whence);
	}

	/**
	* @uses AvroIO::read()
	*/
	private function read(int $len): string
	{
	return $this->io->read($len);
	}

	/**
	* @uses AvroIO::isEof()
	*/
	private function isEof(): bool
	{
	return $this->io->isEof();
	}

	private function skipSync(): bool
	{
	$proposedSyncMarker = $this->read(AvroDataIO::SYNC_SIZE);
	if ($proposedSyncMarker !== $this->sync_marker) {
	$this->seek(-AvroDataIO::SYNC_SIZE, AvroIO::SEEK_CUR);

	return false;
	}

	return true;
	}

	/**
	* Reads the block header (which includes the count of items in the block
	* and the length in bytes of the block)
	* @return int\|string length in bytes of the block. It returns a string if AvroGMP is enabled.
	*/
	private function readBlockHeader(): string\|int
	{
	$this->blockCount = $this->decoder->readLong();

	return $this->decoder->readLong();
	}

	/**
	* @throws AvroException
	*/
	private function gzUncompress(string $compressed): string
	{
	$datum = gzinflate($compressed);

	if (false === $datum) {
	throw new AvroException('gzip uncompression failed.');
	}

	return $datum;
	}

	/**
	* @throws AvroException
	*/
	private function zstdUncompress(string $compressed): string
	{
	if (!extension_loaded('zstd')) {
	throw new AvroException('Please install ext-zstd to use zstandard compression.');
	}
	$datum = zstd_uncompress($compressed);

	if (false === $datum) {
	throw new AvroException('zstd uncompression failed.');
	}

	return $datum;
	}

	/**
	* @throws AvroException
	*/
	private function bzUncompress(string $compressed): string
	{
	if (!extension_loaded('bz2')) {
	throw new AvroException('Please install ext-bz2 to use bzip2 compression.');
	}
	$datum = bzdecompress($compressed);

	if (!is_string($datum)) {
	throw new AvroException('bz2 uncompression failed.');
	}

	return $datum;
	}

	/**
	* @throws AvroException
	*/
	private function snappyUncompress(string $compressed): string
	{
	if (!extension_loaded('snappy')) {
	throw new AvroException('Please install ext-snappy to use snappy compression.');
	}
	$crc32 = unpack('N', substr((string) $compressed, -4))[1];
	$datum = snappy_uncompress(substr((string) $compressed, 0, -4));

	if (false === $datum) {
	throw new AvroException('snappy uncompression failed.');
	}

	if ($crc32 !== crc32($datum)) {
	throw new AvroException('snappy uncompression failed - crc32 mismatch.');
	}

	return $datum;
	}
	}