Repository: duckdb/duckdb (language: python)

Issue:
Macro default variable that is a boolean type fails
### What happens?

Howdy! This is a super tiny issue...

If a macro default variable is a boolean type, the macro will not successfully create. This may be caused by automatically re-writing the literal `true` into `cast('t' as boolean)`.

This is a tiny issue since integer values will perform adequately. The main downside is that the parameters are not as "self-documenting" that there are only 2 possible values.

### To Reproduce

This fails:
```sql
CREATE OR REPLACE MACRO default_bool_test(param1:=true) AS 
    CASE WHEN param1 THEN 'Tis true' ELSE 'Nope' END 
```
```
Parser Error: Invalid parameter: 'CAST('t' AS BOOLEAN)'
```

This succeeds:
```sql
CREATE OR REPLACE MACRO default_bool_test(param1:=1) AS 
    CASE WHEN param1 THEN 'Tis true' ELSE 'Nope' END 
```

### OS:

MacOS

### DuckDB Version:

v1.0.1-dev2074

### DuckDB Client:

Python

### Full Name:

Alex Monahan

### Affiliation:

DuckDB Labs and MotherDuck

### What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have tested with a nightly build

### Did you include all relevant data sets for reproducing the issue?

Not applicable - the reproduction does not require a data set

### Did you include all code required to reproduce the issue?

- [X] Yes, I have

### Did you include all relevant configuration (e.g., CPU architecture, Python version, Linux distribution) to reproduce the issue?

- [X] Yes, I have


Discussion / hints:
Could we add logic deal with `ExpressionType::OPERATOR_CAST` in `Transformer::TransformMacroFunction` during transform phase to solve this issue.

Fix patch:
diff --git a/extension/fts/fts_indexing.cpp b/extension/fts/fts_indexing.cpp
index c098f194901c..bd985f7a6c2f 100644
--- a/extension/fts/fts_indexing.cpp
+++ b/extension/fts/fts_indexing.cpp
@@ -149,7 +149,7 @@ static string IndexingScript(ClientContext &context, QualifiedName &qname, const
             FROM %fts_schema%.docs AS docs
         );
 
-        CREATE MACRO %fts_schema%.match_bm25(docname, query_string, fields := NULL, k := 1.2, b := 0.75, conjunctive := 0) AS (
+        CREATE MACRO %fts_schema%.match_bm25(docname, query_string, fields := NULL, k := 1.2, b := 0.75, conjunctive := false) AS (
             WITH tokens AS (
                 SELECT DISTINCT stem(unnest(%fts_schema%.tokenize(query_string)), '%stemmer%') AS t
             ),
diff --git a/extension/json/include/json_scan.hpp b/extension/json/include/json_scan.hpp
index 9dc96ed438fd..a2ad431b1698 100644
--- a/extension/json/include/json_scan.hpp
+++ b/extension/json/include/json_scan.hpp
@@ -240,7 +240,8 @@ struct JSONScanLocalState {
 
 	void SkipOverArrayStart();
 
-	void ReadAndAutoDetect(JSONScanGlobalState &gstate, AllocatedData &buffer, optional_idx &buffer_index);
+	void ReadAndAutoDetect(JSONScanGlobalState &gstate, AllocatedData &buffer, optional_idx &buffer_index,
+	                       bool &file_done);
 	bool ReconstructFirstObject(JSONScanGlobalState &gstate);
 	void ParseNextChunk(JSONScanGlobalState &gstate);
 
diff --git a/extension/json/json_scan.cpp b/extension/json/json_scan.cpp
index 5289ff05a8f5..c7dba73228cb 100644
--- a/extension/json/json_scan.cpp
+++ b/extension/json/json_scan.cpp
@@ -600,20 +600,18 @@ bool JSONScanLocalState::ReadNextBuffer(JSONScanGlobalState &gstate) {
 		// Open the file if it is not yet open
 		if (!current_reader->IsOpen()) {
 			current_reader->OpenJSONFile();
-			if (current_reader->GetFileHandle().FileSize() == 0 && !current_reader->GetFileHandle().IsPipe()) {
-				current_reader->GetFileHandle().Close();
-				// Skip over empty files
-				if (gstate.enable_parallel_scans) {
-					TryIncrementFileIndex(gstate);
-				}
-				continue;
-			}
 		}
 
 		// Auto-detect if we haven't yet done this during the bind
 		if (gstate.bind_data.options.record_type == JSONRecordType::AUTO_DETECT ||
 		    current_reader->GetFormat() == JSONFormat::AUTO_DETECT) {
-			ReadAndAutoDetect(gstate, buffer, buffer_index);
+			bool file_done = false;
+			ReadAndAutoDetect(gstate, buffer, buffer_index, file_done);
+			if (file_done) {
+				TryIncrementFileIndex(gstate);
+				lock_guard<mutex> reader_guard(current_reader->lock);
+				current_reader->GetFileHandle().Close();
+			}
 		}
 
 		if (gstate.enable_parallel_scans) {
@@ -653,9 +651,8 @@ bool JSONScanLocalState::ReadNextBuffer(JSONScanGlobalState &gstate) {
 }
 
 void JSONScanLocalState::ReadAndAutoDetect(JSONScanGlobalState &gstate, AllocatedData &buffer,
-                                           optional_idx &buffer_index) {
+                                           optional_idx &buffer_index, bool &file_done) {
 	// We have to detect the JSON format - hold the gstate lock while we do this
-	bool file_done = false;
 	if (!ReadNextBufferInternal(gstate, buffer, buffer_index, file_done)) {
 		return;
 	}
diff --git a/src/execution/join_hashtable.cpp b/src/execution/join_hashtable.cpp
index 406cb7dea9fc..3b8352f384ef 100644
--- a/src/execution/join_hashtable.cpp
+++ b/src/execution/join_hashtable.cpp
@@ -24,10 +24,10 @@ JoinHashTable::ProbeState::ProbeState()
       ht_offsets_dense_v(LogicalType::UBIGINT), non_empty_sel(STANDARD_VECTOR_SIZE) {
 }
 
-JoinHashTable::InsertState::InsertState(const unique_ptr<TupleDataCollection> &data_collection,
-                                        const vector<column_t> &equality_predicate_columns)
+JoinHashTable::InsertState::InsertState(const JoinHashTable &ht)
     : SharedState(), remaining_sel(STANDARD_VECTOR_SIZE), key_match_sel(STANDARD_VECTOR_SIZE) {
-	data_collection->InitializeChunkState(chunk_state, equality_predicate_columns);
+	ht.data_collection->InitializeChunk(lhs_data, ht.equality_predicate_columns);
+	ht.data_collection->InitializeChunkState(chunk_state, ht.equality_predicate_columns);
 }
 
 JoinHashTable::JoinHashTable(BufferManager &buffer_manager_p, const vector<JoinCondition> &conditions_p,
@@ -129,21 +129,16 @@ void JoinHashTable::Merge(JoinHashTable &other) {
 }
 
 static void ApplyBitmaskAndGetSaltBuild(Vector &hashes_v, const idx_t &count, const idx_t &bitmask) {
-
 	if (hashes_v.GetVectorType() == VectorType::CONSTANT_VECTOR) {
-
 		D_ASSERT(!ConstantVector::IsNull(hashes_v));
-
 		auto indices = ConstantVector::GetData<hash_t>(hashes_v);
 		hash_t salt = ht_entry_t::ExtractSaltWithNulls(*indices);
 		idx_t offset = *indices & bitmask;
 		*indices = offset | salt;
 		hashes_v.Flatten(count);
-
 	} else {
 		hashes_v.Flatten(count);
 		auto hashes = FlatVector::GetData<hash_t>(hashes_v);
-
 		for (idx_t i = 0; i < count; i++) {
 			idx_t salt = ht_entry_t::ExtractSaltWithNulls(hashes[i]);
 			idx_t offset = hashes[i] & bitmask;
@@ -415,21 +410,22 @@ idx_t JoinHashTable::PrepareKeys(DataChunk &keys, vector<TupleDataVectorFormat>
 	// figure out which keys are NULL, and create a selection vector out of them
 	current_sel = FlatVector::IncrementalSelectionVector();
 	idx_t added_count = keys.size();
-	if (build_side && (PropagatesBuildSide(join_type))) {
+	if (build_side && PropagatesBuildSide(join_type)) {
 		// in case of a right or full outer join, we cannot remove NULL keys from the build side
 		return added_count;
 	}
 
 	for (idx_t col_idx = 0; col_idx < keys.ColumnCount(); col_idx++) {
-		if (!null_values_are_equal[col_idx]) {
-			auto &col_key_data = vector_data[col_idx].unified;
-			if (col_key_data.validity.AllValid()) {
-				continue;
-			}
-			added_count = FilterNullValues(col_key_data, *current_sel, added_count, sel);
-			// null values are NOT equal for this column, filter them out
-			current_sel = &sel;
+		if (null_values_are_equal[col_idx]) {
+			continue;
+		}
+		auto &col_key_data = vector_data[col_idx].unified;
+		if (col_key_data.validity.AllValid()) {
+			continue;
 		}
+		added_count = FilterNullValues(col_key_data, *current_sel, added_count, sel);
+		// null values are NOT equal for this column, filter them out
+		current_sel = &sel;
 	}
 	return added_count;
 }
@@ -438,8 +434,8 @@ idx_t JoinHashTable::PrepareKeys(DataChunk &keys, vector<TupleDataVectorFormat>
 //! happen and we need to return the pointer to the to row with which the new entry would have collided. In any other
 //! case we return a nullptr
 template <bool PARALLEL, bool EXPECT_EMPTY>
-static inline data_ptr_t InsertRowToEntry(atomic<ht_entry_t> &entry, data_ptr_t row_ptr_to_insert, const hash_t salt,
-                                          const idx_t pointer_offset) {
+static inline data_ptr_t InsertRowToEntry(atomic<ht_entry_t> &entry, const data_ptr_t &row_ptr_to_insert,
+                                          const hash_t &salt, const idx_t &pointer_offset) {
 
 	if (PARALLEL) {
 		// if we expect the entry to be empty, if the operation fails we need to cancel the whole operation as another
@@ -473,9 +469,8 @@ static inline data_ptr_t InsertRowToEntry(atomic<ht_entry_t> &entry, data_ptr_t
 
 			return nullptr;
 		}
-	}
-	// if we are not in parallel mode, we can just do the operation without any checks
-	else {
+	} else {
+		// if we are not in parallel mode, we can just do the operation without any checks
 		ht_entry_t current_entry = entry.load(std::memory_order_relaxed);
 		data_ptr_t current_row_pointer = current_entry.GetPointerOrNull();
 		Store<data_ptr_t>(current_row_pointer, row_ptr_to_insert + pointer_offset);
@@ -483,22 +478,18 @@ static inline data_ptr_t InsertRowToEntry(atomic<ht_entry_t> &entry, data_ptr_t
 		return nullptr;
 	}
 }
-static inline void PerformKeyComparison(atomic<ht_entry_t> entries[], JoinHashTable::InsertState &state,
-                                        JoinHashTable *ht, const unique_ptr<TupleDataCollection> &data_collection,
-                                        Vector &row_locations, const idx_t count, idx_t &key_match_count,
-                                        idx_t &key_no_match_count) {
+static inline void PerformKeyComparison(JoinHashTable::InsertState &state, JoinHashTable &ht,
+                                        const TupleDataCollection &data_collection, Vector &row_locations,
+                                        const idx_t count, idx_t &key_match_count, idx_t &key_no_match_count) {
 	// Get the data for the rows that need to be compared
-	DataChunk lhs_data;
-	data_collection->InitializeChunk(lhs_data,
-	                                 ht->equality_predicate_columns); // makes sure DataChunk has the right format
-	lhs_data.SetCardinality(count);                                   // and the right size
+	state.lhs_data.Reset();
+	state.lhs_data.SetCardinality(count); // the right size
 
 	// The target selection vector says where to write the results into the lhs_data, we just want to write
 	// sequentially as otherwise we trigger a bug in the Gather function
-	data_collection->Gather(row_locations, state.salt_match_sel, count, ht->equality_predicate_columns, lhs_data,
-	                        *FlatVector::IncrementalSelectionVector(), state.chunk_state.cached_cast_vectors);
-
-	TupleDataCollection::ToUnifiedFormat(state.chunk_state, lhs_data);
+	data_collection.Gather(row_locations, state.salt_match_sel, count, ht.equality_predicate_columns, state.lhs_data,
+	                       *FlatVector::IncrementalSelectionVector(), state.chunk_state.cached_cast_vectors);
+	TupleDataCollection::ToUnifiedFormat(state.chunk_state, state.lhs_data);
 
 	for (idx_t i = 0; i < count; i++) {
 		state.key_match_sel.set_index(i, i);
@@ -506,40 +497,40 @@ static inline void PerformKeyComparison(atomic<ht_entry_t> entries[], JoinHashTa
 
 	// Perform row comparisons
 	key_match_count =
-	    ht->row_matcher_build.Match(lhs_data, state.chunk_state.vector_data, state.key_match_sel, count, ht->layout,
-	                                state.rhs_row_locations, &state.key_no_match_sel, key_no_match_count);
+	    ht.row_matcher_build.Match(state.lhs_data, state.chunk_state.vector_data, state.key_match_sel, count, ht.layout,
+	                               state.rhs_row_locations, &state.key_no_match_sel, key_no_match_count);
 
 	D_ASSERT(key_match_count + key_no_match_count == count);
 }
 
 template <bool PARALLEL>
 static inline void InsertMatchesAndIncrementMisses(atomic<ht_entry_t> entries[], JoinHashTable::InsertState &state,
-                                                   JoinHashTable *ht, data_ptr_t *lhs_row_locations,
-                                                   idx_t *ht_offsets_and_salts, const idx_t capacity_mask,
+                                                   JoinHashTable &ht, const data_ptr_t lhs_row_locations[],
+                                                   idx_t ht_offsets_and_salts[], const idx_t capacity_mask,
                                                    const idx_t key_match_count, const idx_t key_no_match_count) {
+	if (key_match_count != 0) {
+		ht.chains_longer_than_one = true;
+	}
+
 	// Insert the rows that match
 	for (idx_t i = 0; i < key_match_count; i++) {
 		const auto need_compare_idx = state.key_match_sel.get_index(i);
 		const auto entry_index = state.salt_match_sel.get_index(need_compare_idx);
 
-		idx_t ht_offset = ht_offsets_and_salts[entry_index] & ht_entry_t::POINTER_MASK;
+		const auto &ht_offset = ht_offsets_and_salts[entry_index] & ht_entry_t::POINTER_MASK;
 		auto &entry = entries[ht_offset];
-		data_ptr_t row_ptr_to_insert = lhs_row_locations[entry_index];
+		const data_ptr_t row_ptr_to_insert = lhs_row_locations[entry_index];
 
-		auto salt = ht_offsets_and_salts[entry_index];
-		InsertRowToEntry<PARALLEL, false>(entry, row_ptr_to_insert, salt, ht->pointer_offset);
-
-		ht->chains_longer_than_one = true;
+		const auto salt = ht_offsets_and_salts[entry_index];
+		InsertRowToEntry<PARALLEL, false>(entry, row_ptr_to_insert, salt, ht.pointer_offset);
 	}
 
 	// Linear probing: each of the entries that do not match move to the next entry in the HT
 	for (idx_t i = 0; i < key_no_match_count; i++) {
-
 		const auto need_compare_idx = state.key_no_match_sel.get_index(i);
 		const auto entry_index = state.salt_match_sel.get_index(need_compare_idx);
 
 		idx_t &ht_offset_and_salt = ht_offsets_and_salts[entry_index];
-
 		IncrementAndWrap(ht_offset_and_salt, capacity_mask);
 
 		state.remaining_sel.set_index(i, entry_index);
@@ -547,50 +538,71 @@ static inline void InsertMatchesAndIncrementMisses(atomic<ht_entry_t> entries[],
 }
 
 template <bool PARALLEL>
-static void InsertHashesLoop(atomic<ht_entry_t> entries[], Vector row_locations, Vector &hashes_v, const idx_t &count,
-                             JoinHashTable::InsertState &state, unique_ptr<TupleDataCollection> &data_collection,
-                             JoinHashTable *ht) {
+static void InsertHashesLoop(atomic<ht_entry_t> entries[], Vector &row_locations, Vector &hashes_v, const idx_t &count,
+                             JoinHashTable::InsertState &state, const TupleDataCollection &data_collection,
+                             JoinHashTable &ht) {
 	D_ASSERT(hashes_v.GetType().id() == LogicalType::HASH);
-	ApplyBitmaskAndGetSaltBuild(hashes_v, count, ht->bitmask);
+	ApplyBitmaskAndGetSaltBuild(hashes_v, count, ht.bitmask);
 
 	// the offset for each row to insert
-	idx_t *ht_offsets_and_salts = FlatVector::GetData<idx_t>(hashes_v);
+	const auto ht_offsets_and_salts = FlatVector::GetData<idx_t>(hashes_v);
 	// the row locations of the rows that are already in the hash table
-	data_ptr_t *rhs_row_locations = FlatVector::GetData<data_ptr_t>(state.rhs_row_locations);
+	const auto rhs_row_locations = FlatVector::GetData<data_ptr_t>(state.rhs_row_locations);
 	// the row locations of the rows that are to be inserted
-	data_ptr_t *lhs_row_locations = FlatVector::GetData<data_ptr_t>(row_locations);
+	const auto lhs_row_locations = FlatVector::GetData<data_ptr_t>(row_locations);
 
-	const SelectionVector *remaining_sel = FlatVector::IncrementalSelectionVector();
+	// we start off with the entire chunk
 	idx_t remaining_count = count;
+	const auto *remaining_sel = FlatVector::IncrementalSelectionVector();
+
+	if (PropagatesBuildSide(ht.join_type)) {
+		// if we propagate the build side, we may have added rows with NULL keys to the HT
+		// these may need to be filtered out depending on the comparison type (exactly like PrepareKeys does)
+		for (idx_t col_idx = 0; col_idx < ht.conditions.size(); col_idx++) {
+			// if null values are NOT equal for this column we filter them out
+			if (ht.NullValuesAreEqual(col_idx)) {
+				continue;
+			}
+
+			idx_t entry_idx;
+			idx_t idx_in_entry;
+			ValidityBytes::GetEntryIndex(col_idx, entry_idx, idx_in_entry);
+
+			idx_t new_remaining_count = 0;
+			for (idx_t i = 0; i < remaining_count; i++) {
+				const auto idx = remaining_sel->get_index(i);
+				if (ValidityBytes(lhs_row_locations[idx]).RowIsValidUnsafe(col_idx)) {
+					state.remaining_sel.set_index(new_remaining_count++, idx);
+				}
+			}
+			remaining_count = new_remaining_count;
+			remaining_sel = &state.remaining_sel;
+		}
+	}
 
 	// use the ht bitmask to make the modulo operation faster but keep the salt bits intact
-	idx_t capacity_mask = ht->bitmask | ht_entry_t::SALT_MASK;
+	idx_t capacity_mask = ht.bitmask | ht_entry_t::SALT_MASK;
 	while (remaining_count > 0) {
 		idx_t salt_match_count = 0;
 
-		// iterate over each entry to find out whether it belongs to an existing list or will start
-		// a new list
+		// iterate over each entry to find out whether it belongs to an existing list or will start a new list
 		for (idx_t i = 0; i < remaining_count; i++) {
-
 			const idx_t row_index = remaining_sel->get_index(i);
-
 			idx_t &ht_offset_and_salt = ht_offsets_and_salts[row_index];
 			const hash_t salt = ht_entry_t::ExtractSalt(ht_offset_and_salt);
 
+			// increment the ht_offset_and_salt of the entry as long as next entry is occupied and salt does not match
 			idx_t ht_offset;
 			ht_entry_t entry;
-			bool occupied, salt_match;
-
-			// increment the ht_offset_and_salt of the entry as long as next entry is occupied and salt does not match
+			bool occupied;
 			while (true) {
 				ht_offset = ht_offset_and_salt & ht_entry_t::POINTER_MASK;
 				atomic<ht_entry_t> &atomic_entry = entries[ht_offset];
 				entry = atomic_entry.load(std::memory_order_relaxed);
 				occupied = entry.IsOccupied();
-				salt_match = entry.GetSalt() == salt;
 
 				// condition for incrementing the ht_offset: occupied and row_salt does not match -> move to next entry
-				if (!occupied || salt_match) {
+				if (!occupied || entry.GetSalt() == salt) {
 					break;
 				}
 
@@ -598,10 +610,10 @@ static void InsertHashesLoop(atomic<ht_entry_t> entries[], Vector row_locations,
 			}
 
 			if (!occupied) { // insert into free
-				atomic<ht_entry_t> &atomic_entry = entries[ht_offset];
-				data_ptr_t row_ptr_to_insert = lhs_row_locations[row_index];
-				data_ptr_t potential_collided_ptr =
-				    InsertRowToEntry<PARALLEL, true>(atomic_entry, row_ptr_to_insert, salt, ht->pointer_offset);
+				auto &atomic_entry = entries[ht_offset];
+				const auto row_ptr_to_insert = lhs_row_locations[row_index];
+				const auto potential_collided_ptr =
+				    InsertRowToEntry<PARALLEL, true>(atomic_entry, row_ptr_to_insert, salt, ht.pointer_offset);
 
 				if (PARALLEL) {
 					// if the insertion was not successful, the entry was occupied in the meantime, so we have to
@@ -624,15 +636,11 @@ static void InsertHashesLoop(atomic<ht_entry_t> entries[], Vector row_locations,
 
 		// at this step, for all the rows to insert we stepped either until we found an empty entry or an entry with
 		// a matching salt, we now need to compare the keys for the ones that have a matching salt
-
-		idx_t key_match_count = 0;
 		idx_t key_no_match_count = 0;
-
 		if (salt_match_count != 0) {
-
-			PerformKeyComparison(entries, state, ht, data_collection, row_locations, salt_match_count, key_match_count,
+			idx_t key_match_count = 0;
+			PerformKeyComparison(state, ht, data_collection, row_locations, salt_match_count, key_match_count,
 			                     key_no_match_count);
-
 			InsertMatchesAndIncrementMisses<PARALLEL>(entries, state, ht, lhs_row_locations, ht_offsets_and_salts,
 			                                          capacity_mask, key_match_count, key_no_match_count);
 		}
@@ -644,17 +652,14 @@ static void InsertHashesLoop(atomic<ht_entry_t> entries[], Vector row_locations,
 	}
 }
 
-void JoinHashTable::InsertHashes(Vector &hashes_v, idx_t count, TupleDataChunkState &chunk_state,
+void JoinHashTable::InsertHashes(Vector &hashes_v, const idx_t count, TupleDataChunkState &chunk_state,
                                  InsertState &insert_state, bool parallel) {
 	auto atomic_entries = reinterpret_cast<atomic<ht_entry_t> *>(this->entries);
 	auto row_locations = chunk_state.row_locations;
-
 	if (parallel) {
-		InsertHashesLoop<true>(atomic_entries, row_locations, hashes_v, count, insert_state, this->data_collection,
-		                       this);
+		InsertHashesLoop<true>(atomic_entries, row_locations, hashes_v, count, insert_state, *data_collection, *this);
 	} else {
-		InsertHashesLoop<false>(atomic_entries, row_locations, hashes_v, count, insert_state, this->data_collection,
-		                        this);
+		InsertHashesLoop<false>(atomic_entries, row_locations, hashes_v, count, insert_state, *data_collection, *this);
 	}
 }
 
@@ -697,7 +702,7 @@ void JoinHashTable::Finalize(idx_t chunk_idx_from, idx_t chunk_idx_to, bool para
 	                                chunk_idx_to, false);
 	const auto row_locations = iterator.GetRowLocations();
 
-	InsertState insert_state(this->data_collection, this->equality_predicate_columns);
+	InsertState insert_state(*this);
 	do {
 		const auto count = iterator.GetCurrentChunkCount();
 		for (idx_t i = 0; i < count; i++) {
diff --git a/src/execution/operator/scan/physical_column_data_scan.cpp b/src/execution/operator/scan/physical_column_data_scan.cpp
index 9ab0e474afc8..a89947fa4a23 100644
--- a/src/execution/operator/scan/physical_column_data_scan.cpp
+++ b/src/execution/operator/scan/physical_column_data_scan.cpp
@@ -11,7 +11,8 @@ namespace duckdb {
 PhysicalColumnDataScan::PhysicalColumnDataScan(vector<LogicalType> types, PhysicalOperatorType op_type,
                                                idx_t estimated_cardinality,
                                                optionally_owned_ptr<ColumnDataCollection> collection_p)
-    : PhysicalOperator(op_type, std::move(types), estimated_cardinality), collection(std::move(collection_p)) {
+    : PhysicalOperator(op_type, std::move(types), estimated_cardinality), collection(std::move(collection_p)),
+      cte_index(DConstants::INVALID_INDEX) {
 }
 
 PhysicalColumnDataScan::PhysicalColumnDataScan(vector<LogicalType> types, PhysicalOperatorType op_type,
@@ -21,10 +22,8 @@ PhysicalColumnDataScan::PhysicalColumnDataScan(vector<LogicalType> types, Physic
 
 class PhysicalColumnDataGlobalScanState : public GlobalSourceState {
 public:
-	PhysicalColumnDataGlobalScanState(const ClientContext &context, const ColumnDataCollection &collection)
-	    : max_threads(MaxValue<idx_t>(context.config.verify_parallelism ? collection.ChunkCount()
-	                                                                    : collection.ChunkCount() / CHUNKS_PER_THREAD,
-	                                  1)) {
+	explicit PhysicalColumnDataGlobalScanState(const ColumnDataCollection &collection)
+	    : max_threads(MaxValue<idx_t>(collection.ChunkCount(), 1)) {
 		collection.InitializeScan(global_scan_state);
 	}
 
@@ -35,7 +34,6 @@ class PhysicalColumnDataGlobalScanState : public GlobalSourceState {
 public:
 	ColumnDataParallelScanState global_scan_state;
 
-	static constexpr idx_t CHUNKS_PER_THREAD = 32;
 	const idx_t max_threads;
 };
 
@@ -45,7 +43,7 @@ class PhysicalColumnDataLocalScanState : public LocalSourceState {
 };
 
 unique_ptr<GlobalSourceState> PhysicalColumnDataScan::GetGlobalSourceState(ClientContext &context) const {
-	return make_uniq<PhysicalColumnDataGlobalScanState>(context, *collection);
+	return make_uniq<PhysicalColumnDataGlobalScanState>(*collection);
 }
 
 unique_ptr<LocalSourceState> PhysicalColumnDataScan::GetLocalSourceState(ExecutionContext &,
diff --git a/src/include/duckdb/execution/ht_entry.hpp b/src/include/duckdb/execution/ht_entry.hpp
index 346d94e5c467..73fbd81bf5b9 100644
--- a/src/include/duckdb/execution/ht_entry.hpp
+++ b/src/include/duckdb/execution/ht_entry.hpp
@@ -8,6 +8,9 @@
 
 #pragma once
 
+#include "duckdb/common/assert.hpp"
+#include "duckdb/common/typedefs.hpp"
+
 namespace duckdb {
 
 //! The ht_entry_t struct represents an individual entry within a hash table.
@@ -93,7 +96,7 @@ struct ht_entry_t { // NOLINT
 
 // uses an AND operation to apply the modulo operation instead of an if condition that could be branch mispredicted
 inline void IncrementAndWrap(idx_t &offset, const uint64_t &capacity_mask) {
-	++offset &= capacity_mask; // leave the salt bits unchanged
+	++offset &= capacity_mask;
 }
 
 } // namespace duckdb
diff --git a/src/include/duckdb/execution/join_hashtable.hpp b/src/include/duckdb/execution/join_hashtable.hpp
index 85ddb953ffd7..add5f1e73561 100644
--- a/src/include/duckdb/execution/join_hashtable.hpp
+++ b/src/include/duckdb/execution/join_hashtable.hpp
@@ -150,11 +150,12 @@ class JoinHashTable {
 	};
 
 	struct InsertState : SharedState {
-		InsertState(const unique_ptr<TupleDataCollection> &data_collection,
-		            const vector<column_t> &equality_predicate_columns);
+		explicit InsertState(const JoinHashTable &ht);
 		/// Because of the index hick up
 		SelectionVector remaining_sel;
 		SelectionVector key_match_sel;
+
+		DataChunk lhs_data;
 		TupleDataChunkState chunk_state;
 	};
 
@@ -197,6 +198,9 @@ class JoinHashTable {
 	TupleDataCollection &GetDataCollection() {
 		return *data_collection;
 	}
+	bool NullValuesAreEqual(idx_t col_idx) const {
+		return null_values_are_equal[col_idx];
+	}
 
 	//! BufferManager
 	BufferManager &buffer_manager;
@@ -288,7 +292,7 @@ class JoinHashTable {
 	//! Insert the given set of locations into the HT with the given set of hashes_v
 	void InsertHashes(Vector &hashes_v, idx_t count, TupleDataChunkState &chunk_state, InsertState &insert_statebool,
 	                  bool parallel);
-
+	//! Prepares keys by filtering NULLs
 	idx_t PrepareKeys(DataChunk &keys, vector<TupleDataVectorFormat> &vector_data, const SelectionVector *&current_sel,
 	                  SelectionVector &sel, bool build_side);
 
diff --git a/src/optimizer/filter_pushdown.cpp b/src/optimizer/filter_pushdown.cpp
index b235e8e35fd8..57289c9f8ca4 100644
--- a/src/optimizer/filter_pushdown.cpp
+++ b/src/optimizer/filter_pushdown.cpp
@@ -3,11 +3,11 @@
 #include "duckdb/optimizer/filter_combiner.hpp"
 #include "duckdb/optimizer/optimizer.hpp"
 #include "duckdb/planner/expression_iterator.hpp"
+#include "duckdb/planner/operator/logical_comparison_join.hpp"
 #include "duckdb/planner/operator/logical_filter.hpp"
 #include "duckdb/planner/operator/logical_join.hpp"
-#include "duckdb/planner/operator/logical_window.hpp"
 #include "duckdb/planner/operator/logical_projection.hpp"
-#include "duckdb/planner/operator/logical_comparison_join.hpp"
+#include "duckdb/planner/operator/logical_window.hpp"
 
 namespace duckdb {
 
@@ -89,11 +89,13 @@ unique_ptr<LogicalOperator> FilterPushdown::Rewrite(unique_ptr<LogicalOperator>
 		return PushdownSetOperation(std::move(op));
 	case LogicalOperatorType::LOGICAL_DISTINCT:
 		return PushdownDistinct(std::move(op));
-	case LogicalOperatorType::LOGICAL_ORDER_BY: {
+	case LogicalOperatorType::LOGICAL_ORDER_BY:
 		// we can just push directly through these operations without any rewriting
 		op->children[0] = Rewrite(std::move(op->children[0]));
 		return op;
-	}
+	case LogicalOperatorType::LOGICAL_MATERIALIZED_CTE:
+		op->children[1] = Rewrite(std::move(op->children[1]));
+		return op;
 	case LogicalOperatorType::LOGICAL_GET:
 		return PushdownGet(std::move(op));
 	case LogicalOperatorType::LOGICAL_LIMIT:
diff --git a/src/parallel/task_scheduler.cpp b/src/parallel/task_scheduler.cpp
index 5f959d693cbb..79a3aa2e276a 100644
--- a/src/parallel/task_scheduler.cpp
+++ b/src/parallel/task_scheduler.cpp
@@ -278,9 +278,6 @@ void TaskScheduler::SetThreads(idx_t total_threads, idx_t external_threads) {
 	}
 #endif
 	requested_thread_count = NumericCast<int32_t>(total_threads - external_threads);
-	if (Allocator::SupportsFlush()) {
-		Allocator::FlushAll();
-	}
 }
 
 void TaskScheduler::SetAllocatorFlushTreshold(idx_t threshold) {
@@ -382,6 +379,9 @@ void TaskScheduler::RelaunchThreadsInternal(int32_t n) {
 		}
 	}
 	current_thread_count = NumericCast<int32_t>(threads.size() + config.options.external_threads);
+	if (Allocator::SupportsFlush()) {
+		Allocator::FlushAll();
+	}
 #endif
 }
 
diff --git a/src/parser/transform/statement/transform_create_function.cpp b/src/parser/transform/statement/transform_create_function.cpp
index bd2198d6c4d2..fda65d8b1008 100644
--- a/src/parser/transform/statement/transform_create_function.cpp
+++ b/src/parser/transform/statement/transform_create_function.cpp
@@ -1,11 +1,10 @@
+#include "duckdb/function/scalar_macro_function.hpp"
+#include "duckdb/function/table_macro_function.hpp"
+#include "duckdb/parser/expression/comparison_expression.hpp"
 #include "duckdb/parser/parsed_data/create_macro_info.hpp"
 #include "duckdb/parser/statement/create_statement.hpp"
-#include "duckdb/parser/expression/comparison_expression.hpp"
 #include "duckdb/parser/transformer.hpp"
 
-#include "duckdb/function/scalar_macro_function.hpp"
-#include "duckdb/function/table_macro_function.hpp"
-
 namespace duckdb {
 
 unique_ptr<MacroFunction> Transformer::TransformMacroFunction(duckdb_libpgquery::PGFunctionDefinition &def) {
@@ -25,7 +24,8 @@ unique_ptr<MacroFunction> Transformer::TransformMacroFunction(duckdb_libpgquery:
 	vector<unique_ptr<ParsedExpression>> parameters;
 	TransformExpressionList(*def.params, parameters);
 	for (auto &param : parameters) {
-		if (param->type == ExpressionType::VALUE_CONSTANT) {
+		Value const_param;
+		if (ConstructConstantFromExpression(*param, const_param)) {
 			// parameters with default value (must have an alias)
 			if (param->alias.empty()) {
 				throw ParserException("Invalid parameter: '%s'", param->ToString());
@@ -33,7 +33,9 @@ unique_ptr<MacroFunction> Transformer::TransformMacroFunction(duckdb_libpgquery:
 			if (macro_func->default_parameters.find(param->alias) != macro_func->default_parameters.end()) {
 				throw ParserException("Duplicate default parameter: '%s'", param->alias);
 			}
-			macro_func->default_parameters[param->alias] = std::move(param);
+			auto constructed_constant = make_uniq<ConstantExpression>(std::move(const_param));
+			constructed_constant->alias = param->alias;
+			macro_func->default_parameters[param->alias] = std::move(constructed_constant);
 		} else if (param->GetExpressionClass() == ExpressionClass::COLUMN_REF) {
 			// positional parameters
 			if (!macro_func->default_parameters.empty()) {
diff --git a/src/planner/binder/query_node/plan_cte_node.cpp b/src/planner/binder/query_node/plan_cte_node.cpp
index f221ade536f0..057ee60fc49d 100644
--- a/src/planner/binder/query_node/plan_cte_node.cpp
+++ b/src/planner/binder/query_node/plan_cte_node.cpp
@@ -26,30 +26,34 @@ unique_ptr<LogicalOperator> Binder::CreatePlan(BoundCTENode &node) {
 unique_ptr<LogicalOperator> Binder::CreatePlan(BoundCTENode &node, unique_ptr<LogicalOperator> base) {
 	// Generate the logical plan for the cte_query and child.
 	auto cte_query = CreatePlan(*node.query);
-	unique_ptr<LogicalOperator> cte_child;
+	unique_ptr<LogicalOperator> root;
 	if (node.child && node.child->type == QueryNodeType::CTE_NODE) {
-		cte_child = CreatePlan(node.child->Cast<BoundCTENode>(), std::move(base));
+		root = CreatePlan(node.child->Cast<BoundCTENode>(), std::move(base));
 	} else if (node.child) {
-		cte_child = CreatePlan(*node.child);
+		root = CreatePlan(*node.child);
 	} else {
-		cte_child = std::move(base);
+		root = std::move(base);
 	}
 
 	// Only keep the materialized CTE, if it is used
 	if (node.child_binder->bind_context.cte_references[node.ctename] &&
 	    *node.child_binder->bind_context.cte_references[node.ctename] > 0) {
-		auto root = make_uniq<LogicalMaterializedCTE>(node.ctename, node.setop_index, node.types.size(),
-		                                              std::move(cte_query), std::move(cte_child));
+
+		// Push the CTE through single-child operators so query modifiers appear ABOVE the CTE (internal issue #2652)
+		// Otherwise, we may have a LIMIT on top of the CTE, and an ORDER BY in the query, and we can't make a TopN
+		reference<unique_ptr<LogicalOperator>> cte_child = root;
+		while (cte_child.get()->children.size() == 1 && cte_child.get()->type != LogicalOperatorType::LOGICAL_CTE_REF) {
+			cte_child = cte_child.get()->children[0];
+		}
+		cte_child.get() = make_uniq<LogicalMaterializedCTE>(node.ctename, node.setop_index, node.types.size(),
+		                                                    std::move(cte_query), std::move(cte_child.get()));
 
 		// check if there are any unplanned subqueries left in either child
 		has_unplanned_dependent_joins = has_unplanned_dependent_joins ||
 		                                node.child_binder->has_unplanned_dependent_joins ||
 		                                node.query_binder->has_unplanned_dependent_joins;
-
-		return VisitQueryNode(node, std::move(root));
-	} else {
-		return VisitQueryNode(node, std::move(cte_child));
 	}
+	return VisitQueryNode(node, std::move(root));
 }
 
 } // namespace duckdb


Test patch:
diff --git a/test/sql/catalog/function/test_macro_issue_13104.test b/test/sql/catalog/function/test_macro_issue_13104.test
new file mode 100644
index 000000000000..acdbfee3a07c
--- /dev/null
+++ b/test/sql/catalog/function/test_macro_issue_13104.test
@@ -0,0 +1,36 @@
+# name: test/sql/catalog/function/test_macro_issue_13104.test
+# description: Test Issue 13104 - Macro default variable that is a boolean type fails
+# group: [function]
+
+statement ok
+create or replace macro my_macro(a:=true) as a;
+
+query I
+select my_macro()
+----
+true
+
+statement ok
+create or replace macro my_macro(a:=false) as a;
+
+query I
+select my_macro()
+----
+false
+
+# bonus: we can also use struct/list as default parameters now
+statement ok
+create or replace macro my_macro(a:={duck:42}) as a;
+
+query I
+select my_macro()
+----
+{'duck': 42}
+
+statement ok
+create or replace macro my_macro(a:=[42]) as a;
+
+query I
+select my_macro()
+----
+[42]
diff --git a/test/sql/cte/materialized/materialized_cte_modifiers.test b/test/sql/cte/materialized/materialized_cte_modifiers.test
new file mode 100644
index 000000000000..6dc54e4d79ae
--- /dev/null
+++ b/test/sql/cte/materialized/materialized_cte_modifiers.test
@@ -0,0 +1,37 @@
+# name: test/sql/cte/materialized/materialized_cte_modifiers.test
+# description: TopN should still work with materialized CTE (internal issue 2569)
+# group: [materialized]
+
+require tpcds
+
+statement ok
+call dsdgen(sf=0)
+
+# q01
+query II
+explain
+WITH customer_total_return AS
+  (SELECT sr_customer_sk AS ctr_customer_sk,
+          sr_store_sk AS ctr_store_sk,
+          sum(sr_return_amt) AS ctr_total_return
+   FROM store_returns,
+        date_dim
+   WHERE sr_returned_date_sk = d_date_sk
+     AND d_year = 2000
+   GROUP BY sr_customer_sk,
+            sr_store_sk)
+SELECT c_customer_id
+FROM customer_total_return ctr1,
+     store,
+     customer
+WHERE ctr1.ctr_total_return >
+    (SELECT avg(ctr_total_return)*1.2
+     FROM customer_total_return ctr2
+     WHERE ctr1.ctr_store_sk = ctr2.ctr_store_sk)
+  AND s_store_sk = ctr1.ctr_store_sk
+  AND s_state = 'TN'
+  AND ctr1.ctr_customer_sk = c_customer_sk
+ORDER BY c_customer_id
+LIMIT 100;
+----
+physical_plan	<REGEX>:.*TOP_N.*
diff --git a/test/sql/json/issues/issue13212.test b/test/sql/json/issues/issue13212.test
new file mode 100644
index 000000000000..228343056f9b
--- /dev/null
+++ b/test/sql/json/issues/issue13212.test
@@ -0,0 +1,23 @@
+# name: test/sql/json/issues/issue13212.test
+# description: Test issue 13212 - Trying to read an empty compressed JSON file deadlocks
+# group: [issues]
+
+require json
+
+statement ok
+copy (select range::int64 as id, range::varchar as name from range(0)) to '__TEST_DIR__/t1.json.gz' (format json, compression gzip)
+
+statement ok
+copy (select id, name from values (1, 'bob'), (2, 'tom') tbl(id, name)) to '__TEST_DIR__/t2.json.gz' (format json, compression gzip)
+
+query II
+SELECT * FROM read_ndjson_auto(['__TEST_DIR__/t1.json.gz', '__TEST_DIR__/t2.json.gz']);
+----
+1	bob
+2	tom
+
+query II
+SELECT * FROM read_ndjson(['__TEST_DIR__/t1.json.gz', '__TEST_DIR__/t2.json.gz'], columns={id: 'int64', name: 'varchar'});
+----
+1	bob
+2	tom