FSST compression

CMakeLists.txt

@@ -392,6 +392,7 @@ endif()
 
 
 include_directories(src/include)
+include_directories(third_party/fsst)
 include_directories(third_party/fmt/include)
 include_directories(third_party/hyperloglog)
 include_directories(third_party/fastpforlib)

benchmark/micro/compression/dictionary/dictionary_read_best_case.benchmark

@@ -8,7 +8,7 @@ storage persistent
 
 load
 DROP TABLE IF EXISTS test;
-PRAGMA force_compression='uncompressed';
+PRAGMA force_compression='dictionary';
 CREATE TABLE test AS SELECT (100 + (i%2))::VARCHAR AS i FROM range(0, 200000000) tbl(i);
 checkpoint;
 

benchmark/micro/compression/dictionary/dictionary_store.benchmark

@@ -9,7 +9,7 @@ require_reinit
 
 load
 PRAGMA force_compression='dictionary';
-DROP TABLE IF EXISTS integers;
+DROP TABLE IF EXISTS test;
 
 run
 CREATE TABLE test AS SELECT (100 + (i%1000))::VARCHAR AS i FROM range(0, 100000000) tbl(i);

benchmark/micro/compression/dictionary/dictionary_store_worst_case.benchmark

@@ -9,7 +9,7 @@ require_reinit
 
 load
 PRAGMA force_compression='dictionary';
-DROP TABLE IF EXISTS integers;
+DROP TABLE IF EXISTS test;
 
 run
 CREATE TABLE test AS SELECT i::VARCHAR AS i FROM range(0, 50000000) tbl(i);

benchmark/micro/compression/fsst/fsst_late_decompression.benchmark

@@ -0,0 +1,20 @@
+# name: benchmark/micro/compression/fsst/fsst_late_decompression.benchmark
+# description: Using a filter on another column to make use of late decompression
+# group: [fsst]
+
+name fsst late decompression benefit
+group fsst
+storage persistent
+
+load
+DROP TABLE IF EXISTS test;
+PRAGMA force_compression='fsst';
+CREATE TABLE test AS SELECT i as id, (100 + (i%2))::VARCHAR AS value FROM range(0, 50000000) tbl(i);
+checkpoint;
+SET enable_fsst_vectors=false;
+
+run
+select avg(value::INT) from test where id%10=0;
+
+result I
+100.500000

benchmark/micro/compression/fsst/fsst_read.benchmark

@@ -0,0 +1,19 @@
+# name: benchmark/micro/compression/fsst/fsst_read.benchmark
+# description: Scanning strings at ~3.35x compression
+# group: [fsst]
+
+name fsst Compression Scan
+group fsst
+storage persistent
+
+load
+DROP TABLE IF EXISTS test;
+PRAGMA force_compression='fsst';
+CREATE TABLE test AS SELECT (100 + (i%1000))::VARCHAR AS i FROM range(0, 50000000) tbl(i);
+checkpoint;
+
+run
+select avg(i::INT) from test;
+
+result I
+599.500000

benchmark/micro/compression/fsst/fsst_read_worst_case.benchmark

@@ -0,0 +1,16 @@
+# name: benchmark/micro/compression/fsst/fsst_read_worst_case.benchmark
+# description: Scanning data that is not with fsst encoding, note that compresssion ratio is still 1.9x due to bitpacking
+# group: [fsst]
+
+name fsst Compression Scan
+group aggregate
+storage persistent
+
+load
+DROP TABLE IF EXISTS test;
+PRAGMA force_compression='fsst';
+CREATE TABLE test AS SELECT gen_random_uuid()::VARCHAR AS i FROM range(0, 20000000) tbl(i);
+checkpoint;
+
+run
+select max(i[2]) from test;

benchmark/micro/compression/fsst/fsst_store.benchmark

