Add support for deflate and snappy codecs (#9)

Thanks @smalot for the contribution!

Author: smalot

.travis.yml

@@ -9,6 +9,16 @@ php:
 - '7.4'
 
 before_script:
+- cd /tmp
+- git clone --recursive --depth=1 https://github.com/kjdev/php-ext-snappy.git
+- cd php-ext-snappy
+- export CXXFLAGS=-std=c++11
+- phpize
+- ./configure
+- make
+- make install
+- echo "extension = snappy.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
+- cd $TRAVIS_BUILD_DIR
 - composer self-update
 - composer update
 

Dockerfile

@@ -5,8 +5,8 @@ FROM php:${PHP_VERSION}-cli-alpine
 ARG XDEBUG_VERSION=2.7.0RC1
 
 RUN apk add --no-cache --virtual .build-deps $PHPIZE_DEPS \
-    && apk add --no-cache --virtual .runtime-deps git libzip-dev \
-    && docker-php-ext-install zip \
+    && apk add --no-cache --virtual .runtime-deps git libzip-dev gmp-dev \
+    && docker-php-ext-install zip gmp \
     && pecl install xdebug-$XDEBUG_VERSION \
     && docker-php-ext-enable xdebug \
     && echo "xdebug.max_nesting_level=15000" >> "$PHP_INI_DIR/conf.d/docker-php-ext-xdebug.ini" \
@@ -16,4 +16,11 @@ RUN apk add --no-cache --virtual .build-deps $PHPIZE_DEPS \
     && echo "xdebug.remote_handler=dbgp" >> "$PHP_INI_DIR/conf.d/docker-php-ext-xdebug.ini" \
     && echo "xdebug.remote_autostart=1" >> "$PHP_INI_DIR/conf.d/docker-php-ext-xdebug.ini" \
     && echo "xdebug.remote_connect_back=0" >> "$PHP_INI_DIR/conf.d/docker-php-ext-xdebug.ini" \
+    && git clone --recursive --depth=1 https://github.com/kjdev/php-ext-snappy.git \
+    && cd php-ext-snappy \
+    && phpize \
+    && ./configure \
+    && make \
+    && make install \
+    && docker-php-ext-enable snappy \
     && apk del .build-deps

README.md

@@ -8,6 +8,8 @@ Requirements
  * PHP >= 7.1
  * On 32-bit platforms, the [GMP PHP extension](http://php.net/gmp)
  * For testing, [PHPUnit](http://www.phpunit.de/)
+ * For Deflate codec, [Zlib](https://www.php.net/zlib)
+ * For Snappy codec, [Snappy](http://google.github.io/snappy/)
 
 Both GMP and PHPUnit are often available via package management
 systems as `php5-gmp` and `phpunit`, respectively.

composer.json

@@ -9,7 +9,9 @@
         "phpunit/phpunit": "^5.0"
     },
     "suggest": {
-        "ext-gmp": "Large integer support for 32-bit platforms."
+        "ext-gmp": "Large integer support for 32-bit platforms.",
+        "ext-zlib": "Zlib support for Deflate codec.",
+        "ext-snappy": "Snappy support for Snappy codec (mainly Google)."
     },
     "autoload": {
         "classmap": ["lib/"]

lib/avro/data_file.php

@@ -73,12 +73,16 @@ class AvroDataIO
    */
   const DEFLATE_CODEC = 'deflate';
 
+  /**
+   * @var string codec value for snappy codec
+   *   cf: http://google.github.io/snappy/
+   */
+  const SNAPPY_CODEC = 'snappy';
+
   /**
    * @var array array of valid codec names
-   * @todo Avro implementations are required to implement deflate codec as well,
-   *       so implement it already!
    */
-  private static $valid_codecs = array(self::NULL_CODEC);
+  private static $valid_codecs = array(self::NULL_CODEC, self::DEFLATE_CODEC, self::SNAPPY_CODEC);
 
   /**
    * @var AvroSchema cached version of metadata schema object
@@ -111,13 +115,14 @@ public static function metadata_schema()
    * @param string $file_path file_path of file to open
    * @param string $mode one of AvroFile::READ_MODE or AvroFile::WRITE_MODE
    * @param string $schema_json JSON of writer's schema
+   * @param string $codec compression codec
    * @return AvroDataIOWriter instance of AvroDataIOWriter
    *
    * @throws AvroDataIOException if $writers_schema is not provided
    *         or if an invalid $mode is given.
    */
   public static function open_file($file_path, $mode=AvroFile::READ_MODE,
-                                   $schema_json=null)
+                                   $schema_json=null, $codec=self::NULL_CODEC)
   {
     $schema = !is_null($schema_json)
       ? AvroSchema::parse($schema_json) : null;
@@ -128,7 +133,7 @@ public static function open_file($file_path, $mode=AvroFile::READ_MODE,
         if (is_null($schema))
           throw new AvroDataIOException('Writing an Avro file requires a schema.');
         $file = new AvroFile($file_path, AvroFile::WRITE_MODE);
-        $io = self::open_writer($file, $schema);
+        $io = self::open_writer($file, $schema, $codec);
         break;
       case AvroFile::READ_MODE:
         $file = new AvroFile($file_path, AvroFile::READ_MODE);
@@ -145,7 +150,7 @@ public static function open_file($file_path, $mode=AvroFile::READ_MODE,
   /**
    * @return array array of valid codecs
    */
-  private static function valid_codecs()
+  public static function valid_codecs()
   {
     return self::$valid_codecs;
   }
@@ -162,12 +167,13 @@ public static function is_valid_codec($codec)
   /**
    * @param AvroIO $io
    * @param AvroSchema $schema
+   * @param string $codec
    * @return AvroDataIOWriter
    */
-  protected static function open_writer($io, $schema)
+  protected static function open_writer($io, $schema, $codec=self::NULL_CODEC)
   {
     $writer = new AvroIODatumWriter($schema);
-    return new AvroDataIOWriter($io, $writer, $schema);
+    return new AvroDataIOWriter($io, $writer, $schema, $codec);
   }
 
   /**
@@ -220,11 +226,18 @@ class AvroDataIOReader
    */
   private $block_count;
 
+  /**
+   * @var string compression codec
+   */
+  private $codec;
+
   /**
    * @param AvroIO $io source from which to read
    * @param AvroIODatumReader $datum_reader 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 read_header()
    */
   public function __construct($io, $datum_reader)
@@ -242,6 +255,7 @@ public function __construct($io, $datum_reader)
                                    AvroDataIO::METADATA_CODEC_ATTR);
     if ($codec && !AvroDataIO::is_valid_codec($codec))
       throw new AvroDataIOException(sprintf('Uknown codec: %s', $codec));
+    $this->codec = $codec;
 
     $this->block_count = 0;
     // FIXME: Seems unsanitary to set writers_schema here.
@@ -276,8 +290,10 @@ private function read_header()
   }
 
   /**
-   * @internal Would be nice to implement data() as an iterator, I think
    * @return array of data from object container.
+   * @throws AvroDataIOException
+   * @throws AvroIOException
+   * @internal Would be nice to implement data() as an iterator, I think
    */
   public function data()
   {
@@ -293,9 +309,29 @@ public function data()
           if ($this->is_eof())
             break;
 
-        $this->read_block_header();
+        $length = $this->read_block_header();
+        $decoder = $this->decoder;
+        if ($this->codec == AvroDataIO::DEFLATE_CODEC) {
+          if (!function_exists('gzinflate')) {
+            throw new AvroDataIOException('"gzinflate" function not available, "zlib" extension required.');
+          }
+          $compressed = $decoder->read($length);
+          $datum = gzinflate($compressed);
+          $decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
+        } elseif ($this->codec == AvroDataIO::SNAPPY_CODEC) {
+          if (!function_exists('snappy_uncompress')) {
+            throw new AvroDataIOException('"snappy_uncompress" function not available, "snappy" extension required.');
+          }
+          $compressed = $decoder->read($length-4);
+          $datum = snappy_uncompress($compressed);
+          $crc32 = unpack('N', $decoder->read(4));
+          if ($crc32[1] != crc32($datum)) {
+            throw new AvroDataIOException('Invalid CRC32 checksum.');
+          }
+          $decoder = new AvroIOBinaryDecoder(new AvroStringIO($datum));
+        }
       }
-      $data []= $this->datum_reader->read($this->decoder);
+      $data []= $this->datum_reader->read($decoder);
       $this->block_count -= 1;
     }
     return $data;
@@ -414,13 +450,19 @@ private static function generate_sync_marker()
    */
   private $metadata;
 
+  /**
+   * @var string compression codec
+   */
+  private $codec;
+
   /**
    * @param AvroIO $io
    * @param AvroIODatumWriter $datum_writer
    * @param AvroSchema $writers_schema
+   * @param string $codec
    * @throws AvroDataIOException
    */
-  public function __construct($io, $datum_writer, $writers_schema=null)
+  public function __construct($io, $datum_writer, $writers_schema=null, $codec=AvroDataIO::NULL_CODEC)
   {
     if (!($io instanceof AvroIO))
       throw new AvroDataIOException('io must be instance of AvroIO');
@@ -435,16 +477,21 @@ public function __construct($io, $datum_writer, $writers_schema=null)
 
     if ($writers_schema)
     {
+      if (!AvroDataIO::is_valid_codec($codec))
+        throw new AvroDataIOException(
+          sprintf('codec %s is not supported', $codec));
+
       $this->sync_marker = self::generate_sync_marker();
-      $this->metadata[AvroDataIO::METADATA_CODEC_ATTR] = AvroDataIO::NULL_CODEC;
+      $this->metadata[AvroDataIO::METADATA_CODEC_ATTR] = $this->codec = $codec;
       $this->metadata[AvroDataIO::METADATA_SCHEMA_ATTR] = strval($writers_schema);
       $this->write_header();
     }
     else
     {
       $dfr = new AvroDataIOReader($this->io, new AvroIODatumReader());
       $this->sync_marker = $dfr->sync_marker;
-      $this->metadata[AvroDataIO::METADATA_CODEC_ATTR] = $dfr->metadata[AvroDataIO::METADATA_CODEC_ATTR];
+      $this->metadata[AvroDataIO::METADATA_CODEC_ATTR] = $this->codec
+                                                       = $dfr->metadata[AvroDataIO::METADATA_CODEC_ATTR];
 
       $schema_from_file = $dfr->metadata[AvroDataIO::METADATA_SCHEMA_ATTR];
       $this->metadata[AvroDataIO::METADATA_SCHEMA_ATTR] = $schema_from_file;
@@ -489,27 +536,29 @@ private function flush()
 
   /**
    * Writes a block of data to the AvroIO object container.
-   * @throws AvroDataIOException if the codec provided by the encoder
-   *         is not supported
-   * @internal Should the codec check happen in the constructor?
-   *           Why wait until we're writing data?
    */
   private function write_block()
   {
     if ($this->block_count > 0)
     {
       $this->encoder->write_long($this->block_count);
       $to_write = strval($this->buffer);
-      $this->encoder->write_long(strlen($to_write));
 
-      if (AvroDataIO::is_valid_codec(
-            $this->metadata[AvroDataIO::METADATA_CODEC_ATTR]))
-        $this->write($to_write);
-      else
-        throw new AvroDataIOException(
-          sprintf('codec %s is not supported',
-                  $this->metadata[AvroDataIO::METADATA_CODEC_ATTR]));
+      if ($this->codec == AvroDataIO::DEFLATE_CODEC) {
+        if (!function_exists('gzinflate')) {
+          throw new AvroDataIOException('"gzinflate" function not available, "zlib" extension required.');
+        }
+        $to_write = gzdeflate($to_write);
+      } elseif ($this->codec == AvroDataIO::SNAPPY_CODEC) {
+        if (!function_exists('snappy_compress')) {
+          throw new AvroDataIOException('"snappy_compress" function not available, "snappy" extension required.');
+        }
+        $crc32 = pack('N', crc32($to_write));
+        $to_write = snappy_compress($to_write) . $crc32;
+      }
 
+      $this->encoder->write_long(strlen($to_write));
+      $this->write($to_write);
       $this->write($this->sync_marker);
       $this->buffer->truncate();
       $this->block_count = 0;