fix race condition in spatial join with CTE

Author: Maxxen

src/spatial/operators/spatial_join_physical.cpp

@@ -452,6 +452,8 @@ class SpatialJoinGlobalState final : public GlobalSinkState {
 
 	// This is initialized in the finalize state
 	unique_ptr<FlatRTree> rtree = nullptr;
+
+	mutex combine_lock;
 };
 
 unique_ptr<GlobalSinkState> PhysicalSpatialJoin::GetGlobalSinkState(ClientContext &context) const {
@@ -580,6 +582,7 @@ SinkCombineResultType PhysicalSpatialJoin::Combine(ExecutionContext &context, Op
 	lstate.collection->FinalizePinState(lstate.append_state.pin_state);
 
 	// Append the local collection to the global collection
+	lock_guard<mutex> lock(gstate.combine_lock);
 	gstate.collection->Combine(*lstate.collection);
 
 	// Merge the non-null and non-empty count
@@ -1019,14 +1022,13 @@ class SpatialJoinGlobalSourceState final : public GlobalSourceState {
 		column_ids.push_back(op.build_side_key_types.size() + op.build_side_payload_types.size());
 
 		// We dont need to keep the tuples aroun after scanning
-		state.collection->InitializeScan(scan_state, std::move(column_ids), TupleDataPinProperties::DESTROY_AFTER_DONE);
+		state.collection->InitializeScan(scan_state, std::move(column_ids), TupleDataPinProperties::KEEP_EVERYTHING_PINNED);
 
 		tuples_maximum = state.collection->Count();
 	}
 
 	const PhysicalSpatialJoin &op;
 	TupleDataParallelScanState scan_state;
-
 	// How many tuples we have scanned so far
 	idx_t tuples_maximum = 0;
 	atomic<idx_t> tuples_scanned = {0};
@@ -1056,7 +1058,7 @@ class SpatialJoinLocalSourceState final : public LocalSourceState {
 		column_ids.push_back(op.build_side_key_types.size() + op.build_side_payload_types.size());
 
 		// We dont need to keep the tuples aroun after scanning
-		state.collection->InitializeScan(scan_state, std::move(column_ids), TupleDataPinProperties::DESTROY_AFTER_DONE);
+		state.collection->InitializeScan(scan_state, std::move(column_ids));
 		state.collection->InitializeScanChunk(scan_state, scan_chunk);
 	}
 

test/sql/join/spatial_join_cte.test

@@ -0,0 +1,29 @@
+# name: test/sql/join/spatial_join_cte.test
+# group: [join]
+
+require spatial
+
+query IIII
+with bigger as (
+select 'a' as region, ST_GeomFromText('POLYGON ((0 0, 0 10, 10 10, 10 0 , 0 0 ))') as geom
+      union all
+select 'b' as region, ST_GeomFromText('POLYGON ((50 50, 50 60, 60 60, 60 50 , 50 50 ))') as geom
+      union all
+select 'c' as region, ST_GeomFromText('POLYGON ((50 50, 50 60, 60 60, 60 50 , 50 50 ))') as geom
+      union all
+select 'd' as region, ST_GeomFromText('POLYGON ((50 50, 50 60, 60 60, 60 50 , 50 50 ))') as geom
+      union all
+select 'e' as region, ST_GeomFromText('POLYGON ((50 50, 50 60, 60 60, 60 50 , 50 50 ))') as geom
+      union all
+select 'f' as region, ST_GeomFromText('POLYGON ((50 50, 50 60, 60 60, 60 50 , 50 50 ))') as geom
+      union all
+select 'g' as region, ST_GeomFromText('POLYGON ((50 50, 50 60, 60 60, 60 50 , 50 50 ))') as geom
+), smaller as  (
+  select '1' as region, ST_GeomFromText('POLYGON ((2 2, 2 8, 8 8, 8 2 , 2 2 ))') as geom
+  union all
+  select '2' as region, ST_GeomFromText('POLYGON ((3 3, 3 5, 5 5, 5 3 , 3 3 ))') as geom
+)
+select * from smaller left join bigger on st_intersects(bigger.geom, smaller.geom) order by all;
+----
+1	POLYGON ((2 2, 2 8, 8 8, 8 2, 2 2))	a	POLYGON ((0 0, 0 10, 10 10, 10 0, 0 0))
+2	POLYGON ((3 3, 3 5, 5 5, 5 3, 3 3))	a	POLYGON ((0 0, 0 10, 10 10, 10 0, 0 0))