@@ -0,0 +1,16 @@
+# name: benchmark/micro/compression/fsst/fsst_store.benchmark
+# description: Storing strings compressed at ~3.3x compression
+# group: [fsst]
+
+name fsst Compression Write
+group aggregate
+storage persistent
+require_reinit
+
+load
+PRAGMA force_compression='fsst';
+
+run
+CREATE TABLE test_compressed AS SELECT (100 + (i%1000))::VARCHAR AS i FROM range(0, 2500000) tbl(i);
+checkpoint;
+

benchmark/micro/compression/fsst/fsst_store_worst_case.benchmark

@@ -0,0 +1,16 @@
+# name: benchmark/micro/compression/fsst/fsst_store_worst_case.benchmark
+# description: Storing a column containing only unique strings.
+# group: [fsst]
+
+name name fsst Compression Write
+group fsst
+storage persistent
+require_reinit
+
+load
+PRAGMA force_compression='fsst';
+DROP TABLE IF EXISTS test;
+
+run
+CREATE TABLE test AS SELECT gen_random_uuid()::VARCHAR AS i FROM range(0, 2000000) tbl(i);
+checkpoint;

benchmark/micro/compression/dictionary/dictionary_store_tpch_sf1.benchmark → benchmark/micro/compression/store_tpch_sf1.benchmark

@@ -1,8 +1,8 @@
-# name: benchmark/micro/compression/dictionary/dictionary_store_tpch_sf1.benchmark
+# name: benchmark/micro/compression/store_tpch_sf1.benchmark
 # description: Generating and storing a tpc-h sf1 database using default compression
-# group: [dictionary]
+# group: [compression]
 
-name Dictionary Compression Write
+name TPC-H Write benchmark
 group aggregate
 storage persistent
 require_reinit

scripts/package_build.py

@@ -10,6 +10,7 @@
 def third_party_includes():
     includes = []
     includes += [os.path.join('third_party', 'fmt', 'include')]
+    includes += [os.path.join('third_party', 'fsst')]
     includes += [os.path.join('third_party', 're2')]
     includes += [os.path.join('third_party', 'miniz')]
     includes += [os.path.join('third_party', 'utf8proc', 'include')]
@@ -33,6 +34,7 @@ def third_party_includes():
 def third_party_sources():
     sources = []
     sources += [os.path.join('third_party', 'fmt')]
+    sources += [os.path.join('third_party', 'fsst')]
     sources += [os.path.join('third_party', 'miniz')]
     sources += [os.path.join('third_party', 're2')]
     sources += [os.path.join('third_party', 'hyperloglog')]

src/CMakeLists.txt

@@ -61,6 +61,7 @@ else()
 
   set(DUCKDB_LINK_LIBS
       ${DUCKDB_SYSTEM_LIBS}
+      duckdb_fsst
       duckdb_fmt
       duckdb_pg_query
       duckdb_re2

src/common/types/vector.cpp

@@ -13,6 +13,8 @@
 #include "duckdb/common/vector_operations/vector_operations.hpp"
 #include "duckdb/storage/buffer/buffer_handle.hpp"
 #include "duckdb/function/scalar/nested_functions.hpp"
+#include "duckdb/storage/string_uncompressed.hpp"
+#include "fsst.h"
 
 #include <cstring> // strlen() on Solaris
 
@@ -167,6 +169,12 @@ void Vector::Slice(const SelectionVector &sel, idx_t count) {
 		}
 		return;
 	}
