Repository: duckdb/duckdb (language: python)

Issue:
Add Queries to CSV FROM/TO



Discussion / hints:

Fix patch:
diff --git a/extension/core_functions/aggregate/holistic/mode.cpp b/extension/core_functions/aggregate/holistic/mode.cpp
index e66fa7f4e6bc..bdfa0fc836e3 100644
--- a/extension/core_functions/aggregate/holistic/mode.cpp
+++ b/extension/core_functions/aggregate/holistic/mode.cpp
@@ -17,33 +17,7 @@
 // Returns the most frequent value for the values within expr1.
 // NULL values are ignored. If all the values are NULL, or there are 0 rows, then the function returns NULL.
 
-namespace std {
-
-template <>
-struct hash<duckdb::interval_t> {
-	inline size_t operator()(const duckdb::interval_t &val) const {
-		int64_t months, days, micros;
-		val.Normalize(months, days, micros);
-		return hash<int32_t> {}(duckdb::UnsafeNumericCast<int32_t>(days)) ^
-		       hash<int32_t> {}(duckdb::UnsafeNumericCast<int32_t>(months)) ^ hash<int64_t> {}(micros);
-	}
-};
-
-template <>
-struct hash<duckdb::hugeint_t> {
-	inline size_t operator()(const duckdb::hugeint_t &val) const {
-		return hash<int64_t> {}(val.upper) ^ hash<uint64_t> {}(val.lower);
-	}
-};
-
-template <>
-struct hash<duckdb::uhugeint_t> {
-	inline size_t operator()(const duckdb::uhugeint_t &val) const {
-		return hash<uint64_t> {}(val.upper) ^ hash<uint64_t> {}(val.lower);
-	}
-};
-
-} // namespace std
+namespace std {} // namespace std
 
 namespace duckdb {
 
diff --git a/extension/parquet/column_writer.cpp b/extension/parquet/column_writer.cpp
index 9046dcad80d9..f9354693047d 100644
--- a/extension/parquet/column_writer.cpp
+++ b/extension/parquet/column_writer.cpp
@@ -6,6 +6,8 @@
 #include "parquet_rle_bp_decoder.hpp"
 #include "parquet_rle_bp_encoder.hpp"
 #include "parquet_writer.hpp"
+#include "parquet_statistics.hpp"
+#include "geo_parquet.hpp"
 #ifndef DUCKDB_AMALGAMATION
 #include "duckdb/common/exception.hpp"
 #include "duckdb/common/operator/comparison_operators.hpp"
@@ -25,6 +27,9 @@
 #include "miniz_wrapper.hpp"
 #include "snappy.h"
 #include "zstd.h"
+#include "zstd/common/xxhash.hpp"
+
+#include <cmath>
 
 namespace duckdb {
 
@@ -332,6 +337,8 @@ class BasicColumnWriterState : public ColumnWriterState {
 	vector<PageWriteInformation> write_info;
 	unique_ptr<ColumnWriterStatistics> stats_state;
 	idx_t current_page = 0;
+
+	unique_ptr<ParquetBloomFilter> bloom_filter;
 };
 
 //===--------------------------------------------------------------------===//
@@ -353,10 +360,8 @@ class BasicColumnWriter : public ColumnWriter {
 	//! Dictionary pages must be below 2GB. Unlike data pages, there's only one dictionary page.
 	//! For this reason we go with a much higher, but still a conservative upper bound of 1GB;
 	static constexpr const idx_t MAX_UNCOMPRESSED_DICT_PAGE_SIZE = 1e9;
-	//! If the dictionary has this many entries, but the compression ratio is still below 1,
-	//! we stop creating the dictionary
+	//! If the dictionary has this many entries, we stop creating the dictionary
 	static constexpr const idx_t DICTIONARY_ANALYZE_THRESHOLD = 1e4;
-
 	//! The maximum size a key entry in an RLE page takes
 	static constexpr const idx_t MAX_DICTIONARY_KEY_SIZE = sizeof(uint32_t);
 	//! The size of encoding the string length
@@ -623,6 +628,9 @@ void BasicColumnWriter::Write(ColumnWriterState &state_p, Vector &vector, idx_t
 }
 
 void BasicColumnWriter::SetParquetStatistics(BasicColumnWriterState &state, duckdb_parquet::ColumnChunk &column_chunk) {
+	if (!state.stats_state) {
+		return;
+	}
 	if (max_repeat == 0) {
 		column_chunk.meta_data.statistics.null_count = NumericCast<int64_t>(state.null_count);
 		column_chunk.meta_data.statistics.__isset.null_count = true;
@@ -657,6 +665,11 @@ void BasicColumnWriter::SetParquetStatistics(BasicColumnWriterState &state, duck
 		column_chunk.meta_data.__isset.statistics = true;
 	}
 	for (const auto &write_info : state.write_info) {
+		// only care about data page encodings, data_page_header.encoding is meaningless for dict
+		if (write_info.page_header.type != PageType::DATA_PAGE &&
+		    write_info.page_header.type != PageType::DATA_PAGE_V2) {
+			continue;
+		}
 		column_chunk.meta_data.encodings.push_back(write_info.page_header.data_page_header.encoding);
 	}
 }
@@ -703,6 +716,11 @@ void BasicColumnWriter::FinalizeWrite(ColumnWriterState &state_p) {
 	column_chunk.meta_data.total_compressed_size =
 	    UnsafeNumericCast<int64_t>(column_writer.GetTotalWritten() - start_offset);
 	column_chunk.meta_data.total_uncompressed_size = UnsafeNumericCast<int64_t>(total_uncompressed_size);
+
+	if (state.bloom_filter) {
+		writer.BufferBloomFilter(state.col_idx, std::move(state.bloom_filter));
+	}
+	// which row group is this?
 }
 
 void BasicColumnWriter::FlushDictionary(BasicColumnWriterState &state, ColumnWriterStatistics *stats) {
@@ -775,13 +793,39 @@ class NumericStatisticsState : public ColumnWriterStatistics {
 };
 
 struct BaseParquetOperator {
+
+	template <class SRC, class TGT>
+	static void WriteToStream(const TGT &input, WriteStream &ser) {
+		ser.WriteData(const_data_ptr_cast(&input), sizeof(TGT));
+	}
+
+	template <class SRC, class TGT>
+	static uint64_t XXHash64(const TGT &target_value) {
+		return duckdb_zstd::XXH64(&target_value, sizeof(target_value), 0);
+	}
+
+	template <class SRC, class TGT>
+	static unique_ptr<ColumnWriterStatistics> InitializeStats() {
+		return nullptr;
+	}
+
+	template <class SRC, class TGT>
+	static void HandleStats(ColumnWriterStatistics *stats, TGT target_value) {
+	}
+};
+
+struct ParquetCastOperator : public BaseParquetOperator {
+	template <class SRC, class TGT>
+	static TGT Operation(SRC input) {
+		return TGT(input);
+	}
 	template <class SRC, class TGT>
 	static unique_ptr<ColumnWriterStatistics> InitializeStats() {
 		return make_uniq<NumericStatisticsState<SRC, TGT, BaseParquetOperator>>();
 	}
 
 	template <class SRC, class TGT>
-	static void HandleStats(ColumnWriterStatistics *stats, SRC source_value, TGT target_value) {
+	static void HandleStats(ColumnWriterStatistics *stats, TGT target_value) {
 		auto &numeric_stats = (NumericStatisticsState<SRC, TGT, BaseParquetOperator> &)*stats;
 		if (LessThan::Operation(target_value, numeric_stats.min)) {
 			numeric_stats.min = target_value;
@@ -792,24 +836,165 @@ struct BaseParquetOperator {
 	}
 };
 
-struct ParquetCastOperator : public BaseParquetOperator {
+struct ParquetTimestampNSOperator : public ParquetCastOperator {
 	template <class SRC, class TGT>
 	static TGT Operation(SRC input) {
 		return TGT(input);
 	}
 };
 
-struct ParquetTimestampNSOperator : public BaseParquetOperator {
+struct ParquetTimestampSOperator : public ParquetCastOperator {
 	template <class SRC, class TGT>
 	static TGT Operation(SRC input) {
-		return TGT(input);
+		return Timestamp::FromEpochSecondsPossiblyInfinite(input).value;
+	}
+};
+
+class StringStatisticsState : public ColumnWriterStatistics {
+	static constexpr const idx_t MAX_STRING_STATISTICS_SIZE = 10000;
+
+public:
+	StringStatisticsState() : has_stats(false), values_too_big(false), min(), max() {
+	}
+
+	bool has_stats;
+	bool values_too_big;
+	string min;
+	string max;
+
+public:
+	bool HasStats() override {
+		return has_stats;
+	}
+
+	void Update(const string_t &val) {
+		if (values_too_big) {
+			return;
+		}
+		auto str_len = val.GetSize();
+		if (str_len > MAX_STRING_STATISTICS_SIZE) {
+			// we avoid gathering stats when individual string values are too large
+			// this is because the statistics are copied into the Parquet file meta data in uncompressed format
+			// ideally we avoid placing several mega or giga-byte long strings there
+			// we put a threshold of 10KB, if we see strings that exceed this threshold we avoid gathering stats
+			values_too_big = true;
+			has_stats = false;
+			min = string();
+			max = string();
+			return;
+		}
+		if (!has_stats || LessThan::Operation(val, string_t(min))) {
+			min = val.GetString();
+		}
+		if (!has_stats || GreaterThan::Operation(val, string_t(max))) {
+			max = val.GetString();
+		}
+		has_stats = true;
+	}
+
+	string GetMin() override {
+		return GetMinValue();
+	}
+	string GetMax() override {
+		return GetMaxValue();
+	}
+	string GetMinValue() override {
+		return HasStats() ? min : string();
+	}
+	string GetMaxValue() override {
+		return HasStats() ? max : string();
 	}
 };
 
-struct ParquetTimestampSOperator : public BaseParquetOperator {
+struct ParquetStringOperator : public BaseParquetOperator {
 	template <class SRC, class TGT>
 	static TGT Operation(SRC input) {
-		return Timestamp::FromEpochSecondsPossiblyInfinite(input).value;
+		return input;
+	}
+
+	template <class SRC, class TGT>
+	static unique_ptr<ColumnWriterStatistics> InitializeStats() {
+		return make_uniq<StringStatisticsState>();
+	}
+
+	template <class SRC, class TGT>
+	static void HandleStats(ColumnWriterStatistics *stats, TGT target_value) {
+		auto &string_stats = (StringStatisticsState &)*stats;
+		string_stats.Update(target_value);
+	}
+
+	template <class SRC, class TGT>
+	static void WriteToStream(const TGT &target_value, WriteStream &ser) {
+		ser.Write<uint32_t>(target_value.GetSize());
+		ser.WriteData(const_data_ptr_cast(target_value.GetData()), target_value.GetSize());
+	}
+
+	template <class SRC, class TGT>
+	static uint64_t XXHash64(const TGT &target_value) {
+		return duckdb_zstd::XXH64(target_value.GetData(), target_value.GetSize(), 0);
+	}
+};
+
+struct ParquetIntervalTargetType {
+	static constexpr const idx_t PARQUET_INTERVAL_SIZE = 12;
+	data_t bytes[PARQUET_INTERVAL_SIZE];
+};
+
+struct ParquetIntervalOperator : public BaseParquetOperator {
+	template <class SRC, class TGT>
+	static TGT Operation(SRC input) {
+
+		if (input.days < 0 || input.months < 0 || input.micros < 0) {
+			throw IOException("Parquet files do not support negative intervals");
+		}
+		TGT result;
+		Store<uint32_t>(input.months, result.bytes);
+		Store<uint32_t>(input.days, result.bytes + sizeof(uint32_t));
+		Store<uint32_t>(input.micros / 1000, result.bytes + sizeof(uint32_t) * 2);
+		return result;
+	}
+
+	template <class SRC, class TGT>
+	static void WriteToStream(const TGT &target_value, WriteStream &ser) {
+		ser.WriteData(target_value.bytes, ParquetIntervalTargetType::PARQUET_INTERVAL_SIZE);
+	}
+
+	template <class SRC, class TGT>
+	static uint64_t XXHash64(const TGT &target_value) {
+		return duckdb_zstd::XXH64(target_value.bytes, ParquetIntervalTargetType::PARQUET_INTERVAL_SIZE, 0);
+	}
+};
+
+struct ParquetUUIDTargetType {
+	static constexpr const idx_t PARQUET_UUID_SIZE = 16;
+	data_t bytes[PARQUET_UUID_SIZE];
+};
+
+struct ParquetUUIDOperator : public BaseParquetOperator {
+	template <class SRC, class TGT>
+	static TGT Operation(SRC input) {
+		TGT result;
+		uint64_t high_bytes = input.upper ^ (int64_t(1) << 63);
+		uint64_t low_bytes = input.lower;
+		for (idx_t i = 0; i < sizeof(uint64_t); i++) {
+			auto shift_count = (sizeof(uint64_t) - i - 1) * 8;
+			result.bytes[i] = (high_bytes >> shift_count) & 0xFF;
+		}
+		for (idx_t i = 0; i < sizeof(uint64_t); i++) {
+			auto shift_count = (sizeof(uint64_t) - i - 1) * 8;
+			result.bytes[sizeof(uint64_t) + i] = (low_bytes >> shift_count) & 0xFF;
+		}
+		return result;
+	}
+
+	template <class SRC, class TGT>
+	static void WriteToStream(const TGT &target_value, WriteStream &ser) {
+		ser.WriteData(target_value.bytes, ParquetUUIDTargetType::PARQUET_UUID_SIZE);
+	}
+
+	template <class SRC, class TGT>
+	static uint64_t XXHash64(const TGT &target_value) {
+		return duckdb_zstd::XXH64(target_value.bytes, ParquetUUIDTargetType::PARQUET_UUID_SIZE, 0);
 	}
 };
 
@@ -820,7 +1005,7 @@ struct ParquetTimeTZOperator : public BaseParquetOperator {
 	}
 };
 
-struct ParquetHugeintOperator {
+struct ParquetHugeintOperator : public BaseParquetOperator {
 	template <class SRC, class TGT>
 	static TGT Operation(SRC input) {
 		return Hugeint::Cast<double>(input);
@@ -832,11 +1017,11 @@ struct ParquetHugeintOperator {
 	}
 
 	template <class SRC, class TGT>
-	static void HandleStats(ColumnWriterStatistics *stats, SRC source_value, TGT target_value) {
+	static void HandleStats(ColumnWriterStatistics *stats, TGT target_value) {
 	}
 };
 
-struct ParquetUhugeintOperator {
+struct ParquetUhugeintOperator : public BaseParquetOperator {
 	template <class SRC, class TGT>
 	static TGT Operation(SRC input) {
 		return Uhugeint::Cast<double>(input);
@@ -848,16 +1033,13 @@ struct ParquetUhugeintOperator {
 	}
 
 	template <class SRC, class TGT>
-	static void HandleStats(ColumnWriterStatistics *stats, SRC source_value, TGT target_value) {
+	static void HandleStats(ColumnWriterStatistics *stats, TGT target_value) {
 	}
 };
 
 template <class SRC, class TGT, class OP = ParquetCastOperator>
 static void TemplatedWritePlain(Vector &col, ColumnWriterStatistics *stats, const idx_t chunk_start,
                                 const idx_t chunk_end, const ValidityMask &mask, WriteStream &ser) {
-	static constexpr idx_t WRITE_COMBINER_CAPACITY = 8;
-	TGT write_combiner[WRITE_COMBINER_CAPACITY];
-	idx_t write_combiner_count = 0;
 
 	const auto *ptr = FlatVector::GetData<SRC>(col);
 	for (idx_t r = chunk_start; r < chunk_end; r++) {
@@ -865,16 +1047,12 @@ static void TemplatedWritePlain(Vector &col, ColumnWriterStatistics *stats, cons
 			continue;
 		}
 		TGT target_value = OP::template Operation<SRC, TGT>(ptr[r]);
-		OP::template HandleStats<SRC, TGT>(stats, ptr[r], target_value);
-		write_combiner[write_combiner_count++] = target_value;
-		if (write_combiner_count == WRITE_COMBINER_CAPACITY) {
-			ser.WriteData(const_data_ptr_cast(write_combiner), WRITE_COMBINER_CAPACITY * sizeof(TGT));
-			write_combiner_count = 0;
-		}
+		OP::template HandleStats<SRC, TGT>(stats, target_value);
+		OP::template WriteToStream<SRC, TGT>(target_value, ser);
 	}
-	ser.WriteData(const_data_ptr_cast(write_combiner), write_combiner_count * sizeof(TGT));
 }
 
+template <class T>
 class StandardColumnWriterState : public BasicColumnWriterState {
 public:
 	StandardColumnWriterState(duckdb_parquet::RowGroup &row_group, idx_t col_idx)
@@ -884,16 +1062,84 @@ class StandardColumnWriterState : public BasicColumnWriterState {
 
 	// analysis state for integer values for DELTA_BINARY_PACKED
 	idx_t total_value_count = 0;
+
+	unordered_map<T, uint32_t> dictionary;
+	duckdb_parquet::Encoding::type encoding;
 };
 
+template <class T>
 class StandardWriterPageState : public ColumnWriterPageState {
 public:
-	explicit StandardWriterPageState(const idx_t total_value_count) : encoder(total_value_count), initialized(false) {
+	explicit StandardWriterPageState(const idx_t total_value_count, Encoding::type encoding_p,
+	                                 const unordered_map<T, uint32_t> &dictionary_p)
+	    : encoding(encoding_p), dbp_initialized(false), dbp_encoder(total_value_count), dictionary(dictionary_p),
+	      dict_written_value(false), dict_bit_width(RleBpDecoder::ComputeBitWidth(dictionary.size())),
+	      dict_encoder(dict_bit_width) {
 	}
-	DbpEncoder encoder;
-	bool initialized;
+	duckdb_parquet::Encoding::type encoding;
+
+	bool dbp_initialized;
+	DbpEncoder dbp_encoder;
+
+	const unordered_map<T, uint32_t> &dictionary;
+	bool dict_written_value;
+	uint32_t dict_bit_width;
+	RleBpEncoder dict_encoder;
 };
 
+namespace dbp_encoder {
+template <class T>
+void BeginWrite(DbpEncoder &encoder, WriteStream &writer, const T &first_value) {
+	throw InternalException("Can't write type to DELTA_BINARY_PACKED column");
+}
+
+template <>
+void BeginWrite(DbpEncoder &encoder, WriteStream &writer, const int64_t &first_value) {
+	encoder.BeginWrite(writer, first_value);
+}
+
+template <>
+void BeginWrite(DbpEncoder &encoder, WriteStream &writer, const int32_t &first_value) {
+	BeginWrite(encoder, writer, UnsafeNumericCast<int64_t>(first_value));
+}
+
+template <>
+void BeginWrite(DbpEncoder &encoder, WriteStream &writer, const uint64_t &first_value) {
+	encoder.BeginWrite(writer, UnsafeNumericCast<int64_t>(first_value));
+}
+
+template <>
+void BeginWrite(DbpEncoder &encoder, WriteStream &writer, const uint32_t &first_value) {
+	BeginWrite(encoder, writer, UnsafeNumericCast<int64_t>(first_value));
+}
+
+template <class T>
+void WriteValue(DbpEncoder &encoder, WriteStream &writer, const T &value) {
+	throw InternalException("Can't write type to DELTA_BINARY_PACKED column");
+}
+
+template <>
+void WriteValue(DbpEncoder &encoder, WriteStream &writer, const int64_t &value) {
+	encoder.WriteValue(writer, value);
+}
+
+template <>
+void WriteValue(DbpEncoder &encoder, WriteStream &writer, const int32_t &value) {
+	WriteValue(encoder, writer, UnsafeNumericCast<int64_t>(value));
+}
+
+template <>
+void WriteValue(DbpEncoder &encoder, WriteStream &writer, const uint64_t &value) {
+	encoder.WriteValue(writer, UnsafeNumericCast<int64_t>(value));
+}
+
+template <>
+void WriteValue(DbpEncoder &encoder, WriteStream &writer, const uint32_t &value) {
+	WriteValue(encoder, writer, UnsafeNumericCast<int64_t>(value));
+}
+
+} // namespace dbp_encoder
+
 template <class SRC, class TGT, class OP = ParquetCastOperator>
 class StandardColumnWriter : public BasicColumnWriter {
 public:
@@ -905,86 +1151,156 @@ class StandardColumnWriter : public BasicColumnWriter {
 
 public:
 	unique_ptr<ColumnWriterState> InitializeWriteState(duckdb_parquet::RowGroup &row_group) override {
-		auto result = make_uniq<StandardColumnWriterState>(row_group, row_group.columns.size());
+		auto result = make_uniq<StandardColumnWriterState<SRC>>(row_group, row_group.columns.size());
+		result->encoding = Encoding::RLE_DICTIONARY;
 		RegisterToRowGroup(row_group);
 		return std::move(result);
 	}
 
 	unique_ptr<ColumnWriterPageState> InitializePageState(BasicColumnWriterState &state_p) override {
-		auto &state = state_p.Cast<StandardColumnWriterState>();
-		auto result = make_uniq<StandardWriterPageState>(state.total_value_count);
+		auto &state = state_p.Cast<StandardColumnWriterState<SRC>>();
+
+		auto result =
+		    make_uniq<StandardWriterPageState<SRC>>(state.total_value_count, state.encoding, state.dictionary);
 		return std::move(result);
 	}
 
 	void FlushPageState(WriteStream &temp_writer, ColumnWriterPageState *state_p) override {
-		auto &page_state = state_p->Cast<StandardWriterPageState>();
-		if (!page_state.initialized) {
-			page_state.encoder.BeginWrite(temp_writer, 0);
+		auto &page_state = state_p->Cast<StandardWriterPageState<SRC>>();
+		switch (page_state.encoding) {
+		case Encoding::DELTA_BINARY_PACKED:
+			if (!page_state.dbp_initialized) {
+				dbp_encoder::BeginWrite<int64_t>(page_state.dbp_encoder, temp_writer, 0);
+			}
+			page_state.dbp_encoder.FinishWrite(temp_writer);
+			break;
+		case Encoding::RLE_DICTIONARY:
+			D_ASSERT(page_state.dict_bit_width != 0);
+			if (!page_state.dict_written_value) {
+				// all values are null
+				// just write the bit width
+				temp_writer.Write<uint8_t>(page_state.dict_bit_width);
+				return;
+			}
+			page_state.dict_encoder.FinishWrite(temp_writer);
+
+			break;
+		case Encoding::PLAIN:
+			break;
+		default:
+			throw InternalException("Unknown encoding");
 		}
-		page_state.encoder.FinishWrite(temp_writer);
 	}
 
-	Encoding::type GetEncoding(BasicColumnWriterState &state) override {
-		return HasAnalyze() ? Encoding::DELTA_BINARY_PACKED : Encoding::PLAIN;
+	Encoding::type GetEncoding(BasicColumnWriterState &state_p) override {
+		auto &state = state_p.Cast<StandardColumnWriterState<SRC>>();
+		return state.encoding;
 	}
 
 	bool HasAnalyze() override {
-		// We can only do DELTA_BINARY_PACKED if the target type is int32_t/int64_t
-		const auto type = writer.GetType(schema_idx);
-		return type == Type::type::INT32 || type == Type::type::INT64;
+		return true;
 	}
 
 	void Analyze(ColumnWriterState &state_p, ColumnWriterState *parent, Vector &vector, idx_t count) override {
-		D_ASSERT(HasAnalyze());
-		auto &state = state_p.Cast<StandardColumnWriterState>();
+		auto &state = state_p.Cast<StandardColumnWriterState<SRC>>();
+
+		auto data_ptr = FlatVector::GetData<SRC>(vector);
+		idx_t vector_index = 0;
+		uint32_t new_value_index = state.dictionary.size();
 
 		const bool check_parent_empty = parent && !parent->is_empty.empty();
 		const idx_t parent_index = state.definition_levels.size();
 
 		const idx_t vcount =
 		    check_parent_empty ? parent->definition_levels.size() - state.definition_levels.size() : count;
+
 		const auto &validity = FlatVector::Validity(vector);
 
-		idx_t vector_index = 0;
 		for (idx_t i = 0; i < vcount; i++) {
 			if (check_parent_empty && parent->is_empty[parent_index + i]) {
 				continue;
 			}
-
 			if (validity.RowIsValid(vector_index)) {
+				if (state.dictionary.size() <= writer.DictionarySizeLimit()) {
+					const auto &src_value = data_ptr[vector_index];
+					if (state.dictionary.find(src_value) == state.dictionary.end()) {
+						state.dictionary[src_value] = new_value_index;
+						new_value_index++;
+					}
+				}
 				state.total_value_count++;
 			}
 			vector_index++;
 		}
 	}
 
-	void FinalizeAnalyze(ColumnWriterState &state) override {
-		// NOP
+	void FinalizeAnalyze(ColumnWriterState &state_p) override {
+		const auto type = writer.GetType(schema_idx);
+
+		auto &state = state_p.Cast<StandardColumnWriterState<SRC>>();
+		if (state.dictionary.size() == 0 || state.dictionary.size() > writer.DictionarySizeLimit()) {
+			// special handling for int column: dpb, otherwise plain
+			state.encoding = (type == Type::type::INT32 || type == Type::type::INT64) ? Encoding::DELTA_BINARY_PACKED
+			                                                                          : Encoding::PLAIN;
+			state.dictionary.clear();
+		}
 	}
 
 	unique_ptr<ColumnWriterStatistics> InitializeStatsState() override {
 		return OP::template InitializeStats<SRC, TGT>();
 	}
 
+	bool HasDictionary(BasicColumnWriterState &state_p) override {
+		auto &state = state_p.Cast<StandardColumnWriterState<SRC>>();
+		return state.encoding == Encoding::RLE_DICTIONARY;
+	}
+
+	idx_t DictionarySize(BasicColumnWriterState &state_p) override {
+		auto &state = state_p.Cast<StandardColumnWriterState<SRC>>();
+		return state.dictionary.size();
+	}
+
 	void WriteVector(WriteStream &temp_writer, ColumnWriterStatistics *stats, ColumnWriterPageState *page_state_p,
 	                 Vector &input_column, idx_t chunk_start, idx_t chunk_end) override {
+		auto &page_state = page_state_p->Cast<StandardWriterPageState<SRC>>();
+
 		const auto &mask = FlatVector::Validity(input_column);
-		if (HasAnalyze()) {
-			auto &page_state = page_state_p->Cast<StandardWriterPageState>();
-			auto &encoder = page_state.encoder;
-			const auto *ptr = FlatVector::GetData<SRC>(input_column);
+		const auto *data_ptr = FlatVector::GetData<SRC>(input_column);
+
+		switch (page_state.encoding) {
+		case Encoding::RLE_DICTIONARY: {
+			for (idx_t r = chunk_start; r < chunk_end; r++) {
+				if (!mask.RowIsValid(r)) {
+					continue;
+				}
+				auto &src_val = data_ptr[r];
+				auto value_index = page_state.dictionary.at(src_val);
+				if (!page_state.dict_written_value) {
+					// first value
+					// write the bit-width as a one-byte entry
+					temp_writer.Write<uint8_t>(page_state.dict_bit_width);
+					// now begin writing the actual value
+					page_state.dict_encoder.BeginWrite(temp_writer, value_index);
+					page_state.dict_written_value = true;
+				} else {
+					page_state.dict_encoder.WriteValue(temp_writer, value_index);
+				}
+			}
+			break;
+		}
 
+		case Encoding::DELTA_BINARY_PACKED: {
 			idx_t r = chunk_start;
-			if (!page_state.initialized) {
+			if (!page_state.dbp_initialized) {
 				// find first non-null value
 				for (; r < chunk_end; r++) {
 					if (!mask.RowIsValid(r)) {
 						continue;
 					}
-					const TGT target_value = OP::template Operation<SRC, TGT>(ptr[r]);
-					OP::template HandleStats<SRC, TGT>(stats, ptr[r], target_value);
-					encoder.BeginWrite(temp_writer, target_value);
-					page_state.initialized = true;
+					const TGT target_value = OP::template Operation<SRC, TGT>(data_ptr[r]);
+					OP::template HandleStats<SRC, TGT>(stats, target_value);
+					dbp_encoder::BeginWrite(page_state.dbp_encoder, temp_writer, target_value);
+					page_state.dbp_initialized = true;
 					r++; // skip over
 					break;
 				}
@@ -994,16 +1310,56 @@ class StandardColumnWriter : public BasicColumnWriter {
 				if (!mask.RowIsValid(r)) {
 					continue;
 				}
-				const TGT target_value = OP::template Operation<SRC, TGT>(ptr[r]);
-				OP::template HandleStats<SRC, TGT>(stats, ptr[r], target_value);
-				encoder.WriteValue(temp_writer, target_value);
+				const TGT target_value = OP::template Operation<SRC, TGT>(data_ptr[r]);
+				OP::template HandleStats<SRC, TGT>(stats, target_value);
+				dbp_encoder::WriteValue(page_state.dbp_encoder, temp_writer, target_value);
 			}
-		} else {
+			break;
+		}
+		case Encoding::PLAIN: {
+			D_ASSERT(page_state.encoding == Encoding::PLAIN);
 			TemplatedWritePlain<SRC, TGT, OP>(input_column, stats, chunk_start, chunk_end, mask, temp_writer);
+			break;
+		}
+		default:
+			throw InternalException("Unknown encoding");
 		}
 	}
 
-	idx_t GetRowSize(const Vector &vector, const idx_t index, const BasicColumnWriterState &state) const override {
+	void FlushDictionary(BasicColumnWriterState &state_p, ColumnWriterStatistics *stats) override {
+		auto &state = state_p.Cast<StandardColumnWriterState<SRC>>();
+
+		D_ASSERT(state.encoding == Encoding::RLE_DICTIONARY);
+
+		// first we need to sort the values in index order
+		auto values = vector<SRC>(state.dictionary.size());
+		for (const auto &entry : state.dictionary) {
+			values[entry.second] = entry.first;
+		}
+
+		state.bloom_filter =
+		    make_uniq<ParquetBloomFilter>(state.dictionary.size(), writer.BloomFilterFalsePositiveRatio());
+
+		// first write the contents of the dictionary page to a temporary buffer
+		auto temp_writer = make_uniq<MemoryStream>(MaxValue<idx_t>(
+		    NextPowerOfTwo(state.dictionary.size() * sizeof(TGT)), MemoryStream::DEFAULT_INITIAL_CAPACITY));
+		for (idx_t r = 0; r < values.size(); r++) {
+			const TGT target_value = OP::template Operation<SRC, TGT>(values[r]);
+			// update the statistics
+			OP::template HandleStats<SRC, TGT>(stats, target_value);
+			// update the bloom filter
+			auto hash = OP::template XXHash64<SRC, TGT>(target_value);
+			state.bloom_filter->FilterInsert(hash);
+			// actually write the dictionary value
+			OP::template WriteToStream<SRC, TGT>(target_value, *temp_writer);
+		}
+		// flush the dictionary page and add it to the to-be-written pages
+		WriteDictionary(state, std::move(temp_writer), values.size());
+		// bloom filter will be queued for writing in ParquetWriter::BufferBloomFilter one level up
+	}
+
+	// TODO this now vastly over-estimates the page size
+	idx_t GetRowSize(const Vector &vector, const idx_t index, const BasicColumnWriterState &state_p) const override {
 		return sizeof(TGT);
 	}
 };
@@ -1204,427 +1560,25 @@ class FixedDecimalColumnWriter : public BasicColumnWriter {
 };
 
 //===--------------------------------------------------------------------===//
-// UUID Column Writer
+// WKB Column Writer
 //===--------------------------------------------------------------------===//
-class UUIDColumnWriter : public BasicColumnWriter {
-	static constexpr const idx_t PARQUET_UUID_SIZE = 16;
-
+// Used to store the metadata for a WKB-encoded geometry column when writing
+// GeoParquet files.
+class WKBColumnWriterState final : public StandardColumnWriterState<string_t> {
 public:
-	UUIDColumnWriter(ParquetWriter &writer, idx_t schema_idx, vector<string> schema_path_p, idx_t max_repeat,
-	                 idx_t max_define, bool can_have_nulls)
-	    : BasicColumnWriter(writer, schema_idx, std::move(schema_path_p), max_repeat, max_define, can_have_nulls) {
+	WKBColumnWriterState(ClientContext &context, duckdb_parquet::RowGroup &row_group, idx_t col_idx)
+	    : StandardColumnWriterState(row_group, col_idx), geo_data(), geo_data_writer(context) {
 	}
-	~UUIDColumnWriter() override = default;
 
-public:
-	static void WriteParquetUUID(hugeint_t input, data_ptr_t result) {
-		uint64_t high_bytes = input.upper ^ (int64_t(1) << 63);
-		uint64_t low_bytes = input.lower;
+	GeoParquetColumnMetadata geo_data;
+	GeoParquetColumnMetadataWriter geo_data_writer;
+};
 
-		for (idx_t i = 0; i < sizeof(uint64_t); i++) {
-			auto shift_count = (sizeof(uint64_t) - i - 1) * 8;
-			result[i] = (high_bytes >> shift_count) & 0xFF;
-		}
-		for (idx_t i = 0; i < sizeof(uint64_t); i++) {
-			auto shift_count = (sizeof(uint64_t) - i - 1) * 8;
-			result[sizeof(uint64_t) + i] = (low_bytes >> shift_count) & 0xFF;
-		}
-	}
-
-	void WriteVector(WriteStream &temp_writer, ColumnWriterStatistics *stats_p, ColumnWriterPageState *page_state,
-	                 Vector &input_column, idx_t chunk_start, idx_t chunk_end) override {
-		auto &mask = FlatVector::Validity(input_column);
-		auto *ptr = FlatVector::GetData<hugeint_t>(input_column);
-
-		data_t temp_buffer[PARQUET_UUID_SIZE];
-		for (idx_t r = chunk_start; r < chunk_end; r++) {
-			if (mask.RowIsValid(r)) {
-				WriteParquetUUID(ptr[r], temp_buffer);
-				temp_writer.WriteData(temp_buffer, PARQUET_UUID_SIZE);
-			}
-		}
-	}
-
-	idx_t GetRowSize(const Vector &vector, const idx_t index, const BasicColumnWriterState &state) const override {
-		return PARQUET_UUID_SIZE;
-	}
-};
-
-//===--------------------------------------------------------------------===//
-// Interval Column Writer
-//===--------------------------------------------------------------------===//
-class IntervalColumnWriter : public BasicColumnWriter {
-	static constexpr const idx_t PARQUET_INTERVAL_SIZE = 12;
-
-public:
-	IntervalColumnWriter(ParquetWriter &writer, idx_t schema_idx, vector<string> schema_path_p, idx_t max_repeat,
-	                     idx_t max_define, bool can_have_nulls)
-	    : BasicColumnWriter(writer, schema_idx, std::move(schema_path_p), max_repeat, max_define, can_have_nulls) {
-	}
-	~IntervalColumnWriter() override = default;
-
-public:
-	static void WriteParquetInterval(interval_t input, data_ptr_t result) {
-		if (input.days < 0 || input.months < 0 || input.micros < 0) {
-			throw IOException("Parquet files do not support negative intervals");
-		}
-		Store<uint32_t>(input.months, result);
-		Store<uint32_t>(input.days, result + sizeof(uint32_t));
-		Store<uint32_t>(input.micros / 1000, result + sizeof(uint32_t) * 2);
-	}
-
-	void WriteVector(WriteStream &temp_writer, ColumnWriterStatistics *stats_p, ColumnWriterPageState *page_state,
-	                 Vector &input_column, idx_t chunk_start, idx_t chunk_end) override {
-		auto &mask = FlatVector::Validity(input_column);
-		auto *ptr = FlatVector::GetData<interval_t>(input_column);
-
-		data_t temp_buffer[PARQUET_INTERVAL_SIZE];
-		for (idx_t r = chunk_start; r < chunk_end; r++) {
-			if (mask.RowIsValid(r)) {
-				WriteParquetInterval(ptr[r], temp_buffer);
-				temp_writer.WriteData(temp_buffer, PARQUET_INTERVAL_SIZE);
-			}
-		}
-	}
-
-	idx_t GetRowSize(const Vector &vector, const idx_t index, const BasicColumnWriterState &state) const override {
-		return PARQUET_INTERVAL_SIZE;
-	}
-};
-
-//===--------------------------------------------------------------------===//
-// String Column Writer
-//===--------------------------------------------------------------------===//
-class StringStatisticsState : public ColumnWriterStatistics {
-	static constexpr const idx_t MAX_STRING_STATISTICS_SIZE = 10000;
-
-public:
-	StringStatisticsState() : has_stats(false), values_too_big(false), min(), max() {
-	}
-
-	bool has_stats;
-	bool values_too_big;
-	string min;
-	string max;
-
-public:
-	bool HasStats() override {
-		return has_stats;
-	}
-
-	void Update(const string_t &val) {
-		if (values_too_big) {
-			return;
-		}
-		auto str_len = val.GetSize();
-		if (str_len > MAX_STRING_STATISTICS_SIZE) {
-			// we avoid gathering stats when individual string values are too large
-			// this is because the statistics are copied into the Parquet file meta data in uncompressed format
-			// ideally we avoid placing several mega or giga-byte long strings there
-			// we put a threshold of 10KB, if we see strings that exceed this threshold we avoid gathering stats
-			values_too_big = true;
-			has_stats = false;
-			min = string();
-			max = string();
-			return;
-		}
-		if (!has_stats || LessThan::Operation(val, string_t(min))) {
-			min = val.GetString();
-		}
-		if (!has_stats || GreaterThan::Operation(val, string_t(max))) {
-			max = val.GetString();
-		}
-		has_stats = true;
-	}
-
-	string GetMin() override {
-		return GetMinValue();
-	}
-	string GetMax() override {
-		return GetMaxValue();
-	}
-	string GetMinValue() override {
-		return HasStats() ? min : string();
-	}
-	string GetMaxValue() override {
-		return HasStats() ? max : string();
-	}
-};
-
-class StringColumnWriterState : public BasicColumnWriterState {
-public:
-	StringColumnWriterState(duckdb_parquet::RowGroup &row_group, idx_t col_idx)
-	    : BasicColumnWriterState(row_group, col_idx) {
-	}
-	~StringColumnWriterState() override = default;
-
-	// analysis state
-	idx_t estimated_dict_page_size = 0;
-	idx_t estimated_rle_pages_size = 0;
-	idx_t estimated_plain_size = 0;
-
-	// Dictionary and accompanying string heap
-	string_map_t<uint32_t> dictionary;
-	// key_bit_width== 0 signifies the chunk is written in plain encoding
-	uint32_t key_bit_width;
-
-	bool IsDictionaryEncoded() const {
-		return key_bit_width != 0;
-	}
-};
-
-class StringWriterPageState : public ColumnWriterPageState {
-public:
-	explicit StringWriterPageState(uint32_t bit_width, const string_map_t<uint32_t> &values)
-	    : bit_width(bit_width), dictionary(values), encoder(bit_width), written_value(false) {
-		D_ASSERT(IsDictionaryEncoded() || (bit_width == 0 && dictionary.empty()));
-	}
-
-	bool IsDictionaryEncoded() {
-		return bit_width != 0;
-	}
-	// if 0, we're writing a plain page
-	uint32_t bit_width;
-	const string_map_t<uint32_t> &dictionary;
-	RleBpEncoder encoder;
-	bool written_value;
-};
-
-class StringColumnWriter : public BasicColumnWriter {
-public:
-	StringColumnWriter(ParquetWriter &writer, idx_t schema_idx, vector<string> schema_path_p, idx_t max_repeat,
-	                   idx_t max_define, bool can_have_nulls)
-	    : BasicColumnWriter(writer, schema_idx, std::move(schema_path_p), max_repeat, max_define, can_have_nulls) {
-	}
-	~StringColumnWriter() override = default;
-
-public:
-	unique_ptr<ColumnWriterStatistics> InitializeStatsState() override {
-		return make_uniq<StringStatisticsState>();
-	}
-
-	unique_ptr<ColumnWriterState> InitializeWriteState(duckdb_parquet::RowGroup &row_group) override {
-		auto result = make_uniq<StringColumnWriterState>(row_group, row_group.columns.size());
-		RegisterToRowGroup(row_group);
-		return std::move(result);
-	}
-
-	bool HasAnalyze() override {
-		return true;
-	}
-
-	void Analyze(ColumnWriterState &state_p, ColumnWriterState *parent, Vector &vector, idx_t count) override {
-		auto &state = state_p.Cast<StringColumnWriterState>();
-		if (writer.DictionaryCompressionRatioThreshold() == NumericLimits<double>::Maximum() ||
-		    (state.dictionary.size() > DICTIONARY_ANALYZE_THRESHOLD && WontUseDictionary(state))) {
-			// Early out: compression ratio is less than the specified parameter
-			// after seeing more entries than the threshold
-			return;
-		}
-
-		idx_t vcount = parent ? parent->definition_levels.size() - state.definition_levels.size() : count;
-		idx_t parent_index = state.definition_levels.size();
-		auto &validity = FlatVector::Validity(vector);
-		idx_t vector_index = 0;
-		uint32_t new_value_index = state.dictionary.size();
-		uint32_t last_value_index = -1;
-		idx_t run_length = 0;
-		idx_t run_count = 0;
-		auto strings = FlatVector::GetData<string_t>(vector);
-		for (idx_t i = 0; i < vcount; i++) {
-			if (parent && !parent->is_empty.empty() && parent->is_empty[parent_index + i]) {
-				continue;
-			}
-
-			if (validity.RowIsValid(vector_index)) {
-				run_length++;
-				const auto &value = strings[vector_index];
-				// Try to insert into the dictionary. If it's already there, we get back the value index
-				auto found = state.dictionary.insert(string_map_t<uint32_t>::value_type(value, new_value_index));
-				state.estimated_plain_size += value.GetSize() + STRING_LENGTH_SIZE;
-				if (found.second) {
-					// string didn't exist yet in the dictionary
-					new_value_index++;
-					state.estimated_dict_page_size += value.GetSize() + MAX_DICTIONARY_KEY_SIZE;
-				}
-
-				// if the value changed, we will encode it in the page
-				if (last_value_index != found.first->second) {
-					// we will add the value index size later, when we know the total number of keys
-					state.estimated_rle_pages_size += ParquetDecodeUtils::GetVarintSize(run_length);
-					run_length = 0;
-					run_count++;
-					last_value_index = found.first->second;
-				}
-			}
-			vector_index++;
-		}
-		// Add the costs of keys sizes. We don't know yet how many bytes the keys need as we haven't
-		// seen all the values. therefore we use an over-estimation of
-		state.estimated_rle_pages_size += MAX_DICTIONARY_KEY_SIZE * run_count;
-	}
-
-	void FinalizeAnalyze(ColumnWriterState &state_p) override {
-		auto &state = state_p.Cast<StringColumnWriterState>();
-
-		// check if a dictionary will require more space than a plain write, or if the dictionary page is going to
-		// be too large
-		if (WontUseDictionary(state)) {
-			// clearing the dictionary signals a plain write
-			state.dictionary.clear();
-			state.key_bit_width = 0;
-		} else {
-			state.key_bit_width = RleBpDecoder::ComputeBitWidth(state.dictionary.size());
-		}
-	}
-
-	void WriteVector(WriteStream &temp_writer, ColumnWriterStatistics *stats_p, ColumnWriterPageState *page_state_p,
-	                 Vector &input_column, idx_t chunk_start, idx_t chunk_end) override {
-		auto &page_state = page_state_p->Cast<StringWriterPageState>();
-		auto &mask = FlatVector::Validity(input_column);
-		auto &stats = stats_p->Cast<StringStatisticsState>();
-
-		auto *ptr = FlatVector::GetData<string_t>(input_column);
-		if (page_state.IsDictionaryEncoded()) {
-			// dictionary based page
-			for (idx_t r = chunk_start; r < chunk_end; r++) {
-				if (!mask.RowIsValid(r)) {
-					continue;
-				}
-				auto value_index = page_state.dictionary.at(ptr[r]);
-				if (!page_state.written_value) {
-					// first value
-					// write the bit-width as a one-byte entry
-					temp_writer.Write<uint8_t>(page_state.bit_width);
-					// now begin writing the actual value
-					page_state.encoder.BeginWrite(temp_writer, value_index);
-					page_state.written_value = true;
-				} else {
-					page_state.encoder.WriteValue(temp_writer, value_index);
-				}
-			}
-		} else {
-			// plain page
-			for (idx_t r = chunk_start; r < chunk_end; r++) {
-				if (!mask.RowIsValid(r)) {
-					continue;
-				}
-				stats.Update(ptr[r]);
-				temp_writer.Write<uint32_t>(ptr[r].GetSize());
-				temp_writer.WriteData(const_data_ptr_cast(ptr[r].GetData()), ptr[r].GetSize());
-			}
-		}
-	}
-
-	unique_ptr<ColumnWriterPageState> InitializePageState(BasicColumnWriterState &state_p) override {
-		auto &state = state_p.Cast<StringColumnWriterState>();
-		return make_uniq<StringWriterPageState>(state.key_bit_width, state.dictionary);
-	}
-
-	void FlushPageState(WriteStream &temp_writer, ColumnWriterPageState *state_p) override {
-		auto &page_state = state_p->Cast<StringWriterPageState>();
-		if (page_state.bit_width != 0) {
-			if (!page_state.written_value) {
-				// all values are null
-				// just write the bit width
-				temp_writer.Write<uint8_t>(page_state.bit_width);
-				return;
-			}
-			page_state.encoder.FinishWrite(temp_writer);
-		}
-	}
-
-	duckdb_parquet::Encoding::type GetEncoding(BasicColumnWriterState &state_p) override {
-		auto &state = state_p.Cast<StringColumnWriterState>();
-		return state.IsDictionaryEncoded() ? Encoding::RLE_DICTIONARY : Encoding::PLAIN;
-	}
-
-	bool HasDictionary(BasicColumnWriterState &state_p) override {
-		auto &state = state_p.Cast<StringColumnWriterState>();
-		return state.IsDictionaryEncoded();
-	}
-
-	idx_t DictionarySize(BasicColumnWriterState &state_p) override {
-		auto &state = state_p.Cast<StringColumnWriterState>();
-		D_ASSERT(state.IsDictionaryEncoded());
-		return state.dictionary.size();
-	}
-
-	void FlushDictionary(BasicColumnWriterState &state_p, ColumnWriterStatistics *stats_p) override {
-		auto &stats = stats_p->Cast<StringStatisticsState>();
-		auto &state = state_p.Cast<StringColumnWriterState>();
-		if (!state.IsDictionaryEncoded()) {
-			return;
-		}
-		// first we need to sort the values in index order
-		auto values = vector<string_t>(state.dictionary.size());
-		for (const auto &entry : state.dictionary) {
-			D_ASSERT(values[entry.second].GetSize() == 0);
-			values[entry.second] = entry.first;
-		}
-		// first write the contents of the dictionary page to a temporary buffer
-		auto temp_writer = make_uniq<MemoryStream>(
-		    MaxValue<idx_t>(NextPowerOfTwo(state.estimated_dict_page_size), MemoryStream::DEFAULT_INITIAL_CAPACITY));
-		for (idx_t r = 0; r < values.size(); r++) {
-			auto &value = values[r];
-			// update the statistics
-			stats.Update(value);
-			// write this string value to the dictionary
-			temp_writer->Write<uint32_t>(value.GetSize());
-			temp_writer->WriteData(const_data_ptr_cast((value.GetData())), value.GetSize());
-		}
-		// flush the dictionary page and add it to the to-be-written pages
-		WriteDictionary(state, std::move(temp_writer), values.size());
-	}
-
-	idx_t GetRowSize(const Vector &vector, const idx_t index, const BasicColumnWriterState &state_p) const override {
-		auto &state = state_p.Cast<StringColumnWriterState>();
-		if (state.IsDictionaryEncoded()) {
-			return (state.key_bit_width + 7) / 8;
-		} else {
-			auto strings = FlatVector::GetData<string_t>(vector);
-			return strings[index].GetSize();
-		}
-	}
-
-private:
-	bool WontUseDictionary(StringColumnWriterState &state) const {
-		return state.estimated_dict_page_size > MAX_UNCOMPRESSED_DICT_PAGE_SIZE ||
-		       DictionaryCompressionRatio(state) < writer.DictionaryCompressionRatioThreshold();
-	}
-
-	static double DictionaryCompressionRatio(StringColumnWriterState &state) {
-		// If any are 0, we just return a compression ratio of 1
-		if (state.estimated_plain_size == 0 || state.estimated_rle_pages_size == 0 ||
-		    state.estimated_dict_page_size == 0) {
-			return 1;
-		}
-		// Otherwise, plain size divided by compressed size
-		return double(state.estimated_plain_size) /
-		       double(state.estimated_rle_pages_size + state.estimated_dict_page_size);
-	}
-};
-
-//===--------------------------------------------------------------------===//
-// WKB Column Writer
-//===--------------------------------------------------------------------===//
-// Used to store the metadata for a WKB-encoded geometry column when writing
-// GeoParquet files.
-class WKBColumnWriterState final : public StringColumnWriterState {
-public:
-	WKBColumnWriterState(ClientContext &context, duckdb_parquet::RowGroup &row_group, idx_t col_idx)
-	    : StringColumnWriterState(row_group, col_idx), geo_data(), geo_data_writer(context) {
-	}
-
-	GeoParquetColumnMetadata geo_data;
-	GeoParquetColumnMetadataWriter geo_data_writer;
-};
-
-class WKBColumnWriter final : public StringColumnWriter {
+class WKBColumnWriter final : public StandardColumnWriter<string_t, string_t, ParquetStringOperator> {
 public:
 	WKBColumnWriter(ClientContext &context_p, ParquetWriter &writer, idx_t schema_idx, vector<string> schema_path_p,
 	                idx_t max_repeat, idx_t max_define, bool can_have_nulls, string name)
-	    : StringColumnWriter(writer, schema_idx, std::move(schema_path_p), max_repeat, max_define, can_have_nulls),
+	    : StandardColumnWriter(writer, schema_idx, std::move(schema_path_p), max_repeat, max_define, can_have_nulls),
 	      column_name(std::move(name)), context(context_p) {
 
 		this->writer.GetGeoParquetData().RegisterGeometryColumn(column_name);
@@ -1632,18 +1586,20 @@ class WKBColumnWriter final : public StringColumnWriter {
 
 	unique_ptr<ColumnWriterState> InitializeWriteState(duckdb_parquet::RowGroup &row_group) override {
 		auto result = make_uniq<WKBColumnWriterState>(context, row_group, row_group.columns.size());
+		result->encoding = Encoding::RLE_DICTIONARY;
 		RegisterToRowGroup(row_group);
 		return std::move(result);
 	}
+
 	void Write(ColumnWriterState &state, Vector &vector, idx_t count) override {
-		StringColumnWriter::Write(state, vector, count);
+		StandardColumnWriter::Write(state, vector, count);
 
 		auto &geo_state = state.Cast<WKBColumnWriterState>();
 		geo_state.geo_data_writer.Update(geo_state.geo_data, vector, count);
 	}
 
 	void FinalizeWrite(ColumnWriterState &state) override {
-		StringColumnWriter::FinalizeWrite(state);
+		StandardColumnWriter::FinalizeWrite(state);
 
 		// Add the geodata object to the writer
 		const auto &geo_state = state.Cast<WKBColumnWriterState>();
@@ -2160,6 +2116,42 @@ void ArrayColumnWriter::Write(ColumnWriterState &state_p, Vector &vector, idx_t
 	child_writer->Write(*state.child_state, array_child, count * array_size);
 }
 
+// special double/float class to deal with dictionary encoding and NaN equality
+struct double_na_equal {
+	double_na_equal() : val(0) {
+	}
+	double_na_equal(const double val_p) : val(val_p) {
+	}
+	operator double() const {
+		return val;
+	}
+
+	bool operator==(const double &right) const {
+		if (std::isnan(val) && std::isnan(right)) {
+			return true;
+		}
+		return val == right;
+	}
+	double val;
+};
+
+struct float_na_equal {
+	float_na_equal() : val(0) {
+	}
+	float_na_equal(const float val_p) : val(val_p) {
+	}
+	operator float() const {
+		return val;
+	}
+	bool operator==(const float &right) const {
+		if (std::isnan(val) && std::isnan(right)) {
+			return true;
+		}
+		return val == right;
+	}
+	float val;
+};
+
 //===--------------------------------------------------------------------===//
 // Create Column Writer
 //===--------------------------------------------------------------------===//
@@ -2325,7 +2317,6 @@ unique_ptr<ColumnWriter> ColumnWriter::CreateWriterRecursive(ClientContext &cont
 	ParquetWriter::SetSchemaProperties(type, schema_element);
 	schemas.push_back(std::move(schema_element));
 	schema_path.push_back(name);
-
 	if (type.id() == LogicalTypeId::BLOB && type.GetAlias() == "WKB_BLOB" &&
 	    GeoParquetFileMetadata::IsGeoParquetConversionEnabled(context)) {
 		return make_uniq<WKBColumnWriter>(context, writer, schema_idx, std::move(schema_path), max_repeat, max_define,
@@ -2381,11 +2372,11 @@ unique_ptr<ColumnWriter> ColumnWriter::CreateWriterRecursive(ClientContext &cont
 		return make_uniq<StandardColumnWriter<uint64_t, uint64_t>>(writer, schema_idx, std::move(schema_path),
 		                                                           max_repeat, max_define, can_have_nulls);
 	case LogicalTypeId::FLOAT:
-		return make_uniq<StandardColumnWriter<float, float>>(writer, schema_idx, std::move(schema_path), max_repeat,
-		                                                     max_define, can_have_nulls);
+		return make_uniq<StandardColumnWriter<float_na_equal, float>>(writer, schema_idx, std::move(schema_path),
+		                                                              max_repeat, max_define, can_have_nulls);
 	case LogicalTypeId::DOUBLE:
-		return make_uniq<StandardColumnWriter<double, double>>(writer, schema_idx, std::move(schema_path), max_repeat,
-		                                                       max_define, can_have_nulls);
+		return make_uniq<StandardColumnWriter<double_na_equal, double>>(writer, schema_idx, std::move(schema_path),
+		                                                                max_repeat, max_define, can_have_nulls);
 	case LogicalTypeId::DECIMAL:
 		switch (type.InternalType()) {
 		case PhysicalType::INT16:
@@ -2403,14 +2394,14 @@ unique_ptr<ColumnWriter> ColumnWriter::CreateWriterRecursive(ClientContext &cont
 		}
 	case LogicalTypeId::BLOB:
 	case LogicalTypeId::VARCHAR:
-		return make_uniq<StringColumnWriter>(writer, schema_idx, std::move(schema_path), max_repeat, max_define,
-		                                     can_have_nulls);
+		return make_uniq<StandardColumnWriter<string_t, string_t, ParquetStringOperator>>(
+		    writer, schema_idx, std::move(schema_path), max_repeat, max_define, can_have_nulls);
 	case LogicalTypeId::UUID:
-		return make_uniq<UUIDColumnWriter>(writer, schema_idx, std::move(schema_path), max_repeat, max_define,
-		                                   can_have_nulls);
+		return make_uniq<StandardColumnWriter<hugeint_t, ParquetUUIDTargetType, ParquetUUIDOperator>>(
+		    writer, schema_idx, std::move(schema_path), max_repeat, max_define, can_have_nulls);
 	case LogicalTypeId::INTERVAL:
-		return make_uniq<IntervalColumnWriter>(writer, schema_idx, std::move(schema_path), max_repeat, max_define,
-		                                       can_have_nulls);
+		return make_uniq<StandardColumnWriter<interval_t, ParquetIntervalTargetType, ParquetIntervalOperator>>(
+		    writer, schema_idx, std::move(schema_path), max_repeat, max_define, can_have_nulls);
 	case LogicalTypeId::ENUM:
 		return make_uniq<EnumColumnWriter>(writer, type, schema_idx, std::move(schema_path), max_repeat, max_define,
 		                                   can_have_nulls);
@@ -2419,4 +2410,73 @@ unique_ptr<ColumnWriter> ColumnWriter::CreateWriterRecursive(ClientContext &cont
 	}
 }
 
+template <>
+struct NumericLimits<float_na_equal> {
+	static constexpr float Minimum() {
+		return std::numeric_limits<float>::lowest();
+	};
+	static constexpr float Maximum() {
+		return std::numeric_limits<float>::max();
+	};
+	static constexpr bool IsSigned() {
+		return std::is_signed<float>::value;
+	}
+	static constexpr bool IsIntegral() {
+		return std::is_integral<float>::value;
+	}
+};
+
+template <>
+struct NumericLimits<double_na_equal> {
+	static constexpr double Minimum() {
+		return std::numeric_limits<double>::lowest();
+	};
+	static constexpr double Maximum() {
+		return std::numeric_limits<double>::max();
+	};
+	static constexpr bool IsSigned() {
+		return std::is_signed<double>::value;
+	}
+	static constexpr bool IsIntegral() {
+		return std::is_integral<double>::value;
+	}
+};
+
 } // namespace duckdb
+
+namespace std {
+template <>
+struct hash<duckdb::ParquetIntervalTargetType> {
+	size_t operator()(const duckdb::ParquetIntervalTargetType &val) const {
+		return duckdb::Hash(duckdb::const_char_ptr_cast(val.bytes),
+		                    duckdb::ParquetIntervalTargetType::PARQUET_INTERVAL_SIZE);
+	}
+};
+
+template <>
+struct hash<duckdb::ParquetUUIDTargetType> {
+	size_t operator()(const duckdb::ParquetUUIDTargetType &val) const {
+		return duckdb::Hash(duckdb::const_char_ptr_cast(val.bytes), duckdb::ParquetUUIDTargetType::PARQUET_UUID_SIZE);
+	}
+};
+
+template <>
+struct hash<duckdb::float_na_equal> {
+	size_t operator()(const duckdb::float_na_equal &val) const {
+		if (std::isnan(val.val)) {
+			return duckdb::Hash<float>(std::numeric_limits<float>::quiet_NaN());
+		}
+		return duckdb::Hash<float>(val.val);
+	}
+};
+
+template <>
+struct hash<duckdb::double_na_equal> {
+	inline size_t operator()(const duckdb::double_na_equal &val) const {
+		if (std::isnan(val.val)) {
+			return duckdb::Hash<double>(std::numeric_limits<double>::quiet_NaN());
+		}
+		return duckdb::Hash<double>(val.val);
+	}
+};
+} // namespace std
diff --git a/extension/parquet/include/column_writer.hpp b/extension/parquet/include/column_writer.hpp
index 619370b6df81..b27254f741ec 100644
--- a/extension/parquet/include/column_writer.hpp
+++ b/extension/parquet/include/column_writer.hpp
@@ -17,6 +17,8 @@ class ParquetWriter;
 class ColumnWriterPageState;
 class BasicColumnWriterState;
 struct ChildFieldIDs;
+class ResizeableBuffer;
+class ParquetBloomFilter;
 
 class ColumnWriterState {
 public:
diff --git a/extension/parquet/include/parquet_dbp_encoder.hpp b/extension/parquet/include/parquet_dbp_encoder.hpp
index 36cffb1a1b88..791d10e08bb3 100644
--- a/extension/parquet/include/parquet_dbp_encoder.hpp
+++ b/extension/parquet/include/parquet_dbp_encoder.hpp
@@ -23,7 +23,7 @@ class DbpEncoder {
 	}
 
 public:
-	void BeginWrite(WriteStream &writer, const int64_t first_value) {
+	void BeginWrite(WriteStream &writer, const int64_t &first_value) {
 		// <block size in values> <number of miniblocks in a block> <total value count> <first value>
 
 		// the block size is a multiple of 128; it is stored as a ULEB128 int
diff --git a/extension/parquet/include/parquet_metadata.hpp b/extension/parquet/include/parquet_metadata.hpp
index 2310e1de80e2..09ecd5afa5e0 100644
--- a/extension/parquet/include/parquet_metadata.hpp
+++ b/extension/parquet/include/parquet_metadata.hpp
@@ -33,4 +33,9 @@ class ParquetFileMetadataFunction : public TableFunction {
 	ParquetFileMetadataFunction();
 };
 
+class ParquetBloomProbeFunction : public TableFunction {
+public:
+	ParquetBloomProbeFunction();
+};
+
 } // namespace duckdb
diff --git a/extension/parquet/include/parquet_statistics.hpp b/extension/parquet/include/parquet_statistics.hpp
index 43c47223ff1f..d4e0a4a4d0be 100644
--- a/extension/parquet/include/parquet_statistics.hpp
+++ b/extension/parquet/include/parquet_statistics.hpp
@@ -21,6 +21,7 @@ using duckdb_parquet::SchemaElement;
 
 struct LogicalType;
 class ColumnReader;
+class ResizeableBuffer;
 
 struct ParquetStatisticsUtils {
 
@@ -29,6 +30,77 @@ struct ParquetStatisticsUtils {
 
 	static Value ConvertValue(const LogicalType &type, const duckdb_parquet::SchemaElement &schema_ele,
 	                          const std::string &stats);
+
+	static bool BloomFilterSupported(const LogicalTypeId &type_id);
+
+	static bool BloomFilterExcludes(const TableFilter &filter, const duckdb_parquet::ColumnMetaData &column_meta_data,
+	                                duckdb_apache::thrift::protocol::TProtocol &file_proto, Allocator &allocator);
+};
+
+class ParquetBloomFilter {
+	static constexpr const idx_t DEFAULT_BLOCK_COUNT = 32; // 4k filter
+
+public:
+	ParquetBloomFilter(idx_t num_entries, double bloom_filter_false_positive_ratio);
+	ParquetBloomFilter(unique_ptr<ResizeableBuffer> data_p);
+	void FilterInsert(uint64_t x);
+	bool FilterCheck(uint64_t x);
+	void Shrink(idx_t new_block_count);
+	double OneRatio();
+	ResizeableBuffer *Get();
+
+private:
+	unique_ptr<ResizeableBuffer> data;
+	idx_t block_count;
+};
+
+// see https://github.com/apache/parquet-format/blob/master/BloomFilter.md
+
+struct ParquetBloomBlock {
+	struct ParquetBloomMaskResult {
+		uint8_t bit_set[8] = {0};
+	};
+
+	uint32_t block[8] = {0};
+
+	static bool check_bit(uint32_t &x, const uint8_t i) {
+		D_ASSERT(i < 32);
+		return (x >> i) & (uint32_t)1;
+	}
+
+	static void set_bit(uint32_t &x, const uint8_t i) {
+		D_ASSERT(i < 32);
+		x |= (uint32_t)1 << i;
+		D_ASSERT(check_bit(x, i));
+	}
+
+	static ParquetBloomMaskResult Mask(uint32_t x) {
+		static const uint32_t parquet_bloom_salt[8] = {0x47b6137bU, 0x44974d91U, 0x8824ad5bU, 0xa2b7289dU,
+		                                               0x705495c7U, 0x2df1424bU, 0x9efc4947U, 0x5c6bfb31U};
+		ParquetBloomMaskResult result;
+		for (idx_t i = 0; i < 8; i++) {
+			result.bit_set[i] = (x * parquet_bloom_salt[i]) >> 27;
+		}
+		return result;
+	}
+
+	static void BlockInsert(ParquetBloomBlock &b, uint32_t x) {
+		auto masked = Mask(x);
+		for (idx_t i = 0; i < 8; i++) {
+			set_bit(b.block[i], masked.bit_set[i]);
+			D_ASSERT(check_bit(b.block[i], masked.bit_set[i]));
+		}
+	}
+
+	static bool BlockCheck(ParquetBloomBlock &b, uint32_t x) {
+		auto masked = Mask(x);
+		for (idx_t i = 0; i < 8; i++) {
+			if (!check_bit(b.block[i], masked.bit_set[i])) {
+				return false;
+			}
+		}
+		return true;
+	}
 };
 
 } // namespace duckdb
diff --git a/extension/parquet/include/parquet_writer.hpp b/extension/parquet/include/parquet_writer.hpp
index 7eb5bc85c277..e601926b2690 100644
--- a/extension/parquet/include/parquet_writer.hpp
+++ b/extension/parquet/include/parquet_writer.hpp
@@ -19,6 +19,7 @@
 #include "duckdb/function/copy_function.hpp"
 #endif
 
+#include "parquet_statistics.hpp"
 #include "column_writer.hpp"
 #include "parquet_types.h"
 #include "geo_parquet.hpp"
@@ -61,13 +62,19 @@ struct FieldID {
 	static FieldID Deserialize(Deserializer &source);
 };
 
+struct ParquetBloomFilterEntry {
+	unique_ptr<ParquetBloomFilter> bloom_filter;
+	idx_t row_group_idx;
+	idx_t column_idx;
+};
+
 class ParquetWriter {
 public:
 	ParquetWriter(ClientContext &context, FileSystem &fs, string file_name, vector<LogicalType> types,
 	              vector<string> names, duckdb_parquet::CompressionCodec::type codec, ChildFieldIDs field_ids,
 	              const vector<pair<string, string>> &kv_metadata,
-	              shared_ptr<ParquetEncryptionConfig> encryption_config, double dictionary_compression_ratio_threshold,
-	              int64_t compression_level, bool debug_use_openssl);
+	              shared_ptr<ParquetEncryptionConfig> encryption_config, idx_t dictionary_size_limit,
+	              double bloom_filter_false_positive_ratio, int64_t compression_level, bool debug_use_openssl);
 
 public:
 	void PrepareRowGroup(ColumnDataCollection &buffer, PreparedRowGroup &result);
@@ -97,8 +104,11 @@ class ParquetWriter {
 		lock_guard<mutex> glock(lock);
 		return writer->total_written;
 	}
-	double DictionaryCompressionRatioThreshold() const {
-		return dictionary_compression_ratio_threshold;
+	idx_t DictionarySizeLimit() const {
+		return dictionary_size_limit;
+	}
+	double BloomFilterFalsePositiveRatio() const {
+		return bloom_filter_false_positive_ratio;
 	}
 	int64_t CompressionLevel() const {
 		return compression_level;
@@ -116,6 +126,8 @@ class ParquetWriter {
 	static bool TryGetParquetType(const LogicalType &duckdb_type,
 	                              optional_ptr<duckdb_parquet::Type::type> type = nullptr);
 
+	void BufferBloomFilter(idx_t col_idx, unique_ptr<ParquetBloomFilter> bloom_filter);
+
 private:
 	string file_name;
 	vector<LogicalType> sql_types;
@@ -123,7 +135,8 @@ class ParquetWriter {
 	duckdb_parquet::CompressionCodec::type codec;
 	ChildFieldIDs field_ids;
 	shared_ptr<ParquetEncryptionConfig> encryption_config;
-	double dictionary_compression_ratio_threshold;
+	idx_t dictionary_size_limit;
+	double bloom_filter_false_positive_ratio;
 	int64_t compression_level;
 	bool debug_use_openssl;
 	shared_ptr<EncryptionUtil> encryption_util;
@@ -136,6 +149,7 @@ class ParquetWriter {
 	vector<unique_ptr<ColumnWriter>> column_writers;
 
 	unique_ptr<GeoParquetFileMetadata> geoparquet_data;
+	vector<ParquetBloomFilterEntry> bloom_filters;
 };
 
 } // namespace duckdb
diff --git a/extension/parquet/parquet_extension.cpp b/extension/parquet/parquet_extension.cpp
index ab1ddb8c2170..9042c2216fb7 100644
--- a/extension/parquet/parquet_extension.cpp
+++ b/extension/parquet/parquet_extension.cpp
@@ -192,8 +192,11 @@ struct ParquetWriteBindData : public TableFunctionData {
 	shared_ptr<ParquetEncryptionConfig> encryption_config;
 	bool debug_use_openssl = true;
 
-	//! Dictionary compression is applied only if the compression ratio exceeds this threshold
-	double dictionary_compression_ratio_threshold = 1.0;
+	//! After how many distinct values should we abandon dictionary compression and bloom filters?
+	idx_t dictionary_size_limit = row_group_size / 100;
+
+	//! What false positive rate are we willing to accept for bloom filters
+	double bloom_filter_false_positive_ratio = 0.01;
 
 	//! After how many row groups to rotate to a new file
 	optional_idx row_groups_per_file;
@@ -1278,14 +1281,19 @@ unique_ptr<FunctionData> ParquetWriteBind(ClientContext &context, CopyFunctionBi
 		} else if (loption == "encryption_config") {
 			bind_data->encryption_config = ParquetEncryptionConfig::Create(context, option.second[0]);
 		} else if (loption == "dictionary_compression_ratio_threshold") {
+			// deprecated, ignore setting
+		} else if (loption == "dictionary_size_limit") {
+			auto val = option.second[0].GetValue<int64_t>();
+			if (val < 0) {
+				throw BinderException("dictionary_size_limit must be greater than 0 or 0 to disable");
+			}
+			bind_data->dictionary_size_limit = val;
+		} else if (loption == "bloom_filter_false_positive_ratio") {
 			auto val = option.second[0].GetValue<double>();
-			if (val == -1) {
-				val = NumericLimits<double>::Maximum();
-			} else if (val < 0) {
-				throw BinderException("dictionary_compression_ratio_threshold must be greater than 0, or -1 to disable "
-				                      "dictionary compression");
+			if (val <= 0) {
+				throw BinderException("bloom_filter_false_positive_ratio must be greater than 0");
 			}
-			bind_data->dictionary_compression_ratio_threshold = val;
+			bind_data->bloom_filter_false_positive_ratio = val;
 		} else if (loption == "debug_use_openssl") {
 			auto val = StringUtil::Lower(option.second[0].GetValue<std::string>());
 			if (val == "false") {
@@ -1333,11 +1341,11 @@ unique_ptr<GlobalFunctionData> ParquetWriteInitializeGlobal(ClientContext &conte
 	auto &parquet_bind = bind_data.Cast<ParquetWriteBindData>();
 
 	auto &fs = FileSystem::GetFileSystem(context);
-	global_state->writer =
-	    make_uniq<ParquetWriter>(context, fs, file_path, parquet_bind.sql_types, parquet_bind.column_names,
-	                             parquet_bind.codec, parquet_bind.field_ids.Copy(), parquet_bind.kv_metadata,
-	                             parquet_bind.encryption_config, parquet_bind.dictionary_compression_ratio_threshold,
-	                             parquet_bind.compression_level, parquet_bind.debug_use_openssl);
+	global_state->writer = make_uniq<ParquetWriter>(
+	    context, fs, file_path, parquet_bind.sql_types, parquet_bind.column_names, parquet_bind.codec,
+	    parquet_bind.field_ids.Copy(), parquet_bind.kv_metadata, parquet_bind.encryption_config,
+	    parquet_bind.dictionary_size_limit, parquet_bind.bloom_filter_false_positive_ratio,
+	    parquet_bind.compression_level, parquet_bind.debug_use_openssl);
 	return std::move(global_state);
 }
 
@@ -1471,13 +1479,15 @@ static void ParquetCopySerialize(Serializer &serializer, const FunctionData &bin
 	serializer.WriteProperty(106, "field_ids", bind_data.field_ids);
 	serializer.WritePropertyWithDefault<shared_ptr<ParquetEncryptionConfig>>(107, "encryption_config",
 	                                                                         bind_data.encryption_config, nullptr);
-	serializer.WriteProperty(108, "dictionary_compression_ratio_threshold",
-	                         bind_data.dictionary_compression_ratio_threshold);
+
+	// 108 was dictionary_compression_ratio_threshold, but was deleted
 	const auto compression_level = SerializeCompressionLevel(bind_data.compression_level);
 	D_ASSERT(DeserializeCompressionLevel(compression_level) == bind_data.compression_level);
 	serializer.WritePropertyWithDefault<optional_idx>(109, "compression_level", compression_level);
 	serializer.WriteProperty(110, "row_groups_per_file", bind_data.row_groups_per_file);
 	serializer.WriteProperty(111, "debug_use_openssl", bind_data.debug_use_openssl);
+	serializer.WriteProperty(112, "dictionary_size_limit", bind_data.dictionary_size_limit);
+	serializer.WriteProperty(113, "bloom_filter_false_positive_ratio", bind_data.bloom_filter_false_positive_ratio);
 }
 
 static unique_ptr<FunctionData> ParquetCopyDeserialize(Deserializer &deserializer, CopyFunction &function) {
@@ -1491,8 +1501,8 @@ static unique_ptr<FunctionData> ParquetCopyDeserialize(Deserializer &deserialize
 	data->field_ids = deserializer.ReadProperty<ChildFieldIDs>(106, "field_ids");
 	deserializer.ReadPropertyWithExplicitDefault<shared_ptr<ParquetEncryptionConfig>>(107, "encryption_config",
 	                                                                                  data->encryption_config, nullptr);
-	deserializer.ReadPropertyWithExplicitDefault<double>(108, "dictionary_compression_ratio_threshold",
-	                                                     data->dictionary_compression_ratio_threshold, 1.0);
+	deserializer.ReadDeletedProperty<double>(108, "dictionary_compression_ratio_threshold");
+
 	optional_idx compression_level;
 	deserializer.ReadPropertyWithDefault<optional_idx>(109, "compression_level", compression_level);
 	data->compression_level = DeserializeCompressionLevel(compression_level);
@@ -1500,6 +1510,11 @@ static unique_ptr<FunctionData> ParquetCopyDeserialize(Deserializer &deserialize
 	data->row_groups_per_file =
 	    deserializer.ReadPropertyWithExplicitDefault<optional_idx>(110, "row_groups_per_file", optional_idx::Invalid());
 	data->debug_use_openssl = deserializer.ReadPropertyWithExplicitDefault<bool>(111, "debug_use_openssl", true);
+	data->dictionary_size_limit =
+	    deserializer.ReadPropertyWithExplicitDefault<idx_t>(112, "dictionary_size_limit", data->row_group_size / 10);
+	data->bloom_filter_false_positive_ratio =
+	    deserializer.ReadPropertyWithExplicitDefault<double>(113, "bloom_filter_false_positive_ratio", 0.01);
+
 	return std::move(data);
 }
 // LCOV_EXCL_STOP
@@ -1701,6 +1716,10 @@ void ParquetExtension::Load(DuckDB &db) {
 	ParquetFileMetadataFunction file_meta_fun;
 	ExtensionUtil::RegisterFunction(db_instance, MultiFileReader::CreateFunctionSet(file_meta_fun));
 
+	// parquet_bloom_probe
+	ParquetBloomProbeFunction bloom_probe_fun;
+	ExtensionUtil::RegisterFunction(db_instance, MultiFileReader::CreateFunctionSet(bloom_probe_fun));
+
 	CopyFunction function("parquet");
 	function.copy_to_select = ParquetWriteSelect;
 	function.copy_to_bind = ParquetWriteBind;
diff --git a/extension/parquet/parquet_metadata.cpp b/extension/parquet/parquet_metadata.cpp
index 2ddab818b1dc..1c3b9b2f9773 100644
--- a/extension/parquet/parquet_metadata.cpp
+++ b/extension/parquet/parquet_metadata.cpp
@@ -8,6 +8,7 @@
 #include "duckdb/common/multi_file_reader.hpp"
 #include "duckdb/common/types/blob.hpp"
 #include "duckdb/common/types/column/column_data_collection.hpp"
+#include "duckdb/planner/filter/constant_filter.hpp"
 #include "duckdb/main/config.hpp"
 #endif
 
@@ -19,7 +20,18 @@ struct ParquetMetaDataBindData : public TableFunctionData {
 	unique_ptr<MultiFileReader> multi_file_reader;
 };
 
-enum class ParquetMetadataOperatorType : uint8_t { META_DATA, SCHEMA, KEY_VALUE_META_DATA, FILE_META_DATA };
+struct ParquetBloomProbeBindData : public ParquetMetaDataBindData {
+	string probe_column_name;
+	Value probe_constant;
+};
+
+enum class ParquetMetadataOperatorType : uint8_t {
+	META_DATA,
+	SCHEMA,
+	KEY_VALUE_META_DATA,
+	FILE_META_DATA,
+	BLOOM_PROBE
+};
 
 struct ParquetMetaDataOperatorData : public GlobalTableFunctionState {
 	explicit ParquetMetaDataOperatorData(ClientContext &context, const vector<LogicalType> &types)
@@ -37,11 +49,14 @@ struct ParquetMetaDataOperatorData : public GlobalTableFunctionState {
 	static void BindSchema(vector<LogicalType> &return_types, vector<string> &names);
 	static void BindKeyValueMetaData(vector<LogicalType> &return_types, vector<string> &names);
 	static void BindFileMetaData(vector<LogicalType> &return_types, vector<string> &names);
+	static void BindBloomProbe(vector<LogicalType> &return_types, vector<string> &names);
 
 	void LoadRowGroupMetadata(ClientContext &context, const vector<LogicalType> &return_types, const string &file_path);
 	void LoadSchemaData(ClientContext &context, const vector<LogicalType> &return_types, const string &file_path);
 	void LoadKeyValueMetaData(ClientContext &context, const vector<LogicalType> &return_types, const string &file_path);
 	void LoadFileMetaData(ClientContext &context, const vector<LogicalType> &return_types, const string &file_path);
+	void ExecuteBloomProbe(ClientContext &context, const vector<LogicalType> &return_types, const string &file_path,
+	                       const string &column_name, const Value &probe);
 };
 
 template <class T>
@@ -164,6 +179,12 @@ void ParquetMetaDataOperatorData::BindMetaData(vector<LogicalType> &return_types
 
 	names.emplace_back("key_value_metadata");
 	return_types.emplace_back(LogicalType::MAP(LogicalType::BLOB, LogicalType::BLOB));
+
+	names.emplace_back("bloom_filter_offset");
+	return_types.emplace_back(LogicalType::BIGINT);
+
+	names.emplace_back("bloom_filter_length");
+	return_types.emplace_back(LogicalType::BIGINT);
 }
 
 Value ConvertParquetStats(const LogicalType &type, const duckdb_parquet::SchemaElement &schema_ele, bool stats_is_set,
@@ -298,6 +319,14 @@ void ParquetMetaDataOperatorData::LoadRowGroupMetadata(ClientContext &context, c
 			    23, count,
 			    Value::MAP(LogicalType::BLOB, LogicalType::BLOB, std::move(map_keys), std::move(map_values)));
 
+			// bloom_filter_offset, LogicalType::BIGINT
+			current_chunk.SetValue(
+			    24, count, ParquetElementBigint(col_meta.bloom_filter_offset, col_meta.__isset.bloom_filter_offset));
+
+			// bloom_filter_length, LogicalType::BIGINT
+			current_chunk.SetValue(
+			    25, count, ParquetElementBigint(col_meta.bloom_filter_length, col_meta.__isset.bloom_filter_length));
+
 			count++;
 			if (count >= STANDARD_VECTOR_SIZE) {
 				current_chunk.SetCardinality(count);
@@ -560,12 +589,84 @@ void ParquetMetaDataOperatorData::LoadFileMetaData(ClientContext &context, const
 	collection.InitializeScan(scan_state);
 }
 
+//===--------------------------------------------------------------------===//
+// Bloom Probe
+//===--------------------------------------------------------------------===//
+void ParquetMetaDataOperatorData::BindBloomProbe(vector<LogicalType> &return_types, vector<string> &names) {
+	names.emplace_back("file_name");
+	return_types.emplace_back(LogicalType::VARCHAR);
+
+	names.emplace_back("row_group_id");
+	return_types.emplace_back(LogicalType::BIGINT);
+
+	names.emplace_back("bloom_filter_excludes");
+	return_types.emplace_back(LogicalType::BOOLEAN);
+}
+
+void ParquetMetaDataOperatorData::ExecuteBloomProbe(ClientContext &context, const vector<LogicalType> &return_types,
+                                                    const string &file_path, const string &column_name,
+                                                    const Value &probe) {
+	collection.Reset();
+	ParquetOptions parquet_options(context);
+	auto reader = make_uniq<ParquetReader>(context, file_path, parquet_options);
+	idx_t count = 0;
+	DataChunk current_chunk;
+	current_chunk.Initialize(context, return_types);
+	auto meta_data = reader->GetFileMetadata();
+
+	optional_idx probe_column_idx;
+	for (idx_t column_idx = 0; column_idx < reader->names.size(); column_idx++) {
+		if (reader->names[column_idx] == column_name) {
+			probe_column_idx = column_idx;
+		}
+	}
+
+	if (!probe_column_idx.IsValid()) {
+		throw InvalidInputException("Column %s not found in %s", column_name, file_path);
+	}
+
+	auto &allocator = Allocator::DefaultAllocator();
+	auto transport = std::make_shared<ThriftFileTransport>(allocator, reader->GetHandle(), false);
+	auto protocol =
+	    make_uniq<duckdb_apache::thrift::protocol::TCompactProtocolT<ThriftFileTransport>>(std::move(transport));
+
+	D_ASSERT(!probe.IsNull());
+	ConstantFilter filter(ExpressionType::COMPARE_EQUAL,
+	                      probe.CastAs(context, reader->GetTypes()[probe_column_idx.GetIndex()]));
+
+	for (idx_t row_group_idx = 0; row_group_idx < meta_data->row_groups.size(); row_group_idx++) {
+		auto &row_group = meta_data->row_groups[row_group_idx];
+		auto &column = row_group.columns[probe_column_idx.GetIndex()];
+
+		auto bloom_excludes =
+		    ParquetStatisticsUtils::BloomFilterExcludes(filter, column.meta_data, *protocol, allocator);
+		current_chunk.SetValue(0, count, Value(file_path));
+		current_chunk.SetValue(1, count, Value::BIGINT(row_group_idx));
+		current_chunk.SetValue(2, count, Value::BOOLEAN(bloom_excludes));
+
+		count++;
+		if (count >= STANDARD_VECTOR_SIZE) {
+			current_chunk.SetCardinality(count);
+			collection.Append(current_chunk);
+
+			count = 0;
+			current_chunk.Reset();
+		}
+	}
+
+	current_chunk.SetCardinality(count);
+	collection.Append(current_chunk);
+	collection.InitializeScan(scan_state);
+}
+
 //===--------------------------------------------------------------------===//
 // Bind
 //===--------------------------------------------------------------------===//
 template <ParquetMetadataOperatorType TYPE>
 unique_ptr<FunctionData> ParquetMetaDataBind(ClientContext &context, TableFunctionBindInput &input,
                                              vector<LogicalType> &return_types, vector<string> &names) {
+	auto result = make_uniq<ParquetMetaDataBindData>();
+
 	switch (TYPE) {
 	case ParquetMetadataOperatorType::SCHEMA:
 		ParquetMetaDataOperatorData::BindSchema(return_types, names);
@@ -579,14 +680,26 @@ unique_ptr<FunctionData> ParquetMetaDataBind(ClientContext &context, TableFuncti
 	case ParquetMetadataOperatorType::FILE_META_DATA:
 		ParquetMetaDataOperatorData::BindFileMetaData(return_types, names);
 		break;
+	case ParquetMetadataOperatorType::BLOOM_PROBE: {
+		auto probe_bind_data = make_uniq<ParquetBloomProbeBindData>();
+		D_ASSERT(input.inputs.size() == 3);
+		if (input.inputs[1].IsNull() || input.inputs[2].IsNull()) {
+			throw InvalidInputException("Can't have NULL parameters for parquet_bloom_probe");
+		}
+		probe_bind_data->probe_column_name = input.inputs[1].CastAs(context, LogicalType::VARCHAR).GetValue<string>();
+		probe_bind_data->probe_constant = input.inputs[2];
+		result = std::move(probe_bind_data);
+		ParquetMetaDataOperatorData::BindBloomProbe(return_types, names);
+		break;
+	}
 	default:
 		throw InternalException("Unsupported ParquetMetadataOperatorType");
 	}
 
-	auto result = make_uniq<ParquetMetaDataBindData>();
 	result->return_types = return_types;
 	result->multi_file_reader = MultiFileReader::Create(input.table_function);
 	result->file_list = result->multi_file_reader->CreateFileList(context, input.inputs[0]);
+
 	return std::move(result);
 }
 
@@ -614,6 +727,12 @@ unique_ptr<GlobalTableFunctionState> ParquetMetaDataInit(ClientContext &context,
 	case ParquetMetadataOperatorType::FILE_META_DATA:
 		result->LoadFileMetaData(context, bind_data.return_types, bind_data.file_list->GetFirstFile());
 		break;
+	case ParquetMetadataOperatorType::BLOOM_PROBE: {
+		auto &bloom_probe_bind_data = input.bind_data->Cast<ParquetBloomProbeBindData>();
+		result->ExecuteBloomProbe(context, bind_data.return_types, bind_data.file_list->GetFirstFile(),
+		                          bloom_probe_bind_data.probe_column_name, bloom_probe_bind_data.probe_constant);
+		break;
+	}
 	default:
 		throw InternalException("Unsupported ParquetMetadataOperatorType");
 	}
@@ -647,6 +766,12 @@ void ParquetMetaDataImplementation(ClientContext &context, TableFunctionInput &d
 			case ParquetMetadataOperatorType::FILE_META_DATA:
 				data.LoadFileMetaData(context, bind_data.return_types, data.current_file);
 				break;
+			case ParquetMetadataOperatorType::BLOOM_PROBE: {
+				auto &bloom_probe_bind_data = data_p.bind_data->Cast<ParquetBloomProbeBindData>();
+				data.ExecuteBloomProbe(context, bind_data.return_types, bind_data.file_list->GetFirstFile(),
+				                       bloom_probe_bind_data.probe_column_name, bloom_probe_bind_data.probe_constant);
+				break;
+			}
 			default:
 				throw InternalException("Unsupported ParquetMetadataOperatorType");
 			}
@@ -686,4 +811,11 @@ ParquetFileMetadataFunction::ParquetFileMetadataFunction()
                     ParquetMetaDataInit<ParquetMetadataOperatorType::FILE_META_DATA>) {
 }
 
+ParquetBloomProbeFunction::ParquetBloomProbeFunction()
+    : TableFunction("parquet_bloom_probe", {LogicalType::VARCHAR, LogicalType::VARCHAR, LogicalType::ANY},
+                    ParquetMetaDataImplementation<ParquetMetadataOperatorType::BLOOM_PROBE>,
+                    ParquetMetaDataBind<ParquetMetadataOperatorType::BLOOM_PROBE>,
+                    ParquetMetaDataInit<ParquetMetadataOperatorType::BLOOM_PROBE>) {
+}
+
 } // namespace duckdb
diff --git a/extension/parquet/parquet_reader.cpp b/extension/parquet/parquet_reader.cpp
index 6bc229a8f3c3..840699cb73d2 100644
--- a/extension/parquet/parquet_reader.cpp
+++ b/extension/parquet/parquet_reader.cpp
@@ -750,14 +750,22 @@ void ParquetReader::PrepareRowGroupBuffer(ParquetReaderScanState &state, idx_t c
 		// filters contain output chunk index, not file col idx!
 		auto global_id = reader_data.column_mapping[col_idx];
 		auto filter_entry = reader_data.filters->filters.find(global_id);
+
 		if (stats && filter_entry != reader_data.filters->filters.end()) {
-			bool skip_chunk = false;
 			auto &filter = *filter_entry->second;
 
 			FilterPropagateResult prune_result;
-			if (column_reader.Type().id() == LogicalTypeId::VARCHAR &&
-			    group.columns[column_reader.FileIdx()].meta_data.statistics.__isset.min_value &&
-			    group.columns[column_reader.FileIdx()].meta_data.statistics.__isset.max_value) {
+			// TODO we might not have stats but STILL a bloom filter so move this up
+			// check the bloom filter if present
+			if (!column_reader.Type().IsNested() &&
+			    ParquetStatisticsUtils::BloomFilterSupported(column_reader.Type().id()) &&
+			    ParquetStatisticsUtils::BloomFilterExcludes(filter, group.columns[column_reader.FileIdx()].meta_data,
+			                                                *state.thrift_file_proto, allocator)) {
+				prune_result = FilterPropagateResult::FILTER_ALWAYS_FALSE;
+			} else if (column_reader.Type().id() == LogicalTypeId::VARCHAR &&
+			           group.columns[column_reader.FileIdx()].meta_data.statistics.__isset.min_value &&
+			           group.columns[column_reader.FileIdx()].meta_data.statistics.__isset.max_value) {
+
 				// our StringStats only store the first 8 bytes of strings (even if Parquet has longer string stats)
 				// however, when reading remote Parquet files, skipping row groups is really important
 				// here, we implement a special case to check the full length for string filters
@@ -784,9 +792,6 @@ void ParquetReader::PrepareRowGroupBuffer(ParquetReaderScanState &state, idx_t c
 			}
 
 			if (prune_result == FilterPropagateResult::FILTER_ALWAYS_FALSE) {
-				skip_chunk = true;
-			}
-			if (skip_chunk) {
 				// this effectively will skip this chunk
 				state.group_offset = group.num_rows;
 				return;
diff --git a/extension/parquet/parquet_statistics.cpp b/extension/parquet/parquet_statistics.cpp
index ad6b2ff7df08..d40d2afbfcd8 100644
--- a/extension/parquet/parquet_statistics.cpp
+++ b/extension/parquet/parquet_statistics.cpp
@@ -5,11 +5,14 @@
 #include "parquet_timestamp.hpp"
 #include "string_column_reader.hpp"
 #include "struct_column_reader.hpp"
+#include "zstd/common/xxhash.hpp"
+
 #ifndef DUCKDB_AMALGAMATION
 #include "duckdb/common/types/blob.hpp"
 #include "duckdb/common/types/time.hpp"
 #include "duckdb/common/types/value.hpp"
 #include "duckdb/storage/statistics/struct_stats.hpp"
+#include "duckdb/planner/filter/constant_filter.hpp"
 #endif
 
 namespace duckdb {
@@ -51,7 +54,7 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 	switch (type.id()) {
 	case LogicalTypeId::BOOLEAN: {
 		if (stats.size() != sizeof(bool)) {
-			throw InternalException("Incorrect stats size for type BOOLEAN");
+			throw InvalidInputException("Incorrect stats size for type BOOLEAN");
 		}
 		return Value::BOOLEAN(Load<bool>(stats_data));
 	}
@@ -59,29 +62,29 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 	case LogicalTypeId::USMALLINT:
 	case LogicalTypeId::UINTEGER:
 		if (stats.size() != sizeof(uint32_t)) {
-			throw InternalException("Incorrect stats size for type UINTEGER");
+			throw InvalidInputException("Incorrect stats size for type UINTEGER");
 		}
 		return Value::UINTEGER(Load<uint32_t>(stats_data));
 	case LogicalTypeId::UBIGINT:
 		if (stats.size() != sizeof(uint64_t)) {
-			throw InternalException("Incorrect stats size for type UBIGINT");
+			throw InvalidInputException("Incorrect stats size for type UBIGINT");
 		}
 		return Value::UBIGINT(Load<uint64_t>(stats_data));
 	case LogicalTypeId::TINYINT:
 	case LogicalTypeId::SMALLINT:
 	case LogicalTypeId::INTEGER:
 		if (stats.size() != sizeof(int32_t)) {
-			throw InternalException("Incorrect stats size for type INTEGER");
+			throw InvalidInputException("Incorrect stats size for type INTEGER");
 		}
 		return Value::INTEGER(Load<int32_t>(stats_data));
 	case LogicalTypeId::BIGINT:
 		if (stats.size() != sizeof(int64_t)) {
-			throw InternalException("Incorrect stats size for type BIGINT");
+			throw InvalidInputException("Incorrect stats size for type BIGINT");
 		}
 		return Value::BIGINT(Load<int64_t>(stats_data));
 	case LogicalTypeId::FLOAT: {
 		if (stats.size() != sizeof(float)) {
-			throw InternalException("Incorrect stats size for type FLOAT");
+			throw InvalidInputException("Incorrect stats size for type FLOAT");
 		}
 		auto val = Load<float>(stats_data);
 		if (!Value::FloatIsFinite(val)) {
@@ -99,7 +102,7 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 			break;
 		}
 		if (stats.size() != sizeof(double)) {
-			throw InternalException("Incorrect stats size for type DOUBLE");
+			throw InvalidInputException("Incorrect stats size for type DOUBLE");
 		}
 		auto val = Load<double>(stats_data);
 		if (!Value::DoubleIsFinite(val)) {
@@ -113,13 +116,13 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 		switch (schema_ele.type) {
 		case Type::INT32: {
 			if (stats.size() != sizeof(int32_t)) {
-				throw InternalException("Incorrect stats size for type %s", type.ToString());
+				throw InvalidInputException("Incorrect stats size for type %s", type.ToString());
 			}
 			return Value::DECIMAL(Load<int32_t>(stats_data), width, scale);
 		}
 		case Type::INT64: {
 			if (stats.size() != sizeof(int64_t)) {
-				throw InternalException("Incorrect stats size for type %s", type.ToString());
+				throw InvalidInputException("Incorrect stats size for type %s", type.ToString());
 			}
 			return Value::DECIMAL(Load<int64_t>(stats_data), width, scale);
 		}
@@ -140,7 +143,7 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 				    ParquetDecimalUtils::ReadDecimalValue<hugeint_t>(stats_data, stats.size(), schema_ele), width,
 				    scale);
 			default:
-				throw InternalException("Unsupported internal type for decimal");
+				throw InvalidInputException("Unsupported internal type for decimal");
 			}
 		default:
 			throw InternalException("Unsupported internal type for decimal?..");
@@ -155,7 +158,7 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 		}
 	case LogicalTypeId::DATE:
 		if (stats.size() != sizeof(int32_t)) {
-			throw InternalException("Incorrect stats size for type DATE");
+			throw InvalidInputException("Incorrect stats size for type DATE");
 		}
 		return Value::DATE(date_t(Load<int32_t>(stats_data)));
 	case LogicalTypeId::TIME: {
@@ -165,7 +168,7 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 		} else if (stats.size() == sizeof(int64_t)) {
 			val = Load<int64_t>(stats_data);
 		} else {
-			throw InternalException("Incorrect stats size for type TIME");
+			throw InvalidInputException("Incorrect stats size for type TIME");
 		}
 		if (schema_ele.__isset.logicalType && schema_ele.logicalType.__isset.TIME) {
 			// logical type
@@ -192,7 +195,7 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 		} else if (stats.size() == sizeof(int64_t)) {
 			val = Load<int64_t>(stats_data);
 		} else {
-			throw InternalException("Incorrect stats size for type TIMETZ");
+			throw InvalidInputException("Incorrect stats size for type TIMETZ");
 		}
 		if (schema_ele.__isset.logicalType && schema_ele.logicalType.__isset.TIME) {
 			// logical type
@@ -213,13 +216,13 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 		timestamp_t timestamp_value;
 		if (schema_ele.type == Type::INT96) {
 			if (stats.size() != sizeof(Int96)) {
-				throw InternalException("Incorrect stats size for type TIMESTAMP");
+				throw InvalidInputException("Incorrect stats size for type TIMESTAMP");
 			}
 			timestamp_value = ImpalaTimestampToTimestamp(Load<Int96>(stats_data));
 		} else {
 			D_ASSERT(schema_ele.type == Type::INT64);
 			if (stats.size() != sizeof(int64_t)) {
-				throw InternalException("Incorrect stats size for type TIMESTAMP");
+				throw InvalidInputException("Incorrect stats size for type TIMESTAMP");
 			}
 			auto val = Load<int64_t>(stats_data);
 			if (schema_ele.__isset.logicalType && schema_ele.logicalType.__isset.TIMESTAMP) {
@@ -248,13 +251,13 @@ Value ParquetStatisticsUtils::ConvertValue(const LogicalType &type, const duckdb
 		timestamp_ns_t timestamp_value;
 		if (schema_ele.type == Type::INT96) {
 			if (stats.size() != sizeof(Int96)) {
-				throw InternalException("Incorrect stats size for type TIMESTAMP_NS");
+				throw InvalidInputException("Incorrect stats size for type TIMESTAMP_NS");
 			}
 			timestamp_value = ImpalaTimestampToTimestampNS(Load<Int96>(stats_data));
 		} else {
 			D_ASSERT(schema_ele.type == Type::INT64);
 			if (stats.size() != sizeof(int64_t)) {
-				throw InternalException("Incorrect stats size for type TIMESTAMP_NS");
+				throw InvalidInputException("Incorrect stats size for type TIMESTAMP_NS");
 			}
 			auto val = Load<int64_t>(stats_data);
 			if (schema_ele.__isset.logicalType && schema_ele.logicalType.__isset.TIMESTAMP) {
@@ -388,4 +391,210 @@ unique_ptr<BaseStatistics> ParquetStatisticsUtils::TransformColumnStatistics(con
 	return row_group_stats;
 }
 
+static bool HasFilterConstants(const TableFilter &duckdb_filter) {
+	switch (duckdb_filter.filter_type) {
+	case TableFilterType::CONSTANT_COMPARISON: {
+		auto &constant_filter = duckdb_filter.Cast<ConstantFilter>();
+		return (constant_filter.comparison_type == ExpressionType::COMPARE_EQUAL && !constant_filter.constant.IsNull());
+	}
+	case TableFilterType::CONJUNCTION_AND: {
+		auto &conjunction_and_filter = duckdb_filter.Cast<ConjunctionAndFilter>();
+		bool child_has_constant = false;
+		for (auto &child_filter : conjunction_and_filter.child_filters) {
+			child_has_constant |= HasFilterConstants(*child_filter);
+		}
+		return child_has_constant;
+	}
+	case TableFilterType::CONJUNCTION_OR: {
+		auto &conjunction_or_filter = duckdb_filter.Cast<ConjunctionOrFilter>();
+		bool child_has_constant = false;
+		for (auto &child_filter : conjunction_or_filter.child_filters) {
+			child_has_constant |= HasFilterConstants(*child_filter);
+		}
+		return child_has_constant;
+	}
+	default:
+		return false;
+	}
+}
+
+template <class T>
+uint64_t ValueXH64FixedWidth(const Value &constant) {
+	T val = constant.GetValue<T>();
+	return duckdb_zstd::XXH64(&val, sizeof(val), 0);
+}
+
+// TODO we can only this if the parquet representation of the type exactly matches the duckdb rep!
+// TODO TEST THIS!
+// TODO perhaps we can re-use some writer infra here
+static uint64_t ValueXXH64(const Value &constant) {
+	switch (constant.type().InternalType()) {
+	case PhysicalType::UINT8:
+		return ValueXH64FixedWidth<int32_t>(constant);
+	case PhysicalType::INT8:
+		return ValueXH64FixedWidth<int32_t>(constant);
+	case PhysicalType::UINT16:
+		return ValueXH64FixedWidth<int32_t>(constant);
+	case PhysicalType::INT16:
+		return ValueXH64FixedWidth<int32_t>(constant);
+	case PhysicalType::UINT32:
+		return ValueXH64FixedWidth<uint32_t>(constant);
+	case PhysicalType::INT32:
+		return ValueXH64FixedWidth<int32_t>(constant);
+	case PhysicalType::UINT64:
+		return ValueXH64FixedWidth<uint64_t>(constant);
+	case PhysicalType::INT64:
+		return ValueXH64FixedWidth<int64_t>(constant);
+	case PhysicalType::FLOAT:
+		return ValueXH64FixedWidth<float>(constant);
+	case PhysicalType::DOUBLE:
+		return ValueXH64FixedWidth<double>(constant);
+	case PhysicalType::VARCHAR: {
+		auto val = constant.GetValue<string>();
+		return duckdb_zstd::XXH64(val.c_str(), val.length(), 0);
+	}
+	default:
+		return 0;
+	}
+}
+
+static bool ApplyBloomFilter(const TableFilter &duckdb_filter, ParquetBloomFilter &bloom_filter) {
+	switch (duckdb_filter.filter_type) {
+	case TableFilterType::CONSTANT_COMPARISON: {
+		auto &constant_filter = duckdb_filter.Cast<ConstantFilter>();
+		D_ASSERT(constant_filter.comparison_type == ExpressionType::COMPARE_EQUAL);
+		D_ASSERT(!constant_filter.constant.IsNull());
+		auto hash = ValueXXH64(constant_filter.constant);
+		return hash > 0 && !bloom_filter.FilterCheck(hash);
+	}
+	case TableFilterType::CONJUNCTION_AND: {
+		auto &conjunction_and_filter = duckdb_filter.Cast<ConjunctionAndFilter>();
+		bool any_children_true = false;
+		for (auto &child_filter : conjunction_and_filter.child_filters) {
+			any_children_true |= ApplyBloomFilter(*child_filter, bloom_filter);
+		}
+		return any_children_true;
+	}
+	case TableFilterType::CONJUNCTION_OR: {
+		auto &conjunction_or_filter = duckdb_filter.Cast<ConjunctionOrFilter>();
+		bool all_children_true = true;
+		for (auto &child_filter : conjunction_or_filter.child_filters) {
+			all_children_true &= ApplyBloomFilter(*child_filter, bloom_filter);
+		}
+		return all_children_true;
+	}
+	default:
+		return false;
+	}
+}
+
+bool ParquetStatisticsUtils::BloomFilterSupported(const LogicalTypeId &type_id) {
+	switch (type_id) {
+	case LogicalTypeId::TINYINT:
+	case LogicalTypeId::UTINYINT:
+	case LogicalTypeId::SMALLINT:
+	case LogicalTypeId::USMALLINT:
+	case LogicalTypeId::INTEGER:
+	case LogicalTypeId::UINTEGER:
+	case LogicalTypeId::BIGINT:
+	case LogicalTypeId::UBIGINT:
+	case LogicalTypeId::FLOAT:
+	case LogicalTypeId::DOUBLE:
+	case LogicalTypeId::VARCHAR:
+	case LogicalTypeId::BLOB:
+		return true;
+	default:
+		return false;
+	}
+}
+
+bool ParquetStatisticsUtils::BloomFilterExcludes(const TableFilter &duckdb_filter,
+                                                 const duckdb_parquet::ColumnMetaData &column_meta_data,
+                                                 TProtocol &file_proto, Allocator &allocator) {
+	if (!HasFilterConstants(duckdb_filter) || !column_meta_data.__isset.bloom_filter_offset ||
+	    column_meta_data.bloom_filter_offset <= 0) {
+		return false;
+	}
+	// TODO check length against file length!
+
+	auto &transport = reinterpret_cast<ThriftFileTransport &>(*file_proto.getTransport());
+	transport.SetLocation(column_meta_data.bloom_filter_offset);
+	if (column_meta_data.__isset.bloom_filter_length && column_meta_data.bloom_filter_length > 0) {
+		transport.Prefetch(column_meta_data.bloom_filter_offset, column_meta_data.bloom_filter_length);
+	}
+
+	duckdb_parquet::BloomFilterHeader filter_header;
+	// TODO the bloom filter could be encrypted, too, so need to double check that this is NOT the case
+	filter_header.read(&file_proto);
+	if (!filter_header.algorithm.__isset.BLOCK || !filter_header.compression.__isset.UNCOMPRESSED ||
+	    !filter_header.hash.__isset.XXHASH) {
+		return false;
+	}
+
+	auto new_buffer = make_uniq<ResizeableBuffer>(allocator, filter_header.numBytes);
+	transport.read(new_buffer->ptr, filter_header.numBytes);
+	ParquetBloomFilter bloom_filter(std::move(new_buffer));
+	return ApplyBloomFilter(duckdb_filter, bloom_filter);
+}
+
+ParquetBloomFilter::ParquetBloomFilter(idx_t num_entries, double bloom_filter_false_positive_ratio) {
+
+	// aim for hit ratio of 0.01%
+	// see http://tfk.mit.edu/pdf/bloom.pdf
+	double f = bloom_filter_false_positive_ratio;
+	double k = 8.0;
+	double n = num_entries;
+	double m = -k * n / std::log(1 - std::pow(f, 1 / k));
+	auto b = MaxValue<idx_t>(NextPowerOfTwo(m / k) / 32, 1);
+
+	D_ASSERT(b > 0 && IsPowerOfTwo(b));
+
+	data = make_uniq<ResizeableBuffer>(Allocator::DefaultAllocator(), sizeof(ParquetBloomBlock) * b);
+	data->zero();
+	block_count = data->len / sizeof(ParquetBloomBlock);
+	D_ASSERT(data->len % sizeof(ParquetBloomBlock) == 0);
+}
+
+ParquetBloomFilter::ParquetBloomFilter(unique_ptr<ResizeableBuffer> data_p) {
+	D_ASSERT(data_p->len % sizeof(ParquetBloomBlock) == 0);
+	data = std::move(data_p);
+	block_count = data->len / sizeof(ParquetBloomBlock);
+	D_ASSERT(data->len % sizeof(ParquetBloomBlock) == 0);
+}
+
+void ParquetBloomFilter::FilterInsert(uint64_t x) {
+	auto blocks = (ParquetBloomBlock *)(data->ptr);
+	uint64_t i = ((x >> 32) * block_count) >> 32;
+	auto &b = blocks[i];
+	ParquetBloomBlock::BlockInsert(b, x);
+}
+
+bool ParquetBloomFilter::FilterCheck(uint64_t x) {
+	auto blocks = (ParquetBloomBlock *)(data->ptr);
+	auto i = ((x >> 32) * block_count) >> 32;
+	return ParquetBloomBlock::BlockCheck(blocks[i], x);
+}
+
+// compiler optimizes this into a single instruction (popcnt)
+static uint8_t PopCnt64(uint64_t n) {
+	uint8_t c = 0;
+	for (; n; ++c) {
+		n &= n - 1;
+	}
+	return c;
+}
+
+double ParquetBloomFilter::OneRatio() {
+	auto bloom_ptr = (uint64_t *)data->ptr;
+	idx_t one_count = 0;
+	for (idx_t b_idx = 0; b_idx < data->len / sizeof(uint64_t); ++b_idx) {
+		one_count += PopCnt64(bloom_ptr[b_idx]);
+	}
+	return one_count / (data->len * 8.0);
+}
+
+ResizeableBuffer *ParquetBloomFilter::Get() {
+	return data.get();
+}
+
 } // namespace duckdb
diff --git a/extension/parquet/parquet_writer.cpp b/extension/parquet/parquet_writer.cpp
index 30e62310131e..7a3ddb9a2a6a 100644
--- a/extension/parquet/parquet_writer.cpp
+++ b/extension/parquet/parquet_writer.cpp
@@ -4,6 +4,7 @@
 #include "mbedtls_wrapper.hpp"
 #include "parquet_crypto.hpp"
 #include "parquet_timestamp.hpp"
+#include "resizable_buffer.hpp"
 
 #ifndef DUCKDB_AMALGAMATION
 #include "duckdb/common/file_system.hpp"
@@ -318,13 +319,15 @@ void VerifyUniqueNames(const vector<string> &names) {
 ParquetWriter::ParquetWriter(ClientContext &context, FileSystem &fs, string file_name_p, vector<LogicalType> types_p,
                              vector<string> names_p, CompressionCodec::type codec, ChildFieldIDs field_ids_p,
                              const vector<pair<string, string>> &kv_metadata,
-                             shared_ptr<ParquetEncryptionConfig> encryption_config_p,
-                             double dictionary_compression_ratio_threshold_p, int64_t compression_level_p,
+                             shared_ptr<ParquetEncryptionConfig> encryption_config_p, idx_t dictionary_size_limit_p,
+                             double bloom_filter_false_positive_ratio_p, int64_t compression_level_p,
                              bool debug_use_openssl_p)
     : file_name(std::move(file_name_p)), sql_types(std::move(types_p)), column_names(std::move(names_p)), codec(codec),
       field_ids(std::move(field_ids_p)), encryption_config(std::move(encryption_config_p)),
-      dictionary_compression_ratio_threshold(dictionary_compression_ratio_threshold_p),
-      compression_level(compression_level_p), debug_use_openssl(debug_use_openssl_p) {
+      dictionary_size_limit(dictionary_size_limit_p),
+      bloom_filter_false_positive_ratio(bloom_filter_false_positive_ratio_p), compression_level(compression_level_p),
+      debug_use_openssl(debug_use_openssl_p) {
+
 	// initialize the file writer
 	writer = make_uniq<BufferedFileWriter>(fs, file_name.c_str(),
 	                                       FileFlags::FILE_FLAGS_WRITE | FileFlags::FILE_FLAGS_FILE_CREATE_NEW);
@@ -523,7 +526,35 @@ void ParquetWriter::Flush(ColumnDataCollection &buffer) {
 }
 
 void ParquetWriter::Finalize() {
-	const auto start_offset = writer->GetTotalWritten();
+
+	// dump the bloom filters right before footer, not if stuff is encrypted
+
+	for (auto &bloom_filter_entry : bloom_filters) {
+		D_ASSERT(!encryption_config);
+		// write nonsense bloom filter header
+		duckdb_parquet::BloomFilterHeader filter_header;
+		auto bloom_filter_bytes = bloom_filter_entry.bloom_filter->Get();
+		filter_header.numBytes = bloom_filter_bytes->len;
+		filter_header.algorithm.__set_BLOCK(duckdb_parquet::SplitBlockAlgorithm());
+		filter_header.compression.__set_UNCOMPRESSED(duckdb_parquet::Uncompressed());
+		filter_header.hash.__set_XXHASH(duckdb_parquet::XxHash());
+
+		// set metadata flags
+		auto &column_chunk =
+		    file_meta_data.row_groups[bloom_filter_entry.row_group_idx].columns[bloom_filter_entry.column_idx];
+
+		column_chunk.meta_data.__isset.bloom_filter_offset = true;
+		column_chunk.meta_data.bloom_filter_offset = writer->GetTotalWritten();
+
+		auto bloom_filter_header_size = Write(filter_header);
+		// write actual data
+		WriteData(bloom_filter_bytes->ptr, bloom_filter_bytes->len);
+
+		column_chunk.meta_data.__isset.bloom_filter_length = true;
+		column_chunk.meta_data.bloom_filter_length = bloom_filter_header_size + bloom_filter_bytes->len;
+	}
+
+	const auto metadata_start_offset = writer->GetTotalWritten();
 	if (encryption_config) {
 		// Crypto metadata is written unencrypted
 		FileCryptoMetaData crypto_metadata;
@@ -541,7 +572,7 @@ void ParquetWriter::Finalize() {
 
 	Write(file_meta_data);
 
-	writer->Write<uint32_t>(writer->GetTotalWritten() - start_offset);
+	writer->Write<uint32_t>(writer->GetTotalWritten() - metadata_start_offset);
 
 	if (encryption_config) {
 		// encrypted parquet files also end with the string "PARE"
@@ -563,4 +594,15 @@ GeoParquetFileMetadata &ParquetWriter::GetGeoParquetData() {
 	return *geoparquet_data;
 }
 
+void ParquetWriter::BufferBloomFilter(idx_t col_idx, unique_ptr<ParquetBloomFilter> bloom_filter) {
+	if (encryption_config) {
+		return;
+	}
+	ParquetBloomFilterEntry new_entry;
+	new_entry.bloom_filter = std::move(bloom_filter);
+	new_entry.column_idx = col_idx;
+	new_entry.row_group_idx = file_meta_data.row_groups.size();
+	bloom_filters.push_back(std::move(new_entry));
+}
+
 } // namespace duckdb
diff --git a/src/common/multi_file_reader.cpp b/src/common/multi_file_reader.cpp
index ff91d3601dd8..e81bedcaf114 100644
--- a/src/common/multi_file_reader.cpp
+++ b/src/common/multi_file_reader.cpp
@@ -460,7 +460,7 @@ TablePartitionInfo MultiFileReader::GetPartitionInfo(ClientContext &context, con
 TableFunctionSet MultiFileReader::CreateFunctionSet(TableFunction table_function) {
 	TableFunctionSet function_set(table_function.name);
 	function_set.AddFunction(table_function);
-	D_ASSERT(table_function.arguments.size() == 1 && table_function.arguments[0] == LogicalType::VARCHAR);
+	D_ASSERT(table_function.arguments.size() >= 1 && table_function.arguments[0] == LogicalType::VARCHAR);
 	table_function.arguments[0] = LogicalType::LIST(LogicalType::VARCHAR);
 	function_set.AddFunction(std::move(table_function));
 	return function_set;
diff --git a/src/function/cast/union/from_struct.cpp b/src/function/cast/union/from_struct.cpp
index ca3aecfca0f1..c6fcd2a03c95 100644
--- a/src/function/cast/union/from_struct.cpp
+++ b/src/function/cast/union/from_struct.cpp
@@ -58,6 +58,8 @@ bool StructToUnionCast::Cast(Vector &source, Vector &result, idx_t count, CastPa
 		    cast_data.child_cast_info[i].function(source_child_vector, result_child_vector, count, child_parameters);
 		(void)converted;
 		D_ASSERT(converted);
+		// we flatten the child because we use FlatVector::SetNull below and we may get non-flat from source/cast
+		result_child_vector.Flatten(count);
 	}
 
 	if (source.GetVectorType() == VectorType::CONSTANT_VECTOR) {
@@ -68,15 +70,15 @@ bool StructToUnionCast::Cast(Vector &source, Vector &result, idx_t count, CastPa
 		auto &tag_vec = *target_children[0];
 		ConstantVector::SetNull(result, ConstantVector::IsNull(tag_vec));
 	} else {
-		source.Flatten(count);
-		FlatVector::Validity(result) = FlatVector::Validity(source);
-
 		// if the tag is NULL, the union should be NULL
 		auto &tag_vec = *target_children[0];
-		UnifiedVectorFormat tag_data;
+		UnifiedVectorFormat source_data, tag_data;
+		source.ToUnifiedFormat(count, source_data);
 		tag_vec.ToUnifiedFormat(count, tag_data);
+
 		for (idx_t i = 0; i < count; i++) {
-			if (!tag_data.validity.RowIsValid(tag_data.sel->get_index(i))) {
+			if (!source_data.validity.RowIsValid(source_data.sel->get_index(i)) ||
+			    !tag_data.validity.RowIsValid(tag_data.sel->get_index(i))) {
 				FlatVector::SetNull(result, i, true);
 			}
 		}
diff --git a/src/include/duckdb/common/hugeint.hpp b/src/include/duckdb/common/hugeint.hpp
index 0abc91e6d7da..c9b54bd95053 100644
--- a/src/include/duckdb/common/hugeint.hpp
+++ b/src/include/duckdb/common/hugeint.hpp
@@ -80,3 +80,13 @@ struct hugeint_t { // NOLINT: use numeric casing
 };
 
 } // namespace duckdb
+
+namespace std {
+template <>
+struct hash<duckdb::hugeint_t> {
+	size_t operator()(const duckdb::hugeint_t &val) const {
+		using std::hash;
+		return hash<int64_t> {}(val.upper) ^ hash<uint64_t> {}(val.lower);
+	}
+};
+} // namespace std
diff --git a/src/include/duckdb/common/types/interval.hpp b/src/include/duckdb/common/types/interval.hpp
index d8a9672b0c63..fefc19549e13 100644
--- a/src/include/duckdb/common/types/interval.hpp
+++ b/src/include/duckdb/common/types/interval.hpp
@@ -181,3 +181,17 @@ void interval_t::Normalize(int64_t &months, int64_t &days, int64_t &micros) cons
 }
 
 } // namespace duckdb
+
+namespace std {
+template <>
+struct hash<duckdb::interval_t> {
+	size_t operator()(const duckdb::interval_t &val) const {
+		int64_t months, days, micros;
+		val.Normalize(months, days, micros);
+		using std::hash;
+
+		return hash<int32_t> {}(duckdb::UnsafeNumericCast<int32_t>(days)) ^
+		       hash<int32_t> {}(duckdb::UnsafeNumericCast<int32_t>(months)) ^ hash<int64_t> {}(micros);
+	}
+};
+} // namespace std
diff --git a/src/include/duckdb/common/types/string_type.hpp b/src/include/duckdb/common/types/string_type.hpp
index 882e0625cf0c..0a88e33219e2 100644
--- a/src/include/duckdb/common/types/string_type.hpp
+++ b/src/include/duckdb/common/types/string_type.hpp
@@ -13,6 +13,7 @@
 #include "duckdb/common/helper.hpp"
 #include "duckdb/common/numeric_utils.hpp"
 #include "duckdb/common/limits.hpp"
+#include "duckdb/common/types/hash.hpp"
 
 #include <cstring>
 #include <algorithm>
@@ -234,3 +235,12 @@ struct string_t {
 };
 
 } // namespace duckdb
+
+namespace std {
+template <>
+struct hash<duckdb::string_t> {
+	size_t operator()(const duckdb::string_t &val) const {
+		return Hash(val);
+	}
+};
+} // namespace std
diff --git a/src/include/duckdb/common/uhugeint.hpp b/src/include/duckdb/common/uhugeint.hpp
index ba98df46b743..a579e37e4b2c 100644
--- a/src/include/duckdb/common/uhugeint.hpp
+++ b/src/include/duckdb/common/uhugeint.hpp
@@ -79,3 +79,13 @@ struct uhugeint_t { // NOLINT
 };
 
 } // namespace duckdb
+
+namespace std {
+template <>
+struct hash<duckdb::uhugeint_t> {
+	size_t operator()(const duckdb::uhugeint_t &val) const {
+		using std::hash;
+		return hash<uint64_t> {}(val.upper) ^ hash<uint64_t> {}(val.lower);
+	}
+};
+} // namespace std
diff --git a/src/include/duckdb/main/extension_entries.hpp b/src/include/duckdb/main/extension_entries.hpp
index f30af0b1690e..31c67f73337a 100644
--- a/src/include/duckdb/main/extension_entries.hpp
+++ b/src/include/duckdb/main/extension_entries.hpp
@@ -446,6 +446,7 @@ static constexpr ExtensionFunctionEntry EXTENSION_FUNCTIONS[] = {
     {"nextafter", "core_functions", CatalogType::SCALAR_FUNCTION_ENTRY},
     {"now", "core_functions", CatalogType::SCALAR_FUNCTION_ENTRY},
     {"ord", "core_functions", CatalogType::SCALAR_FUNCTION_ENTRY},
+    {"parquet_bloom_probe", "parquet", CatalogType::TABLE_FUNCTION_ENTRY},
     {"parquet_file_metadata", "parquet", CatalogType::TABLE_FUNCTION_ENTRY},
     {"parquet_kv_metadata", "parquet", CatalogType::TABLE_FUNCTION_ENTRY},
     {"parquet_metadata", "parquet", CatalogType::TABLE_FUNCTION_ENTRY},
diff --git a/src/include/duckdb/planner/table_filter.hpp b/src/include/duckdb/planner/table_filter.hpp
index aa94e5cea154..b58283fc5bc3 100644
--- a/src/include/duckdb/planner/table_filter.hpp
+++ b/src/include/duckdb/planner/table_filter.hpp
@@ -62,7 +62,7 @@ class TableFilter {
 	template <class TARGET>
 	TARGET &Cast() {
 		if (filter_type != TARGET::TYPE) {
-			throw InternalException("Failed to cast table to type - table filter type mismatch");
+			throw InternalException("Failed to cast to type - table filter type mismatch");
 		}
 		return reinterpret_cast<TARGET &>(*this);
 	}
@@ -70,7 +70,7 @@ class TableFilter {
 	template <class TARGET>
 	const TARGET &Cast() const {
 		if (filter_type != TARGET::TYPE) {
-			throw InternalException("Failed to cast table to type - table filter type mismatch");
+			throw InternalException("Failed to cast to type - table filter type mismatch");
 		}
 		return reinterpret_cast<const TARGET &>(*this);
 	}


Test patch:
diff --git a/test/sql/copy/parquet/alltypes-dictionaries.test b/test/sql/copy/parquet/alltypes-dictionaries.test
new file mode 100644
index 000000000000..ea7009e6c451
--- /dev/null
+++ b/test/sql/copy/parquet/alltypes-dictionaries.test
@@ -0,0 +1,27 @@
+# name: test/sql/copy/parquet/alltypes-dictionaries.test
+# group: [parquet]
+
+require parquet
+
+
+foreach type TINYINT SMALLINT INTEGER BIGINT HUGEINT UTINYINT USMALLINT UINTEGER UBIGINT UHUGEINT FLOAT DOUBLE VARCHAR
+
+statement ok
+copy (select (r1.range * 10)::${type} r from range(10) r1, range(1000) r2) to '__TEST_DIR__/dict-${type}.parquet' (row_group_size 2048);
+
+query I
+select first(encodings) from parquet_metadata('__TEST_DIR__/dict-${type}.parquet') group by encodings;
+----
+RLE_DICTIONARY
+
+query III
+select column_id, BOOL_AND(bloom_filter_offset > 4), BOOL_AND(bloom_filter_length > 1)  from parquet_metadata('__TEST_DIR__/dict-${type}.parquet') group by column_id order by column_id;
+----
+0	true	true
+
+#query I
+#SELECT bloom_filter_excludes FROM parquet_bloom_probe('__TEST_DIR__/dict-${type}.parquet', 'r', '11');
+#----
+#true
+
+endloop
\ No newline at end of file
diff --git a/test/sql/copy/parquet/bloom_filters.test b/test/sql/copy/parquet/bloom_filters.test
new file mode 100644
index 000000000000..8ca061e9dbb7
--- /dev/null
+++ b/test/sql/copy/parquet/bloom_filters.test
@@ -0,0 +1,229 @@
+# name: test/sql/copy/parquet/bloom_filters.test
+# group: [parquet]
+
+require parquet
+
+statement ok
+copy (select
+    (r1.range*10)::BIGINT r,
+    r::smallint r_int16,
+    r::integer r_int32,
+    r::double r_double,
+    r::float r_float,
+    'string_' || r::VARCHAR r_string,
+    ('blob_' || r::VARCHAR)::BLOB r_blob
+from range(100) r1, range(1000) order by r) to '__TEST_DIR__/bloom1.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 1000);
+
+# we don't check the actual offsets since they might change due to filters being moved around
+query III
+select column_id, BOOL_AND(bloom_filter_offset > 4), BOOL_AND(bloom_filter_length > 1)  from parquet_metadata('__TEST_DIR__/bloom1.parquet') group by column_id order by column_id;
+----
+0	true	true
+1	true	true
+2	true	true
+3	true	true
+4	true	true
+5	true	true
+6	true	true
+
+# this value is not in the domain but within min/max
+query I
+SELECT BOOL_AND(bloom_filter_excludes) FROM parquet_bloom_probe('__TEST_DIR__/bloom1.parquet', 'r', '201');
+----
+true
+
+# this value is outside min/max
+query I
+SELECT BOOL_AND(bloom_filter_excludes) FROM parquet_bloom_probe('__TEST_DIR__/bloom1.parquet', 'r', '112121212');
+----
+true
+
+statement ok
+CREATE MACRO assert_bloom_filter_hit(file, col, val) AS TABLE
+    SELECT COUNT(*) > 0 AND COUNT(*) < MAX(row_group_id+1) FROM parquet_bloom_probe(file, col, val) WHERE NOT bloom_filter_excludes;
+
+# this in-domain should only be in a subset of row groups since its ordered
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r', '200');
+----
+true
+
+# same dance but with probe not being a string
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r', 200);
+----
+true
+
+# non-existent file
+statement error
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom10000.parquet', 'r', '200');
+----
+No files found
+
+# non-existent column
+statement error
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r2', '200');
+----
+Column r2 not found
+
+# NULL colname
+statement error
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', NULL, '200');
+----
+Can't have NULL parameters
+
+# NULL probe
+statement error
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r', NULL);
+----
+Can't have NULL parameters
+
+statement error
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r', 'a');
+----
+Failed to cast value
+
+# more types
+
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r_int16', 200);
+----
+true
+
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r_int32', 200);
+----
+true
+
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r_float', 200);
+----
+true
+
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r_double', 200);
+----
+true
+
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r_string', 'string_200');
+----
+true
+
+query I
+FROM assert_bloom_filter_hit('__TEST_DIR__/bloom1.parquet', 'r_blob', 'blob_200'::BLOB);
+----
+true
+
+
+
+# some tests for dictionary_size_limit
+
+# no bloom filter, dict limit too low
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom2.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 10);
+
+query III
+select row_group_id, bloom_filter_offset IS NOT NULL, bloom_filter_length IS NOT NULL from parquet_metadata('__TEST_DIR__/bloom2.parquet') order by row_group_id;
+----
+0	false	false
+
+
+# still no bloom filter, limit off-by-one
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom3.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 99);
+
+query III
+select row_group_id, bloom_filter_offset  IS NOT NULL, bloom_filter_length  IS NOT NULL  from parquet_metadata('__TEST_DIR__/bloom3.parquet') order by row_group_id;
+----
+0	false	false
+
+
+# should have a filter here!
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom4.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 100);
+
+query III
+select row_group_id, bloom_filter_offset IS NOT NULL, bloom_filter_length  IS NOT NULL  from parquet_metadata('__TEST_DIR__/bloom4.parquet') order by row_group_id;
+----
+0	true	true
+
+
+# should have a filter here, too
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom5.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 1000);
+
+query III
+select row_group_id, bloom_filter_offset IS NOT NULL, bloom_filter_length  IS NOT NULL  from parquet_metadata('__TEST_DIR__/bloom5.parquet') order by row_group_id;
+----
+0	true	true
+
+
+# lets mess with the false positive ratio and measue bf size
+
+# the default 0.01
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom6.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 1000, bloom_filter_false_positive_ratio 0.01);
+
+
+query II
+select row_group_id, bloom_filter_length from parquet_metadata('__TEST_DIR__/bloom6.parquet') order by row_group_id;
+----
+0	144
+
+
+# higher prob: 0.5 should lead to a smaller filter
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom7.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 1000, bloom_filter_false_positive_ratio 0.5);
+
+
+query II
+select row_group_id, bloom_filter_length from parquet_metadata('__TEST_DIR__/bloom7.parquet') order by row_group_id;
+----
+0	80
+
+
+# lower prob: 0.001 should lead to a bigger filter
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom8.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 1000, bloom_filter_false_positive_ratio 0.001);
+
+
+query II
+select row_group_id, bloom_filter_length from parquet_metadata('__TEST_DIR__/bloom8.parquet') order by row_group_id;
+----
+0	272
+
+
+# even lower prob: 0.0001 should lead to an even bigger filter
+statement ok
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom8.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 1000, bloom_filter_false_positive_ratio 0.0001);
+
+
+query II
+select row_group_id, bloom_filter_length from parquet_metadata('__TEST_DIR__/bloom8.parquet') order by row_group_id;
+----
+0	528
+
+
+# some error cases for the new parameters
+
+statement error
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom8.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit -1, bloom_filter_false_positive_ratio 0.0001);
+----
+dictionary_size_limit must be greater than 0
+
+
+statement error
+copy (select (r1.range*10)::BIGINT r,
+from range(100) r1, range(100) order by r) to '__TEST_DIR__/bloom8.parquet' (format parquet, ROW_GROUP_SIZE 10000, dictionary_size_limit 1000, bloom_filter_false_positive_ratio 0);
+----
+bloom_filter_false_positive_ratio must be greater than 0
\ No newline at end of file
diff --git a/test/sql/copy/parquet/dictionary_compression_ratio_threshold.test b/test/sql/copy/parquet/dictionary_compression_ratio_threshold.test
index b0bfacd4bbf7..cb26b742cbbb 100644
--- a/test/sql/copy/parquet/dictionary_compression_ratio_threshold.test
+++ b/test/sql/copy/parquet/dictionary_compression_ratio_threshold.test
@@ -2,16 +2,16 @@
 # description: Test Parquet dictionary_compression_ratio_threshold parameter
 # group: [parquet]
 
+# the setting dictionary_compression_ratio_threshold is DEPRECATED, the tests are here to make sure it can be set without issue
+
 require parquet
 
 statement ok
 CREATE TABLE test AS SELECT 'thisisaverylongstringbutitrepeatsmanytimessoitshighlycompressible' || (range % 10) i FROM range(100000)
 
-# cannot be negative
-statement error
+statement ok
 COPY test TO '__TEST_DIR__/dictionary_compression_ratio_threshold.parquet' (dictionary_compression_ratio_threshold -2)
-----
-Binder Error: dictionary_compression_ratio_threshold must be greater than 0, or -1 to disable dictionary compression
+
 
 # default is 1.0
 statement ok
@@ -27,11 +27,11 @@ false
 statement ok
 COPY test TO '__TEST_DIR__/dictionary_compression_ratio_threshold.parquet' (dictionary_compression_ratio_threshold -1)
 
-# disabled, so no dictionary compression
+# ignored, still dictionary compression
 query I
 SELECT dictionary_page_offset IS NULL FROM parquet_metadata('__TEST_DIR__/dictionary_compression_ratio_threshold.parquet')
 ----
-true
+false
 
 # the data compresses more than 10x
 statement ok
@@ -47,17 +47,17 @@ false
 statement ok
 COPY test TO '__TEST_DIR__/dictionary_compression_ratio_threshold.parquet' (dictionary_compression_ratio_threshold 20)
 
-# dictionary compression should be disabled
+# dictionary compression still enabled, setting is deprecated
 query I
 SELECT dictionary_page_offset IS NULL FROM parquet_metadata('__TEST_DIR__/dictionary_compression_ratio_threshold.parquet')
 ----
-true
+false
 
 # create table with all uniques
 statement ok
 CREATE OR REPLACE TABLE test AS SELECT 'coolstring' || range i FROM range(100000)
 
-# shouldn't have dictionary compression with default of 1.0
+# should still have compression, setting is deprecated
 statement ok
 COPY test TO '__TEST_DIR__/dictionary_compression_ratio_threshold.parquet'
 
@@ -73,4 +73,4 @@ COPY test TO '__TEST_DIR__/dictionary_compression_ratio_threshold.parquet' (dict
 query I
 SELECT dictionary_page_offset IS NULL FROM parquet_metadata('__TEST_DIR__/dictionary_compression_ratio_threshold.parquet')
 ----
-false
+true
diff --git a/test/sql/copy/parquet/writer/parquet_write_strings.test b/test/sql/copy/parquet/writer/parquet_write_strings.test
index bdc576606cf2..e92379f81f9d 100644
--- a/test/sql/copy/parquet/writer/parquet_write_strings.test
+++ b/test/sql/copy/parquet/writer/parquet_write_strings.test
@@ -26,7 +26,7 @@ COPY strings TO '__TEST_DIR__/strings.parquet' (FORMAT PARQUET);
 query I
 SELECT encodings FROM parquet_metadata('__TEST_DIR__/strings.parquet')
 ----
-PLAIN, RLE_DICTIONARY
+RLE_DICTIONARY
 
 query I
 SELECT * FROM '__TEST_DIR__/strings.parquet'
@@ -174,7 +174,7 @@ COPY strings TO '__TEST_DIR__/strings.parquet' (FORMAT PARQUET);
 query I
 SELECT encodings FROM parquet_metadata('__TEST_DIR__/strings.parquet')
 ----
-PLAIN
+RLE_DICTIONARY
 
 query I
 SELECT * FROM '__TEST_DIR__/strings.parquet'
@@ -213,7 +213,7 @@ SELECT * FROM '__TEST_DIR__/strings.parquet'
 query I
 SELECT stats_distinct_count FROM parquet_metadata('__TEST_DIR__/strings.parquet')
 ----
-NULL
+30
 
 # non-dictionary table with null
 statement ok
diff --git a/test/sql/copy/parquet/writer/write_stats_min_max.test_slow b/test/sql/copy/parquet/writer/write_stats_min_max.test_slow
index 67844fe8ca14..69c6ff69b3b3 100644
--- a/test/sql/copy/parquet/writer/write_stats_min_max.test_slow
+++ b/test/sql/copy/parquet/writer/write_stats_min_max.test_slow
@@ -66,10 +66,10 @@ statement ok
 INSERT INTO tbl SELECT NULL
 
 statement ok
-COPY tbl TO '__TEST_DIR__/${type}_stats.parquet.parquet' (FORMAT PARQUET);
+COPY tbl TO '__TEST_DIR__/${type}_stats.parquet' (FORMAT PARQUET);
 
 query IIII
-SELECT stats_min_value::${type}, stats_max_value::${type}, stats_min::${type}, stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_min_value::${type}, stats_max_value::${type}, stats_min::${type}, stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL	NULL	NULL	NULL
 
@@ -81,22 +81,22 @@ statement ok
 INSERT INTO tbl SELECT max FROM ${type}_limits
 
 statement ok
-COPY tbl TO '__TEST_DIR__/${type}_stats.parquet.parquet' (FORMAT PARQUET);
+COPY tbl TO '__TEST_DIR__/${type}_stats.parquet' (FORMAT PARQUET);
 
 query I
-SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet') EXCEPT SELECT min FROM ${type}_limits
+SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet') EXCEPT SELECT min FROM ${type}_limits
 ----
 
 query I
-SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet') EXCEPT SELECT max FROM ${type}_limits
+SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet') EXCEPT SELECT max FROM ${type}_limits
 ----
 
 query I
-SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet') EXCEPT SELECT min FROM ${type}_limits
+SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet') EXCEPT SELECT min FROM ${type}_limits
 ----
 
 query I
-SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet') EXCEPT SELECT max FROM ${type}_limits
+SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet') EXCEPT SELECT max FROM ${type}_limits
 ----
 
 statement ok
@@ -127,25 +127,25 @@ statement ok
 INSERT INTO tbl SELECT NULL
 
 statement ok
-COPY tbl TO '__TEST_DIR__/${type}_stats.parquet.parquet' (FORMAT PARQUET);
+COPY tbl TO '__TEST_DIR__/${type}_stats.parquet' (FORMAT PARQUET);
 
 query I
-SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
 query I
-SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
 query I
-SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
 query I
-SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
@@ -157,23 +157,23 @@ statement ok
 INSERT INTO tbl SELECT max FROM ${type}_limits
 
 statement ok
-COPY tbl TO '__TEST_DIR__/${type}_stats.parquet.parquet' (FORMAT PARQUET);
+COPY tbl TO '__TEST_DIR__/${type}_stats.parquet' (FORMAT PARQUET);
 
 query I
-SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet') EXCEPT SELECT min FROM ${type}_limits
+SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet') EXCEPT SELECT min FROM ${type}_limits
 ----
 
 query I
-SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet') EXCEPT SELECT max FROM ${type}_limits
+SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet') EXCEPT SELECT max FROM ${type}_limits
 ----
 
 query I
-SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
 query I
-SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
@@ -198,25 +198,25 @@ statement ok
 INSERT INTO tbl SELECT max FROM ${type}_limits
 
 statement ok
-COPY tbl TO '__TEST_DIR__/${type}_stats.parquet.parquet' (FORMAT PARQUET);
+COPY tbl TO '__TEST_DIR__/${type}_stats.parquet' (FORMAT PARQUET);
 
 query I
-SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_min_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
 query I
-SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_max_value::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
 query I
-SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_min::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL
 
 query I
-SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet.parquet')
+SELECT stats_max::${type} FROM parquet_metadata('__TEST_DIR__/${type}_stats.parquet')
 ----
 NULL