Update vendored DuckDB sources to 885d2b4e8d

Author: duckdblabs-bot

src/duckdb/extension/json/json_functions/json_transform.cpp

@@ -262,8 +262,8 @@ bool JSONTransform::GetStringVector(yyjson_val *vals[], const idx_t count, const
 		if (!unsafe_yyjson_is_str(val)) {
 			validity.SetInvalid(i);
 			if (success && options.strict_cast && !unsafe_yyjson_is_str(val)) {
-				options.error_message = StringUtil::Format("Unable to cast '%s' to " + EnumUtil::ToString(target.id()),
-				                                           JSONCommon::ValToString(val, 50));
+				options.error_message = StringUtil::Format(
+				    "Unable to cast '%s' to %s", JSONCommon::ValToString(val, 50), EnumUtil::ToString(target.id()));
 				options.object_index = i;
 				success = false;
 			}
@@ -424,8 +424,8 @@ bool JSONTransform::TransformObject(yyjson_val *objects[], yyjson_alc *alc, cons
 				if (found_keys[col_idx]) {
 					if (success && options.error_duplicate_key) {
 						options.error_message =
-						    StringUtil::Format("Duplicate key \"" + string(key_ptr, key_len) + "\" in object %s",
-						                       JSONCommon::ValToString(objects[i], 50));
+						    StringUtil::Format("Object %s has duplicate key \"%s\"",
+						                       JSONCommon::ValToString(objects[i], 50), string(key_ptr, key_len));
 						options.object_index = i;
 						success = false;
 					}
@@ -436,8 +436,8 @@ bool JSONTransform::TransformObject(yyjson_val *objects[], yyjson_alc *alc, cons
 				}
 			} else if (success && error_unknown_key && options.error_unknown_key) {
 				options.error_message =
-				    StringUtil::Format("Object %s has unknown key \"" + string(key_ptr, key_len) + "\"",
-				                       JSONCommon::ValToString(objects[i], 50));
+				    StringUtil::Format("Object %s has unknown key \"%s\"", JSONCommon::ValToString(objects[i], 50),
+				                       string(key_ptr, key_len));
 				options.object_index = i;
 				success = false;
 			}
@@ -453,8 +453,8 @@ bool JSONTransform::TransformObject(yyjson_val *objects[], yyjson_alc *alc, cons
 				nested_vals[col_idx][i] = nullptr;
 
 				if (success && options.error_missing_key) {
-					options.error_message = StringUtil::Format("Object %s does not have key \"" + names[col_idx] + "\"",
-					                                           JSONCommon::ValToString(objects[i], 50));
+					options.error_message = StringUtil::Format("Object %s does not have key \"%s\"",
+					                                           JSONCommon::ValToString(objects[i], 50), names[col_idx]);
 					options.object_index = i;
 					success = false;
 				}
@@ -750,7 +750,7 @@ static bool TransformObjectToMap(yyjson_val *objects[], yyjson_alc *alc, Vector
 	// Transform keys
 	if (!JSONTransform::Transform(keys, alc, MapVector::GetKeys(result), list_size, options, nullptr)) {
 		throw ConversionException(
-		    StringUtil::Format(options.error_message + ". Cannot default to NULL, because map keys cannot be NULL"));
+		    StringUtil::Format("%s. Cannot default to NULL, because map keys cannot be NULL", options.error_message));
 	}
 
 	// Transform values

src/duckdb/src/execution/operator/join/physical_hash_join.cpp

@@ -733,9 +733,9 @@ void JoinFilterPushdownInfo::PushInFilter(const JoinFilterPushdownFilter &info,
 	return;
 }
 
-unique_ptr<DataChunk> JoinFilterPushdownInfo::Finalize(ClientContext &context, JoinHashTable &ht,
+unique_ptr<DataChunk> JoinFilterPushdownInfo::Finalize(ClientContext &context, optional_ptr<JoinHashTable> ht,
                                                        JoinFilterGlobalState &gstate,
-                                                       const PhysicalOperator &op) const {
+                                                       const PhysicalComparisonJoin &op) const {
 	// finalize the min/max aggregates
 	vector<LogicalType> min_max_types;
 	for (auto &aggr_expr : min_max_aggregates) {
@@ -753,6 +753,7 @@ unique_ptr<DataChunk> JoinFilterPushdownInfo::Finalize(ClientContext &context, J
 	auto dynamic_or_filter_threshold = ClientConfig::GetSetting<DynamicOrFilterThresholdSetting>(context);
 	// create a filter for each of the aggregates
 	for (idx_t filter_idx = 0; filter_idx < join_condition.size(); filter_idx++) {
+		const auto cmp = op.conditions[join_condition[filter_idx]].comparison;
 		for (auto &info : probe_info) {
 			auto filter_col_idx = info.columns[filter_idx].probe_column_index.column_index;
 			auto min_idx = filter_idx * 2;
@@ -767,22 +768,44 @@ unique_ptr<DataChunk> JoinFilterPushdownInfo::Finalize(ClientContext &context, J
 				continue;
 			}
 			// if the HT is small we can generate a complete "OR" filter
-			if (ht.Count() > 1 && ht.Count() <= dynamic_or_filter_threshold) {
-				PushInFilter(info, ht, op, filter_idx, filter_col_idx);
+			if (ht && ht->Count() > 1 && ht->Count() <= dynamic_or_filter_threshold) {
+				PushInFilter(info, *ht, op, filter_idx, filter_col_idx);
 			}
 
 			if (Value::NotDistinctFrom(min_val, max_val)) {
-				// min = max - generate an equality filter
-				auto constant_filter = make_uniq<ConstantFilter>(ExpressionType::COMPARE_EQUAL, std::move(min_val));
+				// min = max - single value
+				// generate a "one-sided" comparison filter for the LHS
+				// Note that this also works for equalities.
+				auto constant_filter = make_uniq<ConstantFilter>(cmp, std::move(min_val));
 				info.dynamic_filters->PushFilter(op, filter_col_idx, std::move(constant_filter));
 			} else {
 				// min != max - generate a range filter
-				auto greater_equals =
-				    make_uniq<ConstantFilter>(ExpressionType::COMPARE_GREATERTHANOREQUALTO, std::move(min_val));
-				info.dynamic_filters->PushFilter(op, filter_col_idx, std::move(greater_equals));
-				auto less_equals =
-				    make_uniq<ConstantFilter>(ExpressionType::COMPARE_LESSTHANOREQUALTO, std::move(max_val));
-				info.dynamic_filters->PushFilter(op, filter_col_idx, std::move(less_equals));
+				// for non-equalities, the range must be half-open
+				// e.g., for lhs < rhs we can only use lhs <= max
+				switch (cmp) {
+				case ExpressionType::COMPARE_EQUAL:
+				case ExpressionType::COMPARE_GREATERTHAN:
+				case ExpressionType::COMPARE_GREATERTHANOREQUALTO: {
+					auto greater_equals =
+					    make_uniq<ConstantFilter>(ExpressionType::COMPARE_GREATERTHANOREQUALTO, std::move(min_val));
+					info.dynamic_filters->PushFilter(op, filter_col_idx, std::move(greater_equals));
+					break;
+				}
+				default:
+					break;
+				}
+				switch (cmp) {
+				case ExpressionType::COMPARE_EQUAL:
+				case ExpressionType::COMPARE_LESSTHAN:
+				case ExpressionType::COMPARE_LESSTHANOREQUALTO: {
+					auto less_equals =
+					    make_uniq<ConstantFilter>(ExpressionType::COMPARE_LESSTHANOREQUALTO, std::move(max_val));
+					info.dynamic_filters->PushFilter(op, filter_col_idx, std::move(less_equals));
+					break;
+				}
+				default:
+					break;
+				}
 			}
 		}
 	}
@@ -857,7 +880,7 @@ SinkFinalizeType PhysicalHashJoin::Finalize(Pipeline &pipeline, Event &event, Cl
 	Value min;
 	Value max;
 	if (filter_pushdown && !sink.skip_filter_pushdown && ht.Count() > 0) {
-		auto final_min_max = filter_pushdown->Finalize(context, ht, *sink.global_filter_state, *this);
+		auto final_min_max = filter_pushdown->Finalize(context, &ht, *sink.global_filter_state, *this);
 		min = final_min_max->data[0].GetValue(0);
 		max = final_min_max->data[1].GetValue(0);
 	} else if (TypeIsIntegral(conditions[0].right->return_type.InternalType())) {

src/duckdb/src/execution/operator/join/physical_nested_loop_join.cpp

@@ -11,13 +11,23 @@ namespace duckdb {
 
 PhysicalNestedLoopJoin::PhysicalNestedLoopJoin(LogicalOperator &op, PhysicalOperator &left, PhysicalOperator &right,
                                                vector<JoinCondition> cond, JoinType join_type,
-                                               idx_t estimated_cardinality)
+                                               idx_t estimated_cardinality,
+                                               unique_ptr<JoinFilterPushdownInfo> pushdown_info_p)
     : PhysicalComparisonJoin(op, PhysicalOperatorType::NESTED_LOOP_JOIN, std::move(cond), join_type,
                              estimated_cardinality) {
+
+	filter_pushdown = std::move(pushdown_info_p);
+
 	children.push_back(left);
 	children.push_back(right);
 }
 
+PhysicalNestedLoopJoin::PhysicalNestedLoopJoin(LogicalOperator &op, PhysicalOperator &left, PhysicalOperator &right,
+                                               vector<JoinCondition> cond, JoinType join_type,
+                                               idx_t estimated_cardinality)
+    : PhysicalNestedLoopJoin(op, left, right, std::move(cond), join_type, estimated_cardinality, nullptr) {
+}
+
 bool PhysicalJoin::HasNullValues(DataChunk &chunk) {
 	for (idx_t col_idx = 0; col_idx < chunk.ColumnCount(); col_idx++) {
 		UnifiedVectorFormat vdata;
@@ -130,30 +140,16 @@ bool PhysicalNestedLoopJoin::IsSupported(const vector<JoinCondition> &conditions
 //===--------------------------------------------------------------------===//
 // Sink
 //===--------------------------------------------------------------------===//
-class NestedLoopJoinLocalState : public LocalSinkState {
-public:
-	explicit NestedLoopJoinLocalState(ClientContext &context, const vector<JoinCondition> &conditions)
-	    : rhs_executor(context) {
-		vector<LogicalType> condition_types;
-		for (auto &cond : conditions) {
-			rhs_executor.AddExpression(*cond.right);
-			condition_types.push_back(cond.right->return_type);
-		}
-		right_condition.Initialize(Allocator::Get(context), condition_types);
-	}
-
-	//! The chunk holding the right condition
-	DataChunk right_condition;
-	//! The executor of the RHS condition
-	ExpressionExecutor rhs_executor;
-};
-
 class NestedLoopJoinGlobalState : public GlobalSinkState {
 public:
 	explicit NestedLoopJoinGlobalState(ClientContext &context, const PhysicalNestedLoopJoin &op)
 	    : right_payload_data(context, op.children[1].get().GetTypes()),
 	      right_condition_data(context, op.GetJoinTypes()), has_null(false),
 	      right_outer(PropagatesBuildSide(op.join_type)) {
+		if (op.filter_pushdown) {
+			skip_filter_pushdown = op.filter_pushdown->probe_info.empty();
+			global_filter_state = op.filter_pushdown->GetGlobalState(context, op);
+		}
 	}
 
 	mutex nj_lock;
@@ -165,6 +161,35 @@ class NestedLoopJoinGlobalState : public GlobalSinkState {
 	atomic<bool> has_null;
 	//! A bool indicating for each tuple in the RHS if they found a match (only used in FULL OUTER JOIN)
 	OuterJoinMarker right_outer;
+	//! Should we not bother pushing down filters?
+	bool skip_filter_pushdown = false;
+	//! The global filter states to push down (if any)
+	unique_ptr<JoinFilterGlobalState> global_filter_state;
+};
+
+class NestedLoopJoinLocalState : public LocalSinkState {
+public:
+	explicit NestedLoopJoinLocalState(ClientContext &context, const PhysicalNestedLoopJoin &op,
+	                                  NestedLoopJoinGlobalState &gstate)
+	    : rhs_executor(context) {
+		vector<LogicalType> condition_types;
+		for (auto &cond : op.conditions) {
+			rhs_executor.AddExpression(*cond.right);
+			condition_types.push_back(cond.right->return_type);
+		}
+		right_condition.Initialize(Allocator::Get(context), condition_types);
+
+		if (op.filter_pushdown) {
+			local_filter_state = op.filter_pushdown->GetLocalState(*gstate.global_filter_state);
+		}
+	}
+
+	//! The chunk holding the right condition
+	DataChunk right_condition;
+	//! The executor of the RHS condition
+	ExpressionExecutor rhs_executor;
+	//! Local state for accumulating filter statistics
+	unique_ptr<JoinFilterLocalState> local_filter_state;
 };
 
 vector<LogicalType> PhysicalNestedLoopJoin::GetJoinTypes() const {
@@ -178,40 +203,54 @@ vector<LogicalType> PhysicalNestedLoopJoin::GetJoinTypes() const {
 SinkResultType PhysicalNestedLoopJoin::Sink(ExecutionContext &context, DataChunk &chunk,
                                             OperatorSinkInput &input) const {
 	auto &gstate = input.global_state.Cast<NestedLoopJoinGlobalState>();
-	auto &nlj_state = input.local_state.Cast<NestedLoopJoinLocalState>();
+	auto &lstate = input.local_state.Cast<NestedLoopJoinLocalState>();
 
 	// resolve the join expression of the right side
-	nlj_state.right_condition.Reset();
-	nlj_state.rhs_executor.Execute(chunk, nlj_state.right_condition);
+	lstate.right_condition.Reset();
+	lstate.rhs_executor.Execute(chunk, lstate.right_condition);
+
+	if (filter_pushdown && !gstate.skip_filter_pushdown) {
+		filter_pushdown->Sink(lstate.right_condition, *lstate.local_filter_state);
+	}
 
 	// if we have not seen any NULL values yet, and we are performing a MARK join, check if there are NULL values in
 	// this chunk
 	if (join_type == JoinType::MARK && !gstate.has_null) {
-		if (HasNullValues(nlj_state.right_condition)) {
+		if (HasNullValues(lstate.right_condition)) {
 			gstate.has_null = true;
 		}
 	}
 
 	// append the payload data and the conditions
 	lock_guard<mutex> nj_guard(gstate.nj_lock);
 	gstate.right_payload_data.Append(chunk);
-	gstate.right_condition_data.Append(nlj_state.right_condition);
+	gstate.right_condition_data.Append(lstate.right_condition);
 	return SinkResultType::NEED_MORE_INPUT;
 }
 
 SinkCombineResultType PhysicalNestedLoopJoin::Combine(ExecutionContext &context,
                                                       OperatorSinkCombineInput &input) const {
+	auto &gstate = input.global_state.Cast<NestedLoopJoinGlobalState>();
+	auto &lstate = input.local_state.Cast<NestedLoopJoinLocalState>();
+
 	auto &client_profiler = QueryProfiler::Get(context.client);
 	context.thread.profiler.Flush(*this);
 	client_profiler.Flush(context.thread.profiler);
+	if (filter_pushdown && !gstate.skip_filter_pushdown) {
+		filter_pushdown->Combine(*gstate.global_filter_state, *lstate.local_filter_state);
+	}
 	return SinkCombineResultType::FINISHED;
 }
 
 SinkFinalizeType PhysicalNestedLoopJoin::Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
                                                   OperatorSinkFinalizeInput &input) const {
-	auto &gstate = input.global_state.Cast<NestedLoopJoinGlobalState>();
-	gstate.right_outer.Initialize(gstate.right_payload_data.Count());
-	if (gstate.right_payload_data.Count() == 0 && EmptyResultIfRHSIsEmpty()) {
+	auto &gsink = input.global_state.Cast<NestedLoopJoinGlobalState>();
+	if (filter_pushdown && !gsink.skip_filter_pushdown) {
+		(void)filter_pushdown->Finalize(context, nullptr, *gsink.global_filter_state, *this);
+	}
+
+	gsink.right_outer.Initialize(gsink.right_payload_data.Count());
+	if (gsink.right_payload_data.Count() == 0 && EmptyResultIfRHSIsEmpty()) {
 		return SinkFinalizeType::NO_OUTPUT_POSSIBLE;
 	}
 	return SinkFinalizeType::READY;
@@ -222,7 +261,8 @@ unique_ptr<GlobalSinkState> PhysicalNestedLoopJoin::GetGlobalSinkState(ClientCon
 }
 
 unique_ptr<LocalSinkState> PhysicalNestedLoopJoin::GetLocalSinkState(ExecutionContext &context) const {
-	return make_uniq<NestedLoopJoinLocalState>(context.client, conditions);
+	auto &gstate = sink_state->Cast<NestedLoopJoinGlobalState>();
+	return make_uniq<NestedLoopJoinLocalState>(context.client, *this, gstate);
 }
 
 //===--------------------------------------------------------------------===//

src/duckdb/src/execution/physical_plan/plan_comparison_join.cpp

@@ -94,7 +94,7 @@ PhysicalOperator &PhysicalPlanGenerator::PlanComparisonJoin(LogicalComparisonJoi
 	if (PhysicalNestedLoopJoin::IsSupported(op.conditions, op.join_type)) {
 		// inequality join: use nested loop
 		return Make<PhysicalNestedLoopJoin>(op, left, right, std::move(op.conditions), op.join_type,
-		                                    op.estimated_cardinality);
+		                                    op.estimated_cardinality, std::move(op.filter_pushdown));
 	}
 
 	for (auto &cond : op.conditions) {

src/duckdb/src/function/table/version/pragma_version.cpp

@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "0-dev2940"
+#define DUCKDB_PATCH_VERSION "0-dev2947"
 #endif
 #ifndef DUCKDB_MINOR_VERSION
 #define DUCKDB_MINOR_VERSION 3
@@ -8,10 +8,10 @@
 #define DUCKDB_MAJOR_VERSION 1
 #endif
 #ifndef DUCKDB_VERSION
-#define DUCKDB_VERSION "v1.3.0-dev2940"
+#define DUCKDB_VERSION "v1.3.0-dev2947"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "56d42bc46a"
+#define DUCKDB_SOURCE_ID "885d2b4e8d"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"

src/duckdb/src/include/duckdb/execution/operator/join/join_filter_pushdown.hpp

@@ -17,6 +17,7 @@ class DataChunk;
 class DynamicTableFilterSet;
 class LogicalGet;
 class JoinHashTable;
+class PhysicalComparisonJoin;
 struct GlobalUngroupedAggregateState;
 struct LocalUngroupedAggregateState;
 
@@ -69,8 +70,8 @@ struct JoinFilterPushdownInfo {
 
 	void Sink(DataChunk &chunk, JoinFilterLocalState &lstate) const;
 	void Combine(JoinFilterGlobalState &gstate, JoinFilterLocalState &lstate) const;
-	unique_ptr<DataChunk> Finalize(ClientContext &context, JoinHashTable &ht, JoinFilterGlobalState &gstate,
-	                               const PhysicalOperator &op) const;
+	unique_ptr<DataChunk> Finalize(ClientContext &context, optional_ptr<JoinHashTable> ht,
+	                               JoinFilterGlobalState &gstate, const PhysicalComparisonJoin &op) const;
 
 private:
 	void PushInFilter(const JoinFilterPushdownFilter &info, JoinHashTable &ht, const PhysicalOperator &op,

src/duckdb/src/include/duckdb/execution/operator/join/physical_nested_loop_join.hpp

@@ -18,6 +18,9 @@ class PhysicalNestedLoopJoin : public PhysicalComparisonJoin {
 	static constexpr const PhysicalOperatorType TYPE = PhysicalOperatorType::NESTED_LOOP_JOIN;
 
 public:
+	PhysicalNestedLoopJoin(LogicalOperator &op, PhysicalOperator &left, PhysicalOperator &right,
+	                       vector<JoinCondition> cond, JoinType join_type, idx_t estimated_cardinality,
+	                       unique_ptr<JoinFilterPushdownInfo> pushdown_info);
 	PhysicalNestedLoopJoin(LogicalOperator &op, PhysicalOperator &left, PhysicalOperator &right,
 	                       vector<JoinCondition> cond, JoinType join_type, idx_t estimated_cardinality);
 

src/duckdb/src/optimizer/join_filter_pushdown_optimizer.cpp

@@ -170,8 +170,17 @@ void JoinFilterPushdownOptimizer::GenerateJoinFilters(LogicalComparisonJoin &joi
 	vector<JoinFilterPushdownColumn> pushdown_columns;
 	for (idx_t cond_idx = 0; cond_idx < join.conditions.size(); cond_idx++) {
 		auto &cond = join.conditions[cond_idx];
-		if (cond.comparison != ExpressionType::COMPARE_EQUAL) {
-			// only equality supported for now
+		switch (cond.comparison) {
+		case ExpressionType::COMPARE_EQUAL:
+		case ExpressionType::COMPARE_LESSTHAN:
+		case ExpressionType::COMPARE_LESSTHANOREQUALTO:
+		case ExpressionType::COMPARE_GREATERTHAN:
+		case ExpressionType::COMPARE_GREATERTHANOREQUALTO:
+			break;
+		case ExpressionType::COMPARE_NOT_DISTINCT_FROM:
+			// TODO: Need OR predicates
+			continue;
+		default:
 			continue;
 		}
 		if (cond.left->GetExpressionType() != ExpressionType::BOUND_COLUMN_REF) {
@@ -215,8 +224,10 @@ void JoinFilterPushdownOptimizer::GenerateJoinFilters(LogicalComparisonJoin &joi
 
 	// Even if we cannot find any table sources in which we can push down filters,
 	// we still initialize the aggregate states so that we have the possibility of doing a perfect hash join
+	// TODO: Can ExpressionType::COMPARE_NOT_DISTINCT_FROM be used with perfect hash joins?
 	const auto compute_aggregates_anyway = join.join_type == JoinType::INNER && join.conditions.size() == 1 &&
 	                                       pushdown_info->join_condition.size() == 1 &&
+	                                       join.conditions[0].comparison == ExpressionType::COMPARE_EQUAL &&
 	                                       TypeIsIntegral(join.conditions[0].right->return_type.InternalType());
 	if (pushdown_info->probe_info.empty() && !compute_aggregates_anyway) {
 		// no table sources found in which we can push down filters