+
+	if (GetVectorType() == VectorType::FSST_VECTOR) {
+		Flatten(sel, count);
+		return;
+	}
+
 	Vector child_vector(*this);
 	auto internal_type = GetType().InternalType();
 	if (internal_type == PhysicalType::STRUCT) {
@@ -416,7 +424,10 @@ Value Vector::GetValueInternal(const Vector &v_p, idx_t index_p) {
 		case VectorType::FLAT_VECTOR:
 			finished = true;
 			break;
-			// dictionary: apply dictionary and forward to child
+		case VectorType::FSST_VECTOR:
+			finished = true;
+			break;
+		// dictionary: apply dictionary and forward to child
 		case VectorType::DICTIONARY_VECTOR: {
 			auto &sel_vector = DictionaryVector::SelVector(*vector);
 			auto &child = DictionaryVector::Child(*vector);
@@ -440,6 +451,24 @@ Value Vector::GetValueInternal(const Vector &v_p, idx_t index_p) {
 	if (!validity.RowIsValid(index)) {
 		return Value(vector->GetType());
 	}
+
+	if (vector->GetVectorType() == VectorType::FSST_VECTOR) {
+		if (vector->GetType().id() == LogicalTypeId::VARCHAR) {
+			unsigned char decompress_buffer[StringUncompressed::STRING_BLOCK_LIMIT + 1];
+
+			auto str_compressed = ((string_t *)data)[index];
+			auto decompressed_string_size =
+			    duckdb_fsst_decompress((duckdb_fsst_decoder_t *)FSSTVector::GetDecoder(const_cast<Vector &>(*vector)),
+			                           str_compressed.GetSize(), (unsigned char *)str_compressed.GetDataUnsafe(),
+			                           StringUncompressed::STRING_BLOCK_LIMIT + 1, &decompress_buffer[0]);
+			D_ASSERT(decompressed_string_size <= StringUncompressed::STRING_BLOCK_LIMIT);
+
+			return Value(string((char *)decompress_buffer, decompressed_string_size));
+		} else {
+			throw InternalException("FSST Vector with non-string datatype found!");
+		}
+	}
+
 	switch (vector->GetType().id()) {
 	case LogicalTypeId::BOOLEAN:
 		return Value::BOOLEAN(((bool *)data)[index]);
@@ -579,6 +608,8 @@ string VectorTypeToString(VectorType type) {
 	switch (type) {
 	case VectorType::FLAT_VECTOR:
 		return "FLAT";
+	case VectorType::FSST_VECTOR:
+		return "FSST";
 	case VectorType::SEQUENCE_VECTOR:
 		return "SEQUENCE";
 	case VectorType::DICTIONARY_VECTOR:
@@ -600,6 +631,21 @@ string Vector::ToString(idx_t count) const {
 			retval += GetValue(i).ToString() + (i == count - 1 ? "" : ", ");
 		}
 		break;
+	case VectorType::FSST_VECTOR: {
+		for (idx_t i = 0; i < count; i++) {
+			string_t compressed_string = ((string_t *)data)[i];
+
+			// Decompress string
+			unsigned char decompress_buffer[StringUncompressed::STRING_BLOCK_LIMIT + 1];
+			auto decompressed_string_size =
+			    duckdb_fsst_decompress((duckdb_fsst_decoder_t *)FSSTVector::GetDecoder(const_cast<Vector &>(*this)),
+			                           compressed_string.GetSize(), (unsigned char *)compressed_string.GetDataUnsafe(),
+			                           StringUncompressed::STRING_BLOCK_LIMIT + 1, &decompress_buffer[0]);
+			D_ASSERT(decompressed_string_size <= StringUncompressed::STRING_BLOCK_LIMIT);
+
+			retval += string((const char *)decompress_buffer, decompressed_string_size) + (i == count - 1 ? "" : ", ");
+		}
+	} break;
 	case VectorType::CONSTANT_VECTOR:
 		retval += GetValue(0).ToString();
 		break;
@@ -662,6 +708,15 @@ void Vector::Flatten(idx_t count) {
 	case VectorType::FLAT_VECTOR:
 		// already a flat vector
 		break;
+	case VectorType::FSST_VECTOR: {
+		// create vector to decompress into
+		Vector other(GetType(), count);
+		// now copy the data of this vector to the other vector, decompressing the strings in the process
+		VectorOperations::Copy(*this, other, count, 0, 0);
+		// create a reference to the data in the other vector
+		this->Reference(other);
+		break;
+	}
 	case VectorType::DICTIONARY_VECTOR: {
 		// create a new flat vector of this type
 		Vector other(GetType(), count);
@@ -771,6 +826,15 @@ void Vector::Flatten(const SelectionVector &sel, idx_t count) {
 	case VectorType::FLAT_VECTOR:
 		// already a flat vector
 		break;
+	case VectorType::FSST_VECTOR: {
+		// create a new flat vector of this type
+		Vector other(GetType());
+		// now copy the data of this vector to the other vector, removing the selection vector in the process
+		VectorOperations::Copy(*this, other, sel, count, 0, 0);
+		// create a reference to the data in the other vector
+		this->Reference(other);
+		break;
+	}
 	case VectorType::SEQUENCE_VECTOR: {
 		int64_t start, increment;
 		SequenceVector::GetSequence(*this, start, increment);
@@ -1357,6 +1421,46 @@ void StringVector::AddHeapReference(Vector &vector, Vector &other) {
 	StringVector::AddBuffer(vector, other.auxiliary);
 }
 
+string_t FSSTVector::AddCompressedString(Vector &vector, const char *data, idx_t len) {
+	return FSSTVector::AddCompressedString(vector, string_t(data, len));
+}
+
+string_t FSSTVector::AddCompressedString(Vector &vector, string_t data) {
+	D_ASSERT(vector.GetType().InternalType() == PhysicalType::VARCHAR);
+	if (data.IsInlined()) {
+		// string will be inlined: no need to store in string heap
+		return data;
+	}
+	if (!vector.auxiliary) {
+		vector.auxiliary = make_buffer<VectorFSSTStringBuffer>();
+	}
+	D_ASSERT(vector.auxiliary->GetBufferType() == VectorBufferType::FSST_BUFFER);
+	auto &fsst_string_buffer = (VectorFSSTStringBuffer &)*vector.auxiliary;
+	return fsst_string_buffer.AddBlob(data);
+}
+
+void *FSSTVector::GetDecoder(Vector &vector) {
+	D_ASSERT(vector.GetType().InternalType() == PhysicalType::VARCHAR);
+	if (!vector.auxiliary) {
+		throw InternalException("GetDecoder called on FSST Vector without registered buffer");
+	}
+	D_ASSERT(vector.auxiliary->GetBufferType() == VectorBufferType::FSST_BUFFER);
+	auto &fsst_string_buffer = (VectorFSSTStringBuffer &)*vector.auxiliary;
+	return (duckdb_fsst_decoder_t *)fsst_string_buffer.GetDecoder();
+}
+
+void FSSTVector::RegisterDecoder(Vector &vector, buffer_ptr<void> &duckdb_fsst_decoder) {
+	D_ASSERT(vector.GetType().InternalType() == PhysicalType::VARCHAR);
+
+	if (!vector.auxiliary) {
+		vector.auxiliary = make_buffer<VectorFSSTStringBuffer>();
+	}
+	D_ASSERT(vector.auxiliary->GetBufferType() == VectorBufferType::FSST_BUFFER);
+
+	auto &fsst_string_buffer = (VectorFSSTStringBuffer &)*vector.auxiliary;
+	fsst_string_buffer.AddDecoder(duckdb_fsst_decoder);
+}
+
 vector<unique_ptr<Vector>> &StructVector::GetEntries(Vector &vector) {
 	D_ASSERT(vector.GetType().id() == LogicalTypeId::STRUCT || vector.GetType().id() == LogicalTypeId::MAP);
 	if (vector.GetVectorType() == VectorType::DICTIONARY_VECTOR) {

src/common/types/vector_buffer.cpp

@@ -27,6 +27,12 @@ buffer_ptr<VectorBuffer> VectorBuffer::CreateStandardVector(const LogicalType &t
 VectorStringBuffer::VectorStringBuffer() : VectorBuffer(VectorBufferType::STRING_BUFFER) {
 }
 
+VectorStringBuffer::VectorStringBuffer(VectorBufferType type) : VectorBuffer(type) {
+}
+
+VectorFSSTStringBuffer::VectorFSSTStringBuffer() : VectorStringBuffer(VectorBufferType::FSST_BUFFER) {
+}
+
 VectorStructBuffer::VectorStructBuffer() : VectorBuffer(VectorBufferType::STRUCT_BUFFER) {
 }