+ * 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");
+ }
}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java
index 8fe45e01ef..6c1b4f2ec4 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java
+++ b/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 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.
+ *
+ * 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 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.
+ *
+ * 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.
*
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java
index 2b3e47116c..298f7de831 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnReaderBase.java
+++ b/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);
}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterBase.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterBase.java
index 408627404d..6f33de1ed6 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterBase.java
+++ b/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();
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/page/PageReader.java b/parquet-column/src/main/java/org/apache/parquet/column/page/PageReader.java
index 0b4321ca78..63c04c2af5 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/page/PageReader.java
+++ b/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
*/
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/page/PageWriter.java b/parquet-column/src/main/java/org/apache/parquet/column/page/PageWriter.java
index 1d82db8c32..e184c8befd 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/page/PageWriter.java
+++ b/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
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/page/SymbolTablePage.java b/parquet-column/src/main/java/org/apache/parquet/column/page/SymbolTablePage.java
new file mode 100644
index 0000000000..b38476df30
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/page/SymbolTablePage.java
@@ -0,0 +1,92 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.page;
+
+import java.io.IOException;
+import java.util.Objects;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.values.symboltable.SymbolTable;
+import org.apache.parquet.column.values.symboltable.SymbolTableType;
+import org.apache.parquet.column.values.symboltable.SymbolTables;
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * Data for a symbol table page.
+ *
+ * A symbol table belongs to a column chunk the way a dictionary does: every page of the chunk is
+ * compressed against it, and it has to be readable before any of them. This page is that page's
+ * mirror, one per chunk instead of one per row of a data page.
+ */
+public class SymbolTablePage extends Page {
+
+ private final BytesInput bytes;
+ private final SymbolTableType type;
+
+ /**
+ * creates an uncompressed page
+ *
+ * @param bytes the content of the page
+ * @param type the symbol table representation
+ */
+ public SymbolTablePage(BytesInput bytes, SymbolTableType type) {
+ this(bytes, (int) bytes.size(), type);
+ }
+
+ /**
+ * creates a symbol table page
+ *
+ * @param bytes the (possibly compressed) content of the page
+ * @param uncompressedSize the size uncompressed
+ * @param type the symbol table representation
+ */
+ public SymbolTablePage(BytesInput bytes, int uncompressedSize, SymbolTableType type) {
+ super(Math.toIntExact(bytes.size()), uncompressedSize);
+ this.bytes = Objects.requireNonNull(bytes, "bytes cannot be null");
+ this.type = Objects.requireNonNull(type, "type cannot be null");
+ }
+
+ public BytesInput getBytes() {
+ return bytes;
+ }
+
+ public SymbolTableType getType() {
+ return type;
+ }
+
+ public SymbolTablePage copy() throws IOException {
+ return new SymbolTablePage(BytesInput.copy(bytes), getUncompressedSize(), type);
+ }
+
+ /**
+ * @return the decoded symbol table
+ */
+ public SymbolTable decode() {
+ try {
+ return SymbolTables.deserialize(type, bytes.toByteArray(), 0, (int) bytes.size());
+ } catch (IOException e) {
+ throw new ParquetDecodingException("could not decode the symbol table", e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "SymbolTablePage [bytes.size=" + bytes.size() + ", type=" + type + ", uncompressedSize="
+ + getUncompressedSize() + "]";
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java b/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java
index ecea4a7520..5ab4bdc45b 100755
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java
@@ -21,6 +21,7 @@
import org.apache.parquet.bytes.BytesInput;
import org.apache.parquet.column.Encoding;
import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.SymbolTablePage;
import org.apache.parquet.io.api.Binary;
/**
@@ -72,6 +73,17 @@ public DictionaryPage toDictPageAndClose() {
return null;
}
+ /**
+ * Returns the symbol table generated by this writer if one was created.
+ * As part of this operation the table is closed and will not have
+ * any new values written into it.
+ *
+ * @return the symbol table page or null if not symbol table based
+ */
+ public SymbolTablePage toSymbolTablePageAndClose() {
+ return null;
+ }
+
/**
* reset the dictionary when a new block starts
*/
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java b/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java
index e0d12c2878..f56ce95c64 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java
@@ -93,10 +93,13 @@ private ValuesWriter getFixedLenByteArrayValuesWriter(ColumnDescriptor path) {
}
private ValuesWriter getBinaryValuesWriter(ColumnDescriptor path) {
- ValuesWriter fallbackWriter = new PlainValuesWriter(
- parquetProperties.getInitialSlabSize(),
- parquetProperties.getPageSizeThreshold(),
- parquetProperties.getAllocator());
+ ValuesWriter fallbackWriter = DefaultValuesWriterFactory.symbolTableWriterWithFallBack(
+ path,
+ parquetProperties,
+ new PlainValuesWriter(
+ parquetProperties.getInitialSlabSize(),
+ parquetProperties.getPageSizeThreshold(),
+ parquetProperties.getAllocator()));
return DefaultValuesWriterFactory.dictWriterWithFallBack(
path, parquetProperties, getEncodingForDictionaryPage(), getEncodingForDataPage(), fallbackWriter);
}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV2ValuesWriterFactory.java b/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV2ValuesWriterFactory.java
index c50b4e49c5..8185ad588a 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV2ValuesWriterFactory.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV2ValuesWriterFactory.java
@@ -105,10 +105,13 @@ private ValuesWriter getFixedLenByteArrayValuesWriter(ColumnDescriptor path) {
}
private ValuesWriter getBinaryValuesWriter(ColumnDescriptor path) {
- ValuesWriter fallbackWriter = new DeltaByteArrayWriter(
- parquetProperties.getInitialSlabSize(),
- parquetProperties.getPageSizeThreshold(),
- parquetProperties.getAllocator());
+ ValuesWriter fallbackWriter = DefaultValuesWriterFactory.symbolTableWriterWithFallBack(
+ path,
+ parquetProperties,
+ new DeltaByteArrayWriter(
+ parquetProperties.getInitialSlabSize(),
+ parquetProperties.getPageSizeThreshold(),
+ parquetProperties.getAllocator()));
return DefaultValuesWriterFactory.dictWriterWithFallBack(
path, parquetProperties, getEncodingForDictionaryPage(), getEncodingForDataPage(), fallbackWriter);
}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultValuesWriterFactory.java b/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultValuesWriterFactory.java
index 4c03e6b65e..467737a5ca 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultValuesWriterFactory.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultValuesWriterFactory.java
@@ -25,6 +25,8 @@
import org.apache.parquet.column.values.ValuesWriter;
import org.apache.parquet.column.values.dictionary.DictionaryValuesWriter;
import org.apache.parquet.column.values.fallback.FallbackValuesWriter;
+import org.apache.parquet.column.values.symboltable.SymbolTableType;
+import org.apache.parquet.column.values.symboltable.SymbolTableValuesWriter;
/**
* Handles ValuesWriter creation statically based on the types of the columns and the writer version.
@@ -103,6 +105,29 @@ static DictionaryValuesWriter dictionaryWriter(
}
}
+ /**
+ * Wraps a writer so that a symbol table encoding gets the first attempt at the column, falling back
+ * to the given writer for a chunk whose codes do not come out smaller than the values.
+ *
+ * Returns the writer unchanged when the column is not configured for a symbol table, which is the
+ * default for every column.
+ */
+ static ValuesWriter symbolTableWriterWithFallBack(
+ ColumnDescriptor path, ParquetProperties properties, ValuesWriter writerToFallBackTo) {
+ SymbolTableType type = properties.getSymbolTableType(path);
+ if (type == null) {
+ return writerToFallBackTo;
+ }
+ return FallbackValuesWriter.of(
+ new SymbolTableValuesWriter(
+ type,
+ properties.getSymbolTableOffsetEncoding(),
+ properties.getInitialSlabSize(),
+ properties.getPageSizeThreshold(),
+ properties.getAllocator()),
+ writerToFallBackTo);
+ }
+
static ValuesWriter dictWriterWithFallBack(
ColumnDescriptor path,
ParquetProperties parquetProperties,
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/fallback/FallbackValuesWriter.java b/parquet-column/src/main/java/org/apache/parquet/column/values/fallback/FallbackValuesWriter.java
index 41fe484f37..a9402b0d42 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/fallback/FallbackValuesWriter.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/fallback/FallbackValuesWriter.java
@@ -21,6 +21,7 @@
import org.apache.parquet.bytes.BytesInput;
import org.apache.parquet.column.Encoding;
import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.SymbolTablePage;
import org.apache.parquet.column.values.RequiresFallback;
import org.apache.parquet.column.values.ValuesWriter;
import org.apache.parquet.io.api.Binary;
@@ -51,6 +52,8 @@ public static The decoder works on one value's slice of the code stream, which is what makes a value
+ * addressable without expanding the ones before it.
+ */
+public interface CodeStreamDecoder {
+
+ /**
+ * How many bytes one value's codes expand to, so a caller can size a destination buffer.
+ *
+ * Cheaper than expanding, because it reads the codes without copying symbol bytes.
+ */
+ int expandedLength(byte[] codes, int codesOffset, int codesLength);
+
+ /**
+ * Expands one value's codes.
+ *
+ * @param codes the code stream
+ * @param codesOffset where this value's codes start
+ * @param codesLength how many bytes of codes belong to this value
+ * @param destination where the value bytes are written
+ * @param destinationOffset where to write the value bytes
+ * @return the number of bytes written
+ */
+ int expand(byte[] codes, int codesOffset, int codesLength, byte[] destination, int destinationOffset);
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/CodeStreamEncoder.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/CodeStreamEncoder.java
new file mode 100644
index 0000000000..e06ca7ed0c
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/CodeStreamEncoder.java
@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+/**
+ * Compresses values into a stream of codes over a symbol table.
+ *
+ * Kept separate from the table itself because a table can be paired with more than one code
+ * stream framing, and because the encode side needs lookup structures that never reach a file.
+ */
+public interface CodeStreamEncoder {
+
+ /**
+ * Upper bound on the compressed size of a value, so a caller can size a destination buffer.
+ *
+ * Must account for escapes, which make the worst case larger than the input.
+ */
+ int maxCompressedLength(int rawLength);
+
+ /**
+ * Compresses one value.
+ *
+ * @param source the value bytes
+ * @param sourceOffset where the value starts
+ * @param sourceLength the value's length in bytes
+ * @param destination where the codes are written; must hold {@link #maxCompressedLength} more
+ * bytes at {@code destinationOffset}
+ * @param destinationOffset where to write the codes
+ * @return the number of bytes written
+ */
+ int compress(byte[] source, int sourceOffset, int sourceLength, byte[] destination, int destinationOffset);
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTable.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTable.java
new file mode 100644
index 0000000000..64f207b272
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTable.java
@@ -0,0 +1,54 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+import org.apache.parquet.bytes.BytesInput;
+
+/**
+ * A table mapping codes to the byte sequences they stand for, in the form that is written to a file
+ * and read back from one.
+ *
+ * This is the decode-side and serialization view of a table. It deliberately knows nothing about
+ * how the table was chosen or how values are compressed against it, so that a variant with wider
+ * codes or a differently built table is a new implementation here rather than a change to the
+ * writer and reader above it.
+ */
+public interface SymbolTable {
+
+ SymbolTableType type();
+
+ /** Number of symbols, excluding the escape mechanism. */
+ int symbolCount();
+
+ /** Length in bytes of the symbol with this code. */
+ int symbolLength(int code);
+
+ /**
+ * Appends the symbol with this code to {@code destination} at {@code position}.
+ *
+ * @return the number of bytes written
+ */
+ int copySymbol(int code, byte[] destination, int position);
+
+ /** The serialized table body, as it appears in a file. */
+ BytesInput serialize();
+
+ /** A decoder for code streams written against this table. */
+ CodeStreamDecoder decoder();
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTablePayload.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTablePayload.java
new file mode 100644
index 0000000000..be99940954
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTablePayload.java
@@ -0,0 +1,217 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.delta.DeltaBinaryPackingValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * A data page body holding a code stream and the boundaries between its values.
+ *
+ * The layout is the same for every symbol table codec:
+ *
+ * Offsets are ends rather than starts, so a value's codes run from the previous end to its own
+ * and the first value starts at zero. That is what makes a value readable without expanding the
+ * ones in front of it, and it is why a reader that only wants the tenth value pays for the offsets
+ * alone.
+ *
+ * The number of values restates a field the page header already carries. Both copies are read
+ * and the page header's is treated as the bound: a payload may describe fewer values than the page
+ * holds, never more.
+ *
+ * This class is the read view. {@link SymbolTablePayloadWriter} produces the same layout.
+ */
+public final class SymbolTablePayload {
+
+ /**
+ * How the offset section is encoded.
+ *
+ * Delta encoding is worth having here rather than being a tuning knob: offsets grow with the
+ * page, so on a page whose values compress to a handful of bytes each, four bytes of offset per
+ * value can outweigh the codes they point at.
+ */
+ public enum OffsetEncoding {
+ PLAIN(0),
+ DELTA_BINARY_PACKED(1);
+
+ private final int value;
+
+ OffsetEncoding(int value) {
+ this.value = value;
+ }
+
+ /** The value written as the payload's first byte. */
+ public int value() {
+ return value;
+ }
+
+ static OffsetEncoding fromValue(int value) {
+ for (OffsetEncoding encoding : values()) {
+ if (encoding.value == value) {
+ return encoding;
+ }
+ }
+ throw new ParquetDecodingException("Unsupported symbol table offset encoding: " + value);
+ }
+ }
+
+ /** Size of the fixed part: the offset encoding byte, the value count, the offset section length. */
+ public static final int HEADER_SIZE = 9;
+
+ private final int valueCount;
+ private final byte[] codes;
+ private final int codesOffset;
+ private final int[] endOffsets;
+
+ private SymbolTablePayload(int valueCount, byte[] codes, int codesOffset, int[] endOffsets) {
+ this.valueCount = valueCount;
+ this.codes = codes;
+ this.codesOffset = codesOffset;
+ this.endOffsets = endOffsets;
+ }
+
+ /**
+ * Reads a payload, consuming the rest of the stream.
+ *
+ * @param in positioned at the payload's first byte
+ * @param pageValueCount the value count from the page header, an upper bound on the payload's own
+ */
+ public static SymbolTablePayload parse(ByteBufferInputStream in, int pageValueCount) throws IOException {
+ int length = in.available();
+ if (length < HEADER_SIZE) {
+ throw new ParquetDecodingException(
+ "Symbol table page body is shorter than its " + HEADER_SIZE + "-byte header: " + length);
+ }
+ ByteBuffer header = in.slice(HEADER_SIZE).order(ByteOrder.LITTLE_ENDIAN);
+ OffsetEncoding offsetEncoding = OffsetEncoding.fromValue(header.get(header.position()) & 0xFF);
+ int valueCount = header.getInt(header.position() + 1);
+ int offsetSectionSize = header.getInt(header.position() + 5);
+
+ if (valueCount < 0 || valueCount > pageValueCount) {
+ throw new ParquetDecodingException("Symbol table page describes " + valueCount
+ + " values, but the page header says the page holds " + pageValueCount);
+ }
+ if (offsetSectionSize < 0 || offsetSectionSize > length - HEADER_SIZE) {
+ throw new ParquetDecodingException("Invalid symbol table offset section length: " + offsetSectionSize);
+ }
+
+ int[] endOffsets = readOffsets(in, offsetEncoding, valueCount, offsetSectionSize);
+
+ int codeSectionSize = length - HEADER_SIZE - offsetSectionSize;
+ ByteBuffer codeSection = in.slice(codeSectionSize);
+ byte[] codes;
+ int codesOffset;
+ if (codeSection.hasArray()) {
+ codes = codeSection.array();
+ codesOffset = codeSection.arrayOffset() + codeSection.position();
+ } else {
+ codes = new byte[codeSectionSize];
+ codeSection.get(codes);
+ codesOffset = 0;
+ }
+
+ checkOffsets(endOffsets, codeSectionSize);
+ return new SymbolTablePayload(valueCount, codes, codesOffset, endOffsets);
+ }
+
+ private static int[] readOffsets(
+ ByteBufferInputStream in, OffsetEncoding offsetEncoding, int valueCount, int offsetSectionSize)
+ throws IOException {
+ if (valueCount == 0) {
+ if (offsetSectionSize != 0) {
+ throw new ParquetDecodingException(
+ "Symbol table page holds no values but has a " + offsetSectionSize + "-byte offset section");
+ }
+ return new int[0];
+ }
+ int[] endOffsets = new int[valueCount];
+ if (offsetEncoding == OffsetEncoding.PLAIN) {
+ long expected = (long) valueCount * Integer.BYTES;
+ if (offsetSectionSize != expected) {
+ throw new ParquetDecodingException("Symbol table PLAIN offset section is " + offsetSectionSize
+ + " bytes; expected " + expected + " for " + valueCount + " values");
+ }
+ ByteBuffer offsets = in.slice(offsetSectionSize).order(ByteOrder.LITTLE_ENDIAN);
+ for (int i = 0; i < valueCount; i++) {
+ endOffsets[i] = offsets.getInt(offsets.position() + i * Integer.BYTES);
+ }
+ } else {
+ DeltaBinaryPackingValuesReader offsets = new DeltaBinaryPackingValuesReader();
+ offsets.initFromPage(valueCount, in.sliceStream(offsetSectionSize));
+ for (int i = 0; i < valueCount; i++) {
+ endOffsets[i] = offsets.readInteger();
+ }
+ }
+ return endOffsets;
+ }
+
+ /**
+ * Rejects offsets that do not describe a partition of the code section.
+ *
+ * Every later read trusts these bounds, so they are checked once here rather than per value.
+ * A payload whose offsets stop short of the code section is rejected too: the leftover bytes
+ * would be unreachable, which means the payload is not the one the writer produced.
+ */
+ private static void checkOffsets(int[] endOffsets, int codeSectionSize) {
+ int previous = 0;
+ for (int endOffset : endOffsets) {
+ if (endOffset < previous || endOffset > codeSectionSize) {
+ throw new ParquetDecodingException("Symbol table offsets are not monotonic within a " + codeSectionSize
+ + "-byte code section: " + previous + " then " + endOffset);
+ }
+ previous = endOffset;
+ }
+ if (previous != codeSectionSize) {
+ throw new ParquetDecodingException("Symbol table offsets end at " + previous + " but the code section is "
+ + codeSectionSize + " bytes");
+ }
+ }
+
+ /** How many values this payload holds, which may be fewer than the page holds. */
+ public int valueCount() {
+ return valueCount;
+ }
+
+ /** The array holding the code stream; {@link #codeStart} indexes into it. */
+ public byte[] codes() {
+ return codes;
+ }
+
+ /** Where this value's codes begin in {@link #codes}. */
+ public int codeStart(int index) {
+ return codesOffset + (index == 0 ? 0 : endOffsets[index - 1]);
+ }
+
+ /** How many bytes of codes belong to this value. */
+ public int codeLength(int index) {
+ return index == 0 ? endOffsets[0] : endOffsets[index] - endOffsets[index - 1];
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTablePayloadWriter.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTablePayloadWriter.java
new file mode 100644
index 0000000000..f9efc7a617
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTablePayloadWriter.java
@@ -0,0 +1,128 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
+import org.apache.parquet.column.values.ValuesWriter;
+import org.apache.parquet.column.values.delta.DeltaBinaryPackingValuesWriter;
+import org.apache.parquet.column.values.delta.DeltaBinaryPackingValuesWriterForInteger;
+import org.apache.parquet.column.values.plain.PlainValuesWriter;
+import org.apache.parquet.column.values.symboltable.SymbolTablePayload.OffsetEncoding;
+
+/**
+ * Builds the data page body {@link SymbolTablePayload} reads.
+ *
+ * Holds the code stream and the offsets, so a codec's writer only has to compress values and
+ * hand the codes over. The layout is documented on the read side.
+ */
+public final class SymbolTablePayloadWriter implements AutoCloseable {
+
+ private final OffsetEncoding offsetEncoding;
+ private final CapacityByteArrayOutputStream codes;
+
+ /**
+ * The offset section, written through an existing encoder rather than by hand.
+ *
+ * Plain int32 and delta binary packing are both already spelled out elsewhere in this module,
+ * and a second spelling of either would be a second thing to keep byte-compatible.
+ */
+ private final ValuesWriter offsetWriter;
+
+ private int valueCount;
+
+ public SymbolTablePayloadWriter(
+ OffsetEncoding offsetEncoding, int initialSlabSize, int pageSize, ByteBufferAllocator allocator) {
+ this.offsetEncoding = offsetEncoding;
+ this.codes = new CapacityByteArrayOutputStream(initialSlabSize, pageSize, allocator);
+ this.offsetWriter = offsetEncoding == OffsetEncoding.PLAIN
+ ? new PlainValuesWriter(initialSlabSize, pageSize, allocator)
+ : new DeltaBinaryPackingValuesWriterForInteger(
+ DeltaBinaryPackingValuesWriter.DEFAULT_NUM_BLOCK_VALUES,
+ DeltaBinaryPackingValuesWriter.DEFAULT_NUM_MINIBLOCKS,
+ initialSlabSize,
+ pageSize,
+ allocator);
+ }
+
+ /** Appends one value's codes and records where it ends. */
+ public void addValue(byte[] valueCodes, int offset, int length) {
+ codes.write(valueCodes, offset, length);
+ long end = codes.size();
+ if (end > Integer.MAX_VALUE) {
+ // The offsets are int32 on the wire, so this is a format limit rather than a Java one.
+ throw new IllegalStateException("Symbol table code stream exceeds 2 GB: " + end);
+ }
+ offsetWriter.writeInteger((int) end);
+ valueCount++;
+ }
+
+ public int valueCount() {
+ return valueCount;
+ }
+
+ /** Bytes buffered so far, excluding the fixed header, for a caller watching the page size. */
+ public long bufferedSize() {
+ return codes.size() + offsetWriter.getBufferedSize();
+ }
+
+ public long allocatedSize() {
+ return codes.getCapacity() + offsetWriter.getAllocatedSize();
+ }
+
+ /**
+ * The finished page body.
+ *
+ * A page holding no values gets an empty offset section rather than an encoder's empty-input
+ * output, because the offset encoding's own framing would otherwise be the only thing in the
+ * section and a reader has no values to spend it on.
+ */
+ public BytesInput getBytes() {
+ BytesInput offsets = valueCount == 0 ? BytesInput.empty() : offsetWriter.getBytes();
+ long offsetSectionSize = offsets.size();
+ if (offsetSectionSize > Integer.MAX_VALUE) {
+ throw new IllegalStateException("Symbol table offset section exceeds 2 GB: " + offsetSectionSize);
+ }
+ byte[] header = new byte[SymbolTablePayload.HEADER_SIZE];
+ header[0] = (byte) offsetEncoding.value();
+ writeIntLittleEndian(header, 1, valueCount);
+ writeIntLittleEndian(header, 5, (int) offsetSectionSize);
+ return BytesInput.concat(BytesInput.from(header), offsets, BytesInput.from(codes));
+ }
+
+ private static void writeIntLittleEndian(byte[] destination, int position, int value) {
+ destination[position] = (byte) value;
+ destination[position + 1] = (byte) (value >>> 8);
+ destination[position + 2] = (byte) (value >>> 16);
+ destination[position + 3] = (byte) (value >>> 24);
+ }
+
+ public void reset() {
+ codes.reset();
+ offsetWriter.reset();
+ valueCount = 0;
+ }
+
+ @Override
+ public void close() {
+ codes.close();
+ offsetWriter.close();
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableSource.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableSource.java
new file mode 100644
index 0000000000..3f8dca74bd
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableSource.java
@@ -0,0 +1,39 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+/**
+ * Where a reader gets the symbol table a column chunk's pages were compressed against.
+ *
+ * The write side's counterpart is
+ * {@link org.apache.parquet.column.values.ValuesWriter#toSymbolTablePageAndClose()}, which hands the
+ * table to the column's {@link org.apache.parquet.column.page.PageWriter} once per chunk. The table
+ * here arrives already deserialized, because which implementation to build from the bytes depends
+ * on the representation and that decision belongs in one place: {@link SymbolTables}.
+ */
+public interface SymbolTableSource {
+
+ /**
+ * The table for the chunk being read.
+ *
+ * Called before the first page. Implementations are expected to deserialize once and hand back
+ * the same table for every page of the chunk.
+ */
+ SymbolTable getSymbolTable();
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableTrainer.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableTrainer.java
new file mode 100644
index 0000000000..85e570c95f
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableTrainer.java
@@ -0,0 +1,36 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+/**
+ * Chooses a symbol table for a set of values.
+ *
+ * This is the seam that separates how a table is chosen from how it is written and used. A
+ * different code width, or a different way of mining symbols out of the data, is a new
+ * implementation of this interface and of {@link SymbolTable}, with nothing above them changed.
+ */
+public interface SymbolTableTrainer {
+
+ SymbolTableType type();
+
+ /**
+ * Trains a table on the given values, which the trainer may read in any order and more than once.
+ */
+ TrainedSymbolTable train(ValueBuffer values);
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableType.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableType.java
new file mode 100644
index 0000000000..85441ed986
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableType.java
@@ -0,0 +1,66 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+/**
+ * The representation of a symbol table, which fixes the width of a code in the code stream and the
+ * framing of an escaped literal.
+ *
+ * A column's encoding does not identify the representation on its own: a single encoding covers
+ * every variant, and the type is carried with the table. A reader therefore has to read the table
+ * before it can commit to decoding the column, and may reject a type it does not implement.
+ *
+ * The numeric values are not ratified. Only {@link #FSST_8} appears in the format proposal so
+ * far; {@link #FSST_16}'s value is this implementation's proposal and must be confirmed against
+ * parquet-format issue #531 before any file written with it is treated as portable.
+ */
+public enum SymbolTableType {
+ /** Single-byte codes, at most 255 symbols, escape marker 255 followed by one literal byte. */
+ FSST_8(0, 1),
+
+ /** Two-byte codes, at most 65535 symbols, escape marker 65535 followed by a literal. */
+ FSST_16(1, 2);
+
+ private final int typeValue;
+ private final int codeWidth;
+
+ SymbolTableType(int typeValue, int codeWidth) {
+ this.typeValue = typeValue;
+ this.codeWidth = codeWidth;
+ }
+
+ /** The value written to the symbol table page header. */
+ public int typeValue() {
+ return typeValue;
+ }
+
+ /** Width in bytes of one code in the code stream. */
+ public int codeWidth() {
+ return codeWidth;
+ }
+
+ public static SymbolTableType fromTypeValue(int typeValue) {
+ for (SymbolTableType type : values()) {
+ if (type.typeValue == typeValue) {
+ return type;
+ }
+ }
+ throw new IllegalArgumentException("Unsupported symbol table type: " + typeValue);
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableValuesReader.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableValuesReader.java
new file mode 100644
index 0000000000..e53193b045
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableValuesReader.java
@@ -0,0 +1,118 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+import java.io.IOException;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * Reads the pages {@link SymbolTableValuesWriter} writes.
+ *
+ * One reader serves every symbol table representation, for the same reason one writer does: the
+ * page body is the same shape whatever the table holds, and expanding a value's codes is the table's
+ * own business. Which representation a chunk used is carried with the table rather than by the
+ * column's encoding, so this reader learns it from the {@link SymbolTableSource} and never has to
+ * guess.
+ *
+ * A value is addressable without expanding the ones before it, because the page records where
+ * each value's codes end. That is what makes {@link #skip} cost nothing and is the property the
+ * encoding exists for.
+ */
+public class SymbolTableValuesReader extends ValuesReader {
+
+ private final SymbolTableSource source;
+
+ private SymbolTable table;
+ private CodeStreamDecoder decoder;
+ private SymbolTablePayload payload;
+ private int index;
+
+ /**
+ * A reader with nowhere to get its table from, which cannot decode a page.
+ *
+ * The format does not carry a symbol table yet, so there is no place for a page reader to find
+ * one and nothing to hand this constructor. It exists so that the encoding is complete on the read
+ * side up to that one missing piece; supply a source and the reader works. See parquet-format issue
+ * #531.
+ */
+ public SymbolTableValuesReader() {
+ this(null);
+ }
+
+ public SymbolTableValuesReader(SymbolTableSource source) {
+ this.source = source;
+ }
+
+ @Override
+ public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOException {
+ if (table == null) {
+ if (source == null) {
+ throw new ParquetDecodingException(
+ "Cannot decode a symbol table encoded page without the chunk's symbol table, which the "
+ + "format does not carry yet: see parquet-format issue #531");
+ }
+ // Once per chunk. The table outlives the page: every page of the chunk shares it.
+ table = source.getSymbolTable();
+ decoder = table.decoder();
+ }
+ payload = SymbolTablePayload.parse(in, valueCount);
+ index = 0;
+ }
+
+ /** The representation of the table this reader is decoding against. */
+ public SymbolTableType symbolTableType() {
+ return table == null ? null : table.type();
+ }
+
+ @Override
+ public Binary readBytes() {
+ checkHasValue(1);
+ byte[] codes = payload.codes();
+ int offset = payload.codeStart(index);
+ int length = payload.codeLength(index);
+ index++;
+ byte[] value = new byte[decoder.expandedLength(codes, offset, length)];
+ decoder.expand(codes, offset, length, value, 0);
+ return Binary.fromConstantByteArray(value);
+ }
+
+ @Override
+ public void skip() {
+ skip(1);
+ }
+
+ @Override
+ public void skip(int n) {
+ checkHasValue(n);
+ index += n;
+ }
+
+ private void checkHasValue(int n) {
+ if (payload == null) {
+ throw new ParquetDecodingException("Symbol table reader used before a page was read");
+ }
+ if (index + n > payload.valueCount()) {
+ throw new ParquetDecodingException("Read past the end of a symbol table page: asked for value "
+ + (index + n) + " of " + payload.valueCount());
+ }
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableValuesWriter.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableValuesWriter.java
new file mode 100644
index 0000000000..1833df4905
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTableValuesWriter.java
@@ -0,0 +1,212 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.SymbolTablePage;
+import org.apache.parquet.column.values.RequiresFallback;
+import org.apache.parquet.column.values.ValuesWriter;
+import org.apache.parquet.column.values.symboltable.SymbolTablePayload.OffsetEncoding;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * Writes binary values as codes over a symbol table trained on the data.
+ *
+ * One writer serves every symbol table representation. The representation decides how a table is
+ * trained, how it is serialized and how a code stream is framed, and all three sit behind
+ * {@link SymbolTableTrainer}, {@link SymbolTable} and {@link CodeStreamEncoder}; nothing about them
+ * reaches this class beyond the {@link SymbolTableType} it is asked for.
+ *
+ * Values are buffered raw as they arrive. The table is trained at the first {@link #getBytes()},
+ * on that page's values, and kept for the rest of the chunk: a table belongs to a column chunk, so
+ * a later page must not train its own. It is handed off through {@link #toSymbolTablePageAndClose()},
+ * called once per chunk after every page has been written, which is also what keeps a chunk that
+ * falls back to another encoding from leaving a table behind that no page refers to. Training on
+ * the first page rather than on the whole chunk is deliberate — a page is already far more text
+ * than a trainer samples, and buffering the chunk to feed it would cost a second copy of the
+ * column.
+ *
+ * Buffering is the reason values are held at all: a trainer reads them in an order of its own
+ * choosing and more than once, and a fallback to another encoding has to replay them. The cost is
+ * the page held twice, once raw and once as codes, which is what {@link #getAllocatedSize()}
+ * reports.
+ *
+ * Compressing a page can make it bigger — short values, or values that share nothing with the
+ * rest of the column. Wrapping this writer in a
+ * {@link org.apache.parquet.column.values.fallback.FallbackValuesWriter} over a plain writer is what
+ * makes the encoding safe to turn on: {@link #isCompressionSatisfying} answers the only question
+ * that matters, which is whether the codes came out smaller than the values.
+ */
+public class SymbolTableValuesWriter extends ValuesWriter implements RequiresFallback {
+
+ private final SymbolTableType type;
+ private final ValueBuffer values;
+ private final SymbolTablePayloadWriter payload;
+
+ /** The chunk's table, trained at the first {@link #getBytes()} and reused after that. */
+ private TrainedSymbolTable trained;
+
+ /** Scratch for one value's codes, grown as needed and reused across values. */
+ private byte[] codes = new byte[0];
+
+ /**
+ * How many of the buffered values have been compressed into {@link #payload}.
+ *
+ * Compression happens at {@link #getBytes()}, because it cannot start before the table exists.
+ * Counting what is already done rather than flagging it keeps a second call correct whether or not
+ * more values arrived in between.
+ */
+ private int compressedCount;
+
+ public SymbolTableValuesWriter(
+ SymbolTableType type,
+ OffsetEncoding offsetEncoding,
+ int initialSlabSize,
+ int pageSize,
+ ByteBufferAllocator allocator) {
+ this.type = type;
+ this.values = new ValueBuffer();
+ this.payload = new SymbolTablePayloadWriter(offsetEncoding, initialSlabSize, pageSize, allocator);
+ }
+
+ public SymbolTableType symbolTableType() {
+ return type;
+ }
+
+ @Override
+ public void writeBytes(Binary v) {
+ values.add(v);
+ }
+
+ /**
+ * The size of the values buffered for this page, counted as a plain page would count them.
+ *
+ * Not the compressed size, which is not known until the table has been trained. A page boundary
+ * has to be decided while values are still arriving, so it is decided on what the page would cost
+ * unencoded — the same choice {@link org.apache.parquet.column.values.fallback.FallbackValuesWriter}
+ * makes, and for the same reason: a page sized by its compressed length is a page that becomes too
+ * big the moment the encoding is abandoned.
+ */
+ @Override
+ public long getBufferedSize() {
+ return values.byteCount() + 4L * values.valueCount();
+ }
+
+ @Override
+ public BytesInput getBytes() {
+ if (trained == null) {
+ trained = SymbolTables.trainer(type).train(values);
+ }
+ compressBufferedValues();
+ return payload.getBytes();
+ }
+
+ @Override
+ public SymbolTablePage toSymbolTablePageAndClose() {
+ return trained == null ? null : new SymbolTablePage(trained.table().serialize(), type);
+ }
+
+ @Override
+ public Encoding getEncoding() {
+ return Encoding.FSST;
+ }
+
+ @Override
+ public void reset() {
+ values.reset();
+ payload.reset();
+ compressedCount = 0;
+ }
+
+ /**
+ * Drops the table, which is what a new column chunk needs.
+ *
+ * Named for the dictionary because that is the only chunk-scoped state a values writer had
+ * before this one. In the file writer a values writer does not outlive its chunk, so this is
+ * belt-and-braces rather than the path that runs.
+ */
+ @Override
+ public void resetDictionary() {
+ trained = null;
+ }
+
+ @Override
+ public void close() {
+ payload.close();
+ }
+
+ @Override
+ public long getAllocatedSize() {
+ return values.allocatedSize() + payload.allocatedSize() + codes.length;
+ }
+
+ @Override
+ public String memUsageString(String prefix) {
+ return String.format(
+ "%s %s{raw %d bytes, codes %d bytes}", prefix, type, values.byteCount(), payload.bufferedSize());
+ }
+
+ // RequiresFallback
+
+ /**
+ * Never, because there is no state here that can run away.
+ *
+ * This is the check a dictionary needs, where the encoding stops paying once the dictionary
+ * outgrows the data. A symbol table is bounded by its representation whatever the data does, so
+ * the only question worth asking is the one {@link #isCompressionSatisfying} asks, once, when
+ * there is an answer to it.
+ */
+ @Override
+ public boolean shouldFallBack() {
+ return false;
+ }
+
+ @Override
+ public boolean isCompressionSatisfying(long rawSize, long encodedSize) {
+ return encodedSize < rawSize;
+ }
+
+ @Override
+ public void fallBackAllValuesTo(ValuesWriter writer) {
+ byte[] data = values.data();
+ for (int i = 0; i < values.valueCount(); i++) {
+ writer.writeBytes(Binary.fromReusedByteArray(data, values.offset(i), values.length(i)));
+ }
+ }
+
+ private void compressBufferedValues() {
+ CodeStreamEncoder encoder = trained.encoder();
+ byte[] data = values.data();
+ for (int i = compressedCount; i < values.valueCount(); i++) {
+ int length = values.length(i);
+ int bound = encoder.maxCompressedLength(length);
+ if (codes.length < bound) {
+ codes = new byte[Math.max(bound, codes.length * 2)];
+ }
+ payload.addValue(codes, 0, encoder.compress(data, values.offset(i), length, codes, 0));
+ }
+ compressedCount = values.valueCount();
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTables.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTables.java
new file mode 100644
index 0000000000..3ea11380b9
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/SymbolTables.java
@@ -0,0 +1,60 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+import org.apache.parquet.column.values.symboltable.fsst.Fsst8SymbolTable;
+import org.apache.parquet.column.values.symboltable.fsst.FsstTrainer;
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * The one place that maps a symbol table representation to the code that implements it.
+ *
+ * Both directions live here so that a new representation is a new implementation plus two lines
+ * of dispatch, and so that a reader's rejection of a representation it does not implement happens in
+ * one place with one message.
+ */
+public final class SymbolTables {
+
+ private SymbolTables() {}
+
+ /** A trainer producing tables of this representation. */
+ public static SymbolTableTrainer trainer(SymbolTableType type) {
+ switch (type) {
+ case FSST_8:
+ return new FsstTrainer();
+ default:
+ throw new IllegalArgumentException("No symbol table trainer for " + type);
+ }
+ }
+
+ /**
+ * Rebuilds a table from a serialized body.
+ *
+ * @throws ParquetDecodingException if this implementation cannot read the representation, which a
+ * reader is allowed to do and is not the same as the file being corrupt
+ */
+ public static SymbolTable deserialize(SymbolTableType type, byte[] body, int offset, int length) {
+ switch (type) {
+ case FSST_8:
+ return Fsst8SymbolTable.deserialize(body, offset, length);
+ default:
+ throw new ParquetDecodingException("Unsupported symbol table type: " + type);
+ }
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/TrainedSymbolTable.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/TrainedSymbolTable.java
new file mode 100644
index 0000000000..234e6202ac
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/TrainedSymbolTable.java
@@ -0,0 +1,44 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+/**
+ * What training produces: a table to write to the file, and an encoder that compresses against it.
+ *
+ * The two come out together because the encode-side lookup structures are a byproduct of
+ * training and rebuilding them from the serialized table would be wasted work.
+ */
+public final class TrainedSymbolTable {
+
+ private final SymbolTable table;
+ private final CodeStreamEncoder encoder;
+
+ public TrainedSymbolTable(SymbolTable table, CodeStreamEncoder encoder) {
+ this.table = table;
+ this.encoder = encoder;
+ }
+
+ public SymbolTable table() {
+ return table;
+ }
+
+ public CodeStreamEncoder encoder() {
+ return encoder;
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/ValueBuffer.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/ValueBuffer.java
new file mode 100644
index 0000000000..b6209c7f47
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/ValueBuffer.java
@@ -0,0 +1,135 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable;
+
+import java.util.Arrays;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * The values of one page, held as a single byte array plus start offsets.
+ *
+ * Training needs to read values in an order of its own choosing, and more than once, so the
+ * values have to be somewhere addressable when the page is closed. Holding them as one array rather
+ * than as a list of separately allocated values keeps that at one copy of the page with no
+ * per-value object, which matters because a writer is already buffering this data anyway.
+ *
+ * The array carries eight bytes of zero padding past the last value. Symbol matching loads eight
+ * bytes at a time and deliberately reads past the end of a short value; the padding is what makes
+ * that read in-bounds, and zeros are what make it harmless, since no symbol can match beyond the
+ * value's own length once the load is masked.
+ */
+public class ValueBuffer {
+
+ /** Slack past the last value, for eight-byte loads that overshoot. */
+ public static final int TAIL_PADDING = 8;
+
+ private static final int DEFAULT_INITIAL_CAPACITY = 64 * 1024;
+ private static final int DEFAULT_INITIAL_VALUES = 1024;
+
+ private byte[] data;
+ private int size;
+ private int[] offsets;
+ private int valueCount;
+
+ public ValueBuffer() {
+ this(DEFAULT_INITIAL_CAPACITY, DEFAULT_INITIAL_VALUES);
+ }
+
+ public ValueBuffer(int initialCapacity, int initialValues) {
+ this.data = new byte[initialCapacity + TAIL_PADDING];
+ this.offsets = new int[initialValues + 1];
+ this.offsets[0] = 0;
+ }
+
+ public void add(Binary value) {
+ int length = value.length();
+ ensureCapacity(size + length);
+ ensureValues(valueCount + 1);
+ // Read through a duplicate so the value's own buffer position is left alone. This is the one
+ // copy of the value; taking the backing array instead would copy for slice-backed values.
+ value.toByteBuffer().duplicate().get(data, size, length);
+ size += length;
+ offsets[++valueCount] = size;
+ }
+
+ public void add(byte[] source, int offset, int length) {
+ ensureCapacity(size + length);
+ ensureValues(valueCount + 1);
+ System.arraycopy(source, offset, data, size, length);
+ size += length;
+ offsets[++valueCount] = size;
+ }
+
+ public int valueCount() {
+ return valueCount;
+ }
+
+ /** Total length of all values, which is the size of the region {@link #data} holds. */
+ public int byteCount() {
+ return size;
+ }
+
+ /** Bytes held by the two arrays, which is what this buffer costs a writer's memory budget. */
+ public long allocatedSize() {
+ return data.length + 4L * offsets.length;
+ }
+
+ /**
+ * The backing array. Valid from 0 to {@link #byteCount()}, followed by {@link #TAIL_PADDING} zero
+ * bytes that a reader may load but must not interpret.
+ */
+ public byte[] data() {
+ return data;
+ }
+
+ public int offset(int index) {
+ return offsets[index];
+ }
+
+ public int length(int index) {
+ return offsets[index + 1] - offsets[index];
+ }
+
+ public BytesInput asBytesInput() {
+ return BytesInput.from(data, 0, size);
+ }
+
+ public void reset() {
+ // Clear through the padding so the next round of eight-byte overshoot still reads zeros.
+ Arrays.fill(data, 0, Math.min(size + TAIL_PADDING, data.length), (byte) 0);
+ size = 0;
+ valueCount = 0;
+ }
+
+ private void ensureCapacity(int required) {
+ if (required + TAIL_PADDING <= data.length) {
+ return;
+ }
+ int capacity = Math.max(data.length * 2, required + TAIL_PADDING);
+ data = Arrays.copyOf(data, capacity);
+ }
+
+ private void ensureValues(int required) {
+ if (required < offsets.length) {
+ return;
+ }
+ offsets = Arrays.copyOf(offsets, Math.max(offsets.length * 2, required + 1));
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8CodeStreamDecoder.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8CodeStreamDecoder.java
new file mode 100644
index 0000000000..9aeb2766d1
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8CodeStreamDecoder.java
@@ -0,0 +1,93 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable.fsst;
+
+import org.apache.parquet.column.values.symboltable.CodeStreamDecoder;
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * Expands 8-bit FSST codes back into value bytes.
+ *
+ * Decoding needs nothing but the table: each code either names a symbol to copy or announces that
+ * the next byte stands for itself. There is no state carried between values, which is what lets a
+ * reader expand one value without touching the ones before it.
+ */
+class Fsst8CodeStreamDecoder implements CodeStreamDecoder {
+
+ private final byte[] symbolBytes;
+ private final int[] symbolOffsets;
+ private final int symbolCount;
+
+ Fsst8CodeStreamDecoder(byte[] symbolBytes, int[] symbolOffsets, int symbolCount) {
+ this.symbolBytes = symbolBytes;
+ this.symbolOffsets = symbolOffsets;
+ this.symbolCount = symbolCount;
+ }
+
+ @Override
+ public int expandedLength(byte[] codes, int codesOffset, int codesLength) {
+ int length = 0;
+ int position = codesOffset;
+ int end = codesOffset + codesLength;
+ while (position < end) {
+ int code = codes[position++] & 0xFF;
+ if (code == Fsst8SymbolTable.ESCAPE) {
+ requireLiteral(position, end);
+ position++;
+ length++;
+ } else {
+ length += symbolLength(code);
+ }
+ }
+ return length;
+ }
+
+ @Override
+ public int expand(byte[] codes, int codesOffset, int codesLength, byte[] destination, int destinationOffset) {
+ int written = destinationOffset;
+ int position = codesOffset;
+ int end = codesOffset + codesLength;
+ while (position < end) {
+ int code = codes[position++] & 0xFF;
+ if (code == Fsst8SymbolTable.ESCAPE) {
+ requireLiteral(position, end);
+ destination[written++] = codes[position++];
+ } else {
+ int length = symbolLength(code);
+ System.arraycopy(symbolBytes, symbolOffsets[code], destination, written, length);
+ written += length;
+ }
+ }
+ return written - destinationOffset;
+ }
+
+ private int symbolLength(int code) {
+ if (code >= symbolCount) {
+ throw new ParquetDecodingException(
+ "FSST value references symbol code " + code + " but the table holds " + symbolCount + " symbols");
+ }
+ return symbolOffsets[code + 1] - symbolOffsets[code];
+ }
+
+ private static void requireLiteral(int position, int end) {
+ if (position >= end) {
+ throw new ParquetDecodingException("FSST value ends with an escape and no literal byte");
+ }
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8CodeStreamEncoder.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8CodeStreamEncoder.java
new file mode 100644
index 0000000000..5aba4b4fbd
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8CodeStreamEncoder.java
@@ -0,0 +1,120 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable.fsst;
+
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_BASE;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.ICL_FREE;
+
+import org.apache.parquet.column.values.symboltable.CodeStreamEncoder;
+
+/**
+ * Compresses value bytes into 8-bit FSST codes.
+ *
+ * Two things here are not free choices. The first is chunking: a value is compressed 511 bytes at
+ * a time, with the table's separator byte written just past each chunk. That bound is what the
+ * reference implementation's vectorized compressor uses, and matching it is what keeps the code
+ * stream identical whichever compressor produced it. Since no symbol contains the separator, no
+ * match can reach across a chunk boundary, so the boundary is also what makes the eight-byte read at
+ * the end of a chunk safe without a shorter tail path.
+ *
+ * The compression loop is a port of FSST's reference implementation by Peter Boncz, Viktor Leis
+ * and Thomas Neumann (CWI / TU Munich), MIT licensed, at https://github.com/cwida/fsst, commit
+ * 89f49c580c6388acf3b6ed2a49e1bfde6c05e616. Its three hand-tuned variants differ only in branch
+ * layout and produce the same bytes, so one is enough here.
+ *
+ * The second is the code translation. The trainer numbers symbols in an order that lets the
+ * compressor skip work, and a file numbers them by length. So every code leaves this class through
+ * the permutation between the two; an escape passes through untouched, since it is a fixed marker in
+ * both orders rather than a symbol.
+ */
+class Fsst8CodeStreamEncoder implements CodeStreamEncoder {
+
+ /** Bytes compressed at a time, matching the reference implementation's vectorized compressor. */
+ private static final int CHUNK_SIZE = 511;
+
+ /** Room past the chunk for the separator byte and for an eight-byte read at the chunk's end. */
+ private static final int CHUNK_PADDING = 8;
+
+ private final FsstSymbolTable table;
+ private final byte[] fileCodeForTrainerCode;
+ private final byte[] chunk = new byte[CHUNK_SIZE + CHUNK_PADDING];
+
+ Fsst8CodeStreamEncoder(FsstSymbolTable table, Fsst8SymbolTable fileTable) {
+ this.table = table;
+ this.fileCodeForTrainerCode = fileTable.encodeCodeMap();
+ }
+
+ @Override
+ public int maxCompressedLength(int rawLength) {
+ // Every byte could escape, which costs the marker plus the byte itself.
+ return 2 * rawLength;
+ }
+
+ @Override
+ public int compress(byte[] source, int sourceOffset, int sourceLength, byte[] destination, int destinationOffset) {
+ int written = destinationOffset;
+ int consumed = 0;
+ while (consumed < sourceLength) {
+ int chunkLength = Math.min(CHUNK_SIZE, sourceLength - consumed);
+ System.arraycopy(source, sourceOffset + consumed, chunk, 0, chunkLength);
+ // Bytes past the separator keep whatever a longer previous chunk left there. That is harmless,
+ // and is what the reference implementation does: a match reaching past the separator would
+ // have to contain it, and no symbol does.
+ chunk[chunkLength] = (byte) table.terminator;
+ written = compressChunk(chunkLength, destination, written);
+ consumed += chunkLength;
+ }
+ return written - destinationOffset;
+ }
+
+ /** Compresses one chunk out of {@link #chunk}, returning the new write position. */
+ private int compressChunk(int chunkLength, byte[] destination, int writePosition) {
+ char[] shortCodes = table.shortCodes;
+ long[] hashValues = table.hashValues;
+ long[] hashDescriptors = table.hashDescriptors;
+ byte[] codeMap = fileCodeForTrainerCode;
+ int byteLimit = table.byteLimit;
+ int written = writePosition;
+ int position = 0;
+ while (position < chunkLength) {
+ long word = FsstCodes.loadWordUnchecked(chunk, position);
+ int shortCode = shortCodes[(int) (word & 0xFFFF)];
+ int bucket = FsstCodes.hashBucket(word);
+ long descriptor = hashDescriptors[bucket];
+ if (descriptor < ICL_FREE && hashValues[bucket] == FsstCodes.maskWord(word, descriptor)) {
+ // A symbol of three bytes or more.
+ destination[written++] = codeMap[FsstCodes.code(descriptor)];
+ position += FsstCodes.length(descriptor);
+ } else if ((shortCode & 0xFF) < byteLimit) {
+ // A two-byte symbol, and the miss above rules out a longer one starting here.
+ destination[written++] = codeMap[shortCode & 0xFF];
+ position += 2;
+ } else if ((shortCode & CODE_BASE) != 0) {
+ // No symbol matches, so the byte stands for itself behind the escape marker.
+ destination[written++] = (byte) Fsst8SymbolTable.ESCAPE;
+ destination[written++] = (byte) word;
+ position += 1;
+ } else {
+ destination[written++] = codeMap[shortCode & 0xFF];
+ position += 1;
+ }
+ }
+ return written;
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8SymbolTable.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8SymbolTable.java
new file mode 100644
index 0000000000..0f8a86359e
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/Fsst8SymbolTable.java
@@ -0,0 +1,213 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable.fsst;
+
+import java.util.Arrays;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.values.symboltable.CodeStreamDecoder;
+import org.apache.parquet.column.values.symboltable.SymbolTable;
+import org.apache.parquet.column.values.symboltable.SymbolTableType;
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * An 8-bit FSST symbol table in the form a file carries it.
+ *
+ * The codes here are not the codes the trainer works with. The trainer numbers symbols in an order
+ * that suits compression, and the format numbers them in order of length, shortest first, so that a
+ * reader can rebuild the table from a length histogram without storing a length per symbol. The
+ * permutation between the two is built once, when a trained table is converted, and is what the
+ * encoder applies to every code it emits.
+ *
+ * Serialized body: one byte holding the symbol count, then eight bytes counting the symbols of
+ * each length from one to eight, then the symbol bytes end to end in that same order. At most 255
+ * symbols of at most 8 bytes each, so at most 2049 bytes. That layout is single-byte-code only, which
+ * is why the 16-bit variant needs a table format of its own rather than an extra field here.
+ */
+public class Fsst8SymbolTable implements SymbolTable {
+
+ /** Code standing for one literal byte, which follows it in the code stream. */
+ public static final int ESCAPE = 0xFF;
+
+ static final int MAX_SYMBOL_LENGTH = 8;
+
+ /** Symbol count plus the eight histogram entries. */
+ static final int HEADER_SIZE = 9;
+
+ /** Largest serialized body: the header plus 255 symbols of 8 bytes. */
+ static final int MAX_SERIALIZED_SIZE = HEADER_SIZE + FsstCodes.MAX_SYMBOLS * MAX_SYMBOL_LENGTH;
+
+ /** Symbol bytes end to end, in code order. */
+ private final byte[] symbolBytes;
+
+ /** Where each symbol starts in {@link #symbolBytes}, with one extra entry for the end. */
+ private final int[] symbolOffsets;
+
+ private final int symbolCount;
+
+ /**
+ * Maps a trainer code to the code written to a file, or {@link #ESCAPE} where the trainer has no
+ * such symbol. Null on a table read back from a file, which never encodes.
+ */
+ private final byte[] encodeCodeMap;
+
+ private Fsst8SymbolTable(byte[] symbolBytes, int[] symbolOffsets, int symbolCount, byte[] encodeCodeMap) {
+ this.symbolBytes = symbolBytes;
+ this.symbolOffsets = symbolOffsets;
+ this.symbolCount = symbolCount;
+ this.encodeCodeMap = encodeCodeMap;
+ }
+
+ /**
+ * Converts a trained table into the file representation, renumbering symbols into length order.
+ *
+ * Symbols of the same length keep their relative order from the trainer, which is what makes the
+ * renumbering reproducible.
+ */
+ static Fsst8SymbolTable of(FsstSymbolTable trained) {
+ int symbolCount = trained.symbolCount;
+ if (symbolCount > FsstCodes.MAX_SYMBOLS) {
+ throw new IllegalArgumentException(
+ "FSST8 holds at most " + FsstCodes.MAX_SYMBOLS + " symbols, got " + symbolCount);
+ }
+ // A counting sort by length, which is stable and needs no comparator.
+ int[] countByLength = new int[MAX_SYMBOL_LENGTH + 1];
+ for (int code = 0; code < symbolCount; code++) {
+ countByLength[FsstCodes.length(trained.symbolDescriptors[code])]++;
+ }
+ int[] nextCodeForLength = new int[MAX_SYMBOL_LENGTH + 2];
+ int[] nextOffsetForLength = new int[MAX_SYMBOL_LENGTH + 2];
+ int runningCode = 0;
+ int runningOffset = 0;
+ for (int length = 1; length <= MAX_SYMBOL_LENGTH; length++) {
+ nextCodeForLength[length] = runningCode;
+ nextOffsetForLength[length] = runningOffset;
+ runningCode += countByLength[length];
+ runningOffset += countByLength[length] * length;
+ }
+
+ byte[] symbolBytes = new byte[runningOffset];
+ int[] symbolOffsets = new int[symbolCount + 1];
+ byte[] encodeCodeMap = new byte[FsstCodes.MAX_SYMBOLS];
+ Arrays.fill(encodeCodeMap, (byte) ESCAPE);
+
+ for (int trainerCode = 0; trainerCode < symbolCount; trainerCode++) {
+ int length = FsstCodes.length(trained.symbolDescriptors[trainerCode]);
+ int fileCode = nextCodeForLength[length]++;
+ int offset = nextOffsetForLength[length];
+ nextOffsetForLength[length] += length;
+ long value = trained.symbolValues[trainerCode];
+ for (int i = 0; i < length; i++) {
+ symbolBytes[offset + i] = (byte) (value >>> (i * 8));
+ }
+ symbolOffsets[fileCode] = offset;
+ encodeCodeMap[trainerCode] = (byte) fileCode;
+ }
+ symbolOffsets[symbolCount] = runningOffset;
+ return new Fsst8SymbolTable(symbolBytes, symbolOffsets, symbolCount, encodeCodeMap);
+ }
+
+ /** Reads a table back from the bytes a file carries. */
+ public static Fsst8SymbolTable deserialize(byte[] body, int offset, int length) {
+ if (length < HEADER_SIZE || length > MAX_SERIALIZED_SIZE) {
+ throw new ParquetDecodingException("Invalid FSST symbol table body size: " + length);
+ }
+ int symbolCount = body[offset] & 0xFF;
+ int[] histogram = new int[MAX_SYMBOL_LENGTH];
+ int histogramSum = 0;
+ int expectedSymbolBytes = 0;
+ for (int i = 0; i < MAX_SYMBOL_LENGTH; i++) {
+ histogram[i] = body[offset + 1 + i] & 0xFF;
+ histogramSum += histogram[i];
+ expectedSymbolBytes += histogram[i] * (i + 1);
+ }
+ // Validate before allocating, so a corrupt header cannot ask for a large buffer.
+ if (histogramSum != symbolCount) {
+ throw new ParquetDecodingException("FSST length histogram sums to " + histogramSum
+ + " but the table declares " + symbolCount + " symbols");
+ }
+ if (expectedSymbolBytes != length - HEADER_SIZE) {
+ throw new ParquetDecodingException("FSST symbol bytes are " + (length - HEADER_SIZE)
+ + " bytes but the length histogram accounts for " + expectedSymbolBytes);
+ }
+
+ byte[] symbolBytes = new byte[expectedSymbolBytes];
+ System.arraycopy(body, offset + HEADER_SIZE, symbolBytes, 0, expectedSymbolBytes);
+ int[] symbolOffsets = new int[symbolCount + 1];
+ int code = 0;
+ int position = 0;
+ for (int length1 = 1; length1 <= MAX_SYMBOL_LENGTH; length1++) {
+ for (int i = 0; i < histogram[length1 - 1]; i++) {
+ symbolOffsets[code++] = position;
+ position += length1;
+ }
+ }
+ symbolOffsets[symbolCount] = position;
+ return new Fsst8SymbolTable(symbolBytes, symbolOffsets, symbolCount, null);
+ }
+
+ @Override
+ public SymbolTableType type() {
+ return SymbolTableType.FSST_8;
+ }
+
+ @Override
+ public int symbolCount() {
+ return symbolCount;
+ }
+
+ @Override
+ public int symbolLength(int code) {
+ return symbolOffsets[code + 1] - symbolOffsets[code];
+ }
+
+ @Override
+ public int copySymbol(int code, byte[] destination, int position) {
+ int length = symbolLength(code);
+ System.arraycopy(symbolBytes, symbolOffsets[code], destination, position, length);
+ return length;
+ }
+
+ @Override
+ public BytesInput serialize() {
+ int[] histogram = new int[MAX_SYMBOL_LENGTH];
+ for (int code = 0; code < symbolCount; code++) {
+ histogram[symbolLength(code) - 1]++;
+ }
+ byte[] body = new byte[HEADER_SIZE + symbolBytes.length];
+ body[0] = (byte) symbolCount;
+ for (int i = 0; i < MAX_SYMBOL_LENGTH; i++) {
+ body[1 + i] = (byte) histogram[i];
+ }
+ System.arraycopy(symbolBytes, 0, body, HEADER_SIZE, symbolBytes.length);
+ return BytesInput.from(body);
+ }
+
+ @Override
+ public CodeStreamDecoder decoder() {
+ return new Fsst8CodeStreamDecoder(symbolBytes, symbolOffsets, symbolCount);
+ }
+
+ /** The trainer-code to file-code map, for the encoder paired with this table. */
+ byte[] encodeCodeMap() {
+ if (encodeCodeMap == null) {
+ throw new IllegalStateException("a symbol table read from a file cannot encode");
+ }
+ return encodeCodeMap;
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstCodes.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstCodes.java
new file mode 100644
index 0000000000..05b629b993
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstCodes.java
@@ -0,0 +1,161 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable.fsst;
+
+/**
+ * Constants and bit-field helpers for the FSST symbol representation.
+ *
+ * This file is a Java port of parts of the FSST reference implementation by Peter Boncz,
+ * Viktor Leis and Thomas Neumann (CWI / TU Munich), distributed under the MIT license and
+ * available at https://github.com/cwida/fsst. The port follows commit
+ * 89f49c580c6388acf3b6ed2a49e1bfde6c05e616. Encoding decisions are reproduced exactly, because
+ * the symbol table a writer produces has to match what other implementations produce for the
+ * same input.
+ *
+ * A symbol is a byte sequence of length 1 to 8, held as a little-endian {@code long} together
+ * with a packed descriptor called {@code icl}: {@code (length << 28) | (code << 16) |
+ * ignoredBits}, where {@code ignoredBits} is {@code (8 - length) * 8}. The descriptor is kept in
+ * a {@code long} rather than an {@code int} on purpose: at length 8 the length field alone is
+ * {@code 8 << 28}, which overflows a signed 32-bit integer. Because every descriptor fits in 32
+ * bits, signed {@code long} comparison of two descriptors matches the unsigned comparison the
+ * reference implementation performs.
+ */
+final class FsstCodes {
+
+ private FsstCodes() {}
+
+ /** Maximum number of bytes in one symbol. */
+ static final int MAX_SYMBOL_LENGTH = 8;
+
+ /** Width of the length field in the packed values held by the code lookup tables. */
+ static final int LEN_BITS = 12;
+
+ static final int CODE_BITS = 9;
+
+ /** Codes below this value are pseudo codes standing for an escaped single byte. */
+ static final int CODE_BASE = 256;
+
+ /** One past the highest representable code; also marks a symbol with no code assigned yet. */
+ static final int CODE_MAX = 1 << CODE_BITS;
+
+ static final int CODE_MASK = CODE_MAX - 1;
+
+ /** Number of buckets in the symbol hash table, which holds symbols of three bytes or more. */
+ static final int HASH_TAB_SIZE = 1 << 10;
+
+ /** Descriptor value marking a free hash bucket: length 15 and an unassigned code. */
+ static final long ICL_FREE = (15L << 28) | ((long) CODE_MASK << 16);
+
+ /** Highest number of symbols a table may hold, so that every code fits in one byte. */
+ static final int MAX_SYMBOLS = 255;
+
+ private static final long HASH_PRIME = 2971215073L;
+ private static final int HASH_SHIFT = 15;
+
+ /**
+ * The reference implementation's hash, used both for symbol lookup and to drive the sampler.
+ *
+ * The shift must be unsigned. For symbol lookup the input is the next three bytes, so the
+ * product cannot reach the sign bit and the distinction does not arise; the sampler chains this
+ * function over full 64-bit values, where a signed shift would send the whole sequence down a
+ * different path and produce a different symbol table.
+ */
+ static long hash(long w) {
+ long product = w * HASH_PRIME;
+ return product ^ (product >>> HASH_SHIFT);
+ }
+
+ /** Packs a symbol descriptor from a code and a length. */
+ static long icl(int code, int length) {
+ return ((long) length << 28) | ((long) code << 16) | ((long) (MAX_SYMBOL_LENGTH - length) * 8);
+ }
+
+ static int length(long icl) {
+ return (int) (icl >>> 28);
+ }
+
+ static int code(long icl) {
+ return (int) ((icl >>> 16) & CODE_MASK);
+ }
+
+ /** Number of high bits to clear in an input word before comparing it against a symbol. */
+ static int ignoredBits(long icl) {
+ return (int) (icl & 0xFF);
+ }
+
+ /** The symbol's first byte, as an unsigned value. */
+ static int first(long val) {
+ return (int) (val & 0xFF);
+ }
+
+ /** The symbol's first two bytes, as an unsigned value. */
+ static int first2(long val) {
+ return (int) (val & 0xFFFF);
+ }
+
+ /** Hash bucket for a symbol, keyed on its first three bytes. */
+ static int hashBucket(long val) {
+ return (int) (hash(val & 0xFFFFFF) & (HASH_TAB_SIZE - 1));
+ }
+
+ /** Builds the little-endian word for the first {@code length} bytes at {@code offset}. */
+ static long loadSymbolBytes(byte[] src, int offset, int length) {
+ long val = 0;
+ for (int i = 0; i < length; i++) {
+ val |= (long) (src[offset + i] & 0xFF) << (i * 8);
+ }
+ return val;
+ }
+
+ /**
+ * Reads eight bytes little-endian, substituting zeros past {@code limit}.
+ *
+ * The reference implementation reads eight bytes unconditionally and relies on the input
+ * buffer being padded. Java cannot read out of bounds, so the tail is zero-filled here; callers
+ * that must reproduce the reference behaviour byte for byte pad their input instead of relying
+ * on this.
+ */
+ static long loadWord(byte[] src, int offset, int limit) {
+ long val = 0;
+ int n = Math.min(MAX_SYMBOL_LENGTH, limit - offset);
+ for (int i = 0; i < n; i++) {
+ val |= (long) (src[offset + i] & 0xFF) << (i * 8);
+ }
+ return val;
+ }
+
+ /**
+ * Reads eight bytes little-endian with no bounds reasoning beyond the array itself.
+ *
+ * For the compressor's inner loop, which works out of a buffer padded past its logical end so
+ * that the read is always in bounds. A byte-array view is used rather than eight shifts because
+ * this is the single hottest read in the codec.
+ */
+ static long loadWordUnchecked(byte[] src, int offset) {
+ return (long) LITTLE_ENDIAN_LONG.get(src, offset);
+ }
+
+ private static final java.lang.invoke.VarHandle LITTLE_ENDIAN_LONG =
+ java.lang.invoke.MethodHandles.byteArrayViewVarHandle(long[].class, java.nio.ByteOrder.LITTLE_ENDIAN);
+
+ /** Clears the high bits of {@code word} that a symbol with this descriptor ignores. */
+ static long maskWord(long word, long icl) {
+ return word & (-1L >>> ignoredBits(icl));
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstCounters.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstCounters.java
new file mode 100644
index 0000000000..fca687996b
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstCounters.java
@@ -0,0 +1,162 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable.fsst;
+
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_MAX;
+
+import java.util.Arrays;
+
+/**
+ * Occurrence counters used while training a symbol table: how often each symbol occurs, and how
+ * often each ordered pair of symbols occurs back to back.
+ *
+ * Java port of the 64-bit {@code Counters} from the FSST reference implementation by Peter Boncz,
+ * Viktor Leis and Thomas Neumann (CWI / TU Munich), MIT licensed, at https://github.com/cwida/fsst,
+ * commit 89f49c580c6388acf3b6ed2a49e1bfde6c05e616.
+ *
+ * Each counter is split into a low byte and a high part, and the pair counter's high part is only
+ * four bits wide, two counters to a byte. That is not only a space optimization to be simplified
+ * away: four bits saturate, so the split decides which candidate symbols survive a training round
+ * and therefore which table comes out. The reference implementation has a wider variant for 32-bit
+ * platforms that produces different tables; this follows the 64-bit variant, which is what other
+ * implementations run.
+ *
+ * The high part is incremented when the low byte wraps from zero rather than when it saturates,
+ * which is what makes a nonzero high part equivalent to a nonzero count and lets the scans below
+ * skip runs of empty counters eight bytes at a time.
+ *
+ * Those scans deliberately read past the counter they were asked about. The reference
+ * implementation lets the read spill into whichever array is laid out next; here each array carries
+ * eight bytes of zero padding instead. The two agree: a spilled read only changes the result if
+ * every valid counter it covers is zero, and in that case the scan has already advanced past the end
+ * of the range and returns zero either way.
+ */
+final class FsstCounters {
+
+ /** Slack for scans that read eight bytes from the last counter. */
+ private static final int PADDING = 8;
+
+ private final byte[] count1High = new byte[CODE_MAX + PADDING];
+ private final byte[] count1Low = new byte[CODE_MAX + PADDING];
+ private final byte[] count2High = new byte[CODE_MAX * (CODE_MAX / 2) + PADDING];
+ private final byte[] count2Low = new byte[CODE_MAX * CODE_MAX + PADDING];
+
+ /** Position reached by the most recent scan, which the caller must adopt. */
+ private int scanPosition;
+
+ void reset() {
+ Arrays.fill(count1High, (byte) 0);
+ Arrays.fill(count1Low, (byte) 0);
+ Arrays.fill(count2High, (byte) 0);
+ Arrays.fill(count2Low, (byte) 0);
+ }
+
+ void count1Set(int position, int value) {
+ count1Low[position] = (byte) (value & 255);
+ count1High[position] = (byte) (value >>> 8);
+ }
+
+ void count1Increment(int position) {
+ // Post-increment on a byte array wraps, and yields the value before the increment, so this
+ // raises the high part exactly when the low byte wraps back to zero.
+ if (count1Low[position]++ == 0) {
+ count1High[position]++;
+ }
+ }
+
+ void count2Increment(int first, int second) {
+ if (count2Low[first * CODE_MAX + second]++ == 0) {
+ // Add one to the four-bit counter, in the low or the high nibble according to parity.
+ count2High[first * (CODE_MAX / 2) + (second >> 1)] += (byte) (1 << ((second & 1) << 2));
+ }
+ }
+
+ /**
+ * Reads the counter for a single symbol, skipping forward over empty counters.
+ *
+ * @return the count, or zero once the scan leaves the range. Either way the caller must adopt
+ * {@link #scanPosition()} as its new position.
+ */
+ int count1Next(int position) {
+ long high = loadLittleEndianLong(count1High, position);
+ int zeroBytes = (high != 0) ? (Long.numberOfTrailingZeros(high) >>> 3) : 7;
+ high = (high >>> (zeroBytes << 3)) & 255;
+ position += zeroBytes;
+ scanPosition = position;
+ if (position >= CODE_MAX || high == 0) {
+ return 0;
+ }
+ int low = count1Low[position] & 0xFF;
+ if (low != 0) {
+ high--; // the high part was raised early
+ }
+ return (int) ((high << 8) + low);
+ }
+
+ /**
+ * Reads the counter for a pair of symbols, skipping forward over empty counters.
+ *
+ * @return the count, or zero once the scan leaves the range. Either way the caller must adopt
+ * {@link #scanPosition()} as its new second position.
+ */
+ int count2Next(int first, int second) {
+ long high = loadLittleEndianLong(count2High, first * (CODE_MAX / 2) + (second >> 1));
+ high >>>= ((second & 1) << 2); // an odd position starts halfway into its byte
+ int zeroNibbles = (high != 0) ? (Long.numberOfTrailingZeros(high) >>> 2) : (15 - (second & 1));
+ high = (high >>> (zeroNibbles << 2)) & 15;
+ second += zeroNibbles;
+ scanPosition = second;
+ if (second >= CODE_MAX || high == 0) {
+ return 0;
+ }
+ int low = count2Low[first * CODE_MAX + second] & 0xFF;
+ if (low != 0) {
+ high--; // the high part was raised early
+ }
+ return (int) ((high << 8) + low);
+ }
+
+ int scanPosition() {
+ return scanPosition;
+ }
+
+ /** Size of the buffer {@link #backupSingleCounts} needs. */
+ static int backupSize() {
+ return 2 * CODE_MAX;
+ }
+
+ /** Saves the single-symbol counters, so the best round's counts can be brought back. */
+ void backupSingleCounts(byte[] buffer) {
+ System.arraycopy(count1High, 0, buffer, 0, CODE_MAX);
+ System.arraycopy(count1Low, 0, buffer, CODE_MAX, CODE_MAX);
+ }
+
+ void restoreSingleCounts(byte[] buffer) {
+ System.arraycopy(buffer, 0, count1High, 0, CODE_MAX);
+ System.arraycopy(buffer, CODE_MAX, count1Low, 0, CODE_MAX);
+ }
+
+ private static long loadLittleEndianLong(byte[] source, int offset) {
+ long value = 0;
+ for (int i = 0; i < 8; i++) {
+ value |= (long) (source[offset + i] & 0xFF) << (i * 8);
+ }
+ return value;
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstSymbolTable.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstSymbolTable.java
new file mode 100644
index 0000000000..1ea80548f3
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstSymbolTable.java
@@ -0,0 +1,267 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable.fsst;
+
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_BASE;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_MASK;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_MAX;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.HASH_TAB_SIZE;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.ICL_FREE;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.LEN_BITS;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.MAX_SYMBOLS;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.MAX_SYMBOL_LENGTH;
+
+/**
+ * The mutable symbol table the FSST trainer builds and the compressor reads.
+ *
+ * Java port of {@code SymbolTable} from the FSST reference implementation by Peter Boncz,
+ * Viktor Leis and Thomas Neumann (CWI / TU Munich), MIT licensed, at
+ * https://github.com/cwida/fsst, commit 89f49c580c6388acf3b6ed2a49e1bfde6c05e616.
+ *
+ * Symbols are held in parallel {@code long} arrays rather than as objects. The reference
+ * implementation copies symbols by value throughout; parallel arrays reproduce that without the
+ * aliasing that an array of mutable objects would introduce, and keep the inner loops free of
+ * pointer chasing. The two code lookup tables use {@code char}, which is Java's only unsigned
+ * 16-bit type and so matches the reference implementation's {@code u16} without masking.
+ *
+ * Two representations of a code appear here and must not be confused. The lookup tables hold
+ * {@code (length << 12) | code}; a symbol descriptor holds {@code (length << 28) | (code << 16) |
+ * ignoredBits}.
+ */
+final class FsstSymbolTable {
+
+ /** Code for a 2-byte pattern, else the pseudo code for its escaped first byte. */
+ final char[] shortCodes = new char[65536];
+
+ /** Code for a 1-byte symbol, else its escaped pseudo code. Not needed after {@link #finish}. */
+ final char[] byteCodes = new char[256];
+
+ /** Symbol bytes, little-endian. Indices below {@link FsstCodes#CODE_BASE} are pseudo symbols. */
+ final long[] symbolValues = new long[CODE_MAX];
+
+ /** Symbol descriptors, parallel to {@link #symbolValues}. */
+ final long[] symbolDescriptors = new long[CODE_MAX];
+
+ /** Symbols of three bytes or more, replicated here to avoid an indirection. */
+ final long[] hashValues = new long[HASH_TAB_SIZE];
+
+ final long[] hashDescriptors = new long[HASH_TAB_SIZE];
+
+ /** Count of symbols of each byte length, indexed by length minus one. */
+ final int[] lengthHistogram = new int[FsstCodes.CODE_BITS];
+
+ int symbolCount;
+
+ /** Codes at or above this value may have a longer suffix; only meaningful after {@link #finish}. */
+ int suffixLimit = CODE_MAX;
+
+ /** Codes at or above this value are one byte long; only meaningful after {@link #finish}. */
+ int byteLimit;
+
+ /** A 1-byte symbol usable as a separator during compression. */
+ int terminator;
+
+ FsstSymbolTable() {
+ for (int i = 0; i < 256; i++) {
+ // Pseudo symbols: one per byte value, standing for that byte escaped.
+ symbolValues[i] = i;
+ symbolDescriptors[i] = FsstCodes.icl(i | (1 << LEN_BITS), 1);
+ }
+ long unused = FsstCodes.icl(CODE_MASK, 1);
+ for (int i = 256; i < CODE_MAX; i++) {
+ symbolValues[i] = 0;
+ symbolDescriptors[i] = unused;
+ }
+ for (int i = 0; i < HASH_TAB_SIZE; i++) {
+ hashDescriptors[i] = ICL_FREE;
+ }
+ for (int i = 0; i < 256; i++) {
+ byteCodes[i] = (char) ((1 << LEN_BITS) | i);
+ }
+ for (int i = 0; i < 65536; i++) {
+ shortCodes[i] = (char) ((1 << LEN_BITS) | (i & 255));
+ }
+ }
+
+ /** Empties the table, touching only the positions that were used. */
+ void clear() {
+ java.util.Arrays.fill(lengthHistogram, 0);
+ for (int i = CODE_BASE; i < CODE_BASE + symbolCount; i++) {
+ long val = symbolValues[i];
+ int length = FsstCodes.length(symbolDescriptors[i]);
+ if (length == 1) {
+ int b = FsstCodes.first(val);
+ byteCodes[b] = (char) ((1 << LEN_BITS) | b);
+ } else if (length == 2) {
+ int b2 = FsstCodes.first2(val);
+ shortCodes[b2] = (char) ((1 << LEN_BITS) | (b2 & 255));
+ } else {
+ int idx = FsstCodes.hashBucket(val);
+ hashValues[idx] = 0;
+ hashDescriptors[idx] = ICL_FREE;
+ }
+ }
+ symbolCount = 0;
+ }
+
+ private boolean hashInsert(long val, long icl) {
+ int idx = FsstCodes.hashBucket(val);
+ if (hashDescriptors[idx] < ICL_FREE) {
+ return false; // bucket taken
+ }
+ hashDescriptors[idx] = icl;
+ hashValues[idx] = FsstCodes.maskWord(val, icl);
+ return true;
+ }
+
+ /** Copies another table over this one, so the best round's table can be kept aside. */
+ void copyFrom(FsstSymbolTable other) {
+ System.arraycopy(other.shortCodes, 0, shortCodes, 0, shortCodes.length);
+ System.arraycopy(other.byteCodes, 0, byteCodes, 0, byteCodes.length);
+ System.arraycopy(other.symbolValues, 0, symbolValues, 0, symbolValues.length);
+ System.arraycopy(other.symbolDescriptors, 0, symbolDescriptors, 0, symbolDescriptors.length);
+ System.arraycopy(other.hashValues, 0, hashValues, 0, hashValues.length);
+ System.arraycopy(other.hashDescriptors, 0, hashDescriptors, 0, hashDescriptors.length);
+ System.arraycopy(other.lengthHistogram, 0, lengthHistogram, 0, lengthHistogram.length);
+ symbolCount = other.symbolCount;
+ suffixLimit = other.suffixLimit;
+ byteLimit = other.byteLimit;
+ terminator = other.terminator;
+ }
+
+ /** Adds a symbol, returning false when its hash bucket is already taken. */
+ boolean add(long val, long icl) {
+ int length = FsstCodes.length(icl);
+ int code = CODE_BASE + symbolCount;
+ long descriptor = FsstCodes.icl(code, length);
+ if (length == 1) {
+ byteCodes[FsstCodes.first(val)] = (char) (code + (1 << LEN_BITS));
+ } else if (length == 2) {
+ shortCodes[FsstCodes.first2(val)] = (char) (code + (2 << LEN_BITS));
+ } else if (!hashInsert(val, descriptor)) {
+ return false;
+ }
+ symbolValues[code] = val;
+ symbolDescriptors[code] = descriptor;
+ symbolCount++;
+ lengthHistogram[length - 1]++;
+ return true;
+ }
+
+ /** Returns the code of the longest symbol matching the input at {@code position}. */
+ int findLongestSymbol(byte[] input, int position, int end) {
+ int length = Math.min(MAX_SYMBOL_LENGTH, end - position);
+ long val = FsstCodes.loadSymbolBytes(input, position, length);
+ long icl = FsstCodes.icl(CODE_MAX, length);
+ int idx = FsstCodes.hashBucket(val);
+ if (hashDescriptors[idx] <= icl && hashValues[idx] == FsstCodes.maskWord(val, hashDescriptors[idx])) {
+ return FsstCodes.code(hashDescriptors[idx]);
+ }
+ if (length >= 2) {
+ int code = shortCodes[FsstCodes.first2(val)] & CODE_MASK;
+ if (code >= CODE_BASE) {
+ return code;
+ }
+ }
+ return byteCodes[FsstCodes.first(val)] & CODE_MASK;
+ }
+
+ /**
+ * Renumbers codes into a single byte each and groups symbols by length.
+ *
+ * Named {@code finalize} in the reference implementation; renamed here because that name is
+ * reserved on {@link Object}. Afterwards real codes occupy {@code [0, symbolCount)} grouped by
+ * length as 2,3,4,5,6,7,8 then 1, two-byte symbols with no longer suffix come first so the
+ * compressor can take a shortcut, escapes in {@link #shortCodes} carry the eighth bit, and
+ * {@link #byteCodes} is folded into {@link #shortCodes} so the compressor never consults it.
+ *
+ * The reference implementation also supports zero-terminated input, which Parquet never
+ * produces because values carry an explicit length. That mode is fixed off here, so terms that
+ * depend on it fall away.
+ */
+ void finish() {
+ if (symbolCount > MAX_SYMBOLS) {
+ throw new IllegalStateException("FSST symbol table holds " + symbolCount + " symbols, at most "
+ + MAX_SYMBOLS + " can be renumbered into one byte each");
+ }
+ int[] newCode = new int[256];
+ int[] runningSum = new int[8];
+ byteLimit = symbolCount - lengthHistogram[0];
+
+ runningSum[0] = byteLimit; // 1-byte codes sort highest
+ runningSum[1] = 0;
+ for (int i = 1; i < 7; i++) {
+ runningSum[i + 1] = runningSum[i] + lengthHistogram[i];
+ }
+
+ suffixLimit = runningSum[1];
+ newCode[0] = 0;
+ symbolValues[0] = symbolValues[CODE_BASE];
+ symbolDescriptors[0] = symbolDescriptors[CODE_BASE];
+
+ for (int i = 0, j = runningSum[2]; i < symbolCount; i++) {
+ long val = symbolValues[CODE_BASE + i];
+ int length = FsstCodes.length(symbolDescriptors[CODE_BASE + i]);
+ // For 2-byte symbols, scan for a longer symbol sharing the same first two bytes. The scan
+ // bound doubles as the answer: clearing it both records the find and ends the loop, exactly
+ // as the reference implementation does.
+ int scan = (length == 2) ? symbolCount : 0;
+ if (scan != 0) {
+ int first2 = FsstCodes.first2(val);
+ for (int k = 0; k < scan; k++) {
+ long otherIcl = symbolDescriptors[CODE_BASE + k];
+ if (k != i
+ && FsstCodes.length(otherIcl) > 1
+ && first2 == FsstCodes.first2(symbolValues[CODE_BASE + k])) {
+ scan = 0;
+ }
+ }
+ newCode[i] = (scan != 0) ? suffixLimit++ : --j;
+ } else {
+ newCode[i] = runningSum[length - 1]++;
+ }
+ symbolValues[newCode[i]] = val;
+ symbolDescriptors[newCode[i]] = FsstCodes.icl(newCode[i], length);
+ }
+
+ for (int i = 0; i < 256; i++) {
+ if ((byteCodes[i] & CODE_MASK) >= CODE_BASE) {
+ byteCodes[i] = (char) (newCode[byteCodes[i] & 0xFF] + (1 << LEN_BITS));
+ } else {
+ byteCodes[i] = (char) (511 + (1 << LEN_BITS));
+ }
+ }
+
+ for (int i = 0; i < 65536; i++) {
+ if ((shortCodes[i] & CODE_MASK) >= CODE_BASE) {
+ shortCodes[i] = (char) (newCode[shortCodes[i] & 0xFF] + (shortCodes[i] & (15 << LEN_BITS)));
+ } else {
+ shortCodes[i] = byteCodes[i & 0xFF];
+ }
+ }
+
+ for (int i = 0; i < HASH_TAB_SIZE; i++) {
+ if (hashDescriptors[i] < ICL_FREE) {
+ int renumbered = newCode[FsstCodes.code(hashDescriptors[i]) & 0xFF];
+ hashValues[i] = symbolValues[renumbered];
+ hashDescriptors[i] = symbolDescriptors[renumbered];
+ }
+ }
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstTrainer.java b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstTrainer.java
new file mode 100644
index 0000000000..1bde3d3dc5
--- /dev/null
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/symboltable/fsst/FsstTrainer.java
@@ -0,0 +1,410 @@
+/*
+ * 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
+ *
+ * http://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.
+ */
+package org.apache.parquet.column.values.symboltable.fsst;
+
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_BASE;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_MASK;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.CODE_MAX;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.ICL_FREE;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.MAX_SYMBOLS;
+import static org.apache.parquet.column.values.symboltable.fsst.FsstCodes.MAX_SYMBOL_LENGTH;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.parquet.column.values.symboltable.SymbolTableTrainer;
+import org.apache.parquet.column.values.symboltable.SymbolTableType;
+import org.apache.parquet.column.values.symboltable.TrainedSymbolTable;
+import org.apache.parquet.column.values.symboltable.ValueBuffer;
+
+/**
+ * Chooses an FSST symbol table for a set of values.
+ *
+ * Java port of the training half of the FSST reference implementation by Peter Boncz, Viktor Leis
+ * and Thomas Neumann (CWI / TU Munich), MIT licensed, at https://github.com/cwida/fsst, commit
+ * 89f49c580c6388acf3b6ed2a49e1bfde6c05e616. The algorithm is reproduced rather than reinterpreted:
+ * a writer that mines symbols differently still produces readable files, but it produces different
+ * tables, and then the compression ratio can no longer be compared against another implementation
+ * to tell a correct port from a subtly broken one.
+ *
+ * Training runs five rounds over a sample of the data. Each round compresses the sample with the
+ * table built so far, counting how often each symbol occurs and how often each ordered pair of
+ * symbols occurs back to back; frequent pairs are then concatenated into candidate symbols, scored
+ * by how many bytes they would save, and the best 255 become the next round's table. Rounds use a
+ * growing fraction of the sample, and the table kept at the end is the round that scored best rather
+ * than the last one.
+ *
+ * One departure is deliberate and is documented at {@link #CANDIDATE_ORDER}.
+ */
+public class FsstTrainer implements SymbolTableTrainer {
+
+ /** Bytes of sample the trainer aims for. */
+ static final int SAMPLE_TARGET = 1 << 14;
+
+ /** Worst-case sample size, which also bounds the score of a round. */
+ static final int SAMPLE_MAX_SIZE = 2 * SAMPLE_TARGET;
+
+ /** Length of one run of bytes taken from a value into the sample. */
+ static final int SAMPLE_LINE = 512;
+
+ /** Seed of the sampler's hash chain, from the reference implementation. */
+ private static final long SAMPLE_SEED = 4637947;
+
+ /**
+ * Order candidates are offered to the table in: best score first, and among equal scores the
+ * numerically smaller symbol, then the shorter one.
+ *
+ * The reference implementation collects candidates in a hash set and drains them through a
+ * heap, so when two candidates tie on both score and numeric value the winner depends on the hash
+ * set's iteration order, which its own standard library does not fix. That last tie is possible,
+ * because a shorter symbol whose trailing bytes are zero has the same numeric value as a longer
+ * one. Breaking it on length makes this port's output a function of the input alone, which is what
+ * lets a table be compared against another implementation's at all. Any table this produces is
+ * readable by any reader; only the tie-breaking rule would have to be agreed to make two writers
+ * agree byte for byte.
+ */
+ private static final Comparator The choice of runs is driven by a chain of the same hash the symbol lookup uses, seeded by a
+ * constant, so it is already reproducible and is ported as it stands. Replacing it with a
+ * different sampler, however reasonable, would change the table for the same input.
+ */
+ private ValueBuffer makeSample(ValueBuffer values) {
+ int valueCount = values.valueCount();
+ if (valueCount == 0 || values.byteCount() < SAMPLE_TARGET) {
+ return values;
+ }
+ sample.reset();
+ long random = FsstCodes.hash(SAMPLE_SEED);
+ int lineLimit = valueCount + SAMPLE_MAX_SIZE / SAMPLE_LINE;
+ while (sample.byteCount() < SAMPLE_TARGET && sample.valueCount() < lineLimit) {
+ // Choose a value, skipping forward over empty ones.
+ random = FsstCodes.hash(random);
+ int index = (int) Long.remainderUnsigned(random, valueCount);
+ while (values.length(index) == 0) {
+ if (++index == valueCount) {
+ index = 0;
+ }
+ }
+ // Choose one of its runs.
+ int runCount = 1 + ((values.length(index) - 1) / SAMPLE_LINE);
+ random = FsstCodes.hash(random);
+ int runStart = SAMPLE_LINE * (int) Long.remainderUnsigned(random, runCount);
+ int runLength = Math.min(values.length(index) - runStart, SAMPLE_LINE);
+ sample.add(values.data(), values.offset(index) + runStart, runLength);
+ }
+ return sample;
+ }
+
+ private FsstSymbolTable buildSymbolTable(ValueBuffer sampleValues) {
+ FsstSymbolTable table = new FsstSymbolTable();
+ FsstSymbolTable best = new FsstSymbolTable();
+ int bestGain = -SAMPLE_MAX_SIZE; // the score if every byte had to be escaped
+ table.terminator = chooseTerminator(sampleValues);
+
+ for (sampleFraction = 8; ; sampleFraction += 30) {
+ counters.reset();
+ int gain = compressCount(table, sampleValues);
+ if (gain >= bestGain) {
+ counters.backupSingleCounts(bestCounters);
+ best.copyFrom(table);
+ bestGain = gain;
+ }
+ if (sampleFraction >= 128) {
+ break; // five rounds, at fractions 8, 38, 68, 98 and 128
+ }
+ makeTable(table);
+ }
+
+ // Rebuild the best round's table from the counts that round produced. The fraction is 128 here,
+ // so this pass only re-ranks the symbols it already has and creates no new ones.
+ counters.restoreSingleCounts(bestCounters);
+ makeTable(best);
+ best.finish();
+ return best;
+ }
+
+ /**
+ * Picks the least frequent byte as the terminator, preferring the lowest such byte.
+ *
+ * The terminator is appended to each run the compressor works on, which is what lets the
+ * compressor read eight bytes at a time without a bounds check: a symbol containing the
+ * terminator is never in the table, so a match cannot run past the end of the value.
+ */
+ private static int chooseTerminator(ValueBuffer values) {
+ int[] byteHistogram = new int[256];
+ byte[] data = values.data();
+ for (int i = 0; i < values.valueCount(); i++) {
+ int end = values.offset(i) + values.length(i);
+ for (int position = values.offset(i); position < end; position++) {
+ byteHistogram[data[position] & 0xFF]++;
+ }
+ }
+ int terminator = 256;
+ int minimum = SAMPLE_MAX_SIZE;
+ for (int i = 255; i >= 0; i--) {
+ if (byteHistogram[i] > minimum) {
+ continue;
+ }
+ terminator = i;
+ minimum = byteHistogram[i];
+ }
+ return terminator;
+ }
+
+ /**
+ * Compresses the sample with the table as it stands, counting symbols and symbol pairs, and
+ * returns the number of bytes the table saves over escaping everything.
+ */
+ private int compressCount(FsstSymbolTable table, ValueBuffer values) {
+ int gain = 0;
+ for (int index = 0; index < values.valueCount(); index++) {
+ int position = values.offset(index);
+ int end = position + values.length(index);
+ if (sampleFraction < 128 && randomFraction(index) > sampleFraction) {
+ continue; // earlier rounds skip most of the sample, which roughly halves the work
+ }
+ if (position >= end) {
+ continue;
+ }
+ byte[] data = values.data();
+ int start = position;
+ int code1 = table.findLongestSymbol(data, position, end);
+ position += FsstCodes.length(table.symbolDescriptors[code1]);
+ gain += FsstCodes.length(table.symbolDescriptors[code1]) - (1 + escapeCost(code1));
+ while (true) {
+ // Count the symbol as it stands, that is, the option of not extending it.
+ counters.count1Increment(code1);
+ // As an alternative, count just its first byte, unless that is the same thing.
+ if (FsstCodes.length(table.symbolDescriptors[code1]) != 1) {
+ counters.count1Increment(data[start] & 0xFF);
+ }
+ if (position == end) {
+ break;
+ }
+
+ start = position;
+ int code2;
+ if (position < end - 7) {
+ // Eight bytes are available, so the three lookups can be done without bounds checks.
+ long word = FsstCodes.loadSymbolBytes(data, position, MAX_SYMBOL_LENGTH);
+ int bucket = FsstCodes.hashBucket(word);
+ long bucketDescriptor = table.hashDescriptors[bucket];
+ code2 = table.shortCodes[FsstCodes.first2(word)] & CODE_MASK;
+ word = FsstCodes.maskWord(word, bucketDescriptor);
+ if (bucketDescriptor < ICL_FREE && table.hashValues[bucket] == word) {
+ code2 = FsstCodes.code(bucketDescriptor);
+ position += FsstCodes.length(bucketDescriptor);
+ } else if (code2 >= CODE_BASE) {
+ position += 2;
+ } else {
+ code2 = table.byteCodes[FsstCodes.first(word)] & CODE_MASK;
+ position += 1;
+ }
+ } else {
+ code2 = table.findLongestSymbol(data, position, end);
+ position += FsstCodes.length(table.symbolDescriptors[code2]);
+ }
+
+ gain += (position - start) - (1 + escapeCost(code2));
+
+ if (sampleFraction < 128) { // the last round does not need pair counts
+ // Count the pair, that is, the option of concatenating the two symbols.
+ counters.count2Increment(code1, code2);
+ // As an alternative, count extending by just the next byte, unless that is the same thing.
+ if ((position - start) > 1) {
+ counters.count2Increment(code1, data[start] & 0xFF);
+ }
+ }
+ code1 = code2;
+ }
+ }
+ return gain;
+ }
+
+ /** A value between 1 and 128, fixed for a given value index and round. */
+ private int randomFraction(int index) {
+ return 1 + (int) (FsstCodes.hash((long) (index + 1) * sampleFraction) & 127);
+ }
+
+ /** Cost in bytes a code adds beyond the one byte it always costs: one more if it escapes. */
+ private static int escapeCost(int code) {
+ return code < CODE_BASE ? 1 : 0;
+ }
+
+ /**
+ * Replaces the table with the best-scoring candidates from the counts just gathered.
+ *
+ * Candidates are the symbols already in the table, the concatenation of each counted pair, and
+ * each symbol extended by one byte. Single-byte symbols are scored eight times higher than their
+ * frequency warrants, which the reference implementation notes both lowers the escape rate and
+ * speeds up compression and decompression.
+ */
+ private void makeTable(FsstSymbolTable table) {
+ candidates.clear();
+
+ // Force the terminator into the table by making it look like the most frequent symbol.
+ int terminatorPosition = table.symbolCount != 0 ? CODE_BASE : table.terminator;
+ counters.count1Set(terminatorPosition, 65535);
+
+ int positionLimit = CODE_BASE + table.symbolCount;
+ for (int pos1 = 0; pos1 < positionLimit; pos1++) {
+ int count1 = counters.count1Next(pos1);
+ pos1 = counters.scanPosition(); // the scan skips empty counters
+ if (count1 == 0) {
+ continue;
+ }
+ long value1 = table.symbolValues[pos1];
+ int length1 = FsstCodes.length(table.symbolDescriptors[pos1]);
+ addOrIncrement(value1, length1, (length1 == 1 ? 8L : 1L) * count1);
+
+ if (sampleFraction >= 128 // the last round does not create new symbols
+ || length1 == MAX_SYMBOL_LENGTH // this symbol cannot be extended
+ || FsstCodes.first(value1) == table.terminator) { // and none may contain the terminator
+ continue;
+ }
+ for (int pos2 = 0; pos2 < positionLimit; pos2++) {
+ int count2 = counters.count2Next(pos1, pos2);
+ pos2 = counters.scanPosition();
+ if (count2 == 0) {
+ continue;
+ }
+ long value2 = table.symbolValues[pos2];
+ if (FsstCodes.first(value2) != table.terminator) {
+ int length2 = FsstCodes.length(table.symbolDescriptors[pos2]);
+ addOrIncrement(concatenate(value1, length1, value2), concatenatedLength(length1, length2), count2);
+ }
+ }
+ }
+
+ List Rare candidates are dropped, on a threshold that grows with the round. Upstream notes this
+ * improves the compression ratio as well as the speed of training.
+ */
+ private void addOrIncrement(long value, int length, long count) {
+ if (count < (5L * sampleFraction) / 128) {
+ return;
+ }
+ Candidate candidate = new Candidate(value, length);
+ Candidate existing = candidates.get(candidate);
+ if (existing != null) {
+ existing.gain += count * length;
+ } else {
+ candidate.gain = count * length;
+ candidates.put(candidate, candidate);
+ }
+ }
+
+ /**
+ * Joins two symbols into one, truncated to the maximum symbol length.
+ *
+ * The shift is safe because a symbol already at the maximum length is never extended, so
+ * {@code firstLength} is at most seven here. That matters more in Java than in C: a shift count of
+ * 64 would be taken modulo 64 and quietly leave the value unshifted.
+ */
+ private static long concatenate(long first, int firstLength, long second) {
+ return (second << (8 * firstLength)) | first;
+ }
+
+ private static int concatenatedLength(int firstLength, int secondLength) {
+ return Math.min(firstLength + secondLength, MAX_SYMBOL_LENGTH);
+ }
+
+ /**
+ * A symbol under consideration and the bytes it would save.
+ *
+ * Identity is the symbol, not the score, so that repeated proposals of the same symbol
+ * accumulate. Hashing on the value alone matches upstream and stays consistent with equality.
+ */
+ private static final class Candidate {
+ private final long value;
+ private final int length;
+ private long gain;
+
+ Candidate(long value, int length) {
+ this.value = value;
+ this.length = length;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof Candidate)) {
+ return false;
+ }
+ Candidate that = (Candidate) other;
+ return value == that.value && length == that.length;
+ }
+
+ @Override
+ public int hashCode() {
+ return Long.hashCode(value);
+ }
+ }
+
+ static {
+ // Guards the assumption the position arithmetic in makeTable relies on.
+ if (CODE_BASE + MAX_SYMBOLS >= CODE_MAX) {
+ throw new AssertionError("symbol positions must stay below " + CODE_MAX);
+ }
+ }
+}
diff --git a/parquet-column/src/test/java/org/apache/parquet/column/TestParquetProperties.java b/parquet-column/src/test/java/org/apache/parquet/column/TestParquetProperties.java
index 6d51f67cb0..109ef64375 100644
--- a/parquet-column/src/test/java/org/apache/parquet/column/TestParquetProperties.java
+++ b/parquet-column/src/test/java/org/apache/parquet/column/TestParquetProperties.java
@@ -24,6 +24,8 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import org.apache.parquet.column.values.symboltable.SymbolTablePayload.OffsetEncoding;
+import org.apache.parquet.column.values.symboltable.SymbolTableType;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;
import org.junit.jupiter.api.BeforeEach;
@@ -122,6 +124,62 @@ public void withCompressionCodec_nullCodec_throwsNullPointerException() {
.hasMessage("codec cannot be null");
}
+ @Test
+ public void fsst_isOffUnlessAskedFor() {
+ ParquetProperties props = ParquetProperties.builder().build();
+ assertThat(props.isFsstEnabled(colA)).isFalse();
+ assertThat(props.getSymbolTableType(colA)).isNull();
+ }
+
+ @Test
+ public void fsst_appliesToBinaryColumnsOnly() {
+ ParquetProperties props =
+ ParquetProperties.builder().withFsstEncoding(true).build();
+ assertThat(props.getSymbolTableType(colA)).isEqualTo(SymbolTableType.FSST_8);
+ assertThat(props.isFsstEnabled(colB)).isFalse();
+ assertThat(props.isFsstEnabled(colC)).isFalse();
+ }
+
+ @Test
+ public void fsst_canBeSetForOneColumn() {
+ ParquetProperties props =
+ ParquetProperties.builder().withFsstEncoding("col_a", true).build();
+ assertThat(props.isFsstEnabled(colA)).isTrue();
+ assertThat(ParquetProperties.builder()
+ .withFsstEncoding(true)
+ .withFsstEncoding("col_a", false)
+ .build()
+ .isFsstEnabled(colA))
+ .isFalse();
+ }
+
+ @Test
+ public void symbolTableOffsets_areDeltaPackedUnlessAskedOtherwise() {
+ assertThat(ParquetProperties.builder().build().getSymbolTableOffsetEncoding())
+ .isEqualTo(OffsetEncoding.DELTA_BINARY_PACKED);
+ assertThat(ParquetProperties.builder()
+ .withSymbolTableOffsetEncoding(OffsetEncoding.PLAIN)
+ .build()
+ .getSymbolTableOffsetEncoding())
+ .isEqualTo(OffsetEncoding.PLAIN);
+ assertThatThrownBy(() -> ParquetProperties.builder().withSymbolTableOffsetEncoding(null))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("offsetEncoding cannot be null");
+ }
+
+ @Test
+ public void copyBuilder_preservesSymbolTableSettings() {
+ ParquetProperties original = ParquetProperties.builder()
+ .withFsstEncoding("col_a", true)
+ .withSymbolTableOffsetEncoding(OffsetEncoding.PLAIN)
+ .build();
+
+ ParquetProperties copy = ParquetProperties.copy(original).build();
+
+ assertThat(copy.isFsstEnabled(colA)).isTrue();
+ assertThat(copy.getSymbolTableOffsetEncoding()).isEqualTo(OffsetEncoding.PLAIN);
+ }
+
@Test
public void copyBuilder_preservesColumnCodecAndLevel() {
ParquetProperties original = ParquetProperties.builder()
diff --git a/parquet-column/src/test/java/org/apache/parquet/column/page/mem/MemPageReader.java b/parquet-column/src/test/java/org/apache/parquet/column/page/mem/MemPageReader.java
index c44a7b644c..0805c4c5cb 100644
--- a/parquet-column/src/test/java/org/apache/parquet/column/page/mem/MemPageReader.java
+++ b/parquet-column/src/test/java/org/apache/parquet/column/page/mem/MemPageReader.java
@@ -23,6 +23,7 @@
import org.apache.parquet.column.page.DataPage;
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.io.ParquetDecodingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,12 +34,22 @@ public class MemPageReader implements PageReader {
private final long totalValueCount;
private final Iterator {@link SymbolTableValuesRoundTripTest} already covers the writer and reader in isolation; what
+ * only this level can catch is a table that never reaches the writer's chunk-finalize hook, or a
+ * reader built before the table it needs exists.
+ */
+public class SymbolTableEndToEndTest {
+
+ private static final ColumnDescriptor REQUIRED_BINARY =
+ requiredBinaryColumn().getColumnDescription(new String[] {"foo", "bar"});
+
+ private static MessageType requiredBinaryColumn() {
+ return MessageTypeParser.parseMessageType("message msg { required group foo { required binary bar; } }");
+ }
+
+ private static List {@link org.apache.parquet.column.impl.ColumnReaderImpl}'s constructor calls {@code consume()}
+ * to prime the first value, so the failure happens at {@code getColumnReader(path)} itself, not on
+ * some later value read.
+ */
+ @Test
+ public void aReaderCannotSilentlyProceedWhenTheTableIsMissing() {
+ MessageType schema = requiredBinaryColumn();
+ ColumnDescriptor path = REQUIRED_BINARY;
+ List The round trip and reference comparison tests in this package both start from bytes this code
+ * produced: one proves the decoder undoes the encoder, the other proves the trainer picks the same
+ * symbols as the reference implementation. Neither can catch a page layout that is self-consistent
+ * and different from everyone else's -- a swapped header field, an off-by-one in the offset section,
+ * an escape read as a code. Only bytes from a foreign writer catch that, which is what these
+ * fixtures are.
+ *
+ * They come from the C++ implementation's Parquet interop file, and their provenance, layout and
+ * per-case coverage are documented in the header of {@code src/test/resources/fsst/interop/
+ * expected.txt}. The expected values there were taken from that file's conventionally-encoded
+ * columns, which hold the same values as the encoded ones, so nothing in the fixtures depends on any
+ * FSST decoder being right.
+ */
+public class SymbolTableInteropTest {
+
+ private static final String DIRECTORY = "/fsst/interop/";
+
+ /** One case's symbol table and the bodies of its data pages, as the fixture files hold them. */
+ private static final class Chunk {
+ final SymbolTableType type;
+ final byte[] table;
+ final List The framing is the part of the encoding that two implementations have to agree on exactly, so
+ * the write-side assertions here are on bytes rather than on a round trip.
+ */
+public class SymbolTablePayloadTest {
+
+ private static final int SLAB_SIZE = 64;
+ private static final int PAGE_SIZE = 1 << 20;
+
+ private static SymbolTablePayloadWriter writer(OffsetEncoding offsetEncoding) {
+ return new SymbolTablePayloadWriter(
+ offsetEncoding, SLAB_SIZE, PAGE_SIZE, HeapByteBufferAllocator.getInstance());
+ }
+
+ private static byte[] write(OffsetEncoding offsetEncoding, byte[]... values) throws IOException {
+ try (SymbolTablePayloadWriter writer = writer(offsetEncoding)) {
+ for (byte[] value : values) {
+ writer.addValue(value, 0, value.length);
+ }
+ assertThat(writer.valueCount()).isEqualTo(values.length);
+ return writer.getBytes().toByteArray();
+ }
+ }
+
+ private static SymbolTablePayload parse(byte[] body, int pageValueCount) throws IOException {
+ return SymbolTablePayload.parse(ByteBufferInputStream.wrap(ByteBuffer.wrap(body)), pageValueCount);
+ }
+
+ /** Reads back every value, so a bound that is off by one shows up as wrong bytes. */
+ private static List The table lives with the column chunk while a page is what a writer produces, and that mismatch
+ * is where the interesting cases are — a table trained on the first page and used by the fifth, a
+ * chunk that abandons the encoding after the table has already been trained, and a reader that has to
+ * be handed the table from outside the page it is reading.
+ */
+public class SymbolTableValuesRoundTripTest {
+
+ private static final int SLAB_SIZE = 1024;
+ private static final int PAGE_SIZE = 1 << 20;
+
+ /**
+ * Stands in for wherever the deserialized {@link SymbolTablePage} ends up.
+ *
+ * Deserializing on every {@link #getSymbolTable()} call is deliberate: it means the reader is
+ * decoding against a table rebuilt from bytes rather than against the trainer's own object.
+ */
+ private static final class SymbolTableRelay implements SymbolTableSource {
+
+ private SymbolTableType type;
+ private byte[] body;
+
+ void receive(SymbolTablePage page) throws IOException {
+ this.type = page.getType();
+ this.body = page.getBytes().toByteArray();
+ }
+
+ @Override
+ public SymbolTable getSymbolTable() {
+ return SymbolTables.deserialize(type, body, 0, body.length);
+ }
+ }
+
+ private static SymbolTableValuesWriter writer(OffsetEncoding offsetEncoding) {
+ return new SymbolTableValuesWriter(
+ SymbolTableType.FSST_8, offsetEncoding, SLAB_SIZE, PAGE_SIZE, HeapByteBufferAllocator.getInstance());
+ }
+
+ private static List The pages deliberately drift: page one has none of the text page four is made of, so a table
+ * retrained per page would give a smaller page four, and a reader given only the first page's table
+ * would decode it wrongly. The point of the assertion is that the second thing does not happen.
+ */
+ @Test
+ public void oneTableTrainedOnTheFirstPageServesLaterPages() throws IOException {
+ List Single bytes with four-byte offsets: the codes are as long as the values and the offsets cost
+ * what a plain page's lengths cost, so the nine-byte header alone decides it. That is the
+ * configuration to fall back from, and it is also the argument against writing offsets plain — the
+ * same page with packed offsets is smaller than plain and keeps the encoding.
+ *
+ * Also the regression test for a fixed publish-timing defect: a chunk that falls back must not
+ * leave a symbol table page behind that no page refers to.
+ */
+ @Test
+ public void aPageTheEncodingWouldGrowIsWrittenPlain() throws IOException {
+ List Every {@code *WithoutZstd}/{@code *WithZstd} pair isolates encode from decode: the bytes a
+ * decode benchmark reads are built once outside the timed rounds, and the side channel each
+ * encoding needs (a symbol table or a dictionary page) is decoded once as well, matching how a real
+ * reader amortizes it across every page of a chunk.
+ *
+ * Ratios are not a JUnitBenchmarks metric, so they are printed once, up front, rather than folded
+ * into a timed round.
+ */
+@EnableRuleMigrationSupport
+@AxisRange(min = 0, max = 1)
+@BenchmarkMethodChart(filePrefix = "benchmark-symboltable-encoding")
+public class BenchmarkSymbolTableEncoding {
+
+ @Rule
+ public org.junit.rules.TestRule benchmarkRun = new BenchmarkRule();
+
+ private static final int INITIAL_SLAB_SIZE = 64 * 1024;
+ private static final int PAGE_SIZE = 4 * 1024 * 1024;
+
+ private static final Binary[] VALUES = buildCorpus();
+ private static final long RAW_BYTES = totalRawBytes(VALUES);
+
+ // FSST
+ private static final byte[] FSST_TABLE_BYTES;
+ private static final byte[] FSST_PAYLOAD_BYTES;
+ private static final byte[] FSST_PAYLOAD_ZSTD;
+ private static final SymbolTable FSST_TABLE;
+
+ // DELTA_LENGTH_BYTE_ARRAY
+ private static final byte[] DLBA_PAYLOAD_BYTES;
+ private static final byte[] DLBA_PAYLOAD_ZSTD;
+
+ // DELTA_BYTE_ARRAY
+ private static final byte[] DBA_PAYLOAD_BYTES;
+ private static final byte[] DBA_PAYLOAD_ZSTD;
+
+ // RLE_DICTIONARY
+ private static final byte[] DICT_PAGE_BYTES;
+ private static final int DICT_SIZE;
+ private static final byte[] DICT_INDEX_BYTES;
+ private static final byte[] DICT_INDEX_ZSTD;
+ private static final DictionaryPage DICTIONARY_PAGE;
+
+ static {
+ try {
+ SymbolTableValuesWriter fsstWriter = new SymbolTableValuesWriter(
+ SymbolTableType.FSST_8,
+ OffsetEncoding.DELTA_BINARY_PACKED,
+ INITIAL_SLAB_SIZE,
+ PAGE_SIZE,
+ new DirectByteBufferAllocator());
+ writeAll(fsstWriter, VALUES);
+ FSST_PAYLOAD_BYTES = fsstWriter.getBytes().toByteArray();
+ FSST_TABLE_BYTES = fsstWriter.toSymbolTablePageAndClose().getBytes().toByteArray();
+ FSST_TABLE = SymbolTables.deserialize(SymbolTableType.FSST_8, FSST_TABLE_BYTES, 0, FSST_TABLE_BYTES.length);
+ FSST_PAYLOAD_ZSTD = Zstd.compress(FSST_PAYLOAD_BYTES);
+
+ DeltaLengthByteArrayValuesWriter dlbaWriter =
+ new DeltaLengthByteArrayValuesWriter(INITIAL_SLAB_SIZE, PAGE_SIZE, new DirectByteBufferAllocator());
+ writeAll(dlbaWriter, VALUES);
+ DLBA_PAYLOAD_BYTES = dlbaWriter.getBytes().toByteArray();
+ DLBA_PAYLOAD_ZSTD = Zstd.compress(DLBA_PAYLOAD_BYTES);
+
+ DeltaByteArrayWriter dbaWriter =
+ new DeltaByteArrayWriter(INITIAL_SLAB_SIZE, PAGE_SIZE, new DirectByteBufferAllocator());
+ writeAll(dbaWriter, VALUES);
+ DBA_PAYLOAD_BYTES = dbaWriter.getBytes().toByteArray();
+ DBA_PAYLOAD_ZSTD = Zstd.compress(DBA_PAYLOAD_BYTES);
+
+ PlainBinaryDictionaryValuesWriter dictWriter = new PlainBinaryDictionaryValuesWriter(
+ Integer.MAX_VALUE, Encoding.RLE_DICTIONARY, Encoding.PLAIN, new DirectByteBufferAllocator());
+ writeAll(dictWriter, VALUES);
+ DICT_INDEX_BYTES = dictWriter.getBytes().toByteArray();
+ DictionaryPage dictPage = dictWriter.toDictPageAndClose();
+ DICT_PAGE_BYTES = dictPage.getBytes().toByteArray();
+ DICT_SIZE = dictPage.getDictionarySize();
+ DICTIONARY_PAGE = new DictionaryPage(BytesInput.from(DICT_PAGE_BYTES), DICT_SIZE, Encoding.PLAIN);
+ DICT_INDEX_ZSTD = Zstd.compress(DICT_INDEX_BYTES);
+
+ report();
+ } catch (IOException e) {
+ throw new ExceptionInInitializerError(e);
+ }
+ }
+
+ /**
+ * Text with the shared structure real string columns have: repeated URL templates and repeated
+ * vocabulary, the case FSST and dictionary encoding both exist for. Random alphanumeric data, used
+ * elsewhere in this module, would flatter nothing and make the comparison meaningless.
+ */
+ private static Binary[] buildCorpus() {
+ String[] words = {
+ "alpha",
+ "bravo",
+ "charlie",
+ "delta",
+ "echo",
+ "foxtrot",
+ "golf",
+ "hotel",
+ "india",
+ "juliet",
+ "kilo",
+ "lima",
+ "mike",
+ "november",
+ "oscar",
+ "papa"
+ };
+ Random random = new Random(42);
+ List This is the test that makes the port trustworthy. A round trip only proves the decoder undoes
+ * whatever the encoder did, so a trainer that picks worse symbols, or numbers them differently,
+ * passes it while writing files no other implementation would have written. Comparing against bytes
+ * the reference produced catches both.
+ *
+ * The fixtures under {@code src/test/resources/fsst} hold, per corpus, the serialized symbol table
+ * and the concatenated code bytes, produced by the reference implementation. The corpora themselves
+ * are not stored: they are generated here from the arithmetic below, which is simple enough to
+ * restate in any language, and each one's digest is asserted so a drifting generator fails as a
+ * generator rather than as a codec.
+ */
+public class FsstReferenceComparisonTest {
+
+ /**
+ * The generator both sides use, spelled out rather than taken from a library so that regenerating
+ * the fixtures from another language gives the same bytes.
+ */
+ private static final class Lcg {
+ private int state;
+
+ Lcg(int seed) {
+ this.state = seed & 0x7FFFFFFF;
+ }
+
+ int next() {
+ state = (state * 1103515245 + 12345) & 0x7FFFFFFF;
+ return state;
+ }
+
+ int below(int bound) {
+ return next() % bound;
+ }
+
+ int nextByte() {
+ return (next() >> 16) & 0xFF;
+ }
+ }
+
+ private static final String[] WORDS = {
+ "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf",
+ "hotel", "india", "juliet", "kilo", "lima", "mike", "november"
+ };
+
+ /** Ordinary text with heavy shared structure: what the encoding is for. */
+ private static List
+ * [1B] offset encoding: 0 = PLAIN, 1 = DELTA_BINARY_PACKED
+ * [4B] number of values, little-endian
+ * [4B] byte length of the offset section, little-endian
+ * [..] one end offset into the code stream per value
+ * [..] the code stream
+ *
+ *
+ * What happens when
+ *
+ * Falling back
+ *
+ *