Repository: duckdb/duckdb (language: python)

Issue:
JDBC - Enable useage of ENUM type
This PR enables the usage of the ENUM type in the JDBC driver. 

The enums are translated to strings in the JNI part and then can be used inside Java. Instead of using the exact enum/type name in the meta data, the general type ENUM is used.

 Please have a close look if the check for discovering enums and the translation to strings is implemented correctly.




Discussion / hints:
Thanks for the PR! Looks great.

Fix patch:
diff --git a/src/execution/window_segment_tree.cpp b/src/execution/window_segment_tree.cpp
index 651a25ce7b6e..3292c4389228 100644
--- a/src/execution/window_segment_tree.cpp
+++ b/src/execution/window_segment_tree.cpp
@@ -1532,6 +1532,7 @@ class WindowDistinctSortTree : public MergeSortTree<idx_t, idx_t> {
 class WindowDistinctAggregatorGlobalState : public WindowAggregatorGlobalState {
 public:
 	using GlobalSortStatePtr = unique_ptr<GlobalSortState>;
+	using LocalSortStatePtr = unique_ptr<LocalSortState>;
 	using ZippedTuple = WindowDistinctSortTree::ZippedTuple;
 	using ZippedElements = WindowDistinctSortTree::ZippedElements;
 
@@ -1540,6 +1541,9 @@ class WindowDistinctAggregatorGlobalState : public WindowAggregatorGlobalState {
 
 	//! Compute the block starts
 	void MeasurePayloadBlocks();
+	//! Create a new local sort
+	optional_ptr<LocalSortState> InitializeLocalSort() const;
+
 	//! Patch up the previous index block boundaries
 	void PatchPrevIdcs();
 	bool TryPrepareNextStage(WindowDistinctAggregatorLocalState &lstate);
@@ -1549,13 +1553,13 @@ class WindowDistinctAggregatorGlobalState : public WindowAggregatorGlobalState {
 	idx_t memory_per_thread;
 
 	//! Finalize guard
-	mutex lock;
+	mutable mutex lock;
 	//! Finalize stage
 	atomic<PartitionSortStage> stage;
 	//! Tasks launched
 	idx_t total_tasks = 0;
 	//! Tasks launched
-	idx_t tasks_assigned = 0;
+	mutable idx_t tasks_assigned;
 	//! Tasks landed
 	mutable atomic<idx_t> tasks_completed;
 
@@ -1566,6 +1570,8 @@ class WindowDistinctAggregatorGlobalState : public WindowAggregatorGlobalState {
 
 	//! Sorting operations
 	GlobalSortStatePtr global_sort;
+	//! Local sort set
+	mutable vector<LocalSortStatePtr> local_sorts;
 	//! The block starts (the scanner doesn't know this) plus the total count
 	vector<idx_t> block_starts;
 
@@ -1586,7 +1592,7 @@ WindowDistinctAggregatorGlobalState::WindowDistinctAggregatorGlobalState(ClientC
                                                                          const WindowDistinctAggregator &aggregator,
                                                                          idx_t group_count)
     : WindowAggregatorGlobalState(context, aggregator, group_count), context(aggregator.context),
-      stage(PartitionSortStage::INIT), tasks_completed(0), merge_sort_tree(*this, group_count),
+      stage(PartitionSortStage::INIT), tasks_assigned(0), tasks_completed(0), merge_sort_tree(*this, group_count),
       levels_flat_native(aggr) {
 	payload_types.emplace_back(LogicalType::UBIGINT);
 
@@ -1642,6 +1648,16 @@ WindowDistinctAggregatorGlobalState::WindowDistinctAggregatorGlobalState(ClientC
 	}
 }
 
+optional_ptr<LocalSortState> WindowDistinctAggregatorGlobalState::InitializeLocalSort() const {
+	lock_guard<mutex> local_sort_guard(lock);
+	auto local_sort = make_uniq<LocalSortState>();
+	local_sort->Initialize(*global_sort, global_sort->buffer_manager);
+	++tasks_assigned;
+	local_sorts.emplace_back(std::move(local_sort));
+
+	return local_sorts.back().get();
+}
+
 class WindowDistinctAggregatorLocalState : public WindowAggregatorLocalState {
 public:
 	explicit WindowDistinctAggregatorLocalState(const WindowDistinctAggregatorGlobalState &aggregator);
@@ -1655,7 +1671,7 @@ class WindowDistinctAggregatorLocalState : public WindowAggregatorLocalState {
 	              idx_t count, idx_t row_idx);
 
 	//! Thread-local sorting data
-	LocalSortState local_sort;
+	optional_ptr<LocalSortState> local_sort;
 	//! Finalize stage
 	PartitionSortStage stage = PartitionSortStage::INIT;
 	//! Finalize scan block index
@@ -1694,9 +1710,6 @@ WindowDistinctAggregatorLocalState::WindowDistinctAggregatorLocalState(
 	InitSubFrames(frames, gastate.aggregator.exclude_mode);
 	payload_chunk.Initialize(Allocator::DefaultAllocator(), gastate.payload_types);
 
-	auto &global_sort = gastate.global_sort;
-	local_sort.Initialize(*global_sort, global_sort->buffer_manager);
-
 	sort_chunk.Initialize(Allocator::DefaultAllocator(), gastate.sort_types);
 	sort_chunk.data.back().Reference(payload_chunk.data[0]);
 
@@ -1743,10 +1756,14 @@ void WindowDistinctAggregatorLocalState::Sink(DataChunk &sink_chunk, DataChunk &
 		payload_chunk.Slice(*filter_sel, filtered);
 	}
 
-	local_sort.SinkChunk(sort_chunk, payload_chunk);
+	if (!local_sort) {
+		local_sort = gastate.InitializeLocalSort();
+	}
+
+	local_sort->SinkChunk(sort_chunk, payload_chunk);
 
-	if (local_sort.SizeInBytes() > gastate.memory_per_thread) {
-		local_sort.Sort(*gastate.global_sort, true);
+	if (local_sort->SizeInBytes() > gastate.memory_per_thread) {
+		local_sort->Sort(*gastate.global_sort, true);
 	}
 }
 
@@ -1761,9 +1778,8 @@ void WindowDistinctAggregatorLocalState::Finalize(WindowAggregatorGlobalState &g
 void WindowDistinctAggregatorLocalState::ExecuteTask() {
 	auto &global_sort = *gastate.global_sort;
 	switch (stage) {
-	case PartitionSortStage::INIT:
-		//	AddLocalState is thread-safe
-		global_sort.AddLocalState(local_sort);
+	case PartitionSortStage::SCAN:
+		global_sort.AddLocalState(*gastate.local_sorts[block_idx]);
 		break;
 	case PartitionSortStage::MERGE: {
 		MergeSorter merge_sorter(global_sort, global_sort.buffer_manager);
@@ -1795,8 +1811,20 @@ bool WindowDistinctAggregatorGlobalState::TryPrepareNextStage(WindowDistinctAggr
 
 	switch (stage.load()) {
 	case PartitionSortStage::INIT:
-		// Wait for all the local sorts to be processed
-		if (tasks_completed < locals) {
+		//	5: Sort sorted lexicographically increasing
+		total_tasks = local_sorts.size();
+		tasks_assigned = 0;
+		tasks_completed = 0;
+		lstate.stage = stage = PartitionSortStage::SCAN;
+		lstate.block_idx = tasks_assigned++;
+		return true;
+	case PartitionSortStage::SCAN:
+		// Process all the local sorts
+		if (tasks_assigned < total_tasks) {
+			lstate.stage = PartitionSortStage::SCAN;
+			lstate.block_idx = tasks_assigned++;
+			return true;
+		} else if (tasks_completed < tasks_assigned) {
 			return false;
 		}
 		global_sort->PrepareMergePhase();
@@ -1873,10 +1901,7 @@ void WindowDistinctAggregator::Finalize(WindowAggregatorState &gsink, WindowAggr
 	auto &ldstate = lstate.Cast<WindowDistinctAggregatorLocalState>();
 	ldstate.Finalize(gdsink, collection);
 
-	//	5: Sort sorted lexicographically increasing
-	ldstate.ExecuteTask();
-
-	// Merge in parallel
+	// Sort, merge and build the tree in parallel
 	while (gdsink.stage.load() != PartitionSortStage::FINISHED) {
 		if (gdsink.TryPrepareNextStage(ldstate)) {
 			ldstate.ExecuteTask();


Test patch:
diff --git a/test/sql/window/test_window_distinct.test_slow b/test/sql/window/test_window_distinct.test_slow
index 1eb8cf88cbca..bcea205e7b30 100644
--- a/test/sql/window/test_window_distinct.test_slow
+++ b/test/sql/window/test_window_distinct.test_slow
@@ -2,17 +2,11 @@
 # description: Windowed distinct aggregates at scale
 # group: [window]
 
-# FIXME - deadlock causing CI issues
-mode skip
-
 require tpch
 
 statement ok
 CALL dbgen(sf=1);
 
-statement ok
-PRAGMA enable_verification
-
 statement ok
 PRAGMA temp_directory='__TEST_DIR__/window_distinct'
 
@@ -28,9 +22,6 @@ ORDER BY ALL
 ----
 24004860 values hashing to d39903ddf93011916ff3354d8bded7ff
 
-statement ok
-PRAGMA disable_verification
-
 # Trigger external sorting of a partition
 statement ok
 PRAGMA threads=4