Repository: duckdb/duckdb (language: python)

Issue:
Overflow Prevention



Discussion / hints:
Should be mostly done in 55637be213ce57c536139699e22fe3290fe5788d, although there are likely more edge cases to consider

Fix patch:
diff --git a/benchmark/micro/compression/zstd/zstd_read.benchmark b/benchmark/micro/compression/zstd/zstd_read.benchmark
new file mode 100644
index 000000000000..4684249e67fe
--- /dev/null
+++ b/benchmark/micro/compression/zstd/zstd_read.benchmark
@@ -0,0 +1,17 @@
+# name: benchmark/micro/compression/zstd/zstd_read.benchmark
+# description: ZSTD decompression speed of relatively big (8000 byte) strings
+# group: [zstd]
+
+name ZSTD Scan
+group zstd
+storage persistent
+
+load
+DROP TABLE IF EXISTS zstd_strings;
+PRAGMA force_compression='zstd';
+set variable my_string = (list_reduce([chr(((i % 26) + ord('a'))::INTEGER) for i in range(8000)], (x, y) -> concat(x, y)));
+create table zstd_strings as select getvariable('my_string') as data from range(2_500_000) tbl(i);
+checkpoint;
+
+run
+select avg(strlen(data)) from zstd_strings;
diff --git a/benchmark/micro/compression/zstd/zstd_store.benchmark b/benchmark/micro/compression/zstd/zstd_store.benchmark
new file mode 100644
index 000000000000..bab765e2f89a
--- /dev/null
+++ b/benchmark/micro/compression/zstd/zstd_store.benchmark
@@ -0,0 +1,17 @@
+# name: benchmark/micro/compression/zstd/zstd_store.benchmark
+# description: ZSTD decompression speed of relatively big (>= overflow string) strings
+# group: [zstd]
+
+name ZSTD Compression Write
+group zstd
+storage persistent
+require_reinit
+
+load
+DROP TABLE IF EXISTS zstd_strings;
+PRAGMA force_compression='zstd';
+set variable my_string = (list_reduce([chr(((i % 26) + ord('a'))::INTEGER) for i in range(4096)], (x, y) -> concat(x, y)));
+
+run
+create table zstd_strings as select getvariable('my_string') as data from range(2_500_000) tbl(i);
+checkpoint;
diff --git a/src/common/enum_util.cpp b/src/common/enum_util.cpp
index 4a43c72f1486..83bae999c7ed 100644
--- a/src/common/enum_util.cpp
+++ b/src/common/enum_util.cpp
@@ -795,6 +795,7 @@ const StringUtil::EnumStringLiteral *GetCompressionTypeValues() {
 		{ static_cast<uint32_t>(CompressionType::COMPRESSION_PATAS), "COMPRESSION_PATAS" },
 		{ static_cast<uint32_t>(CompressionType::COMPRESSION_ALP), "COMPRESSION_ALP" },
 		{ static_cast<uint32_t>(CompressionType::COMPRESSION_ALPRD), "COMPRESSION_ALPRD" },
+		{ static_cast<uint32_t>(CompressionType::COMPRESSION_ZSTD), "COMPRESSION_ZSTD" },
 		{ static_cast<uint32_t>(CompressionType::COMPRESSION_COUNT), "COMPRESSION_COUNT" }
 	};
 	return values;
