Implemented parallel execution of PredictExpressions

Author: ThatUdeshUK

src/execution/expression_executor/execute_conjunction.cpp

@@ -3,7 +3,11 @@
 #include "duckdb/planner/expression/bound_conjunction_expression.hpp"
 #include "duckdb/execution/adaptive_filter.hpp"
 
+#include <iostream>
 #include <random>
+#include <future>
+
+#define EXECUTE_PARALLEL 0
 
 namespace duckdb {
 
@@ -56,6 +60,10 @@ void ExpressionExecutor::Execute(const BoundConjunctionExpression &expr, Express
 idx_t ExpressionExecutor::Select(const BoundConjunctionExpression &expr, ExpressionState *state_p,
                                  const SelectionVector *sel, idx_t count, SelectionVector *true_sel,
                                  SelectionVector *false_sel) {
+#if EXECUTE_PARALLEL == 1
+	return SelectParallel(expr, state_p, sel, count, true_sel, false_sel);
+#endif
+
 	auto &state = state_p->Cast<ConjunctionState>();
 
 	if (expr.GetExpressionType() == ExpressionType::CONJUNCTION_AND) {
@@ -140,4 +148,104 @@ idx_t ExpressionExecutor::Select(const BoundConjunctionExpression &expr, Express
 	}
 }
 
+idx_t ExpressionExecutor::AsyncSelect(const Expression &expr, ExpressionState *state, const SelectionVector *sel,
+									  idx_t count, SelectionVector *true_sel, SelectionVector *false_sel) {
+	idx_t out = Select(expr, state, sel, count, true_sel, false_sel);
+	// for (idx_t j = 0; j < out; j++) {
+	// 	std::cout << "AsyncSelect- " << " idx- " << j << ", mapping- " << true_sel->get_index(j) << std::endl;
+	// }
+	return out;
+}
+
+idx_t ExpressionExecutor::SelectParallel(const BoundConjunctionExpression &expr, ExpressionState *state_p,
+                                 const SelectionVector *sel, idx_t count, SelectionVector *true_sel,
+                                 SelectionVector *false_sel) {
+	auto &state = state_p->Cast<ConjunctionState>();
+
+	if (expr.GetExpressionType() == ExpressionType::CONJUNCTION_AND) {
+		// get runtime statistics
+		auto filter_state = state.adaptive_filter->BeginFilter();
+		const SelectionVector *current_sel = sel;
+		idx_t false_count = 0;
+
+		unique_ptr<SelectionVector> temp_true, temp_false;
+		if (false_sel) {
+			temp_false = make_uniq<SelectionVector>(STANDARD_VECTOR_SIZE);
+		}
+		if (!true_sel) {
+			temp_true = make_uniq<SelectionVector>(STANDARD_VECTOR_SIZE);
+			true_sel = temp_true.get();
+		}
+
+		vector<std::future<idx_t>> futures;
+		// vector<idx_t> futures_i;
+		vector<unique_ptr<SelectionVector>> children_true;
+		for (idx_t i = 0; i < expr.children.size(); i++) {
+			children_true.push_back(make_uniq<SelectionVector>(STANDARD_VECTOR_SIZE));
+		}
+
+		vector<SelectionVector> children_false{expr.children.size(), SelectionVector{STANDARD_VECTOR_SIZE}};
+		vector<idx_t> true_counts{expr.children.size(), 0};
+
+		for (idx_t i = 0; i < expr.children.size(); i++) {
+			// futures_i.push_back(AsyncSelect(*expr.children[state.adaptive_filter->permutation[i]],
+			// 								state.child_states[state.adaptive_filter->permutation[i]].get(),
+			// 								current_sel, count, children_true[i].get(), &children_false[i]));
+
+			// for (idx_t j = 0; j < futures_i[i]; j++) {
+			// 	std::cout << "InLoopFilter- " << i << " , idx- " << j << ", mapping- " << children_true[i]->get_index(j) << std::endl;
+			// }
+			futures.push_back(std::async(std::launch::async, &ExpressionExecutor::AsyncSelect,
+								this, std::cref(*expr.children[state.adaptive_filter->permutation[i]]),
+								state.child_states[state.adaptive_filter->permutation[i]].get(),
+								std::cref(current_sel), count, children_true[i].get(), &children_false[i]));
+		}
+
+		for (idx_t i = 0; i < expr.children.size(); i++) {
+			// idx_t tcount = futures_i[i];
+			idx_t tcount = futures[i].get();
+			idx_t fcount = count - tcount;
+			if (fcount > 0 && false_sel) {
+				// move failing tuples into the false_sel
+				// tuples passed, move them into the actual result vector
+				for (idx_t j = 0; j < fcount; j++) {
+					false_sel->set_index(false_count++, children_false[i].get_index(j));
+				}
+			}
+			true_counts[i] = tcount;
+			// std::cout << "Filtered: " << tcount << std::endl;
+		}
+		idx_t true_count = 0;
+		idx_t lp = 0;
+		idx_t rp = 0;
+		while (lp < true_counts[0] && rp < true_counts[1]) {
+			idx_t lidx = children_true[0]->get_index(lp);
+			idx_t ridx = children_true[1]->get_index(rp);
+			if (lidx == ridx) {
+				// std::cout << "Match: " << lidx << std::endl;
+				true_sel->set_index(true_count, lidx);
+				true_count++;
+				lp++;
+				rp++;
+			} else if (lidx < ridx) {
+				lp++;
+			} else if (ridx < lidx) {
+				rp++;
+			}
+		}
+		// for (idx_t i = 0; i < children_true.size(); i++) {
+		// 	for (idx_t j = 0; j < true_counts[i]; j++) {
+		// 		std::cout << "Filter- " << i << " , idx- " << j << ", mapping- " << children_true[i]->get_index(j) << std::endl;
+		// 		true_sel->set_index(true_count, children_true[i]->get_index(j));
+		// 		true_count++;
+		// 	}
+		// }
+		// adapt runtime statistics
+		state.adaptive_filter->EndFilter(filter_state);
+		return true_count;
+	} else {
+		throw std::runtime_error("Expression executor does not support select_parallel() with !AND");
+	}
+}
+
 } // namespace duckdb

src/include/duckdb/execution/expression_executor.hpp

@@ -165,6 +165,10 @@ class ExpressionExecutor {
 	             SelectionVector *true_sel, SelectionVector *false_sel);
 	idx_t Select(const BoundConjunctionExpression &expr, ExpressionState *state, const SelectionVector *sel,
 	             idx_t count, SelectionVector *true_sel, SelectionVector *false_sel);
+	idx_t SelectParallel(const BoundConjunctionExpression &expr, ExpressionState *state, const SelectionVector *sel,
+						 idx_t count, SelectionVector *true_sel, SelectionVector *false_sel);
+	idx_t AsyncSelect(const Expression &expr, ExpressionState *state, const SelectionVector *sel,
+					  idx_t count, SelectionVector *true_sel, SelectionVector *false_sel);
 
 	//! Verify that the output of a step in the ExpressionExecutor is correct
 	void Verify(const Expression &expr, Vector &result, idx_t count);

third_party/predictors/common/common.hpp

@@ -15,7 +15,7 @@ class PromptUtil {
 	static std::string extract_json(const std::string &text) {
 		const size_t start = text.find_first_of("{[");
 		if (start == std::string::npos) {
-			throw std::runtime_error("No JSON start found");
+			throw std::runtime_error("No JSON start found: " + text);
 		}
 
 		const char open = text[start];

third_party/predictors/llm_api/duckdb_llm_api.cpp

@@ -202,6 +202,7 @@ std::unique_ptr<BatchResult> LlmApiPredictor::PredictBatch(OpenAI &api, const ve
 		request["model"] = this->model_path;
 		request["messages"] = {{{"content", GenerateSystemMessage(true)}, {"role", "system"}},
 							   {{"content", rewritten}, {"role", "user"}}};
+		request["temperature"] = 0.2;
 #if IS_SCHEMA
 		std::stringstream sch;
 		sch << "{\"type\":\"json_schema\",\"json_schema\":{\"name\":\"json_response\",\"strict\":true,";
@@ -224,6 +225,7 @@ std::unique_ptr<BatchResult> LlmApiPredictor::PredictBatch(OpenAI &api, const ve
 				LLM_LOG( "Too much requests!\n");
 			}
 		} else {
+			LLM_LOG("LLM Output: " + completion.dump());
 			tokens += completion["usage"]["total_tokens"].get<int>();
 			in_tokens += completion["usage"]["prompt_tokens"].get<int>();
 			out_tokens += completion["usage"]["completion_tokens"].get<int>();

third_party/predictors/llm_api/https_client.cpp

@@ -147,30 +147,6 @@ unique_ptr<HTTPParams> HTTPSUtil::InitializeParameters(optional_ptr<FileOpener>
 	                                 info);
 	FileOpener::TryGetCurrentSetting(opener, "ca_cert_file", result->ca_cert_file, info);
 
-	// HTTP Secret lookups
-	KeyValueSecretReader settings_reader(*opener, info, "http");
-
-	if (string proxy_setting;
-	    settings_reader.TryGetSecretKey<string>("http_proxy", proxy_setting) && !proxy_setting.empty()) {
-		idx_t port;
-		string host;
-		ParseHTTPProxyHost(proxy_setting, host, port);
-		result->http_proxy = host;
-		result->http_proxy_port = port;
-	}
-	settings_reader.TryGetSecretKey<string>("http_proxy_username", result->http_proxy_username);
-	settings_reader.TryGetSecretKey<string>("http_proxy_password", result->http_proxy_password);
-	settings_reader.TryGetSecretKey<string>("bearer_token", result->bearer_token);
-
-	if (Value extra_headers; settings_reader.TryGetSecretKey("extra_http_headers", extra_headers)) {
-		auto children = MapValue::GetChildren(extra_headers);
-		for (const auto &child : children) {
-			auto kv = StructValue::GetChildren(child);
-			D_ASSERT(kv.size() == 2);
-			result->extra_headers[kv[0].GetValue<string>()] = kv[1].GetValue<string>();
-		}
-	}
-
 	return std::move(result);
 }