Add lock and task wait time logging

benchmark/micro/string/ilike_regular.benchmark

@@ -0,0 +1,19 @@
+# name: benchmark/micro/string/ilike_regular.benchmark
+# description: Case-insensitive contains word 'regular' in the l_comment (11.5%~)
+# group: [string]
+
+name ILike ('%REGULAR%')
+group string
+
+require tpch
+
+cache tpch_sf1.duckdb
+
+load
+CALL dbgen(sf=1);
+
+run
+SELECT COUNT(*) FROM lineitem WHERE l_comment ILIKE '%REGULAR%'
+
+result I
+687323

src/function/scalar/string/like.cpp

@@ -419,6 +419,10 @@ bool ILikeOperatorFunction(string_t &str, string_t &pattern, char escape = '\0')
 	LowerCase(pat_data, pat_size, pat_ldata.get());
 	string_t str_lcase(str_ldata.get(), UnsafeNumericCast<uint32_t>(str_llength));
 	string_t pat_lcase(pat_ldata.get(), UnsafeNumericCast<uint32_t>(pat_llength));
+	// '\0' is the "no escape" sentinel: use the non-escape matcher so embedded NUL bytes are matched literally
+	if (escape == '\0') {
+		return LikeOperatorFunction(str_lcase, pat_lcase);
+	}
 	return LikeOperatorFunction(str_lcase, pat_lcase, escape);
 }
 
@@ -491,6 +495,69 @@ void LikeEscapeFunction(DataChunk &args, ExpressionState &state, Vector &result)
 	    str, pattern, escape, result, args.size(), FUNC::template Operation<string_t, string_t, string_t>);
 }
 