@@ -802,12 +803,12 @@ const StringUtil::EnumStringLiteral *GetCompressionTypeValues() {
 
 template<>
 const char* EnumUtil::ToChars<CompressionType>(CompressionType value) {
-	return StringUtil::EnumToString(GetCompressionTypeValues(), 13, "CompressionType", static_cast<uint32_t>(value));
+	return StringUtil::EnumToString(GetCompressionTypeValues(), 14, "CompressionType", static_cast<uint32_t>(value));
 }
 
 template<>
 CompressionType EnumUtil::FromString<CompressionType>(const char *value) {
-	return static_cast<CompressionType>(StringUtil::StringToEnum(GetCompressionTypeValues(), 13, "CompressionType", value));
+	return static_cast<CompressionType>(StringUtil::StringToEnum(GetCompressionTypeValues(), 14, "CompressionType", value));
 }
 
 const StringUtil::EnumStringLiteral *GetConflictManagerModeValues() {
diff --git a/src/common/enums/compression_type.cpp b/src/common/enums/compression_type.cpp
index 3b9930cfaecc..c34e6a2555ec 100644
--- a/src/common/enums/compression_type.cpp
+++ b/src/common/enums/compression_type.cpp
@@ -40,6 +40,8 @@ CompressionType CompressionTypeFromString(const string &str) {
 		return CompressionType::COMPRESSION_CHIMP;
 	} else if (compression == "patas") {
 		return CompressionType::COMPRESSION_PATAS;
+	} else if (compression == "zstd") {
+		return CompressionType::COMPRESSION_ZSTD;
 	} else if (compression == "alp") {
 		return CompressionType::COMPRESSION_ALP;
 	} else if (compression == "alprd") {
@@ -71,6 +73,8 @@ string CompressionTypeToString(CompressionType type) {
 		return "Chimp";
 	case CompressionType::COMPRESSION_PATAS:
 		return "Patas";
+	case CompressionType::COMPRESSION_ZSTD:
+		return "ZSTD";
 	case CompressionType::COMPRESSION_ALP:
 		return "ALP";
 	case CompressionType::COMPRESSION_ALPRD:
diff --git a/src/common/settings.json b/src/common/settings.json
index 17e4da37eb11..590edc82e1ec 100644
--- a/src/common/settings.json
+++ b/src/common/settings.json
@@ -717,5 +717,11 @@
         "aliases": [
             "user"
         ]
+    },
+    {
+        "name": "zstd_min_string_length",
+        "description": "The (average) length at which to enable ZSTD compression, defaults to 4096",
+        "type": "UBIGINT",
+        "scope": "global"
     }
 ]
\ No newline at end of file
diff --git a/src/function/compression_config.cpp b/src/function/compression_config.cpp
index a5e686f36f25..5ee991eb8b52 100644
--- a/src/function/compression_config.cpp
+++ b/src/function/compression_config.cpp
@@ -26,6 +26,7 @@ static const DefaultCompressionMethod internal_compression_methods[] = {
     {CompressionType::COMPRESSION_ALP, AlpCompressionFun::GetFunction, AlpCompressionFun::TypeIsSupported},
     {CompressionType::COMPRESSION_ALPRD, AlpRDCompressionFun::GetFunction, AlpRDCompressionFun::TypeIsSupported},
     {CompressionType::COMPRESSION_FSST, FSSTFun::GetFunction, FSSTFun::TypeIsSupported},
+    {CompressionType::COMPRESSION_ZSTD, ZSTDFun::GetFunction, ZSTDFun::TypeIsSupported},
     {CompressionType::COMPRESSION_AUTO, nullptr, nullptr}};
 
 static optional_ptr<CompressionFunction> FindCompressionFunction(CompressionFunctionSet &set, CompressionType type,
@@ -79,6 +80,7 @@ vector<reference<CompressionFunction>> DBConfig::GetCompressionFunctions(const P
 	TryLoadCompression(*this, result, CompressionType::COMPRESSION_ALP, physical_type);
 	TryLoadCompression(*this, result, CompressionType::COMPRESSION_ALPRD, physical_type);
 	TryLoadCompression(*this, result, CompressionType::COMPRESSION_FSST, physical_type);
+	TryLoadCompression(*this, result, CompressionType::COMPRESSION_ZSTD, physical_type);
 	return result;
 }
 
diff --git a/src/function/table/system/pragma_storage_info.cpp b/src/function/table/system/pragma_storage_info.cpp
index 0c33ebf01949..5500c1c5d522 100644
--- a/src/function/table/system/pragma_storage_info.cpp
+++ b/src/function/table/system/pragma_storage_info.cpp
@@ -79,6 +79,9 @@ static unique_ptr<FunctionData> PragmaStorageInfoBind(ClientContext &context, Ta
 	names.emplace_back("segment_info");
 	return_types.emplace_back(LogicalType::VARCHAR);
 
+	names.emplace_back("additional_block_ids");
+	return_types.emplace_back(LogicalType::LIST(LogicalTypeId::BIGINT));
+
 	auto qname = QualifiedName::Parse(input.inputs[0].GetValue<string>());
 
 	// look up the table name in the catalog
@@ -93,6 +96,14 @@ unique_ptr<GlobalTableFunctionState> PragmaStorageInfoInit(ClientContext &contex
 	return make_uniq<PragmaStorageOperatorData>();
 }
 
+static Value ValueFromBlockIdList(const vector<block_id_t> &block_ids) {
+	vector<Value> blocks;
+	for (auto &block_id : block_ids) {
+		blocks.push_back(Value::BIGINT(block_id));
+	}
+	return Value::LIST(LogicalTypeId::BIGINT, blocks);
+}
+
 static void PragmaStorageInfoFunction(ClientContext &context, TableFunctionInput &data_p, DataChunk &output) {
 	auto &bind_data = data_p.bind_data->Cast<PragmaStorageFunctionData>();
 	auto &data = data_p.global_state->Cast<PragmaStorageOperatorData>();
@@ -138,6 +149,12 @@ static void PragmaStorageInfoFunction(ClientContext &context, TableFunctionInput
 		}
 		// segment_info
 		output.SetValue(col_idx++, count, Value(entry.segment_info));
+		// additional_block_ids
+		if (entry.persistent) {
+			output.SetValue(col_idx++, count, ValueFromBlockIdList(entry.additional_blocks));
+		} else {
+			output.SetValue(col_idx++, count, Value());
+		}
 		count++;
 	}
 	output.SetCardinality(count);
diff --git a/src/include/duckdb/common/allocator.hpp b/src/include/duckdb/common/allocator.hpp
index f4550ef61e03..cd058645f930 100644
--- a/src/include/duckdb/common/allocator.hpp
+++ b/src/include/duckdb/common/allocator.hpp
@@ -65,6 +65,9 @@ class AllocatedData {
 	data_ptr_t get() { // NOLINT: matching std style
 		return pointer;
 	}
+	operator bool() const { // NOLINT: missing explicit
+		return pointer != nullptr;
+	}
 	const_data_ptr_t get() const { // NOLINT: matching std style
 		return pointer;
 	}
diff --git a/src/include/duckdb/common/enums/compression_type.hpp b/src/include/duckdb/common/enums/compression_type.hpp
index 19bfa4cc87b7..e6c7191af9a9 100644
--- a/src/include/duckdb/common/enums/compression_type.hpp
+++ b/src/include/duckdb/common/enums/compression_type.hpp
@@ -26,6 +26,7 @@ enum class CompressionType : uint8_t {
 	COMPRESSION_PATAS = 9,
 	COMPRESSION_ALP = 10,
 	COMPRESSION_ALPRD = 11,
+	COMPRESSION_ZSTD = 12,
 	COMPRESSION_COUNT // This has to stay the last entry of the type!
 };
 
diff --git a/src/include/duckdb/common/enums/order_preservation_type.hpp b/src/include/duckdb/common/enums/order_preservation_type.hpp
index ef590c03b11b..d28edb20ef47 100644
--- a/src/include/duckdb/common/enums/order_preservation_type.hpp
+++ b/src/include/duckdb/common/enums/order_preservation_type.hpp
@@ -1,7 +1,7 @@
 //===----------------------------------------------------------------------===//
 //                         DuckDB
 //
-// src/include/duckdb/common/enums/order_preservation_type.hpp
+// duckdb/common/enums/order_preservation_type.hpp
 //
 //
 //===----------------------------------------------------------------------===//
diff --git a/src/include/duckdb/function/compression/compression.hpp b/src/include/duckdb/function/compression/compression.hpp
index cddfc8067f4d..df19c392a8a7 100644
--- a/src/include/duckdb/function/compression/compression.hpp
+++ b/src/include/duckdb/function/compression/compression.hpp
@@ -63,4 +63,9 @@ struct FSSTFun {
 	static bool TypeIsSupported(const PhysicalType physical_type);
 };
 
+struct ZSTDFun {
+	static CompressionFunction GetFunction(PhysicalType type);
+	static bool TypeIsSupported(PhysicalType type);
+};
+
 } // namespace duckdb
diff --git a/src/include/duckdb/function/compression_function.hpp b/src/include/duckdb/function/compression_function.hpp
index 40f50653b3da..0ead5825c30f 100644
--- a/src/include/duckdb/function/compression_function.hpp
+++ b/src/include/duckdb/function/compression_function.hpp
@@ -95,6 +95,11 @@ struct CompressedSegmentState {
 		return "";
 	} // LCOV_EXCL_STOP
 
+	//! Get the block ids of additional pages created by the segment
+	virtual vector<block_id_t> GetAdditionalBlocks() const { // LCOV_EXCL_START
+		return vector<block_id_t>();
+	} // LCOV_EXCL_STOP
+
 	template <class TARGET>
 	TARGET &Cast() {
 		DynamicCastCheck<TARGET>(this);
diff --git a/src/include/duckdb/main/config.hpp b/src/include/duckdb/main/config.hpp
index 11b0941f710a..3e9e687368c1 100644
--- a/src/include/duckdb/main/config.hpp
+++ b/src/include/duckdb/main/config.hpp
@@ -196,6 +196,8 @@ struct DBConfigOptions {
 	bool initialize_default_database = true;
 	//! The set of disabled optimizers (default empty)
 	set<OptimizerType> disabled_optimizers;
+	//! The average string length required to use ZSTD compression.
+	uint64_t zstd_min_string_length = 4096;
 	//! Force a specific compression method to be used when checkpointing (if available)
 	CompressionType force_compression = CompressionType::COMPRESSION_AUTO;
 	//! Force a specific bitpacking mode to be used when using the bitpacking compression method
diff --git a/src/include/duckdb/main/settings.hpp b/src/include/duckdb/main/settings.hpp
index ab33eab1651a..741c8c0b49c0 100644
--- a/src/include/duckdb/main/settings.hpp
+++ b/src/include/duckdb/main/settings.hpp
@@ -1092,6 +1092,17 @@ struct UsernameSetting {
 	static Value GetSetting(const ClientContext &context);
 };
 
+struct ZstdMinStringLengthSetting {
+	using RETURN_TYPE = idx_t;
+	static constexpr const char *Name = "zstd_min_string_length";
+	static constexpr const char *Description =
+	    "The (average) length at which to enable ZSTD compression, defaults to 4096";
+	static constexpr const char *InputType = "UBIGINT";
+	static void SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &parameter);
+	static void ResetGlobal(DatabaseInstance *db, DBConfig &config);
+	static Value GetSetting(const ClientContext &context);
+};
+
 //===----------------------------------------------------------------------===//
 // End of the auto-generated list of settings structures
 //===--------------------------------------------------------------------===//
diff --git a/src/include/duckdb/storage/checkpoint/string_checkpoint_state.hpp b/src/include/duckdb/storage/checkpoint/string_checkpoint_state.hpp
index 59eb2efd42df..22621148a852 100644
--- a/src/include/duckdb/storage/checkpoint/string_checkpoint_state.hpp
+++ b/src/include/duckdb/storage/checkpoint/string_checkpoint_state.hpp
@@ -73,6 +73,10 @@ struct UncompressedStringSegmentState : public CompressedSegmentState {
 		return "Overflow String Block Ids: " + result;
 	}
 
+	vector<block_id_t> GetAdditionalBlocks() const override {
+		return on_disk_blocks;
+	}
+
 private:
 	mutex block_lock;
 	unordered_map<block_id_t, shared_ptr<BlockHandle>> handles;
diff --git a/src/include/duckdb/storage/string_uncompressed.hpp b/src/include/duckdb/storage/string_uncompressed.hpp
index 9d82335975a9..f47f61ef1b81 100644
--- a/src/include/duckdb/storage/string_uncompressed.hpp
+++ b/src/include/duckdb/storage/string_uncompressed.hpp
@@ -41,6 +41,18 @@ struct StringScanState : public SegmentScanState {
 	BufferHandle handle;
 };
 
+//===--------------------------------------------------------------------===//
+// Append
+//===--------------------------------------------------------------------===//
+struct SerializedStringSegmentState : public ColumnSegmentState {
+public:
+	SerializedStringSegmentState();
+	explicit SerializedStringSegmentState(vector<block_id_t> blocks_p);
+
+public:
+	void Serialize(Serializer &serializer) const override;
+};
+
 struct UncompressedStringStorage {
 public:
 	//! Dictionary header size at the beginning of the string segment (offset + length)
diff --git a/src/include/duckdb/storage/table_storage_info.hpp b/src/include/duckdb/storage/table_storage_info.hpp
index 49b3089c657a..493dd320a81a 100644
--- a/src/include/duckdb/storage/table_storage_info.hpp
+++ b/src/include/duckdb/storage/table_storage_info.hpp
@@ -31,6 +31,7 @@ struct ColumnSegmentInfo {
 	bool has_updates;
 	bool persistent;
 	block_id_t block_id;
+	vector<block_id_t> additional_blocks;
 	idx_t block_offset;
 	string segment_info;
 };
diff --git a/src/main/config.cpp b/src/main/config.cpp
index b72ec07cb89f..f1c595c3a071 100644
--- a/src/main/config.cpp
+++ b/src/main/config.cpp
@@ -159,6 +159,7 @@ static const ConfigurationOption internal_options[] = {
     DUCKDB_GLOBAL_ALIAS("worker_threads", ThreadsSetting),
     DUCKDB_GLOBAL(UsernameSetting),
     DUCKDB_GLOBAL_ALIAS("user", UsernameSetting),
+    DUCKDB_GLOBAL(ZstdMinStringLengthSetting),
     FINAL_SETTING};
 
 vector<ConfigurationOption> DBConfig::GetOptions() {
diff --git a/src/main/settings/autogenerated_settings.cpp b/src/main/settings/autogenerated_settings.cpp
index 1fac3ffe4a68..f7bd49bce9fe 100644
--- a/src/main/settings/autogenerated_settings.cpp
+++ b/src/main/settings/autogenerated_settings.cpp
@@ -1,9 +1,9 @@
 //===----------------------------------------------------------------------===//
 //                         DuckDB
-// This code is autogenerated from 'update_settings_definitions.py'.
+// This code is autogenerated from 'scripts/settings_scripts/update_settings_src_code.py'.
 // Please do not make any changes directly here, as they will be overwritten.
 // If you need to implement a custom function for a new setting, enable the
-// 'custom_implementation' in 'settings.json' for this setting.
+// 'custom_implementation' in 'src/common/settings.json' for this setting.
 //
 //===----------------------------------------------------------------------===//
 
@@ -1065,4 +1065,20 @@ Value ScalarSubqueryErrorOnMultipleRowsSetting::GetSetting(const ClientContext &
 	return Value::BOOLEAN(config.scalar_subquery_error_on_multiple_rows);
 }
 
+//===----------------------------------------------------------------------===//
+// Zstd Min String Length
+//===----------------------------------------------------------------------===//
+void ZstdMinStringLengthSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
+	config.options.zstd_min_string_length = input.GetValue<idx_t>();
+}
+
+void ZstdMinStringLengthSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
+	config.options.zstd_min_string_length = DBConfig().options.zstd_min_string_length;
+}
+
+Value ZstdMinStringLengthSetting::GetSetting(const ClientContext &context) {
+	auto &config = DBConfig::GetConfig(context);
+	return Value::UBIGINT(config.options.zstd_min_string_length);
+}
+
 } // namespace duckdb
diff --git a/src/main/settings/custom_settings.cpp b/src/main/settings/custom_settings.cpp
index 887c3ce1ec2e..c08077e27871 100644
--- a/src/main/settings/custom_settings.cpp
+++ b/src/main/settings/custom_settings.cpp
@@ -1,7 +1,7 @@
 //===----------------------------------------------------------------------===//
 //                         DuckDB
 // This file will not be overwritten. To implement a custom function for
-// a new setting, enable 'custom_implementation' in 'settings.json'
+// a new setting, enable 'custom_implementation' in 'src/common/settings.json'
 // for this setting. The 'update_settings_definitions.py' may include new
 // setting methods' signatures that need to be implemented in this file. You
 // can check the functions declaration in 'settings.hpp' and what is
diff --git a/src/storage/compression/CMakeLists.txt b/src/storage/compression/CMakeLists.txt
index fa30a00f041d..befc2b814376 100644
--- a/src/storage/compression/CMakeLists.txt
+++ b/src/storage/compression/CMakeLists.txt
@@ -14,6 +14,7 @@ add_library_unity(
   bitpacking.cpp
   bitpacking_hugeint.cpp
   patas.cpp
+  zstd.cpp
   alprd.cpp
   fsst.cpp)
 set(ALL_OBJECT_FILES
diff --git a/src/storage/compression/string_uncompressed.cpp b/src/storage/compression/string_uncompressed.cpp
index aa340f781ca3..6b8ce948c8cd 100644
--- a/src/storage/compression/string_uncompressed.cpp
+++ b/src/storage/compression/string_uncompressed.cpp
@@ -158,17 +158,16 @@ void UncompressedStringStorage::StringFetchRow(ColumnSegment &segment, ColumnFet
 //===--------------------------------------------------------------------===//
 // Append
 //===--------------------------------------------------------------------===//
-struct SerializedStringSegmentState : public ColumnSegmentState {
-	SerializedStringSegmentState() {
-	}
-	explicit SerializedStringSegmentState(vector<block_id_t> blocks_p) {
-		blocks = std::move(blocks_p);
-	}
+SerializedStringSegmentState::SerializedStringSegmentState() {
+}
 
-	void Serialize(Serializer &serializer) const override {
-		serializer.WriteProperty(1, "overflow_blocks", blocks);
-	}
-};
+SerializedStringSegmentState::SerializedStringSegmentState(vector<block_id_t> blocks_p) {
+	blocks = std::move(blocks_p);
+}
+
+void SerializedStringSegmentState::Serialize(Serializer &serializer) const {
+	serializer.WriteProperty(1, "overflow_blocks", blocks);
+}
 
 unique_ptr<CompressedSegmentState>
 UncompressedStringStorage::StringInitSegment(ColumnSegment &segment, block_id_t block_id,
diff --git a/src/storage/compression/zstd.cpp b/src/storage/compression/zstd.cpp
new file mode 100644
index 000000000000..161eaf215f43
--- /dev/null
+++ b/src/storage/compression/zstd.cpp
@@ -0,0 +1,1048 @@
+#include "duckdb/common/bitpacking.hpp"
+#include "duckdb/storage/string_uncompressed.hpp"
+#include "duckdb/function/compression/compression.hpp"
+#include "duckdb/storage/table/column_data_checkpointer.hpp"
+#include "duckdb/storage/block_manager.hpp"
+#include "duckdb/main/config.hpp"
+#include "duckdb/common/constants.hpp"
+#include "duckdb/common/allocator.hpp"
+#include "duckdb/common/serializer/deserializer.hpp"
+#include "duckdb/storage/segment/uncompressed.hpp"
+
+#include "zstd.h"
+
+/*
+Data layout per segment:
++--------------------------------------------+
+|            Vector Metadata                 |
+|   +------------------------------------+   |
+|   |   int64_t  page_id[]               |   |
+|   |   uint32_t page_offset[]           |   |
+|   |   uint64_t uncompressed_size[]     |   |
+|   |   uint64_t compressed_size[]       |   |
+|   +------------------------------------+   |
+|                                            |
++--------------------------------------------+
+|              [Vector Data]+                |
+|   +------------------------------------+   |
+|   |   uint32_t lengths[]               |   |
+|   |   void    *compressed_data         |   |
+|   +------------------------------------+   |
+|                                            |
++--------------------------------------------+
+*/
+
+using page_id_t = int64_t;
+using page_offset_t = uint32_t;
+using uncompressed_size_t = uint64_t;
+using compressed_size_t = uint64_t;
+using string_length_t = uint32_t;
+
+static int32_t GetCompressionLevel() {
+	return duckdb_zstd::ZSTD_defaultCLevel();
+}
+
+static constexpr idx_t ZSTD_VECTOR_SIZE = STANDARD_VECTOR_SIZE > 2048 ? STANDARD_VECTOR_SIZE : 2048;
+
+namespace duckdb {
+
+static idx_t GetWritableSpace(const CompressionInfo &info) {
+	return info.GetBlockSize() - sizeof(block_id_t);
+}
+
+static idx_t GetVectorCount(idx_t count) {
+	idx_t vector_count = count / ZSTD_VECTOR_SIZE;
+	vector_count += (count % ZSTD_VECTOR_SIZE) != 0;
+	return vector_count;
+}
+
+static idx_t GetVectorMetadataSize(idx_t vector_count) {
+	idx_t vector_metadata_size = 0;
+	vector_metadata_size += sizeof(page_id_t) * vector_count;
+
+	vector_metadata_size = AlignValue<idx_t, sizeof(page_offset_t)>(vector_metadata_size);
+	vector_metadata_size += sizeof(page_offset_t) * vector_count;
+
+	vector_metadata_size = AlignValue<idx_t, sizeof(uncompressed_size_t)>(vector_metadata_size);
+	vector_metadata_size += sizeof(uncompressed_size_t) * vector_count;
+
+	vector_metadata_size = AlignValue<idx_t, sizeof(compressed_size_t)>(vector_metadata_size);
+	vector_metadata_size += sizeof(compressed_size_t) * vector_count;
+	return vector_metadata_size;
+}
+
+struct ZSTDStorage {
+	static unique_ptr<AnalyzeState> StringInitAnalyze(ColumnData &col_data, PhysicalType type);
+	static bool StringAnalyze(AnalyzeState &state_p, Vector &input, idx_t count);
+	static idx_t StringFinalAnalyze(AnalyzeState &state_p);
+
+	static unique_ptr<CompressionState> InitCompression(ColumnDataCheckpointer &checkpointer,
+	                                                    unique_ptr<AnalyzeState> analyze_state_p);
+	static void Compress(CompressionState &state_p, Vector &scan_vector, idx_t count);
+	static void FinalizeCompress(CompressionState &state_p);
+
+	static unique_ptr<SegmentScanState> StringInitScan(ColumnSegment &segment);
+	static void StringScanPartial(ColumnSegment &segment, ColumnScanState &state, idx_t scan_count, Vector &result,
+	                              idx_t result_offset);
+	static void StringScan(ColumnSegment &segment, ColumnScanState &state, idx_t scan_count, Vector &result);
+	static void StringFetchRow(ColumnSegment &segment, ColumnFetchState &state, row_t row_id, Vector &result,
+	                           idx_t result_idx);
+	static void StringSkip(ColumnSegment &segment, ColumnScanState &state, idx_t skip_count) {
+		// NO OP
+	}
+
+	// Segment state metadata
+	// Required because we are creating additional pages that have to be cleaned up
+
+	static unique_ptr<CompressedSegmentState> StringInitSegment(ColumnSegment &segment, block_id_t block_id,
+	                                                            optional_ptr<ColumnSegmentState> segment_state);
+	static unique_ptr<ColumnSegmentState> SerializeState(ColumnSegment &segment);
+	static unique_ptr<ColumnSegmentState> DeserializeState(Deserializer &deserializer);
+	static void CleanupState(ColumnSegment &segment);
+};
+
+//===--------------------------------------------------------------------===//
+// Analyze
+//===--------------------------------------------------------------------===//
+
+struct ZSTDAnalyzeState : public AnalyzeState {
+public:
+	ZSTDAnalyzeState(CompressionInfo &info, DBConfig &config) : AnalyzeState(info), config(config), context(nullptr) {
+		context = duckdb_zstd::ZSTD_createCCtx();
+	}
+	~ZSTDAnalyzeState() override {
+		duckdb_zstd::ZSTD_freeCCtx(context);
+	}
+
+public:
+	inline void AppendString(const string_t &str) {
+		auto string_size = str.GetSize();
+		total_size += string_size;
+	}
+
+public:
+	DBConfig &config;
+
+	duckdb_zstd::ZSTD_CCtx *context;
+	//! The combined string lengths for all values in the segment
+	idx_t total_size = 0;
+	//! The total amount of values in the segment
+	idx_t count = 0;
+
+	//! The amount of vectors per filled segment
+	idx_t vectors_per_segment = 0;
+	//! The total amount of segments we will create
+	idx_t segment_count = 0;
+	//! Current vector in the segment
+	idx_t vectors_in_segment = 0;
+	//! Current amount of values in the vector
+	idx_t values_in_vector = 0;
+};
+
+unique_ptr<AnalyzeState> ZSTDStorage::StringInitAnalyze(ColumnData &col_data, PhysicalType type) {
+	CompressionInfo info(col_data.GetBlockManager().GetBlockSize());
+	auto &data_table_info = col_data.info;
+	auto &attached_db = data_table_info.GetDB();
+	auto &config = DBConfig::Get(attached_db);
+
+	return make_uniq<ZSTDAnalyzeState>(info, config);
+}
+
+bool ZSTDStorage::StringAnalyze(AnalyzeState &state_p, Vector &input, idx_t count) {
+	auto &state = state_p.Cast<ZSTDAnalyzeState>();
+	UnifiedVectorFormat vdata;
+	input.ToUnifiedFormat(count, vdata);
+
+	auto data = UnifiedVectorFormat::GetData<string_t>(vdata);
+	for (idx_t i = 0; i < count; i++) {
+		auto idx = vdata.sel->get_index(i);
+		if (!vdata.validity.RowIsValid(idx)) {
+			continue;
+		}
+		auto &str = data[idx];
+		auto string_size = str.GetSize();
+		state.total_size += string_size;
+	}
+	state.values_in_vector += count;
+	while (state.values_in_vector >= ZSTD_VECTOR_SIZE) {
+		if (GetVectorMetadataSize(state.vectors_in_segment + 1) > GetWritableSpace(state.info)) {
+			state.vectors_per_segment = state.vectors_in_segment;
+			state.segment_count++;
+			state.vectors_in_segment = 0;
+		} else {
+			state.vectors_in_segment++;
+		}
+		state.values_in_vector -= ZSTD_VECTOR_SIZE;
+	}
+	state.count += count;
+	return true;
+}
+
+// Compression score to determine which compression to use
+idx_t ZSTDStorage::StringFinalAnalyze(AnalyzeState &state_p) {
+	auto &state = state_p.Cast<ZSTDAnalyzeState>();
+
+	if (!state.count) {
+		return NumericLimits<idx_t>::Maximum();
+	}
+
+	if (state.values_in_vector) {
+		D_ASSERT(state.values_in_vector < ZSTD_VECTOR_SIZE);
+		state.segment_count++;
+	}
+
+	double penalty;
+	idx_t average_length = state.total_size / state.count;
+	auto threshold = state.config.options.zstd_min_string_length;
+	if (average_length >= threshold) {
+		penalty = 1.0;
+	} else {
+		// Inbetween these two points you're better off using uncompressed or a different compression algorithm.
+		return NumericLimits<idx_t>::Maximum();
+	}
+
+	auto expected_compressed_size = (double)state.total_size / 2.0;
+
+	idx_t estimated_size = 0;
+	estimated_size += LossyNumericCast<idx_t>(expected_compressed_size);
+
+	estimated_size += state.count * sizeof(string_length_t);
+	estimated_size += GetVectorMetadataSize(GetVectorCount(state.count));
+
+	return LossyNumericCast<idx_t>((double)estimated_size * penalty);
+}
+
+//===--------------------------------------------------------------------===//
+// Compress
+//===--------------------------------------------------------------------===//
+
+class ZSTDCompressionState : public CompressionState {
+public:
+	explicit ZSTDCompressionState(ColumnDataCheckpointer &checkpointer, unique_ptr<ZSTDAnalyzeState> &&analyze_state_p)
+	    : CompressionState(analyze_state_p->info), analyze_state(std::move(analyze_state_p)),
+	      checkpointer(checkpointer), partial_block_manager(checkpointer.GetCheckpointState().GetPartialBlockManager()),
+	      function(checkpointer.GetCompressionFunction(CompressionType::COMPRESSION_ZSTD)) {
+
+		total_vector_count = GetVectorCount(analyze_state->count);
+		total_segment_count = analyze_state->segment_count;
+		vectors_per_segment = analyze_state->vectors_per_segment;
+
+		segment_count = 0;
+		vector_count = 0;
+		vector_in_segment_count = 0;
+		tuple_count = 0;
+
+		idx_t offset = NewSegment();
+		SetCurrentBuffer(segment_handle);
+		current_buffer_ptr = segment_handle.Ptr() + offset;
+		D_ASSERT(GetCurrentOffset() <= GetWritableSpace(info));
+	}
+
+public:
+	void ResetOutBuffer() {
+		out_buffer.dst = current_buffer_ptr;
+		out_buffer.pos = 0;
+
+		auto remaining_space = info.GetBlockSize() - GetCurrentOffset() - sizeof(block_id_t);
+		out_buffer.size = remaining_space;
+	}
+
+	void SetCurrentBuffer(BufferHandle &handle) {
+		current_buffer = &handle;
+		current_buffer_ptr = handle.Ptr();
+	}
+
+	BufferHandle &GetExtraPageBuffer(block_id_t current_block_id) {
+		auto &block_manager = partial_block_manager.GetBlockManager();
+		auto &buffer_manager = block_manager.buffer_manager;
+
+		optional_ptr<BufferHandle> to_use;
+
+		if (in_vector) {
+			// Currently in a Vector, we have to be mindful of the buffer that the string_lengths lives on
+			// as that will have to stay writable until the Vector is finished
+			bool already_separated = current_buffer != vector_lengths_buffer;
+			if (already_separated) {
+				// Already separated, can keep using the other buffer (flush it first)
+				FlushPage(*current_buffer, current_block_id);
+				to_use = current_buffer;
+			} else {
+				// Not already separated, have to use the other page
+				to_use = current_buffer == &extra_pages[0] ? &extra_pages[1] : &extra_pages[0];
+			}
+		} else {
+			// Start of a new Vector, the string_lengths did not fit on the previous page
+			bool previous_page_is_segment = current_buffer == &segment_handle;
+			if (!previous_page_is_segment) {
+				// We're asking for a fresh buffer to start the vectors data
+				// that means the previous vector is finished - so we can flush the current page and reuse it
+				D_ASSERT(current_block_id != INVALID_BLOCK);
+				FlushPage(*current_buffer, current_block_id);
+				to_use = current_buffer;
+			} else {
+				// Previous buffer was the segment, take the first extra page in this case
+				to_use = &extra_pages[0];
+			}
+		}
+
+		if (!to_use->IsValid()) {
+			*to_use = buffer_manager.Allocate(MemoryTag::OVERFLOW_STRINGS, block_manager.GetBlockSize());
+		}
+		return *to_use;
+	}
+
+	idx_t NewSegment() {
+		if (current_buffer == &segment_handle) {
+			// This should never happen, the string lengths + vector metadata size should always exceed a page size,
+			// even if the strings are all empty
+			throw InternalException("We are asking for a new segment, but somehow we're still writing vector data onto "
+			                        "the initial (segment) page");
+		}
+		idx_t row_start;
+		if (segment) {
+			row_start = segment->start + segment->count;
+			FlushSegment();
+		} else {
+			row_start = checkpointer.GetRowGroup().start;
+		}
+		CreateEmptySegment(row_start);
+
+		// Figure out how many vectors we are storing in this segment
+		idx_t vectors_in_segment;
+		if (segment_count + 1 >= total_segment_count) {
+			vectors_in_segment = total_vector_count - vector_count;
+		} else {
+			vectors_in_segment = vectors_per_segment;
+		}
+
+		idx_t offset = 0;
+		page_ids = reinterpret_cast<page_id_t *>(segment_handle.Ptr() + offset);
+		offset += (sizeof(page_id_t) * vectors_in_segment);
+
+		offset = AlignValue<idx_t, sizeof(page_offset_t)>(offset);
+		page_offsets = reinterpret_cast<page_offset_t *>(segment_handle.Ptr() + offset);
+		offset += (sizeof(page_offset_t) * vectors_in_segment);
+
+		offset = AlignValue<idx_t, sizeof(uncompressed_size_t)>(offset);
+		uncompressed_sizes = reinterpret_cast<uncompressed_size_t *>(segment_handle.Ptr() + offset);
+		offset += (sizeof(uncompressed_size_t) * vectors_in_segment);
+
+		offset = AlignValue<idx_t, sizeof(compressed_size_t)>(offset);
+		compressed_sizes = reinterpret_cast<compressed_size_t *>(segment_handle.Ptr() + offset);
+		offset += (sizeof(compressed_size_t) * vectors_in_segment);
+
+		D_ASSERT(offset == GetVectorMetadataSize(vectors_in_segment));
+		return offset;
+	}
+
+	void InitializeVector() {
+		D_ASSERT(!in_vector);
+		if (vector_count + 1 >= total_vector_count) {
+			vector_size = analyze_state->count - (ZSTD_VECTOR_SIZE * vector_count);
+		} else {
+			vector_size = ZSTD_VECTOR_SIZE;
+		}
+		auto current_offset = GetCurrentOffset();
+		current_offset = UnsafeNumericCast<page_offset_t>(
+		    AlignValue<idx_t, sizeof(string_length_t)>(UnsafeNumericCast<idx_t>(current_offset)));
+		current_buffer_ptr = current_buffer->Ptr() + current_offset;
+		compressed_size = 0;
+		uncompressed_size = 0;
+
+		if (GetVectorMetadataSize(vector_in_segment_count + 1) > GetWritableSpace(info)) {
+			D_ASSERT(vector_in_segment_count <= vectors_per_segment);
+			// Can't fit this vector on this segment anymore, have to flush and a grab new one
+			NewSegment();
+		}
+
+		if (current_offset + (vector_size * sizeof(string_length_t)) >= GetWritableSpace(info)) {
+			// Check if there is room on the current page for the vector data
+			NewPage();
+		}
+		current_offset = GetCurrentOffset();
+		starting_offset = current_offset;
+		starting_page = GetCurrentId();
+
+		vector_lengths_buffer = current_buffer;
+		string_lengths = reinterpret_cast<string_length_t *>(current_buffer->Ptr() + current_offset);
+		current_buffer_ptr = reinterpret_cast<data_ptr_t>(string_lengths);
+		current_buffer_ptr += vector_size * sizeof(string_length_t);
+		// 'out_buffer' should be set to point directly after the string_lengths
+		ResetOutBuffer();
+
+		// Initialize the context for streaming compression
+		duckdb_zstd::ZSTD_CCtx_reset(analyze_state->context, duckdb_zstd::ZSTD_reset_session_only);
+		duckdb_zstd::ZSTD_CCtx_refCDict(analyze_state->context, nullptr);
+		duckdb_zstd::ZSTD_CCtx_setParameter(analyze_state->context, duckdb_zstd::ZSTD_c_compressionLevel,
+		                                    GetCompressionLevel());
+		in_vector = true;
+	}
+
+	void CompressString(const string_t &string, bool end_of_vector) {
+		duckdb_zstd::ZSTD_inBuffer in_buffer = {/*data = */ string.GetData(),
+		                                        /*length = */ string.GetSize(),
+		                                        /*pos = */ 0};
+
+		if (!end_of_vector && string.GetSize() == 0) {
+			return;
+		}
+		uncompressed_size += string.GetSize();
+		const auto end_mode = end_of_vector ? duckdb_zstd::ZSTD_e_end : duckdb_zstd::ZSTD_e_continue;
+
+		size_t compress_result;
+		while (true) {
+			idx_t old_pos = out_buffer.pos;
+
+			compress_result =
+			    duckdb_zstd::ZSTD_compressStream2(analyze_state->context, &out_buffer, &in_buffer, end_mode);
+			D_ASSERT(out_buffer.pos >= old_pos);
+			auto diff = out_buffer.pos - old_pos;
+			compressed_size += diff;
+			current_buffer_ptr += diff;
+
+			if (duckdb_zstd::ZSTD_isError(compress_result)) {
+				throw InvalidInputException("ZSTD Compression failed: %s",
+				                            duckdb_zstd::ZSTD_getErrorName(compress_result));
+			}
+			if (compress_result == 0) {
+				// Finished
+				break;
+			}
+			if (out_buffer.pos != out_buffer.size) {
+				throw InternalException("Expected ZSTD_compressStream2 to fully utilize the current buffer, but pos is "
+				                        "%d, while size is %d",
+				                        out_buffer.pos, out_buffer.size);
+			}
+			NewPage();
+		}
+	}
+
+	void AddString(const string_t &string) {
+		if (!tuple_count) {
+			InitializeVector();
+		}
+
+		string_lengths[tuple_count] = UnsafeNumericCast<string_length_t>(string.GetSize());
+		bool final_tuple = tuple_count + 1 >= vector_size;
+		CompressString(string, final_tuple);
+
+		tuple_count++;
+		if (tuple_count == vector_size) {
+			// Reached the end of this vector
+			FlushVector();
+		}
+
+		UncompressedStringStorage::UpdateStringStats(segment->stats, string);
+	}
+
+	void NewPage(bool additional_data_page = false) {
+		block_id_t new_id = FinalizePage();
+		block_id_t current_block_id = block_id;
+		auto &buffer = GetExtraPageBuffer(current_block_id);
+		block_id = new_id;
+		SetCurrentBuffer(buffer);
+		ResetOutBuffer();
+	}
+
+	block_id_t FinalizePage() {
+		auto &block_manager = partial_block_manager.GetBlockManager();
+		auto new_id = block_manager.GetFreeBlockId();
+		auto &state = segment->GetSegmentState()->Cast<UncompressedStringSegmentState>();
+		state.RegisterBlock(block_manager, new_id);
+
+		D_ASSERT(GetCurrentOffset() <= GetWritableSpace(info));
+
+		// Write the new id at the end of the last page
+		Store<block_id_t>(new_id, current_buffer_ptr);
+		current_buffer_ptr += sizeof(block_id_t);
+		return new_id;
+	}
+
+	void FlushPage(BufferHandle &buffer, block_id_t block_id) {
+		if (block_id == INVALID_BLOCK) {
+			return;
+		}
+
+		// Write the current page to disk
+		auto &block_manager = partial_block_manager.GetBlockManager();
+		block_manager.Write(buffer.GetFileBuffer(), block_id);
+		{
+			auto lock = partial_block_manager.GetLock();
+			partial_block_manager.AddWrittenBlock(block_id);
+		}
+	}
+
+	void FlushVector() {
+		// Write the metadata for this Vector
+		page_ids[vector_in_segment_count] = starting_page;
+		page_offsets[vector_in_segment_count] = starting_offset;
+		compressed_sizes[vector_in_segment_count] = compressed_size;
+		uncompressed_sizes[vector_in_segment_count] = uncompressed_size;
+		vector_count++;
+		vector_in_segment_count++;
+		in_vector = false;
+		segment->count += tuple_count;
+
+		const bool is_last_vector = vector_count == total_vector_count;
+		tuple_count = 0;
+		if (is_last_vector) {
+			FlushPage(*current_buffer, block_id);
+			if (starting_page != block_id) {
+				FlushPage(*vector_lengths_buffer, starting_page);
+			}
+		} else {
+			if (vector_lengths_buffer == current_buffer) {
+				// We did not cross a page boundary writing this vector
+				return;
+			}
+			// Flush the page that holds the vector lengths
+			FlushPage(*vector_lengths_buffer, starting_page);
+		}
+	}
+
+	page_id_t GetCurrentId() {
+		if (&segment_handle == current_buffer.get()) {
+			return INVALID_BLOCK;
+		}
+		return block_id;
+	}
+
+	page_offset_t GetCurrentOffset() {
+		auto &handle = *current_buffer;
+		auto start_of_buffer = handle.Ptr();
+		D_ASSERT(current_buffer_ptr >= start_of_buffer);
+		auto res = (page_offset_t)(current_buffer_ptr - start_of_buffer);
+		D_ASSERT(res <= GetWritableSpace(info));
+		return res;
+	}
+
+	void CreateEmptySegment(idx_t row_start) {
+		auto &db = checkpointer.GetDatabase();
+		auto &type = checkpointer.GetType();
+		auto compressed_segment =
+		    ColumnSegment::CreateTransientSegment(db, type, row_start, info.GetBlockSize(), info.GetBlockSize());
+		segment = std::move(compressed_segment);
+		segment->function = function;
+
+		auto &buffer_manager = BufferManager::GetBufferManager(checkpointer.GetDatabase());
+		segment_handle = buffer_manager.Pin(segment->block);
+	}
+
+	void FlushSegment() {
+		auto &state = checkpointer.GetCheckpointState();
+		idx_t segment_block_size;
+
+		if (current_buffer.get() == &segment_handle) {
+			segment_block_size = GetCurrentOffset();
+		} else {
+			// Block is fully used
+			segment_block_size = info.GetBlockSize();
+		}
+
+		state.FlushSegment(std::move(segment), std::move(segment_handle), segment_block_size);
+		segment_count++;
+		vector_in_segment_count = 0;
+	}
+
+	void Finalize() {
+		D_ASSERT(!tuple_count);
+		FlushSegment();
+		segment.reset();
+	}
+
+	void AddNull() {
+		AddString("");
+	}
+
+public:
+	unique_ptr<ZSTDAnalyzeState> analyze_state;
+	ColumnDataCheckpointer &checkpointer;
+	PartialBlockManager &partial_block_manager;
+	CompressionFunction &function;
+
+	// The segment state
+	//! Current segment index we're at
+	idx_t segment_count = 0;
+	//! The total amount of segments we're writing
+	idx_t total_segment_count = 0;
+	//! The vectors to store in the last segment
+	idx_t vectors_in_last_segment = 0;
+	//! The vectors to store in a segment (not the last one)
+	idx_t vectors_per_segment = 0;
+	unique_ptr<ColumnSegment> segment;
+	BufferHandle segment_handle;
+
+	// Non-segment buffers
+	BufferHandle extra_pages[2];
+	block_id_t block_id = INVALID_BLOCK;
+
+	// Current block state
+	optional_ptr<BufferHandle> current_buffer;
+	//! The buffer that contains the vector lengths
+	optional_ptr<BufferHandle> vector_lengths_buffer;
+	data_ptr_t current_buffer_ptr;
+
+	//===--------------------------------------------------------------------===//
+	// Vector metadata
+	//===--------------------------------------------------------------------===//
+	page_id_t starting_page;
+	page_offset_t starting_offset;
+
+	page_id_t *page_ids;
+	page_offset_t *page_offsets;
+	uncompressed_size_t *uncompressed_sizes;
+	compressed_size_t *compressed_sizes;
+	//! The amount of vectors we've seen so far
+	idx_t vector_count = 0;
+	//! The amount of vectors we've seen in the current segment
+	idx_t vector_in_segment_count = 0;
+	//! The amount of vectors we're writing
+	idx_t total_vector_count = 0;
+	//! Whether we are currently in a Vector
+	bool in_vector = false;
+	//! The compression context indicating where we are in the output buffer
+	duckdb_zstd::ZSTD_outBuffer out_buffer;
+	idx_t uncompressed_size = 0;
+	idx_t compressed_size = 0;
+	string_length_t *string_lengths;
+
+	//! Amount of tuples we have seen for the current vector
+	idx_t tuple_count = 0;
+	//! The expected size of this vector (ZSTD_VECTOR_SIZE except for the last one)
+	idx_t vector_size;
+};
+
+unique_ptr<CompressionState> ZSTDStorage::InitCompression(ColumnDataCheckpointer &checkpointer,
+                                                          unique_ptr<AnalyzeState> analyze_state_p) {
+	return make_uniq<ZSTDCompressionState>(checkpointer,
+	                                       unique_ptr_cast<AnalyzeState, ZSTDAnalyzeState>(std::move(analyze_state_p)));
+}
+
+void ZSTDStorage::Compress(CompressionState &state_p, Vector &input, idx_t count) {
+	auto &state = state_p.Cast<ZSTDCompressionState>();
+
+	// Get vector data
+	UnifiedVectorFormat vdata;
+	input.ToUnifiedFormat(count, vdata);
+	auto data = UnifiedVectorFormat::GetData<string_t>(vdata);
+
+	for (idx_t i = 0; i < count; i++) {
+		auto idx = vdata.sel->get_index(i);
+		// Note: we treat nulls and empty strings the same
+		if (!vdata.validity.RowIsValid(idx) || data[idx].GetSize() == 0) {
+			state.AddNull();
+			continue;
+		}
+		state.AddString(data[idx]);
+	}
+}
+
+void ZSTDStorage::FinalizeCompress(CompressionState &state_p) {
+	auto &state = state_p.Cast<ZSTDCompressionState>();
+	state.Finalize();
+}
+
+struct ZSTDVectorScanMetadata {
+	//! The index of the (internal) vector being read
+	idx_t vector_idx;
+	block_id_t block_id;
+	page_offset_t block_offset;
+
+	uncompressed_size_t uncompressed_size;
+	compressed_size_t compressed_size;
+
+	//! The amount of tuples that are in this Vector
+	idx_t count;
+};
+
+struct ZSTDVectorScanState {
+public:
+	ZSTDVectorScanState() {
+	}
+	ZSTDVectorScanState(ZSTDVectorScanState &&other) = default;
+	ZSTDVectorScanState(const ZSTDVectorScanState &other) = delete;
+
+public:
+	//! The metadata of the vector
+	ZSTDVectorScanMetadata metadata;
+	//! The (pinned) buffer handle(s) for this vectors data
+	vector<BufferHandle> buffer_handles;
+	//! The current pointer at which we're reading the vectors data
+	data_ptr_t current_buffer_ptr;
+	//! The (uncompressed) string lengths for this vector
+	string_length_t *string_lengths;
+	//! The amount of values already consumed from the state
+	idx_t scanned_count = 0;
+	//! The amount of compressed data read
+	idx_t compressed_scan_count = 0;
+	//! The inBuffer that ZSTD_decompressStream reads the compressed data from
+	duckdb_zstd::ZSTD_inBuffer in_buffer;
+};
+
+//===--------------------------------------------------------------------===//
+// Scan
+//===--------------------------------------------------------------------===//
+struct ZSTDScanState : public SegmentScanState {
+public:
+	explicit ZSTDScanState(ColumnSegment &segment)
+	    : state(segment.GetSegmentState()->Cast<UncompressedStringSegmentState>()),
+	      block_manager(segment.GetBlockManager()), buffer_manager(BufferManager::GetBufferManager(segment.db)),
+	      segment_block_offset(segment.GetBlockOffset()) {
+		decompression_context = duckdb_zstd::ZSTD_createDCtx();
+		segment_handle = buffer_manager.Pin(segment.block);
+
+		auto data = segment_handle.Ptr() + segment.GetBlockOffset();
+		idx_t offset = 0;
+
+		segment_count = segment.count.load();
+		idx_t amount_of_vectors = (segment_count / ZSTD_VECTOR_SIZE) + ((segment_count % ZSTD_VECTOR_SIZE) != 0);
+
+		// Set pointers to the Vector Metadata
+		offset = AlignValue<idx_t, sizeof(page_id_t)>(offset);
+		page_ids = reinterpret_cast<page_id_t *>(data + offset);
+		offset += (sizeof(page_id_t) * amount_of_vectors);
+
+		offset = AlignValue<idx_t, sizeof(page_offset_t)>(offset);
+		page_offsets = reinterpret_cast<page_offset_t *>(data + offset);
+		offset += (sizeof(page_offset_t) * amount_of_vectors);
+
+		offset = AlignValue<idx_t, sizeof(uncompressed_size_t)>(offset);
+		uncompressed_sizes = reinterpret_cast<uncompressed_size_t *>(data + offset);
+		offset += (sizeof(uncompressed_size_t) * amount_of_vectors);
+
+		offset = AlignValue<idx_t, sizeof(compressed_size_t)>(offset);
+		compressed_sizes = reinterpret_cast<compressed_size_t *>(data + offset);
+		offset += (sizeof(compressed_size_t) * amount_of_vectors);
+
+		scanned_count = 0;
+	}
+	~ZSTDScanState() override {
+		duckdb_zstd::ZSTD_freeDCtx(decompression_context);
+	}
+
+public:
+	idx_t GetVectorIndex(idx_t start_index, idx_t &offset) {
+		idx_t vector_idx = start_index / ZSTD_VECTOR_SIZE;
+		offset = start_index % ZSTD_VECTOR_SIZE;
+		return vector_idx;
+	}
+
+	ZSTDVectorScanMetadata GetVectorMetadata(idx_t vector_idx) {
+		idx_t previous_value_count = vector_idx * ZSTD_VECTOR_SIZE;
+		idx_t value_count = MinValue<idx_t>(segment_count - previous_value_count, ZSTD_VECTOR_SIZE);
+
+		return ZSTDVectorScanMetadata {/* vector_idx = */ vector_idx,
+		                               /* block_id = */ page_ids[vector_idx],
+		                               /* block_offset = */ page_offsets[vector_idx],
+		                               /* uncompressed_size = */ uncompressed_sizes[vector_idx],
+		                               /* compressed_size = */ compressed_sizes[vector_idx],
+		                               /* count = */ value_count};
+	}
+
+	shared_ptr<BlockHandle> LoadPage(block_id_t block_id) {
+		return state.GetHandle(block_manager, block_id);
+	}
+
+	bool UseVectorStateCache(idx_t vector_idx, idx_t internal_offset) {
+		if (!current_vector) {
+			// No vector loaded yet
+			return false;
+		}
+		if (current_vector->metadata.vector_idx != vector_idx) {
+			// Not the same vector
+			return false;
+		}
+		if (current_vector->scanned_count != internal_offset) {
+			// Not the same scan offset
+			return false;
+		}
+		return true;
+	}
+
+	ZSTDVectorScanState &LoadVector(idx_t vector_idx, idx_t internal_offset) {
+		if (UseVectorStateCache(vector_idx, internal_offset)) {
+			return *current_vector;
+		}
+		current_vector = make_uniq<ZSTDVectorScanState>();
+		current_vector->metadata = GetVectorMetadata(vector_idx);
+		auto &metadata = current_vector->metadata;
+		auto &scan_state = *current_vector;
+		data_ptr_t handle_start;
+		idx_t ptr_offset = 0;
+		if (metadata.block_id == INVALID_BLOCK) {
+			// Data lives on the segment's page
+			handle_start = segment_handle.Ptr();
+			ptr_offset += segment_block_offset;
+		} else {
+			// Data lives on an extra page, have to load the block first
+			auto block = LoadPage(metadata.block_id);
+			auto data_handle = buffer_manager.Pin(block);
+			handle_start = data_handle.Ptr();
+			scan_state.buffer_handles.push_back(std::move(data_handle));
+		}
+
+		ptr_offset += metadata.block_offset;
+		ptr_offset = AlignValue<idx_t, sizeof(string_length_t)>(ptr_offset);
+		scan_state.current_buffer_ptr = handle_start + ptr_offset;
+
+		auto vector_size = metadata.count;
+
+		scan_state.string_lengths = reinterpret_cast<string_length_t *>(scan_state.current_buffer_ptr);
+		scan_state.current_buffer_ptr += (sizeof(string_length_t) * vector_size);
+
+		// Update the in_buffer to point to the start of the compressed data frame
+		idx_t current_offset = UnsafeNumericCast<idx_t>(scan_state.current_buffer_ptr - handle_start);
+		scan_state.in_buffer.src = scan_state.current_buffer_ptr;
+		scan_state.in_buffer.pos = 0;
+		scan_state.in_buffer.size = block_manager.GetBlockSize() - sizeof(block_id_t) - current_offset;
+
+		// Initialize the context for streaming decompression
+		duckdb_zstd::ZSTD_DCtx_reset(decompression_context, duckdb_zstd::ZSTD_reset_session_only);
+		duckdb_zstd::ZSTD_DCtx_refDDict(decompression_context, nullptr);
+
+		if (internal_offset) {
+			Skip(scan_state, internal_offset);
+		}
+		return scan_state;
+	}
+
+	void LoadNextPageForVector(ZSTDVectorScanState &scan_state) {
+		if (scan_state.in_buffer.pos != scan_state.in_buffer.size) {
+			throw InternalException(
+			    "(ZSTDScanState::LoadNextPageForVector) Trying to load the next page before consuming the current one");
+		}
+		// Read the next block id from the end of the page
+		auto base_ptr =
+		    reinterpret_cast<data_ptr_t>(const_cast<void *>(scan_state.in_buffer.src)); // NOLINT: const cast
+		auto next_id_ptr = base_ptr + scan_state.in_buffer.size;
+		block_id_t next_id = Load<block_id_t>(next_id_ptr);
+
+		// Load the next page
+		auto block = LoadPage(next_id);
+		auto handle = buffer_manager.Pin(block);
+		auto ptr = handle.Ptr();
+		scan_state.buffer_handles.push_back(std::move(handle));
+		scan_state.current_buffer_ptr = ptr;
+
+		// Update the in_buffer to point to the new page
+		scan_state.in_buffer.src = ptr;
+		scan_state.in_buffer.pos = 0;
+
+		idx_t page_size = block_manager.GetBlockSize() - sizeof(block_id_t);
+		idx_t remaining_compressed_data = scan_state.metadata.compressed_size - scan_state.compressed_scan_count;
+		scan_state.in_buffer.size = MinValue<idx_t>(page_size, remaining_compressed_data);
+	}
+
+	void DecompressString(ZSTDVectorScanState &scan_state, data_ptr_t destination, idx_t uncompressed_length) {
+		if (uncompressed_length == 0) {
+			return;
+		}
+
+		duckdb_zstd::ZSTD_outBuffer out_buffer;
+
+		out_buffer.dst = destination;
+		out_buffer.pos = 0;
+		out_buffer.size = uncompressed_length;
+
+		while (true) {
+			idx_t old_pos = scan_state.in_buffer.pos;
+			size_t res = duckdb_zstd::ZSTD_decompressStream(
+			    /* zds = */ decompression_context,
+			    /* output =*/&out_buffer,
+			    /* input =*/&scan_state.in_buffer);
+			scan_state.compressed_scan_count += scan_state.in_buffer.pos - old_pos;
+			if (duckdb_zstd::ZSTD_isError(res)) {
+				throw InvalidInputException("ZSTD Decompression failed: %s", duckdb_zstd::ZSTD_getErrorName(res));
+			}
+			if (out_buffer.pos == out_buffer.size) {
+				break;
+			}
+			// Did not fully decompress, it needs a new page to read from
+			LoadNextPageForVector(scan_state);
+		}
+	}
+
+	void Skip(ZSTDVectorScanState &scan_state, idx_t count) {
+		if (!skip_buffer) {
+			skip_buffer = Allocator::DefaultAllocator().Allocate(duckdb_zstd::ZSTD_DStreamOutSize());
+		}
+
+		D_ASSERT(scan_state.scanned_count + count <= scan_state.metadata.count);
+
+		// Figure out how much we need to skip
+		string_length_t *string_lengths = &scan_state.string_lengths[scan_state.scanned_count];
+		idx_t uncompressed_length = 0;
+		for (idx_t i = 0; i < count; i++) {
+			uncompressed_length += string_lengths[i];
+		}
+
+		// Skip that many bytes by decompressing into the skip_buffer
+		idx_t remaining = uncompressed_length;
+		while (remaining) {
+			idx_t to_scan = MinValue<idx_t>(skip_buffer.GetSize(), remaining);
+			DecompressString(scan_state, skip_buffer.get(), to_scan);
+			remaining -= to_scan;
+		}
+		scan_state.scanned_count += count;
+		scanned_count += count;
+	}
+
+	void ScanInternal(ZSTDVectorScanState &scan_state, idx_t count, Vector &result, idx_t result_offset) {
+		D_ASSERT(scan_state.scanned_count + count <= scan_state.metadata.count);
+		D_ASSERT(result.GetType().id() == LogicalTypeId::VARCHAR);
+
+		string_length_t *string_lengths = &scan_state.string_lengths[scan_state.scanned_count];
+		idx_t uncompressed_length = 0;
+		for (idx_t i = 0; i < count; i++) {
+			uncompressed_length += string_lengths[i];
+		}
+		auto empty_string = StringVector::EmptyString(result, uncompressed_length);
+		auto uncompressed_data = empty_string.GetDataWriteable();
+		auto string_data = FlatVector::GetData<string_t>(result);
+
+		DecompressString(scan_state, reinterpret_cast<data_ptr_t>(uncompressed_data), uncompressed_length);
+
+		idx_t offset = 0;
+		auto uncompressed_data_const = empty_string.GetData();
+		for (idx_t i = 0; i < count; i++) {
+			string_data[result_offset + i] = string_t(uncompressed_data_const + offset, string_lengths[i]);
+			offset += string_lengths[i];
+		}
+		scan_state.scanned_count += count;
+		scanned_count += count;
+	}
+
+	void ScanPartial(idx_t start_idx, Vector &result, idx_t offset, idx_t count) {
+		idx_t remaining = count;
+		idx_t scanned = 0;
+		while (remaining) {
+			idx_t internal_offset;
+			idx_t vector_idx = GetVectorIndex(start_idx + scanned, internal_offset);
+			auto &scan_state = LoadVector(vector_idx, internal_offset);
+			idx_t remaining_in_vector = scan_state.metadata.count - scan_state.scanned_count;
+			idx_t to_scan = MinValue<idx_t>(remaining, remaining_in_vector);
+			ScanInternal(scan_state, to_scan, result, offset + scanned);
+			remaining -= to_scan;
+			scanned += to_scan;
+		}
+		D_ASSERT(scanned == count);
+	}
+
+public:
+	UncompressedStringSegmentState &state;
+	BlockManager &block_manager;
+	BufferManager &buffer_manager;
+
+	duckdb_zstd::ZSTD_DCtx *decompression_context = nullptr;
+
+	idx_t segment_block_offset;
+	BufferHandle segment_handle;
+
+	//===--------------------------------------------------------------------===//
+	// Vector metadata
+	//===--------------------------------------------------------------------===//
+	page_id_t *page_ids;
+	page_offset_t *page_offsets;
+	uncompressed_size_t *uncompressed_sizes;
+	compressed_size_t *compressed_sizes;
+
+	//! Cache of (the scan state of) the current vector being read
+	unique_ptr<ZSTDVectorScanState> current_vector;
+
+	//! The amount of tuples stored in the segment
+	idx_t segment_count;
+	//! The amount of tuples consumed
+	idx_t scanned_count = 0;
+
+	//! Buffer for skipping data
+	AllocatedData skip_buffer;
+};
+
+unique_ptr<SegmentScanState> ZSTDStorage::StringInitScan(ColumnSegment &segment) {
+	auto result = make_uniq<ZSTDScanState>(segment);
+	return std::move(result);
+}
+
+//===--------------------------------------------------------------------===//
+// Scan base data
+//===--------------------------------------------------------------------===//
+void ZSTDStorage::StringScanPartial(ColumnSegment &segment, ColumnScanState &state, idx_t scan_count, Vector &result,
+                                    idx_t result_offset) {
+	auto &scan_state = state.scan_state->template Cast<ZSTDScanState>();
+	auto start = segment.GetRelativeIndex(state.row_index);
+
+	scan_state.ScanPartial(start, result, result_offset, scan_count);
+}
+
+void ZSTDStorage::StringScan(ColumnSegment &segment, ColumnScanState &state, idx_t scan_count, Vector &result) {
+	StringScanPartial(segment, state, scan_count, result, 0);
+}
+
+//===--------------------------------------------------------------------===//
+// Fetch
+//===--------------------------------------------------------------------===//
+void ZSTDStorage::StringFetchRow(ColumnSegment &segment, ColumnFetchState &state, row_t row_id, Vector &result,
+                                 idx_t result_idx) {
+	ZSTDScanState scan_state(segment);
+	scan_state.ScanPartial(UnsafeNumericCast<idx_t>(row_id), result, result_idx, 1);
+}
+
+//===--------------------------------------------------------------------===//
+// Serialization & Cleanup
+//===--------------------------------------------------------------------===//
+
+unique_ptr<CompressedSegmentState> ZSTDStorage::StringInitSegment(ColumnSegment &segment, block_id_t block_id,
+                                                                  optional_ptr<ColumnSegmentState> segment_state) {
+	auto result = make_uniq<UncompressedStringSegmentState>();
+	if (segment_state) {
+		auto &serialized_state = segment_state->Cast<SerializedStringSegmentState>();
+		result->on_disk_blocks = std::move(serialized_state.blocks);
+	}
+	return std::move(result);
+}
+
+unique_ptr<ColumnSegmentState> ZSTDStorage::SerializeState(ColumnSegment &segment) {
+	auto &state = segment.GetSegmentState()->Cast<UncompressedStringSegmentState>();
+	if (state.on_disk_blocks.empty()) {
+		// no on-disk blocks - nothing to write
+		return nullptr;
+	}
+	return make_uniq<SerializedStringSegmentState>(state.on_disk_blocks);
+}
+
+unique_ptr<ColumnSegmentState> ZSTDStorage::DeserializeState(Deserializer &deserializer) {
+	auto result = make_uniq<SerializedStringSegmentState>();
+	deserializer.ReadProperty(1, "overflow_blocks", result->blocks);
+	return std::move(result);
+}
+
+void ZSTDStorage::CleanupState(ColumnSegment &segment) {
+	auto &state = segment.GetSegmentState()->Cast<UncompressedStringSegmentState>();
+	auto &block_manager = segment.GetBlockManager();
+	for (auto &block_id : state.on_disk_blocks) {
+		block_manager.MarkBlockAsModified(block_id);
+	}
+}
+
+//===--------------------------------------------------------------------===//
+// Get Function
+//===--------------------------------------------------------------------===//
+CompressionFunction ZSTDFun::GetFunction(PhysicalType data_type) {
+	D_ASSERT(data_type == PhysicalType::VARCHAR);
+	auto zstd = CompressionFunction(
+	    CompressionType::COMPRESSION_ZSTD, data_type, ZSTDStorage::StringInitAnalyze, ZSTDStorage::StringAnalyze,
+	    ZSTDStorage::StringFinalAnalyze, ZSTDStorage::InitCompression, ZSTDStorage::Compress,
+	    ZSTDStorage::FinalizeCompress, ZSTDStorage::StringInitScan, ZSTDStorage::StringScan,
+	    ZSTDStorage::StringScanPartial, ZSTDStorage::StringFetchRow, ZSTDStorage::StringSkip);
+	zstd.init_segment = ZSTDStorage::StringInitSegment;
+	zstd.serialize_state = ZSTDStorage::SerializeState;
+	zstd.deserialize_state = ZSTDStorage::DeserializeState;
+	zstd.cleanup_state = ZSTDStorage::CleanupState;
+	return zstd;
+}
+
+bool ZSTDFun::TypeIsSupported(PhysicalType type) {
+	return type == PhysicalType::VARCHAR;
+}
+
+} // namespace duckdb
diff --git a/src/storage/table/column_data.cpp b/src/storage/table/column_data.cpp
index 7e778b8687fa..64899a26a432 100644
--- a/src/storage/table/column_data.cpp
+++ b/src/storage/table/column_data.cpp
@@ -833,6 +833,7 @@ void ColumnData::GetColumnSegmentInfo(idx_t row_group_index, vector<idx_t> col_p
 		auto segment_state = segment->GetSegmentState();
 		if (segment_state) {
 			column_info.segment_info = segment_state->GetSegmentInfo();
+			column_info.additional_blocks = segment_state->GetAdditionalBlocks();
 		}
 		result.emplace_back(column_info);
 


Test patch:
diff --git a/test/sql/storage/compression/alp/alp_min_max.test b/test/sql/storage/compression/alp/alp_min_max.test
index d25d697c6156..2b30c17f7a73 100644
--- a/test/sql/storage/compression/alp/alp_min_max.test
+++ b/test/sql/storage/compression/alp/alp_min_max.test
@@ -24,8 +24,8 @@ INSERT INTO all_types SELECT ${type} FROM all_types;
 statement ok
 checkpoint
 
-query IIIIIIIIIIIIIII
-SELECT * FROM pragma_storage_info('all_types') WHERE segment_type == '${type}' AND compression != 'ALP';
+query I
+SELECT compression FROM pragma_storage_info('all_types') WHERE segment_type == '${type}' AND compression != 'ALP';
 ----
 
 # i
diff --git a/test/sql/storage/compression/alprd/alprd_min_max.test b/test/sql/storage/compression/alprd/alprd_min_max.test
index 6e031b320452..4d9c94ae0403 100644
--- a/test/sql/storage/compression/alprd/alprd_min_max.test
+++ b/test/sql/storage/compression/alprd/alprd_min_max.test
@@ -24,8 +24,8 @@ INSERT INTO all_types SELECT ${type} FROM all_types;
 statement ok
 checkpoint
 
-query IIIIIIIIIIIIIII
-SELECT * FROM pragma_storage_info('all_types') WHERE segment_type == '${type}' AND compression != 'ALPRD';
+query I
+SELECT compression FROM pragma_storage_info('all_types') WHERE segment_type == '${type}' AND compression != 'ALPRD';
 ----
 
 # i
diff --git a/test/sql/storage/compression/zstd/fetch_row.test b/test/sql/storage/compression/zstd/fetch_row.test
new file mode 100644
index 000000000000..a87528a63486
--- /dev/null
+++ b/test/sql/storage/compression/zstd/fetch_row.test
@@ -0,0 +1,66 @@
+# name: test/sql/storage/compression/zstd/fetch_row.test
+# group: [zstd]
+
+# load the DB from disk
+load __TEST_DIR__/test_zstd_compression_fetch_row.db
+
+statement ok
+CREATE TABLE big_string (
+	a VARCHAR,
+	id INT
+);
+
+statement ok
+pragma force_compression='zstd'
+
+statement ok
+INSERT INTO big_string values (repeat('a', 8000), 1);
+INSERT INTO big_string values (repeat('b', 10), 2);
+INSERT INTO big_string values (repeat('c', 8000), 3);
+INSERT INTO big_string values (repeat('d', 10), 4);
+INSERT INTO big_string values (repeat('a', 8000), 1);
+INSERT INTO big_string values (repeat('b', 10), 2);
+INSERT INTO big_string values (repeat('c', 8000), 3);
+INSERT INTO big_string values (repeat('d', 10), 4);
+INSERT INTO big_string values (repeat('a', 8000), 1);
+INSERT INTO big_string values (repeat('b', 10), 2);
+INSERT INTO big_string values (repeat('c', 8000), 3);
+INSERT INTO big_string values (repeat('d', 10), 4);
+
+# Test with a string that is larger than a block size after compression
+# uncompressed size: 3888890 compressed size: 1068813
+statement ok
+INSERT INTO big_string values (concat(range(0,500000)::VARCHAR), 5);
+
+statement ok
+INSERT INTO big_string values (repeat('f', 1), 6);
+INSERT INTO big_string values (repeat('g', 8000), 7);
+INSERT INTO big_string values (repeat('h', 10000), 8);
+
+statement ok
+checkpoint
+
+restart
+
+statement ok
+pragma verify_fetch_row;
+
+query II
+SELECT a[1], strlen(a) from big_string;
+----
+a	8000
+b	10
+c	8000
+d	10
+a	8000
+b	10
+c	8000
+d	10
+a	8000
+b	10
+c	8000
+d	10
+[	3888890
+f	1
+g	8000
+h	10000
diff --git a/test/sql/storage/compression/zstd/nulls.test b/test/sql/storage/compression/zstd/nulls.test
new file mode 100644
index 000000000000..665986a7005e
--- /dev/null
+++ b/test/sql/storage/compression/zstd/nulls.test
@@ -0,0 +1,35 @@
+# name: test/sql/storage/compression/zstd/nulls.test
+# group: [zstd]
+
+load __TEST_DIR__/zstd_nulls.db
+
+require block_size 262144
+
+statement ok
+create table tbl (
+	a varchar
+);
+
+statement ok
+set variable my_string = (
+	select concat(range(0,1000)::VARCHAR)
+);
+
+statement ok
+INSERT INTO tbl (a)
+SELECT CASE
+	WHEN (i % 7) = 0 THEN NULL
+	ELSE getvariable('my_string') || i
+END
+FROM range(5000) t(i);
+
+statement ok
+pragma force_compression='zstd'
+
+statement ok
+checkpoint
+
+query I
+select count(*) from tbl where a IS NULL;
+----
+715
diff --git a/test/sql/storage/compression/zstd/page_flushing_test1.test b/test/sql/storage/compression/zstd/page_flushing_test1.test
new file mode 100644
index 000000000000..0d42bf47982d
--- /dev/null
+++ b/test/sql/storage/compression/zstd/page_flushing_test1.test
@@ -0,0 +1,91 @@
+# name: test/sql/storage/compression/zstd/page_flushing_test1.test
+# group: [zstd]
+
+load __TEST_DIR__/zstd_page_flushing_test1.db
+
+require block_size 262144
+
+statement ok
+create table tbl (
+	a VARCHAR
+);
+
+statement ok
+set variable my_string = (
+	select concat(range(0,500000)::VARCHAR)
+);
+
+statement ok
+SET checkpoint_threshold = '10.0 GB';
+
+# Feed it enough data to actually use zstd (enough to train a dictionary)
+statement ok
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+insert into tbl values ('aaaabbbb');
+
+# ------- VECTOR 1 -------
+
+# This will start on the segment page and write the remainder on 'extra_pages[0]' buffer
+statement ok
+insert into tbl select getvariable('my_string');
+
+# ------- VECTOR 2 -------
+
+# This starts on 'extra_pages[0]' and writes onto 'extra_pages[1]'
+statement ok
+insert into tbl select getvariable('my_string');
+
+# Insert 2036 (STANDARD_VECTOR_SIZE - 12) to finish the vector of values
+statement ok
+insert into tbl select 'bbbbaaaa' from range(2036)
+
+# ------- VECTOR 3 -------
+
+# This starts on 'extra_pages[1]' and writes onto 'extra_pages[0]'
+statement ok
+insert into tbl select getvariable('my_string');
+
+statement ok
+insert into tbl select 'bbbbaaaa' from range(2047)
+
+# ------- VECTOR 4 -------
+
+# This starts on 'extra_pages[0]', overflows to 'extra_pages[1]' twice
+statement ok
+insert into tbl select concat(getvariable('my_string'), getvariable('my_string'));
+
+statement ok
+insert into tbl select 'bbbbaaaa' from range(2047)
+
+# ------- VECTOR 5 -------
+
+# This starts on 'extra_pages[1]', overflows to 'extra_pages[0]' twice
+statement ok
+insert into tbl select concat(getvariable('my_string'), getvariable('my_string'));
+
+statement ok
+insert into tbl select 'bbbbaaaa' from range(2047)
+
+# ------------------------
+
+statement ok
+pragma force_compression='zstd'
+
+statement ok
+checkpoint
+
+query II
+select strlen(a), count(a) from tbl group by strlen(a) order by all;
+----
+8	8188
+3888890	3
+7777780	2
diff --git a/test/sql/storage/compression/zstd/reclaim_space_column.test_slow b/test/sql/storage/compression/zstd/reclaim_space_column.test_slow
new file mode 100644
index 000000000000..68598ec7054d
--- /dev/null
+++ b/test/sql/storage/compression/zstd/reclaim_space_column.test_slow
@@ -0,0 +1,93 @@
+# name: test/sql/storage/compression/zstd/reclaim_space_column.test_slow
+# description: Test that we reclaim space when dropping columns containing overflow strings
+# group: [zstd]
+
+load __TEST_DIR__/reclaim_space_drop_column_overflow_strings.db
+
+statement ok
+PRAGMA force_compression='ZSTD';
+
+statement ok
+PRAGMA force_checkpoint;
+
+statement ok
+CREATE TABLE strings AS SELECT i, repeat('X', case when i%17=0 then 5000 else i%7 end) AS s FROM generate_series(0,150000) tbl(i);
+
+statement ok
+CHECKPOINT;
+
+statement ok
+CHECKPOINT;
+
+query IIIIII
+SELECT AVG(STRLEN(s)), MIN(STRLEN(S)), MAX(STRLEN(S)), SUM(STRLEN(S)), MIN(S[1]), MAX(S[1]) FROM strings
+----
+296.955	0	5000	44543527	(empty)	X
+
+# For smaller block sizes (16KB) the total blocks alternate between a few values in the loop,
+# therefore, we need to compare to a range of total block counts.
+statement ok
+CREATE TABLE total_blocks_tbl AS SELECT total_blocks FROM pragma_database_size();
+
+statement ok
+create type test_result as UNION(
+	ok BOOL,
+	err STRUCT(
+		old BIGINT,
+		allowed_max DECIMAL(21,1),
+		actual BIGINT
+	)
+);
+
+loop i 0 30
+
+statement ok
+ALTER TABLE strings DROP COLUMN s;
+
+statement ok
+ALTER TABLE strings ADD COLUMN s VARCHAR;
+
+statement ok
+UPDATE strings SET s=repeat('X', case when i%17=0 then 5000 else i%7 end);
+
+query IIIIII
+SELECT AVG(STRLEN(s)), MIN(STRLEN(S)), MAX(STRLEN(S)), SUM(STRLEN(S)), MIN(S[1]), MAX(S[1]) FROM strings
+----
+296.955	0	5000	44543527	(empty)	X
+
+statement ok
+CHECKPOINT;
+
+# Ensure that the total blocks don't exceed the total blocks by more than 1.2.
+
+query I
+SELECT
+	CASE WHEN ${i} < 10
+		THEN True::test_result
+	WHEN current.total_blocks <= total_blocks_tbl.total_blocks * 1.2
+		THEN True::test_result
+	ELSE {
+			'old': total_blocks_tbl.total_blocks,
+			'allowed_max': total_blocks_tbl.total_blocks * 1.2,
+			'actual': current.total_blocks
+		}::test_result
+	END
+FROM pragma_database_size() AS current, total_blocks_tbl;
+----
+true
+
+# Adjust total_blocks_tbl to the count after 10 warm-up iterations.
+
+statement ok
+UPDATE total_blocks_tbl SET total_blocks = (
+	SELECT CASE WHEN ${i} < 10 THEN (SELECT current.total_blocks FROM pragma_database_size() AS current)
+		ELSE (total_blocks) END);
+
+restart
+
+query IIIIII
+SELECT AVG(STRLEN(s)), MIN(STRLEN(S)), MAX(STRLEN(S)), SUM(STRLEN(S)), MIN(S[1]), MAX(S[1]) FROM strings
+----
+296.955	0	5000	44543527	(empty)	X
+
+endloop
diff --git a/test/sql/storage/compression/zstd/reclaim_space_table.test_slow b/test/sql/storage/compression/zstd/reclaim_space_table.test_slow
new file mode 100644
index 000000000000..8a6fb3fe2b38
--- /dev/null
+++ b/test/sql/storage/compression/zstd/reclaim_space_table.test_slow
@@ -0,0 +1,130 @@
+# name: test/sql/storage/compression/zstd/reclaim_space_table.test_slow
+# description: Test that we reclaim space when dropping tables containing overflow strings
+# group: [zstd]
+
+load __TEST_DIR__/reclaim_space_overflow_strings.db
+
+statement ok
+PRAGMA force_compression='ZSTD';
+
+statement ok
+PRAGMA force_checkpoint;
+
+# Every 17th value has length 5000
+# Other values are between 0-6 in length
+statement ok
+CREATE TABLE strings AS SELECT
+	repeat('X', case when i%17=0 then 5000 else i%7 end) AS s
+FROM generate_series(0,150000) tbl(i);
+
+statement ok
+CHECKPOINT;
+
+statement ok
+CHECKPOINT;
+
+query IIIIII
+SELECT
+	AVG(STRLEN(s)),
+	MIN(STRLEN(S)),
+	MAX(STRLEN(S)),
+	SUM(STRLEN(S)),
+	MIN(S[1]),
+	MAX(S[1])
+FROM strings
+----
+296.955	0	5000	44543527	(empty)	X
+
+# for smaller block sizes (16KB) the total blocks alternate between a few values in the loop,
+# therefore, we need to compare to a range of total block counts
+statement ok
+CREATE TABLE total_blocks_tbl AS SELECT
+	total_blocks
+FROM pragma_database_size();
+
+statement ok
+create type test_result as UNION(
+	ok BOOL,
+	err STRUCT(
+		old BIGINT,
+		allowed_max DECIMAL(21,1),
+		actual BIGINT
+	)
+);
+
+loop i 0 10
+
+statement ok
+DROP TABLE strings;
+
+# Recreate the table
+statement ok
+CREATE TABLE strings AS SELECT
+	repeat('X', case when i%17=0 then 5000 else i%7 end) AS s
+FROM generate_series(0,150000) tbl(i);
+
+query IIIIII
+SELECT
+	AVG(STRLEN(s)),
+	MIN(STRLEN(S)),
+	MAX(STRLEN(S)),
+	SUM(STRLEN(S)),
+	MIN(S[1]),
+	MAX(S[1])
+FROM strings
+----
+296.955	0	5000	44543527	(empty)	X
+
+statement ok
+CHECKPOINT;
+
+# ensure that the total blocks don't exceed the total blocks after the first iteration
+# by more than 1.2
+
+query I
+SELECT
+	CASE WHEN ${i} = 0
+		THEN True::test_result
+	WHEN current.total_blocks <= total_blocks_tbl.total_blocks * 1.2
+		THEN True::test_result
+	ELSE {
+			'old': total_blocks_tbl.total_blocks,
+			'allowed_max': total_blocks_tbl.total_blocks * 1.2,
+			'actual': current.total_blocks
+		}::test_result
+	END
+FROM pragma_database_size() AS current, total_blocks_tbl;
+----
+true
+
+# adjust total_blocks_tbl once to the count after the first iteration
+
+statement ok
+UPDATE total_blocks_tbl SET total_blocks = (
+	SELECT
+		CASE WHEN ${i} = 0
+			THEN (SELECT current.total_blocks FROM pragma_database_size() AS current)
+		ELSE
+			(total_blocks)
+		END
+	);
+
+statement ok
+CHECKPOINT;
+
+restart
+
+query IIIIII
+SELECT
+	AVG(STRLEN(s)),
+	MIN(STRLEN(S)),
+	MAX(STRLEN(S)),
+	SUM(STRLEN(S)),
+	MIN(S[1]),
+	MAX(S[1])
+FROM strings
+----
+296.955	0	5000	44543527	(empty)	X
+
+# i
+endloop
diff --git a/test/sql/storage/compression/zstd/test_giant_lists.test_slow b/test/sql/storage/compression/zstd/test_giant_lists.test_slow
new file mode 100644
index 000000000000..e55bdd068e83
--- /dev/null
+++ b/test/sql/storage/compression/zstd/test_giant_lists.test_slow
@@ -0,0 +1,39 @@
+# name: test/sql/storage/compression/zstd/test_giant_lists.test_slow
+# group: [zstd]
+
+# With the vector metadata size taking between 24-32 bytes (because of padding), this equates to around 584 vectors per segment
+# which is 1_196_032 values
+statement ok
+SET default_block_size = '16384';
+
+statement ok
+attach '__TEST_DIR__/giant_list.db' as db2
+
+statement ok
+USE db2;
+
+statement ok
+pragma force_compression='zstd';
+
+statement ok
+pragma checkpoint_threshold='1gb';
+
+statement ok
+set zstd_min_string_length=1;
+
+# With 1.2m values per segment, this roughly creates 25 segments
+statement ok
+create table tbl as
+select repeat('a', 30_000_000).split('') lst;
+
+statement ok
+force checkpoint;
+
+query III
+select len(lst), lst[1], lst[-1] from tbl;
+----
+30_000_000	a	a
+
+query I
+SELECT compression FROM pragma_storage_info('tbl') WHERE segment_type = 'VARCHAR' AND compression != 'ZSTD';
+----
diff --git a/test/sql/storage/compression/zstd/test_skipping.test b/test/sql/storage/compression/zstd/test_skipping.test
new file mode 100644
index 000000000000..adfc76a4c952
--- /dev/null
+++ b/test/sql/storage/compression/zstd/test_skipping.test
@@ -0,0 +1,34 @@
+# name: test/sql/storage/compression/zstd/test_skipping.test
+# group: [zstd]
+
+load __TEST_DIR__/zstd_vector_skipping.db
+
+statement ok
+pragma force_compression='zstd';
+
+statement ok
+select setseed(0.42);
+
+statement ok
+create table tbl as
+select
+	i // 5_000 as num,
+	num::VARCHAR || list_reduce([uuid()::varchar for x in range(10)], (x, y) -> concat(x, y)) str
+from range(20_000) t(i) order by num
+
+statement ok
+force checkpoint;
+
+query I
+select str[0:1]::BIGINT from tbl where num = 1 limit 10;
+----
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
diff --git a/test/sql/storage/compression/zstd/zstd.test b/test/sql/storage/compression/zstd/zstd.test
new file mode 100644
index 000000000000..cf1db384ee0c
--- /dev/null
+++ b/test/sql/storage/compression/zstd/zstd.test
@@ -0,0 +1,27 @@
+# name: test/sql/storage/compression/zstd/zstd.test
+# description: Test zstd storage
+# group: [zstd]
+
+# load the DB from disk
+load __TEST_DIR__/test_zstd_storage.db
+
+statement ok
+PRAGMA force_compression = 'zstd'
+
+statement ok
+CREATE TABLE test (a VARCHAR);
+
+statement ok
+INSERT INTO test VALUES ('11'), ('11'), ('12'), (NULL);
+
+statement ok
+checkpoint;
+
+query I
+SELECT * FROM test;
+----
+11
+11
+12
+NULL
+
diff --git a/test/sql/storage/compression/zstd/zstd_compression_ratio.test_slow b/test/sql/storage/compression/zstd/zstd_compression_ratio.test_slow
new file mode 100644
index 000000000000..6381b84ad678
--- /dev/null
+++ b/test/sql/storage/compression/zstd/zstd_compression_ratio.test_slow
@@ -0,0 +1,112 @@
+# name: test/sql/storage/compression/zstd/zstd_compression_ratio.test_slow
+# description: Assert bitpacking compression ratio is within reasonable margins for each mode
+# group: [zstd]
+
+require block_size 262144
+
+load __TEST_DIR__/test_zstd_compression_ratio.db
+
+statement ok
+PRAGMA force_compression='zstd'
+
+statement ok
+select setseed(0.42);
+
+statement ok
+set variable dataset_size = 100_000;
+
+# strings of 4096 characters, created from uuid(), making them semi-random
+statement ok
+create table test_compressed as (
+	with cte as (
+		select list_reduce([uuid()::varchar for x in range(128)], (x, y) -> concat(x, y)) str
+		from range(getvariable('dataset_size'))
+	)
+	select
+		str
+	from cte
+);
+
+statement ok
+checkpoint
+
+statement ok
+PRAGMA force_compression='uncompressed'
+
+statement ok
+select setseed(0.42);
+
+statement ok
+CREATE TABLE test_uncompressed as (
+	with cte as (
+		select list_reduce([uuid()::varchar for x in range(128)], (x, y) -> concat(x, y)) str
+		from range(getvariable('dataset_size'))
+	)
+	select
+		str
+	from cte
+);
+
+statement ok
+checkpoint
+
+query I
+SELECT compression FROM pragma_storage_info('test_compressed') WHERE segment_type != 'VALIDITY' AND compression != 'ZSTD';
+----
+
+query I
+SELECT compression FROM pragma_storage_info('test_uncompressed') WHERE segment_type != 'VALIDITY' AND compression != 'Uncompressed';
+----
+
+statement ok
+CREATE TYPE test_result AS UNION (
+    ok BOOL,
+    err STRUCT(
+        uncompressed HUGEINT,
+        compressed HUGEINT,
+        allowed_minimum_ratio DECIMAL(2,1),
+        allowed_maximum_ratio DECIMAL(2,1),
+        actual_ratio FLOAT
+    )
+);
+
+statement ok
+set variable min_ratio = 1.8;
+set variable max_ratio = 2.5;
+
+query I
+WITH compressed_intermediate AS (
+	SELECT
+		1 + sum(len(additional_block_ids)) as blocks
+	FROM
+		pragma_storage_info('test_compressed')
+	WHERE segment_type NOT IN ('VALIDITY')
+	GROUP BY block_id
+),
+uncompressed_intermediate AS (
+	SELECT
+		1 + sum(len(additional_block_ids)) as blocks
+	FROM
+		pragma_storage_info('test_uncompressed')
+	WHERE segment_type NOT IN ('VALIDITY')
+	GROUP BY block_id
+)
+SELECT
+    CASE 
+        WHEN (uncompressed::FLOAT / compressed::FLOAT) > getvariable('min_ratio') AND (uncompressed::FLOAT / compressed::FLOAT) < getvariable('max_ratio')
+            THEN True::test_result
+        ELSE {
+            'uncompressed': uncompressed,
+            'compressed': compressed,
+            'allowed_minimum_ratio': getvariable('min_ratio'),
+            'allowed_maximum_ratio': getvariable('max_ratio'),
+            'actual_ratio': uncompressed::FLOAT / compressed::FLOAT
+        }::test_result
+    END
+FROM (
+    SELECT
+        (SELECT sum(blocks) FROM compressed_intermediate) as compressed,
+        (SELECT sum(blocks) FROM uncompressed_intermediate) AS uncompressed
+) AS blocks_tbl;
+----
+true
diff --git a/test/sql/storage/overflow_strings/load_overflow_strings_slowly.test b/test/sql/storage/overflow_strings/load_overflow_strings_slowly.test
index 9eec86b94d73..61f76591ebf3 100644
--- a/test/sql/storage/overflow_strings/load_overflow_strings_slowly.test
+++ b/test/sql/storage/overflow_strings/load_overflow_strings_slowly.test
@@ -9,21 +9,38 @@ loop x 0 10
 statement ok
 CREATE TABLE strings(s VARCHAR);
 
+statement ok
+pragma force_compression='uncompressed'
+
 # load large strings into the table iteratively
 loop it 0 10
 
 statement ok
-INSERT INTO strings SELECT repeat('X', CASE WHEN i % 17 = 0 THEN 5000 ELSE i % 7 END) AS s FROM generate_series(0, 2500) tbl(i);
+INSERT INTO strings
+SELECT
+	repeat(
+		'X',
+		CASE WHEN i % 17 = 0
+		THEN 5000
+		ELSE i % 7
+		END
+	) AS s
+FROM generate_series(0, 2500) tbl(i);
 
 statement ok
 CHECKPOINT;
 
 endloop
 
+# And verify that no other compression is used
+query I
+SELECT compression FROM pragma_storage_info('strings') WHERE segment_type == 'VARCHAR' AND compression != 'Uncompressed';
+----
+
 query I
 SELECT EXISTS (SELECT * FROM pragma_storage_info('strings') WHERE contains(segment_info, 'Overflow String'));
 ----
-1
+true
 
 # ensure that the expected total storage size is the same as in the first iteration of the loop
 
diff --git a/test/sql/storage/reclaim_space/reclaim_space_drop_column_overflow_strings.test_slow b/test/sql/storage/reclaim_space/reclaim_space_drop_column_overflow_strings.test_slow
index ad43eabbe7ab..2b6ae0d0fcac 100644
--- a/test/sql/storage/reclaim_space/reclaim_space_drop_column_overflow_strings.test_slow
+++ b/test/sql/storage/reclaim_space/reclaim_space_drop_column_overflow_strings.test_slow
@@ -4,6 +4,9 @@
 
 load __TEST_DIR__/reclaim_space_drop_column_overflow_strings.db
 
+statement ok
+PRAGMA force_compression='Uncompressed';
+
 statement ok
 PRAGMA force_checkpoint;
 
@@ -26,6 +29,16 @@ SELECT AVG(STRLEN(s)), MIN(STRLEN(S)), MAX(STRLEN(S)), SUM(STRLEN(S)), MIN(S[1])
 statement ok
 CREATE TABLE total_blocks_tbl AS SELECT total_blocks FROM pragma_database_size();
 
+statement ok
+create type test_result as UNION(
+	ok BOOL,
+	err STRUCT(
+		old BIGINT,
+		allowed_max DECIMAL(21,1),
+		actual BIGINT
+	)
+);
+
 loop i 0 30
 
 statement ok
@@ -48,12 +61,20 @@ CHECKPOINT;
 # Ensure that the total blocks don't exceed the total blocks by more than 1.2.
 
 query I
-SELECT CASE WHEN ${i} < 10 THEN True
-	WHEN current.total_blocks <= total_blocks_tbl.total_blocks * 1.2 THEN True
-	ELSE False END
+SELECT
+	CASE WHEN ${i} < 10
+		THEN True::test_result
+	WHEN current.total_blocks <= total_blocks_tbl.total_blocks * 1.2
+		THEN True::test_result
+	ELSE {
+			'old': total_blocks_tbl.total_blocks,
+			'allowed_max': total_blocks_tbl.total_blocks * 1.2,
+			'actual': current.total_blocks
+		}::test_result
+	END
 FROM pragma_database_size() AS current, total_blocks_tbl;
 ----
-1
+true
 
 # Adjust total_blocks_tbl to the count after 10 warm-up iterations.
 
diff --git a/test/sql/storage/reclaim_space/reclaim_space_drop_overflow_strings.test_slow b/test/sql/storage/reclaim_space/reclaim_space_drop_overflow_strings.test_slow
index 44b27c4dba54..7772e2a1ccd1 100644
--- a/test/sql/storage/reclaim_space/reclaim_space_drop_overflow_strings.test_slow
+++ b/test/sql/storage/reclaim_space/reclaim_space_drop_overflow_strings.test_slow
@@ -4,11 +4,18 @@
 
 load __TEST_DIR__/reclaim_space_overflow_strings.db
 
+statement ok
+PRAGMA force_compression='Uncompressed';
+
 statement ok
 PRAGMA force_checkpoint;
 
+# Every 17th value has length 5000
+# Other values are between 0-6 in length
 statement ok
-CREATE TABLE strings AS SELECT repeat('X', case when i%17=0 then 5000 else i%7 end) AS s FROM generate_series(0,150000) tbl(i);
+CREATE TABLE strings AS SELECT
+	repeat('X', case when i%17=0 then 5000 else i%7 end) AS s
+FROM generate_series(0,150000) tbl(i);
 
 statement ok
 CHECKPOINT;
@@ -17,25 +24,54 @@ statement ok
 CHECKPOINT;
 
 query IIIIII
-SELECT AVG(STRLEN(s)), MIN(STRLEN(S)), MAX(STRLEN(S)), SUM(STRLEN(S)), MIN(S[1]), MAX(S[1]) FROM strings
+SELECT
+	AVG(STRLEN(s)),
+	MIN(STRLEN(S)),
+	MAX(STRLEN(S)),
+	SUM(STRLEN(S)),
+	MIN(S[1]),
+	MAX(S[1])
+FROM strings
 ----
 296.955	0	5000	44543527	(empty)	X
 
 # for smaller block sizes (16KB) the total blocks alternate between a few values in the loop,
 # therefore, we need to compare to a range of total block counts
 statement ok
-CREATE TABLE total_blocks_tbl AS SELECT total_blocks FROM pragma_database_size();
+CREATE TABLE total_blocks_tbl AS SELECT
+	total_blocks
+FROM pragma_database_size();
+
+statement ok
+create type test_result as UNION(
+	ok BOOL,
+	err STRUCT(
+		old BIGINT,
+		allowed_max DECIMAL(21,1),
+		actual BIGINT
+	)
+);
 
 loop i 0 10
 
 statement ok
 DROP TABLE strings;
 
+# Recreate the table
 statement ok
-CREATE TABLE strings AS SELECT repeat('X', case when i%17=0 then 5000 else i%7 end) AS s FROM generate_series(0,150000) tbl(i);
+CREATE TABLE strings AS SELECT
+	repeat('X', case when i%17=0 then 5000 else i%7 end) AS s
+FROM generate_series(0,150000) tbl(i);
 
 query IIIIII
-SELECT AVG(STRLEN(s)), MIN(STRLEN(S)), MAX(STRLEN(S)), SUM(STRLEN(S)), MIN(S[1]), MAX(S[1]) FROM strings
+SELECT
+	AVG(STRLEN(s)),
+	MIN(STRLEN(S)),
+	MAX(STRLEN(S)),
+	SUM(STRLEN(S)),
+	MIN(S[1]),
+	MAX(S[1])
+FROM strings
 ----
 296.955	0	5000	44543527	(empty)	X
 
@@ -46,19 +82,32 @@ CHECKPOINT;
 # by more than 1.2
 
 query I
-SELECT CASE WHEN ${i} = 0 THEN True
-	WHEN current.total_blocks <= total_blocks_tbl.total_blocks * 1.2 THEN True
-	ELSE False END
+SELECT
+	CASE WHEN ${i} = 0
+		THEN True::test_result
+	WHEN current.total_blocks <= total_blocks_tbl.total_blocks * 1.2
+		THEN True::test_result
+	ELSE {
+			'old': total_blocks_tbl.total_blocks,
+			'allowed_max': total_blocks_tbl.total_blocks * 1.2,
+			'actual': current.total_blocks
+		}::test_result
+	END
 FROM pragma_database_size() AS current, total_blocks_tbl;
 ----
-1
+true
 
 # adjust total_blocks_tbl once to the count after the first iteration
 
 statement ok
 UPDATE total_blocks_tbl SET total_blocks = (
-	SELECT CASE WHEN ${i} = 0 THEN (SELECT current.total_blocks FROM pragma_database_size() AS current)
-		ELSE (total_blocks) END);
+	SELECT
+		CASE WHEN ${i} = 0
+			THEN (SELECT current.total_blocks FROM pragma_database_size() AS current)
+		ELSE
+			(total_blocks)
+		END
+	);
 
 statement ok
 CHECKPOINT;
@@ -66,8 +115,16 @@ CHECKPOINT;
 restart
 
 query IIIIII
-SELECT AVG(STRLEN(s)), MIN(STRLEN(S)), MAX(STRLEN(S)), SUM(STRLEN(S)), MIN(S[1]), MAX(S[1]) FROM strings
+SELECT
+	AVG(STRLEN(s)),
+	MIN(STRLEN(S)),
+	MAX(STRLEN(S)),
+	SUM(STRLEN(S)),
+	MIN(S[1]),
+	MAX(S[1])
+FROM strings
 ----
 296.955	0	5000	44543527	(empty)	X
 
+# i
 endloop
diff --git a/test/sql/storage/test_buffer_manager.cpp b/test/sql/storage/test_buffer_manager.cpp
index b5e333536906..9f2cfc101dc0 100644
--- a/test/sql/storage/test_buffer_manager.cpp
+++ b/test/sql/storage/test_buffer_manager.cpp
@@ -14,6 +14,8 @@ TEST_CASE("Test storing a big string that exceeds buffer manager size", "[storag
 	auto config = GetTestConfig();
 	config->options.default_block_alloc_size = DEFAULT_BLOCK_ALLOC_SIZE;
 	config->options.maximum_threads = 1;
+	// ZSTD can store this in a smaller way, force uncompressed so the 5mb max test correctly fails
+	config->options.force_compression = CompressionType::COMPRESSION_UNCOMPRESSED;
 
 	uint64_t string_length = 64;
 	uint64_t desired_size = 10000000; // desired size is 10MB
@@ -28,8 +30,7 @@ TEST_CASE("Test storing a big string that exceeds buffer manager size", "[storag
 		REQUIRE_NO_FAIL(con.Query("CREATE TABLE test (a VARCHAR, j BIGINT);"));
 		REQUIRE_NO_FAIL(con.Query("INSERT INTO test VALUES ('" + big_string + "', 1)"));
 		while (string_length < desired_size) {
-			REQUIRE_NO_FAIL(con.Query("INSERT INTO test SELECT a||a||a||a||a||a||a||a||a||a, " + to_string(iteration) +
-			                          " FROM test"));
+			REQUIRE_NO_FAIL(con.Query("INSERT INTO test SELECT repeat(a, 10), " + to_string(iteration) + " FROM test"));
 			REQUIRE_NO_FAIL(con.Query("DELETE FROM test WHERE j=" + to_string(iteration - 1)));
 			iteration++;
 			string_length *= 10;
diff --git a/test/sql/types/nested/array/array_storage.test b/test/sql/types/nested/array/array_storage.test
index 897adf847123..118dc9514b28 100644
--- a/test/sql/types/nested/array/array_storage.test
+++ b/test/sql/types/nested/array/array_storage.test
@@ -19,7 +19,7 @@ statement ok
 checkpoint;
 
 query IIIIIIIIIIIIIII rowsort
-SELECT * FROM pragma_storage_info('t1');
+SELECT * EXCLUDE additional_block_ids FROM pragma_storage_info('t1');
 ----
 0	a	1	[1, 0]		0	VALIDITY	0	1	Constant		[Has Null: false, Has No Null: true]					false	true	-1	0	(empty)
 0	a	1	[1, 1, 0]	0	VALIDITY	0	3	Constant		[Has Null: false, Has No Null: true]					false	true	-1	0	(empty)