[WIP][POC] Add FSST encoding

.gitattributes

@@ -19,3 +19,8 @@
 * text eol=lf
 *.png binary
 core.autocrlf=false
+
+# FSST test fixtures are byte streams, not text: line-ending normalization would corrupt them.
+parquet-column/src/test/resources/fsst/*.table binary
+parquet-column/src/test/resources/fsst/*.codes binary
+parquet-column/src/test/resources/fsst/interop/*.pages binary

LICENSE

@@ -206,3 +206,39 @@ Copyright: 2012-2014 Twitter
 Home page: https://github.com/twitter/elephant-bird
 License: http://www.apache.org/licenses/LICENSE-2.0
 
+
+--------------------------------------------------------------------------------
+
+This project includes code from the FSST project. parquet-column's FSST symbol
+table trainer and compressor are a Java port of FSST's reference
+implementation.
+
+* parquet-column's org.apache.parquet.column.values.symboltable.fsst package
+  is derived from FSST's libfsst.hpp and libfsst.cpp, at commit
+  89f49c580c6388acf3b6ed2a49e1bfde6c05e616.
+
+Copyright: 2018-2020 CWI, TU Munich, FSU Jena
+Home page: https://github.com/cwida/fsst
+License: MIT License
+
+MIT License
+
+Copyright (c) 2018-2020, CWI, TU Munich, FSU Jena
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

parquet-column/pom.xml

@@ -112,6 +112,12 @@
       <version>${junit.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.github.luben</groupId>
+      <artifactId>zstd-jni</artifactId>
+      <version>${zstd-jni.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

parquet-column/src/main/java/org/apache/parquet/column/Encoding.java

@@ -54,6 +54,8 @@
 import org.apache.parquet.column.values.plain.PlainValuesReader.LongPlainValuesReader;
 import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesReader;
 import org.apache.parquet.column.values.rle.ZeroIntegerValuesReader;
+import org.apache.parquet.column.values.symboltable.SymbolTable;
+import org.apache.parquet.column.values.symboltable.SymbolTableValuesReader;
 import org.apache.parquet.io.ParquetDecodingException;
 
 /**
@@ -253,6 +255,40 @@ public ValuesReader getDictionaryBasedValuesReader(
     public boolean usesDictionary() {
       return true;
     }
+  },
+
+  /**
+   * Values are replaced by codes over a table of the byte sequences that recur in the column, with
+   * one table per column chunk. The table's own representation decides the width of a code and how a
+   * byte that no symbol covers is escaped, so this one encoding covers every such representation and
+   * a reader has to read the table before it can commit to decoding the column.
+   * <p>
+   * Not ratified: parquet-format issue #531. A writer will not produce this encoding unless it is
+   * turned on explicitly, and no file written by this library carries it yet, because the format has
+   * nowhere to put the table.
+   */
+  FSST {
+    @Override
+    public ValuesReader getValuesReader(ColumnDescriptor descriptor, ValuesType valuesType) {
+      if (descriptor.getType() != BINARY) {
+        throw new ParquetDecodingException("Encoding FSST is only supported for type BINARY");
+      }
+      return new SymbolTableValuesReader();
+    }
+
+    @Override
+    public boolean usesSymbolTable() {
+      return true;
+    }
+
+    @Override
+    public ValuesReader getSymbolTableBasedValuesReader(
+        ColumnDescriptor descriptor, ValuesType valuesType, SymbolTable symbolTable) {
+      if (descriptor.getType() != BINARY) {
+        throw new ParquetDecodingException("Encoding FSST is only supported for type BINARY");
+      }
+      return new SymbolTableValuesReader(() -> symbolTable);
+    }
   };
 
   int getMaxLevel(ColumnDescriptor descriptor, ValuesType valuesType) {
@@ -321,4 +357,25 @@ public ValuesReader getDictionaryBasedValuesReader(
       ColumnDescriptor descriptor, ValuesType valuesType, Dictionary dictionary) {
     throw new UnsupportedOperationException(this.name() + " is not dictionary based");
   }
+
+  /**
+   * @return whether this encoding requires a symbol table
+   */
+  public boolean usesSymbolTable() {
+    return false;
+  }
+
+  /**
+   * To read decoded values that require a symbol table
+   *
+   * @param descriptor  the column to read
+   * @param valuesType  the type of values
+   * @param symbolTable the symbol table for the chunk being read
+   * @return the proper values reader for the given column
+   * @throws UnsupportedOperationException if the encoding is not symbol table based
+   */
+  public ValuesReader getSymbolTableBasedValuesReader(
+      ColumnDescriptor descriptor, ValuesType valuesType, SymbolTable symbolTable) {
+    throw new UnsupportedOperationException(this.name() + " is not symbol table based");
+  }
 }

parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java

@@ -39,6 +39,8 @@
 import org.apache.parquet.column.values.factory.ValuesWriterFactory;
 import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder;
 import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesWriter;
+import org.apache.parquet.column.values.symboltable.SymbolTablePayload.OffsetEncoding;
+import org.apache.parquet.column.values.symboltable.SymbolTableType;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
 import org.apache.parquet.schema.MessageType;
 
@@ -51,6 +53,8 @@ public class ParquetProperties {
   public static final int DEFAULT_DICTIONARY_PAGE_SIZE = DEFAULT_PAGE_SIZE;
   public static final boolean DEFAULT_IS_DICTIONARY_ENABLED = true;
   public static final boolean DEFAULT_IS_BYTE_STREAM_SPLIT_ENABLED = false;
+  public static final boolean DEFAULT_IS_FSST_ENABLED = false;
+  public static final OffsetEncoding DEFAULT_SYMBOL_TABLE_OFFSET_ENCODING = OffsetEncoding.DELTA_BINARY_PACKED;
   public static final WriterVersion DEFAULT_WRITER_VERSION = WriterVersion.PARQUET_1_0;
   public static final boolean DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK = true;
   public static final int DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK = 100;
@@ -133,6 +137,8 @@ public static WriterVersion fromString(String name) {
   private final int pageRowCountLimit;
   private final boolean pageWriteChecksumEnabled;
   private final ColumnProperty<ByteStreamSplitMode> byteStreamSplitEnabled;
+  private final ColumnProperty<Boolean> fsstEnabled;
+  private final OffsetEncoding symbolTableOffsetEncoding;
   private final Map<String, String> extraMetaData;
   private final ColumnProperty<Boolean> statistics;
   private final ColumnProperty<Boolean> sizeStatistics;
@@ -167,6 +173,8 @@ private ParquetProperties(Builder builder) {
     this.pageRowCountLimit = builder.pageRowCountLimit;
     this.pageWriteChecksumEnabled = builder.pageWriteChecksumEnabled;
     this.byteStreamSplitEnabled = builder.byteStreamSplitEnabled.build();
+    this.fsstEnabled = builder.fsstEnabled.build();
+    this.symbolTableOffsetEncoding = builder.symbolTableOffsetEncoding;
     this.extraMetaData = builder.extraMetaData;
     this.statistics = builder.statistics.build();
     this.sizeStatistics = builder.sizeStatistics.build();
@@ -264,6 +272,37 @@ public boolean isByteStreamSplitEnabled(ColumnDescriptor column) {
     }
   }
 
+  /**
+   * Whether a symbol table encoding may be used for this column, which is only ever true for BINARY.
+   *
+   * <p>Off by default, and it has to stay off by default until the format carries a symbol table:
+   * see parquet-format issue #531.
+   */
+  public boolean isFsstEnabled(ColumnDescriptor column) {
+    return getSymbolTableType(column) != null;
+  }
+
+  /**
+   * The symbol table representation to write this column with, or null to not use one.
+   *
+   * <p>One encoding covers every representation, so this is what decides the width of a code and how a
+   * byte that no symbol covers is escaped, and it is where a choice between representations attaches
+   * once there is more than one to choose from. Only single-byte codes are implemented.
+   */
+  public SymbolTableType getSymbolTableType(ColumnDescriptor column) {
+    switch (column.getPrimitiveType().getPrimitiveTypeName()) {
+      case BINARY:
+        return fsstEnabled.getValue(column) ? SymbolTableType.FSST_8 : null;
+      default:
+        return null;
+    }
+  }
+
+  /** How the per-value offsets into a symbol table page's code stream are written. */
+  public OffsetEncoding getSymbolTableOffsetEncoding() {
+    return symbolTableOffsetEncoding;
+  }
+
   public ByteBufferAllocator getAllocator() {
     return allocator;
   }
@@ -455,6 +494,8 @@ public static class Builder {
     private int pageRowCountLimit = DEFAULT_PAGE_ROW_COUNT_LIMIT;
     private boolean pageWriteChecksumEnabled = DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED;
     private final ColumnProperty.Builder<ByteStreamSplitMode> byteStreamSplitEnabled;
+    private final ColumnProperty.Builder<Boolean> fsstEnabled;
+    private OffsetEncoding symbolTableOffsetEncoding = DEFAULT_SYMBOL_TABLE_OFFSET_ENCODING;
     private Map<String, String> extraMetaData = new HashMap<>();
     private final ColumnProperty.Builder<Boolean> statistics;
     private final ColumnProperty.Builder<Boolean> sizeStatistics;
@@ -468,6 +509,7 @@ private Builder() {
               DEFAULT_IS_BYTE_STREAM_SPLIT_ENABLED
                   ? ByteStreamSplitMode.FLOATING_POINT
                   : ByteStreamSplitMode.NONE);
+      fsstEnabled = ColumnProperty.<Boolean>builder().withDefaultValue(DEFAULT_IS_FSST_ENABLED);
       bloomFilterEnabled = ColumnProperty.<Boolean>builder().withDefaultValue(DEFAULT_BLOOM_FILTER_ENABLED);
       bloomFilterNDVs = ColumnProperty.<Long>builder().withDefaultValue(null);
       bloomFilterFPPs = ColumnProperty.<Double>builder().withDefaultValue(DEFAULT_BLOOM_FILTER_FPP);
@@ -504,6 +546,8 @@ private Builder(ParquetProperties toCopy) {
       this.numBloomFilterCandidates = ColumnProperty.builder(toCopy.numBloomFilterCandidates);
       this.maxBloomFilterBytes = toCopy.maxBloomFilterBytes;
       this.byteStreamSplitEnabled = ColumnProperty.builder(toCopy.byteStreamSplitEnabled);
+      this.fsstEnabled = ColumnProperty.builder(toCopy.fsstEnabled);
+      this.symbolTableOffsetEncoding = toCopy.symbolTableOffsetEncoding;
       this.extraMetaData = toCopy.extraMetaData;
       this.statistics = ColumnProperty.builder(toCopy.statistics);
       this.statisticsEnabled = toCopy.statisticsEnabled;
@@ -573,6 +617,51 @@ public Builder withByteStreamSplitEncoding(String columnPath, boolean enable) {
       return this;
     }
 
+    /**
+     * Enable or disable the symbol table encoding for BINARY columns.
+     *
+     * <p>The encoding is not ratified and no file written with it is portable yet, because the format
+     * has nowhere to put the symbol table a column chunk's pages are compressed against: see
+     * parquet-format issue #531. Turning it on without supplying a values writer factory that knows
+     * where to keep the table fails when the first page is written rather than writing a file that
+     * cannot be read.
+     *
+     * @param enable whether the symbol table encoding should be enabled
+     * @return this builder for method chaining.
+     */
+    public Builder withFsstEncoding(boolean enable) {
+      this.fsstEnabled.withDefaultValue(enable);
+      return this;
+    }
+
+    /**
+     * Enable or disable the symbol table encoding for the specified column.
+     *
+     * @param columnPath the path of the column (dot-string)
+     * @param enable     whether the symbol table encoding should be enabled
+     * @return this builder for method chaining.
+     */
+    public Builder withFsstEncoding(String columnPath, boolean enable) {
+      this.fsstEnabled.withValue(columnPath, enable);
+      return this;
+    }
+
+    /**
+     * Set how the per-value offsets into a symbol table page's code stream are written.
+     *
+     * <p>Delta packing by default. Writing them plain costs four bytes a value, which on text that
+     * the encoding halves is around a fifth of the page, so it gives away much of the ratio the
+     * encoding is there for; it is worth having only for a reader that wants the offsets addressable
+     * without decoding them.
+     *
+     * @param offsetEncoding how to write the offset section
+     * @return this builder for method chaining.
+     */
+    public Builder withSymbolTableOffsetEncoding(OffsetEncoding offsetEncoding) {
+      this.symbolTableOffsetEncoding = Objects.requireNonNull(offsetEncoding, "offsetEncoding cannot be null");
+      return this;
+    }
+
     /**
      * Enable or disable BYTE_STREAM_SPLIT encoding for FLOAT, DOUBLE, INT32, INT64 and FIXED_LEN_BYTE_ARRAY columns.
      *

parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java

@@ -39,9 +39,11 @@
 import org.apache.parquet.column.page.DataPageV2;
 import org.apache.parquet.column.page.DictionaryPage;
 import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.page.SymbolTablePage;
 import org.apache.parquet.column.values.RequiresPreviousReader;
 import org.apache.parquet.column.values.ValuesReader;
 import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder;
+import org.apache.parquet.column.values.symboltable.SymbolTable;
 import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.io.api.PrimitiveConverter;
@@ -138,6 +140,7 @@ public double getDouble() {
   private final long totalValueCount;
   private final PageReader pageReader;
   private final Dictionary dictionary;
+  private final SymbolTable symbolTable;
 
   private IntIterator repetitionLevelColumn;
   private IntIterator definitionLevelColumn;
@@ -457,6 +460,8 @@ void writeValue() {
     if (dictionary != null && converter.hasDictionarySupport()) {
       converter.setDictionary(dictionary);
     }
+    SymbolTablePage symbolTablePage = pageReader.readSymbolTablePage();
+    this.symbolTable = symbolTablePage == null ? null : symbolTablePage.decode();
     this.totalValueCount = pageReader.getTotalValueCount();
     if (totalValueCount <= 0) {
       throw new ParquetDecodingException("totalValueCount '" + totalValueCount + "' <= 0");
@@ -703,6 +708,12 @@ private void initDataReader(Encoding dataEncoding, ByteBufferInputStream in, int
             + " as the dictionary was missing for encoding " + dataEncoding);
       }
       this.dataColumn = dataEncoding.getDictionaryBasedValuesReader(path, VALUES, dictionary);
+    } else if (dataEncoding.usesSymbolTable()) {
+      if (symbolTable == null) {
+        throw new ParquetDecodingException("could not read page in col " + path
+            + " as the symbol table was missing for encoding " + dataEncoding);
+      }
+      this.dataColumn = dataEncoding.getSymbolTableBasedValuesReader(path, VALUES, symbolTable);
     } else {
       this.dataColumn = dataEncoding.getValuesReader(path, VALUES);
     }

parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterBase.java

@@ -25,6 +25,7 @@
 import org.apache.parquet.column.ParquetProperties;
 import org.apache.parquet.column.page.DictionaryPage;
 import org.apache.parquet.column.page.PageWriter;
+import org.apache.parquet.column.page.SymbolTablePage;
 import org.apache.parquet.column.statistics.SizeStatistics;
 import org.apache.parquet.column.statistics.Statistics;
 import org.apache.parquet.column.statistics.geospatial.GeospatialStatistics;
@@ -294,6 +295,17 @@ void finalizeColumnChunk() {
         dataColumn.resetDictionary();
       }
 
+      final SymbolTablePage symbolTablePage = dataColumn.toSymbolTablePageAndClose();
+      if (symbolTablePage != null) {
+        if (DEBUG) LOG.debug("write symbol table");
+        try {
+          pageWriter.writeSymbolTablePage(symbolTablePage);
+        } catch (IOException e) {
+          throw new ParquetEncodingException("could not write symbol table page for " + path, e);
+        }
+        dataColumn.resetDictionary();
+      }
+
       collector.finalizeColumnChunk();
     } catch (Throwable t) {
       statusManager.abort();

parquet-column/src/main/java/org/apache/parquet/column/page/PageReader.java

@@ -28,6 +28,13 @@ public interface PageReader {
    */
   DictionaryPage readDictionaryPage();
 
+  /**
+   * @return the symbol table page in that chunk or null if none
+   */
+  default SymbolTablePage readSymbolTablePage() {
+    return null;
+  }
+
   /**
    * @return the total number of values in the column chunk
    */

parquet-column/src/main/java/org/apache/parquet/column/page/PageWriter.java

@@ -226,6 +226,16 @@ default void writePageV2(
    */
   void writeDictionaryPage(DictionaryPage dictionaryPage) throws IOException;
 
+  /**
+   * writes a symbol table page
+   *
+   * @param symbolTablePage the symbol table page containing the table data
+   * @throws IOException if there was an exception while writing
+   */
+  default void writeSymbolTablePage(SymbolTablePage symbolTablePage) throws IOException {
+    throw new UnsupportedOperationException("writeSymbolTablePage is not implemented");
+  }
+
   /**
    * @param prefix a prefix header to add at every line
    * @return a string presenting a summary of how memory is used