+// Execution function for ILIKE / NOT ILIKE ... ESCAPE. Mirrors ILikeFunction: when the pattern and escape are
+// both constant, lowercase the pattern once and reuse a scratch buffer for the per-row string lowercasing. If the
+// escape character does not occur in the pattern, escape handling is a no-op, so we build the case-folded
+// LikeMatcher and use the fast SIMD contains path; otherwise we fall back to the generic escape-aware matcher on
+// the lowercased values. Non-constant pattern/escape falls back to the per-row ternary path.
+template <bool INVERT>
+void ILikeEscapeFunction(DataChunk &args, ExpressionState &state, Vector &result) {
+	auto &str_vec = args.data[0];
+	auto &pattern_vec = args.data[1];
+	auto &escape_vec = args.data[2];
+
+	if (pattern_vec.GetVectorType() == VectorType::CONSTANT_VECTOR && !ConstantVector::IsNull(pattern_vec) &&
+	    escape_vec.GetVectorType() == VectorType::CONSTANT_VECTOR && !ConstantVector::IsNull(escape_vec)) {
+		auto pattern = *ConstantVector::GetData<string_t>(pattern_vec);
+		auto escape = *ConstantVector::GetData<string_t>(escape_vec);
+		char escape_char = GetEscapeChar(escape);
+
+		// lowercase the pattern exactly once, up front
+		idx_t pat_llength = LowerLength(pattern.GetData(), pattern.GetSize());
+		auto pat_ldata = make_unsafe_uniq_array_uninitialized<char>(pat_llength);
+		LowerCase(pattern.GetData(), pattern.GetSize(), pat_ldata.get());
+		string_t pat_lcase(pat_ldata.get(), UnsafeNumericCast<uint32_t>(pat_llength));
+
+		// the matcher cannot honor escape semantics, so only use it when the escape char never appears in the
+		// (lowercased) pattern, in which case escape is irrelevant and the pattern is a plain LIKE pattern
+		unique_ptr<LikeMatcher> matcher;
+		bool escape_active =
+		    escape_char != '\0' && memchr(pat_lcase.GetData(), escape_char, pat_lcase.GetSize()) != nullptr;
+		if (!escape_active) {
+			matcher = LikeMatcher::CreateLikeMatcher(string(pat_lcase.GetData(), pat_lcase.GetSize()));
+		}
+
+		// reusable scratch buffer for lowercasing each string value (grown on demand)
+		idx_t scratch_size = 0;
+		unsafe_unique_array<char> scratch;
+		UnaryExecutor::Execute<string_t, bool>(str_vec, result, args.size(), [&](string_t str) {
+			idx_t str_llength = LowerLength(str.GetData(), str.GetSize());
+			if (str_llength > scratch_size) {
+				scratch = make_unsafe_uniq_array_uninitialized<char>(str_llength);
+				scratch_size = str_llength;
+			}
+			LowerCase(str.GetData(), str.GetSize(), scratch.get());
+			string_t str_lcase(scratch.get(), UnsafeNumericCast<uint32_t>(str_llength));
+			// '\0' escape means no escape: use the non-escape matcher so embedded NUL bytes are matched literally
+			bool match = matcher ? matcher->Match(str_lcase)
+			                     : (escape_char == '\0' ? LikeOperatorFunction(str_lcase, pat_lcase)
+			                                            : LikeOperatorFunction(str_lcase, pat_lcase, escape_char));
+			return INVERT ? !match : match;
+		});
+		return;
+	}
+	// non-constant pattern/escape: fall back to the generic per-row implementation
+	if (INVERT) {
+		TernaryExecutor::Execute<string_t, string_t, string_t, bool>(
+		    str_vec, pattern_vec, escape_vec, result, args.size(),
+		    NotILikeEscapeOperator::Operation<string_t, string_t, string_t>);
+	} else {
+		TernaryExecutor::Execute<string_t, string_t, string_t, bool>(
+		    str_vec, pattern_vec, escape_vec, result, args.size(),
+		    ILikeEscapeOperator::Operation<string_t, string_t, string_t>);
+	}
+}
+
 template <class ASCII_OP>
 unique_ptr<BaseStatistics> ILikePropagateStats(ClientContext &context, FunctionStatisticsInput &input) {
 	auto &child_stats = input.child_stats;
@@ -503,6 +570,48 @@ unique_ptr<BaseStatistics> ILikePropagateStats(ClientContext &context, FunctionS
 	return nullptr;
 }
 
+// Execution function for ILIKE / NOT ILIKE on the (possibly) Unicode path.
+// When the pattern is constant we lowercase it exactly once instead of once per row, and we reuse a single
+// scratch buffer to lowercase each string value instead of heap-allocating per row. This avoids two heap
+// allocations + a redundant pattern case-fold on every row that the generic ILikeOperatorFunction incurs.
+// (The ASCII-only fast path installed by ILikePropagateStats already avoids allocations, so it is unaffected.)
+template <class OP, bool INVERT>
+void ILikeFunction(DataChunk &input, ExpressionState &state, Vector &result) {
+	auto &pattern_vec = input.data[1];
+	if (pattern_vec.GetVectorType() == VectorType::CONSTANT_VECTOR && !ConstantVector::IsNull(pattern_vec)) {
+		// constant pattern: lowercase it exactly once, up front
+		auto pattern = *ConstantVector::GetData<string_t>(pattern_vec);
+		idx_t pat_llength = LowerLength(pattern.GetData(), pattern.GetSize());
+		auto pat_ldata = make_unsafe_uniq_array_uninitialized<char>(pat_llength);
+		LowerCase(pattern.GetData(), pattern.GetSize(), pat_ldata.get());
+		string_t pat_lcase(pat_ldata.get(), UnsafeNumericCast<uint32_t>(pat_llength));
+
+		// Build a case-insensitive matcher from the lowercased pattern. Because both the pattern and each string
+		// value are lowercased, a case-sensitive LikeMatcher over lowercased data is equivalent to ILIKE, and it
+		// uses the fast SIMD FindStrInStr/memcmp contains path. Returns null for patterns with '_' or only '%';
+		// in that case we fall back to the generic recursive matcher on the lowercased values.
+		auto matcher = LikeMatcher::CreateLikeMatcher(string(pat_lcase.GetData(), pat_lcase.GetSize()));
+
+		// reusable scratch buffer for lowercasing each string value (grown on demand)
+		idx_t scratch_size = 0;
+		unsafe_unique_array<char> scratch;
+		UnaryExecutor::Execute<string_t, bool>(input.data[0], result, input.size(), [&](string_t str) {
+			idx_t str_llength = LowerLength(str.GetData(), str.GetSize());
+			if (str_llength > scratch_size) {
+				scratch = make_unsafe_uniq_array_uninitialized<char>(str_llength);
+				scratch_size = str_llength;
+			}
+			LowerCase(str.GetData(), str.GetSize(), scratch.get());
+			string_t str_lcase(scratch.get(), UnsafeNumericCast<uint32_t>(str_llength));
+			bool match = matcher ? matcher->Match(str_lcase) : LikeOperatorFunction(str_lcase, pat_lcase);
+			return INVERT ? !match : match;
+		});
+		return;
+	}
+	// non-constant pattern: fall back to the generic per-row implementation
+	BinaryExecutor::ExecuteStandard<string_t, string_t, bool, OP>(input.data[0], input.data[1], result, input.size());
+}
+
 template <class OP, bool INVERT>
 void RegularLikeFunction(DataChunk &input, ExpressionState &state, Vector &result) {
 	auto &func_expr = state.expr.Cast<BoundFunctionExpression>();
@@ -537,16 +646,16 @@ ScalarFunction GlobPatternFun::GetFunction() {
 
 ScalarFunction ILikeFun::GetFunction() {
 	ScalarFunction ilike("~~*", {LogicalType::VARCHAR, LogicalType::VARCHAR}, LogicalType::BOOLEAN,
-	                     ScalarFunction::BinaryFunction<string_t, string_t, bool, ILikeOperator>, nullptr, nullptr,
+	                     ILikeFunction<ILikeOperator, false>, nullptr, nullptr,
 	                     ILikePropagateStats<ILikeOperatorASCII>);
 	ilike.SetCollationHandling(FunctionCollationHandling::PUSH_COMBINABLE_COLLATIONS);
 	return ilike;
 }
 
 ScalarFunction NotILikeFun::GetFunction() {
 	ScalarFunction not_ilike("!~~*", {LogicalType::VARCHAR, LogicalType::VARCHAR}, LogicalType::BOOLEAN,
-	                         ScalarFunction::BinaryFunction<string_t, string_t, bool, NotILikeOperator>, nullptr,
-	                         nullptr, ILikePropagateStats<NotILikeOperatorASCII>);
+	                         ILikeFunction<NotILikeOperator, true>, nullptr, nullptr,
+	                         ILikePropagateStats<NotILikeOperatorASCII>);
 	not_ilike.SetCollationHandling(FunctionCollationHandling::PUSH_COMBINABLE_COLLATIONS);
 	return not_ilike;
 }
@@ -568,15 +677,15 @@ ScalarFunction NotLikeEscapeFun::GetFunction() {
 
 ScalarFunction IlikeEscapeFun::GetFunction() {
 	ScalarFunction ilike_escape("ilike_escape", {LogicalType::VARCHAR, LogicalType::VARCHAR, LogicalType::VARCHAR},
-	                            LogicalType::BOOLEAN, LikeEscapeFunction<ILikeEscapeOperator>);
+	                            LogicalType::BOOLEAN, ILikeEscapeFunction<false>);
 	ilike_escape.SetCollationHandling(FunctionCollationHandling::PUSH_COMBINABLE_COLLATIONS);
 	return ilike_escape;
 }
 
 ScalarFunction NotIlikeEscapeFun::GetFunction() {
 	ScalarFunction not_ilike_escape("not_ilike_escape",
 	                                {LogicalType::VARCHAR, LogicalType::VARCHAR, LogicalType::VARCHAR},
-	                                LogicalType::BOOLEAN, LikeEscapeFunction<NotILikeEscapeOperator>);
+	                                LogicalType::BOOLEAN, ILikeEscapeFunction<true>);
 	not_ilike_escape.SetCollationHandling(FunctionCollationHandling::PUSH_COMBINABLE_COLLATIONS);
 	return not_ilike_escape;
 }

src/include/duckdb/parallel/pipeline.hpp

@@ -33,9 +33,7 @@ class PipelineTask : public ExecutorTask {
 	Pipeline &pipeline;
 	unique_ptr<PipelineExecutor> pipeline_executor;
 
-	string TaskType() const override {
-		return "PipelineTask";
-	}
+	string TaskType() const override;
 
 public:
 	const PipelineExecutor &GetPipelineExecutor() const;

src/include/duckdb/parallel/task.hpp

@@ -10,6 +10,7 @@
 
 #include "duckdb/common/common.hpp"
 #include "duckdb/common/optional_ptr.hpp"
+#include <chrono>
 
 namespace duckdb {
 class ClientContext;
@@ -58,6 +59,8 @@ class Task : public enable_shared_from_this<Task> {
 
 public:
 	optional_ptr<ProducerToken> token;
+	//! Timestamp when the task was enqueued, used for measuring queue wait time
+	std::chrono::steady_clock::time_point enqueue_time;
 };
 
 } // namespace duckdb

src/include/duckdb/parallel/task_scheduler.hpp

@@ -110,6 +110,21 @@ class TaskScheduler {
 	atomic<int32_t> requested_thread_count;
 	//! The amount of threads currently running
 	atomic<int32_t> current_thread_count;
+	//! The number of background workers currently executing a task (busy). Used for observability only.
+	atomic<int32_t> busy_workers;
+
+public:
+	//! Returns the number of background workers currently executing a task. Observability only.
+	int32_t GetBusyWorkerCount() const {
+		return busy_workers.load();
+	}
+	//! RAII helper to mark a worker busy/idle around task execution.
+	void IncrementBusyWorkers() {
+		++busy_workers;
+	}
+	void DecrementBusyWorkers() {
+		--busy_workers;
+	}
 };
 
 } // namespace duckdb

src/parallel/pipeline.cpp

@@ -20,6 +20,17 @@ PipelineTask::PipelineTask(Pipeline &pipeline_p, shared_ptr<Event> event_p)
     : ExecutorTask(pipeline_p.executor, std::move(event_p)), pipeline(pipeline_p) {
 }
 
+string PipelineTask::TaskType() const {
+	string result = "PipelineTask";
+	if (pipeline.GetSource()) {
+		result += "[src=" + pipeline.GetSource()->GetName() + "]";
+	}
+	if (pipeline.GetSink()) {
+		result += "[sink=" + pipeline.GetSink()->GetName() + "]";
+	}
+	return result;
+}
+
 bool PipelineTask::TaskBlockedOnResult() const {
 	// If this returns true, it means the pipeline this task belongs to has a cached chunk
 	// that was the result of the Sink method returning BLOCKED

src/parallel/pipeline_finish_event.cpp

@@ -61,6 +61,10 @@ class PipelineFinishTask : public ExecutorTask {
 	}
 
 	string TaskType() const override {
+		auto sink = pipeline.GetSink();
+		if (sink) {
+			return "PipelineFinishTask[sink=" + sink->GetName() + "]";
+		}
 		return "PipelineFinishTask";
 	}
 

src/parallel/task_scheduler.cpp

@@ -3,6 +3,7 @@
 #include "duckdb/common/chrono.hpp"
 #include "duckdb/common/exception.hpp"
 #include "duckdb/common/numeric_utils.hpp"
+#include "duckdb/logging/logger.hpp"
 #include "duckdb/main/client_context.hpp"
 #include "duckdb/main/database.hpp"
 #include "duckdb/main/settings.hpp"
@@ -72,6 +73,7 @@ struct QueueProducerToken {
 };
 
 void ConcurrentQueue::Enqueue(ProducerToken &token, shared_ptr<Task> task) {
+	task->enqueue_time = std::chrono::steady_clock::now();
 	lock_guard<mutex> producer_lock(token.producer_lock);
 	task->token = token;
 	if (q.enqueue(token.token->queue_token, std::move(task))) {
@@ -84,9 +86,11 @@ void ConcurrentQueue::Enqueue(ProducerToken &token, shared_ptr<Task> task) {
 
 void ConcurrentQueue::EnqueueBulk(ProducerToken &token, vector<shared_ptr<Task>> &tasks) {
 	typedef std::make_signed<std::size_t>::type ssize_t;
+	auto enqueue_time = std::chrono::steady_clock::now();
 	lock_guard<mutex> producer_lock(token.producer_lock);
 	for (auto &task : tasks) {
 		task->token = token;
+		task->enqueue_time = enqueue_time;
 	}
 	if (q.enqueue_bulk(token.token->queue_token, std::make_move_iterator(tasks.begin()), tasks.size())) {
 		tasks_in_queue += tasks.size();
@@ -228,7 +232,7 @@ TaskScheduler::TaskScheduler(DatabaseInstance &db)
     : db(db), queue(make_uniq<ConcurrentQueue>()),
       allocator_flush_threshold(db.config.options.allocator_flush_threshold),
       allocator_background_threads(Settings::Get<AllocatorBackgroundThreadsSetting>(db)), requested_thread_count(0),
-      current_thread_count(1) {
+      current_thread_count(1), busy_workers(0) {
 	SetAllocatorBackgroundThreads(allocator_background_threads);
 }
 
@@ -300,11 +304,69 @@ void TaskScheduler::ExecuteForever(atomic<bool> *marker) {
 			}
 		}
 		if (queue->Dequeue(task)) {
+			// This worker just transitioned idle->busy by picking up a task. Count it as busy before logging
+			// so the snapshot below can reach the busy==total saturation signature.
+			++busy_workers;
+			// Only log if enqueue_time was properly initialized (not epoch)
+			if (task->enqueue_time != std::chrono::steady_clock::time_point {}) {
+				auto dequeue_time = std::chrono::steady_clock::now();
+				auto queue_wait_ms =
+				    std::chrono::duration_cast<std::chrono::milliseconds>(dequeue_time - task->enqueue_time).count();
+				if (queue_wait_ms > 100) {
+					try {
+						// LOG 1: queue waiting time + task type.
+						// busy_workers/current_thread_count = workers currently inside Execute() vs total workers.
+						// tasks_in_queue = remaining queue depth at the moment this task was picked up.
+						// Read the table in the design notes: busy==total + depth>0 -> saturation/head-of-line
+						// blocking; busy<total + depth>0 -> scheduler bug (free worker, task not picked up);
+						// low busy + small depth + long wait -> OS/cgroup CPU throttling.
+						DUCKDB_LOG_WARNING(db,
+						                   "Task waited " + to_string(queue_wait_ms) +
+						                       "ms in queue before worker picked it up (task_type=" + task->TaskType() +
+						                       ", busy_workers=" + to_string(busy_workers.load()) + "/" +
+						                       to_string(current_thread_count.load()) +
+						                       ", queue_depth=" + to_string(queue->GetTasksInQueue()) + ")");
+					} catch (...) {
+						// Silently ignore if logging fails
+					}
+				}
+			}
 			auto process_mode = TaskExecutionMode::PROCESS_ALL;
 			if (Settings::Get<SchedulerProcessPartialSetting>(config)) {
 				process_mode = TaskExecutionMode::PROCESS_PARTIAL;
 			}
-			auto execute_result = task->Execute(process_mode);
+			// Capture the task type before Execute(): a TASK_FINISHED task may be reset right after, and in
+			// PROCESS_ALL mode this measures the whole task, while in PROCESS_PARTIAL mode it measures one slice.
+			auto task_type = task->TaskType();
+			auto execute_begin = std::chrono::steady_clock::now();
+			TaskExecutionResult execute_result;
+			try {
+				execute_result = task->Execute(process_mode);
+			} catch (...) {
+				--busy_workers;
+				throw;
+			}
+			--busy_workers;
+			{
+				auto execute_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+				                      std::chrono::steady_clock::now() - execute_begin)
+				                      .count();
+				if (execute_ms > 500) {
+					try {
+						// LOG 2: execution time + task type. A long-running, non-preemptive task here is what
+						// holds a worker and causes the queue waits reported by LOG 1.
+						DUCKDB_LOG_WARNING(
+						    db, "Task ran for " + to_string(execute_ms) +
+						            "ms before releasing the worker (task_type=" + task_type + ", mode=" +
+						            (process_mode == TaskExecutionMode::PROCESS_PARTIAL ? "partial" : "all") +
+						            ", busy_workers=" + to_string(busy_workers.load()) + "/" +
+						            to_string(current_thread_count.load()) +
+						            ", queue_depth=" + to_string(queue->GetTasksInQueue()) + ")");
+					} catch (...) {
+						// Silently ignore if logging fails
+					}
+				}
+			}
 
 			switch (execute_result) {
 			case TaskExecutionResult::TASK_FINISHED: