Repository: duckdb/duckdb (language: python)

Issue:
python UDF: NULLs intercepted for scalars but not columnar args
### What happens?

According to https://duckdb.org/docs/api/python/function.html#null-handling, if you pass a NULL into a python UDF, duckdb will short-circuit and immediately resolve to NULL, and won't ever call the python with None.

This short-circuiting appears to happen when the input is a scalar, but doesn't when it is a column.

PS I think this short-circuiting (with a configurable override) is fantastic UX

### To Reproduce

```python
import duckdb
from duckdb.typing import INTEGER


def add_one(x):
    return x + 1


duckdb.create_function("add_one", add_one, [INTEGER], INTEGER, type="native")
duckdb.sql("SELECT add_one(1), add_one(NULL)").fetchall()
# [(2, None)]
duckdb.sql("SELECT add_one(x) FROM (SELECT UNNEST([1, NULL]) as x)").fetchall()
# Invalid Input Error: Python exception occurred while executing the UDF: TypeError: unsupported operand type(s) for +: 'NoneType' and 'int'
```

### OS:

MacM1, MacOS 14.5

### DuckDB Version:

1.0.0

### DuckDB Client:

python

### Full Name:

Nick Crews

### Affiliation:

Ship Creek Group

### What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have tested with a stable release

### Did you include all relevant data sets for reproducing the issue?

Yes

### Did you include all code required to reproduce the issue?

- [X] Yes, I have

### Did you include all relevant configuration (e.g., CPU architecture, Python version, Linux distribution) to reproduce the issue?

- [X] Yes, I have


Discussion / hints:
I was testing the very same thing today and noticed the same thing for user defined scalar functions. It works if you explicitly pass NULL as one of the arguments but doesn't if the NULL value is passed as a column value.
Repro with C API. I used 1.0.1-dev3196 as scalar functions aren't available in v1.

```c
state = duckdb_query(connection, "CREATE TABLE big_table_1 AS SELECT (greatest(random(), 0.1) * 1000)::BIGINT i FROM range(10) t(i);", &result);
state = duckdb_query(connection, "Insert into big_table_1 values (NULL);", &result);

auto function = duckdb_create_scalar_function();
duckdb_scalar_function_set_name(function, "my_null_count");

// set the variable arguments
auto any_type = duckdb_create_logical_type(DUCKDB_TYPE_ANY);
duckdb_scalar_function_set_varargs(function, any_type);
duckdb_destroy_logical_type(&any_type);

// set the return type uto bigint
auto return_type = duckdb_create_logical_type(DUCKDB_TYPE_UBIGINT);
duckdb_scalar_function_set_return_type(function, return_type);
duckdb_destroy_logical_type(&return_type);

// set up the function
duckdb_scalar_function_set_function(function, CountNULLValues);

// register and cleanup
state = duckdb_register_scalar_function(connection, function);
duckdb_destroy_scalar_function(&function);

state = duckdb_query(connection, "Select my_null_count(i, 42) from big_table_1", &result);

void CountNULLValues(duckdb_function_info, duckdb_data_chunk input, duckdb_vector output) {

	// Get the total number of rows and columns in this chunk.
	auto input_size = duckdb_data_chunk_get_size(input);
	auto column_count = duckdb_data_chunk_get_column_count(input);

	// Extract the validity masks.
	std::vector<uint64_t*> validity_masks;
	for (idx_t col_idx = 0; col_idx < column_count; col_idx++) {
		auto col = duckdb_data_chunk_get_vector(input, col_idx);
		auto validity_mask = duckdb_vector_get_validity(col);
		validity_masks.push_back(validity_mask);
	}

	// Execute the function.
	auto result_data = (uint64_t*)duckdb_vector_get_data(output);
	for (idx_t row_idx = 0; row_idx < input_size; row_idx++) {
		idx_t null_count = 0;
		for (idx_t col_idx = 0; col_idx < column_count; col_idx++) {
			if (!duckdb_validity_row_is_valid(validity_masks[col_idx], row_idx)) {
				null_count++;
			}
		}
		result_data[row_idx] = null_count;
	}
}
```

When `CountNULLValues` is executed `input_size` is 11: 10 rows in the table without NULL and one row with NULL
This is expected behavior, constant NULLs can be short-circuited because we know the input is NULL, a column can contain anything, we won't know what it returns until runtime
If the custom UDF still has to handle NULL input, what's the point of short-circuiting?
The initial PR adding this: <https://github.com/duckdb/duckdb/pull/3866>
The changelist is quite small, on `main` the relevant code acting on this that I could find is this:
```c++
    if (bound_function.null_handling == FunctionNullHandling::DEFAULT_NULL_HANDLING) {
        for (auto &child : children) {
            if (child->return_type == LogicalTypeId::SQLNULL) {
                return make_uniq<BoundConstantExpression>(Value(return_type_if_null));
            }
            if (!child->IsFoldable()) {
                continue;
            }
            Value result;
            if (!ExpressionExecutor::TryEvaluateScalar(context, *child, result)) {
                continue;
            }
            if (result.IsNull()) {
                return make_uniq<BoundConstantExpression>(Value(return_type_if_null));
            }
        }
    }
```
I see two potential benefits of short-circuiting:
1. UX, the python programmer doesn't have to worry about this case.
2. performance, we can avoid slow python computation in some cases

If we only short-circuit on scalars, we lose the benefit of 1. We almost entirely lose the benefit of 2 as well, since I think it is reasonable to assume that calling a UDF with a column of N elements is O(N), whereas calling with a scalar is O(1), so we are only getting a performance boost in the O(1) case, which is going to only account for a tiny fraction of a queries runtime.

So, unless we can short-circuit in the columnar case too, I think we should remove the complexity and NEVER short-circuit.

The other thing confounding this is that short-circuiting a python UDF is probably much easier than short-circuiting a pyarrow UDF. But ideally these two flavors should have the same behavior. So unless we can get consistent behavior, I think we should err on the side of usability/simplicity and never short-circuit.

Fix patch:
diff --git a/src/common/vector_operations/vector_copy.cpp b/src/common/vector_operations/vector_copy.cpp
index 7315a4e591dc..0880f23be632 100644
--- a/src/common/vector_operations/vector_copy.cpp
+++ b/src/common/vector_operations/vector_copy.cpp
@@ -35,10 +35,7 @@ static const ValidityMask &ExtractValidityMask(const Vector &v) {
 }
 
 void VectorOperations::Copy(const Vector &source_p, Vector &target, const SelectionVector &sel_p, idx_t source_count,
-                            idx_t source_offset, idx_t target_offset) {
-	D_ASSERT(source_offset <= source_count);
-	D_ASSERT(source_p.GetType() == target.GetType());
-	idx_t copy_count = source_count - source_offset;
+                            idx_t source_offset, idx_t target_offset, idx_t copy_count) {
 
 	SelectionVector owned_sel;
 	const SelectionVector *sel = &sel_p;
@@ -173,7 +170,7 @@ void VectorOperations::Copy(const Vector &source_p, Vector &target, const Select
 		D_ASSERT(source_children.size() == target_children.size());
 		for (idx_t i = 0; i < source_children.size(); i++) {
 			VectorOperations::Copy(*source_children[i], *target_children[i], sel_p, source_count, source_offset,
-			                       target_offset);
+			                       target_offset, copy_count);
 		}
 		break;
 	}
@@ -267,6 +264,14 @@ void VectorOperations::Copy(const Vector &source_p, Vector &target, const Select
 	}
 }
 
+void VectorOperations::Copy(const Vector &source_p, Vector &target, const SelectionVector &sel_p, idx_t source_count,
+                            idx_t source_offset, idx_t target_offset) {
+	D_ASSERT(source_offset <= source_count);
+	D_ASSERT(source_p.GetType() == target.GetType());
+	idx_t copy_count = source_count - source_offset;
+	VectorOperations::Copy(source_p, target, sel_p, source_count, source_offset, target_offset, copy_count);
+}
+
 void VectorOperations::Copy(const Vector &source, Vector &target, idx_t source_count, idx_t source_offset,
                             idx_t target_offset) {
 	VectorOperations::Copy(source, target, *FlatVector::IncrementalSelectionVector(), source_count, source_offset,
diff --git a/src/include/duckdb/common/vector_operations/vector_operations.hpp b/src/include/duckdb/common/vector_operations/vector_operations.hpp
index 514c703c9004..bd772fcad3ba 100644
--- a/src/include/duckdb/common/vector_operations/vector_operations.hpp
+++ b/src/include/duckdb/common/vector_operations/vector_operations.hpp
@@ -192,6 +192,8 @@ struct VectorOperations {
 	                 idx_t target_offset);
 	static void Copy(const Vector &source, Vector &target, const SelectionVector &sel, idx_t source_count,
 	                 idx_t source_offset, idx_t target_offset);
+	static void Copy(const Vector &source, Vector &target, const SelectionVector &sel, idx_t source_count,
+	                 idx_t source_offset, idx_t target_offset, idx_t copy_count);
 
 	// Copy the data of <source> to the target location, setting null values to
 	// NullValue<T>. Used to store data without separate NULL mask.
diff --git a/tools/pythonpkg/src/include/duckdb_python/pybind11/pybind_wrapper.hpp b/tools/pythonpkg/src/include/duckdb_python/pybind11/pybind_wrapper.hpp
index 4ab31a90ef23..6717648d0b03 100644
--- a/tools/pythonpkg/src/include/duckdb_python/pybind11/pybind_wrapper.hpp
+++ b/tools/pythonpkg/src/include/duckdb_python/pybind11/pybind_wrapper.hpp
@@ -79,7 +79,7 @@ template <class T>
 bool try_cast(const handle &object, T &result) {
 	try {
 		result = cast<T>(object);
-	} catch (cast_error &) {
+	} catch (pybind11::cast_error &) {
 		return false;
 	}
 	return true;
diff --git a/tools/pythonpkg/src/python_udf.cpp b/tools/pythonpkg/src/python_udf.cpp
index 53e3eb72c6c9..bca316f09e36 100644
--- a/tools/pythonpkg/src/python_udf.cpp
+++ b/tools/pythonpkg/src/python_udf.cpp
@@ -43,8 +43,7 @@ static py::object ConvertDataChunkToPyArrowTable(DataChunk &input, const ClientP
 	return pyarrow::ToArrowTable(types, names, ConvertToSingleBatch(types, names, input, options), options);
 }
 
-static void ConvertPyArrowToDataChunk(const py::object &table, Vector &out, ClientContext &context, idx_t count) {
-
+static void ConvertArrowTableToVector(const py::object &table, Vector &out, ClientContext &context, idx_t count) {
 	// Create the stream factory from the Table object
 	auto stream_factory = make_uniq<PythonTableArrowArrayStreamFactory>(table.ptr(), context.GetClientProperties());
 	auto stream_factory_produce = PythonTableArrowArrayStreamFactory::Produce;
@@ -68,7 +67,7 @@ static void ConvertPyArrowToDataChunk(const py::object &table, Vector &out, Clie
 
 	TableFunctionRef empty;
 	TableFunction dummy_table_function;
-	dummy_table_function.name = "ConvertPyArrowToDataChunk";
+	dummy_table_function.name = "ConvertArrowTableToVector";
 	TableFunctionBindInput bind_input(children, named_params, input_types, input_names, nullptr, nullptr,
 	                                  dummy_table_function, empty);
 	vector<LogicalType> return_types;
@@ -81,9 +80,6 @@ static void ConvertPyArrowToDataChunk(const py::object &table, Vector &out, Clie
 		    "The returned table from a pyarrow scalar udf should only contain one column, found %d",
 		    return_types.size());
 	}
-	// if (return_types[0] != out.GetType()) {
-	//	throw InvalidInputException("The type of the returned array (%s) does not match the expected type: '%s'", )
-	//}
 
 	DataChunk result;
 	// Reserve for STANDARD_VECTOR_SIZE instead of count, in case the returned table contains too many tuples
@@ -101,14 +97,51 @@ static void ConvertPyArrowToDataChunk(const py::object &table, Vector &out, Clie
 	}
 
 	VectorOperations::Cast(context, result.data[0], out, count);
+	out.Flatten(count);
 }
 
-static scalar_function_t CreateVectorizedFunction(PyObject *function, PythonExceptionHandling exception_handling) {
+static string NullHandlingError() {
+	return R"(
+The returned result contained NULL values, but the 'null_handling' was set to DEFAULT.
+If you want more control over NULL values then 'null_handling' should be set to SPECIAL.
+
+With DEFAULT all rows containing NULL have been filtered from the UDFs input.
+Those rows are automatically set to NULL in the final result.
+The UDF is not expected to return NULL values.
+	)";
+}
+
+static ValidityMask &GetResultValidity(Vector &result) {
+	auto vector_type = result.GetVectorType();
+	if (vector_type == VectorType::CONSTANT_VECTOR) {
+		return ConstantVector::Validity(result);
+	} else if (vector_type == VectorType::FLAT_VECTOR) {
+		return FlatVector::Validity(result);
+	} else {
+		throw InternalException("VectorType %s was not expected here (GetResultValidity)",
+		                        EnumUtil::ToString(vector_type));
+	}
+}
+
+static void VerifyVectorizedNullHandling(Vector &result, idx_t count) {
+	auto &validity = GetResultValidity(result);
+
+	if (validity.AllValid()) {
+		return;
+	}
+
+	throw InvalidInputException(NullHandlingError());
+}
+
+static scalar_function_t CreateVectorizedFunction(PyObject *function, PythonExceptionHandling exception_handling,
+                                                  FunctionNullHandling null_handling) {
 	// Through the capture of the lambda, we have access to the function pointer
 	// We just need to make sure that it doesn't get garbage collected
 	scalar_function_t func = [=](DataChunk &input, ExpressionState &state, Vector &result) -> void {
 		py::gil_scoped_acquire gil;
 
+		const bool default_null_handling = null_handling == FunctionNullHandling::DEFAULT_NULL_HANDLING;
+
 		// owning references
 		py::object python_object;
 		// Convert the input datachunk to pyarrow
@@ -119,6 +152,36 @@ static scalar_function_t CreateVectorizedFunction(PyObject *function, PythonExce
 			options = context.GetClientProperties();
 		}
 
+		auto result_validity = FlatVector::Validity(result);
+		SelectionVector selvec(input.size());
+		idx_t input_size = input.size();
+		if (default_null_handling) {
+			vector<UnifiedVectorFormat> vec_data(input.ColumnCount());
+			for (idx_t i = 0; i < input.ColumnCount(); i++) {
+				input.data[i].ToUnifiedFormat(input.size(), vec_data[i]);
+			}
+
+			idx_t index = 0;
+			for (idx_t i = 0; i < input.size(); i++) {
+				bool any_null = false;
+				for (idx_t col_idx = 0; col_idx < input.ColumnCount(); col_idx++) {
+					auto &vec = vec_data[col_idx];
+					if (!vec.validity.RowIsValid(vec.sel->get_index(i))) {
+						any_null = true;
+						break;
+					}
+				}
+				if (any_null) {
+					result_validity.SetInvalid(i);
+					continue;
+				}
+				selvec.set_index(index++, i);
+			}
+			if (index != input.size()) {
+				input.Slice(selvec, index);
+			}
+		}
+
 		auto pyarrow_table = ConvertDataChunkToPyArrowTable(input, options);
 		py::tuple column_list = pyarrow_table.attr("columns");
 
@@ -126,7 +189,9 @@ static scalar_function_t CreateVectorizedFunction(PyObject *function, PythonExce
 
 		// Call the function
 		auto ret = PyObject_CallObject(function, column_list.ptr());
+		bool exception_occurred = false;
 		if (ret == nullptr && PyErr_Occurred()) {
+			exception_occurred = true;
 			if (exception_handling == PythonExceptionHandling::FORWARD_ERROR) {
 				auto exception = py::error_already_set();
 				throw InvalidInputException("Python exception occurred while executing the UDF: %s", exception.what());
@@ -154,9 +219,43 @@ static scalar_function_t CreateVectorizedFunction(PyObject *function, PythonExce
 			}
 		}
 		// Convert the pyarrow result back to a DuckDB datachunk
-		ConvertPyArrowToDataChunk(python_object, result, state.GetContext(), count);
+		if (count != input_size) {
+			D_ASSERT(default_null_handling);
+			// We filtered out some NULLs, now we need to reconstruct the final result by adding the nulls back
+			Vector temp(result.GetType(), count);
+			// Convert the table into a temporary Vector
+			ConvertArrowTableToVector(python_object, temp, state.GetContext(), count);
+			if (!exception_occurred) {
+				VerifyVectorizedNullHandling(temp, count);
+			}
+			if (count) {
+				SelectionVector inverted(input_size);
+				// Create a SelVec that inverts the filtering
+				// example: count: 6, null_indices: 1,3
+				// input selvec: [0, 2, 4, 5]
+				// inverted selvec: [0, 0, 1, 1, 2, 3]
+				idx_t src_index = 0;
+				for (idx_t i = 0; i < input_size; i++) {
+					// Fill the gaps with the previous index
+					inverted.set_index(i, src_index);
+					if (src_index + 1 < count && selvec.get_index(src_index) == i) {
+						src_index++;
+					}
+				}
+				VectorOperations::Copy(temp, result, inverted, count, 0, 0, input_size);
+			}
+			for (idx_t i = 0; i < input_size; i++) {
+				FlatVector::SetNull(result, i, !result_validity.RowIsValid(i));
+			}
+			result.Verify(input_size);
+		} else {
+			ConvertArrowTableToVector(python_object, result, state.GetContext(), count);
+			if (default_null_handling && !exception_occurred) {
+				VerifyVectorizedNullHandling(result, count);
+			}
+		}
 
-		if (input.size() == 1) {
+		if (input_size == 1) {
 			result.SetVectorType(VectorType::CONSTANT_VECTOR);
 		}
 	};
@@ -164,12 +263,15 @@ static scalar_function_t CreateVectorizedFunction(PyObject *function, PythonExce
 }
 
 static scalar_function_t CreateNativeFunction(PyObject *function, PythonExceptionHandling exception_handling,
-                                              const ClientProperties &client_properties) {
+                                              const ClientProperties &client_properties,
+                                              FunctionNullHandling null_handling) {
 	// Through the capture of the lambda, we have access to the function pointer
 	// We just need to make sure that it doesn't get garbage collected
 	scalar_function_t func = [=](DataChunk &input, ExpressionState &state, Vector &result) -> void { // NOLINT
 		py::gil_scoped_acquire gil;
 
+		const bool default_null_handling = null_handling == FunctionNullHandling::DEFAULT_NULL_HANDLING;
+
 		// owning references
 		vector<py::object> python_objects;
 		vector<PyObject *> python_results;
@@ -177,12 +279,23 @@ static scalar_function_t CreateNativeFunction(PyObject *function, PythonExceptio
 		for (idx_t row = 0; row < input.size(); row++) {
 
 			auto bundled_parameters = py::tuple((int)input.ColumnCount());
+			bool contains_null = false;
 			for (idx_t i = 0; i < input.ColumnCount(); i++) {
 				// Fill the tuple with the arguments for this row
 				auto &column = input.data[i];
 				auto value = column.GetValue(row);
+				if (value.IsNull() && default_null_handling) {
+					contains_null = true;
+					break;
+				}
 				bundled_parameters[i] = PythonObject::FromValue(value, column.GetType(), client_properties);
 			}
+			if (contains_null) {
+				// Immediately insert None, no need to call the function
+				python_objects.push_back(py::none());
+				python_results[row] = py::none().ptr();
+				continue;
+			}
 
 			// Call the function
 			auto ret = PyObject_CallObject(function, bundled_parameters.ptr());
@@ -197,6 +310,8 @@ static scalar_function_t CreateNativeFunction(PyObject *function, PythonExceptio
 				} else {
 					throw NotImplementedException("Exception handling type not implemented");
 				}
+			} else if ((!ret || ret == Py_None) && default_null_handling) {
+				throw InvalidInputException(NullHandlingError());
 			}
 			python_objects.push_back(py::reinterpret_steal<py::object>(ret));
 			python_results[row] = ret;
@@ -304,7 +419,8 @@ struct PythonUDFData {
 		auto signature = GetSignature(udf);
 		auto sig_params = signature.attr("parameters");
 		auto return_annotation = signature.attr("return_annotation");
-		if (!py::none().is(return_annotation)) {
+		auto empty = py::module_::import("inspect").attr("Signature").attr("empty");
+		if (!py::none().is(return_annotation) && !empty.is(return_annotation)) {
 			shared_ptr<DuckDBPyType> pytype;
 			if (py::try_cast<shared_ptr<DuckDBPyType>>(return_annotation, pytype)) {
 				return_type = pytype->Type();
@@ -338,9 +454,9 @@ struct PythonUDFData {
 
 		scalar_function_t func;
 		if (vectorized) {
-			func = CreateVectorizedFunction(udf.ptr(), exception_handling);
+			func = CreateVectorizedFunction(udf.ptr(), exception_handling, null_handling);
 		} else {
-			func = CreateNativeFunction(udf.ptr(), exception_handling, client_properties);
+			func = CreateNativeFunction(udf.ptr(), exception_handling, client_properties, null_handling);
 		}
 		FunctionStability function_side_effects =
 		    side_effects ? FunctionStability::VOLATILE : FunctionStability::CONSISTENT;
diff --git a/tools/pythonpkg/src/typing/pytype.cpp b/tools/pythonpkg/src/typing/pytype.cpp
index 39172a12eb4a..7b03f167121c 100644
--- a/tools/pythonpkg/src/typing/pytype.cpp
+++ b/tools/pythonpkg/src/typing/pytype.cpp
@@ -127,6 +127,9 @@ static bool FromNumpyType(const py::object &type, LogicalType &result) {
 	auto obj = type();
 	// We convert these to string because the underlying physical
 	// types of a numpy type aren't consistent on every platform
+	if (!py::hasattr(obj, "dtype")) {
+		return false;
+	}
 	string type_str = py::str(obj.attr("dtype"));
 	if (type_str == "bool") {
 		result = LogicalType::BOOLEAN;
@@ -185,7 +188,7 @@ static LogicalType FromType(const py::type &obj) {
 		return result;
 	}
 
-	throw py::type_error("Could not convert from unknown 'type' to DuckDBPyType");
+	throw py::cast_error("Could not convert from unknown 'type' to DuckDBPyType");
 }
 
 static bool IsMapType(const py::tuple &args) {


Test patch:
diff --git a/tools/pythonpkg/tests/fast/udf/test_null_filtering.py b/tools/pythonpkg/tests/fast/udf/test_null_filtering.py
new file mode 100644
index 000000000000..f7bdb828fd79
--- /dev/null
+++ b/tools/pythonpkg/tests/fast/udf/test_null_filtering.py
@@ -0,0 +1,237 @@
+import duckdb
+import pytest
+
+pd = pytest.importorskip("pandas")
+pa = pytest.importorskip("pyarrow")
+from typing import Union
+import pyarrow.compute as pc
+import uuid
+import datetime
+import numpy as np
+import cmath
+from typing import NamedTuple, Any, List
+
+from duckdb.typing import *
+
+
+class Candidate(NamedTuple):
+    type: duckdb.typing.DuckDBPyType
+    variant_one: Any
+    variant_two: Any
+
+
+def layout(index: int):
+    return [
+        ['x', 'x', 'y'],
+        ['x', None, 'y'],
+        [None, 'y', None],
+        ['x', None, None],
+        [None, None, 'y'],
+        [None, None, None],
+    ][index]
+
+
+def get_table_data():
+    def add_variations(data, index: int):
+        data.extend(
+            [
+                {
+                    'a': layout(index),
+                    'b': layout(0),
+                    'c': layout(0),
+                },
+                {
+                    'a': layout(0),
+                    'b': layout(0),
+                    'c': layout(index),
+                },
+            ]
+        )
+
+    data = []
+    add_variations(data, 1)
+    add_variations(data, 2)
+    add_variations(data, 3)
+    add_variations(data, 4)
+    add_variations(data, 5)
+    return data
+
+
+def get_types():
+    return [
+        Candidate(TINYINT, -42, -21),
+        Candidate(SMALLINT, -512, -256),
+        Candidate(INTEGER, -131072, -65536),
+        Candidate(
+            BIGINT,
+            -17179869184,
+            -8589934592,
+        ),
+        Candidate(
+            UTINYINT,
+            254,
+            127,
+        ),
+        Candidate(
+            USMALLINT,
+            65535,
+            32767,
+        ),
+        Candidate(
+            UINTEGER,
+            4294967295,
+            2147483647,
+        ),
+        Candidate(UBIGINT, 18446744073709551615, 9223372036854776000),
+        Candidate(VARCHAR, 'long_string_test', 'smallstring'),
+        Candidate(
+            UUID, uuid.UUID('ffffffff-ffff-ffff-ffff-ffffffffffff'), uuid.UUID('ffffffff-ffff-ffff-ffff-000000000000')
+        ),
+        Candidate(
+            FLOAT,
+            0.12246409803628922,
+            0.24492819607257843,
+        ),
+        Candidate(
+            DOUBLE,
+            123142.12312416293784721232344,
+            246284.2462483259,
+        ),
+        Candidate(DATE, datetime.date(2005, 3, 11), datetime.date(1989, 1, 7)),
+        Candidate(TIMESTAMP, datetime.datetime(2009, 2, 13, 11, 5, 53), datetime.datetime(1989, 11, 20, 5, 3, 25)),
+        Candidate(
+            TIME,
+            datetime.time(14, 1, 12),
+            datetime.time(6, 3, 6),
+        ),
+        Candidate(
+            BLOB,
+            b'\xF6\x96\xB0\x85',
+            b'\x85\xB0\x96\xF6',
+        ),
+        Candidate(
+            INTERVAL,
+            datetime.timedelta(days=30969, seconds=999, microseconds=999999),
+            datetime.timedelta(days=786, seconds=53, microseconds=651),
+        ),
+        Candidate(
+            BOOLEAN,
+            True,
+            False,
+        ),
+        Candidate(
+            duckdb.struct_type(['BIGINT[]', 'VARCHAR[]']),
+            {'v1': [1, 2, 3], 'v2': ['a', 'non-inlined string', 'duckdb']},
+            {'v1': [5, 4, 3, 2, 1], 'v2': ['non-inlined-string', 'a', 'b', 'c', 'duckdb']},
+        ),
+        Candidate(duckdb.list_type('VARCHAR'), ['the', 'duck', 'non-inlined string'], ['non-inlined-string', 'test']),
+    ]
+
+
+def construct_query(tuples) -> str:
+    def construct_values_list(row, start_param_idx):
+        parameter_count = len(row)
+        parameters = [f'${x+start_param_idx}' for x in range(parameter_count)]
+        parameters = '(' + ', '.join(parameters) + ')'
+        return parameters
+
+    row_size = len(tuples[0])
+    values_list = [construct_values_list(x, 1 + (i * row_size)) for i, x in enumerate(tuples)]
+    values_list = ', '.join(values_list)
+
+    query = f"""
+        select * from (values {values_list})
+    """
+    return query
+
+
+def construct_parameters(tuples, dbtype):
+    parameters = []
+    for row in tuples:
+        parameters.extend(list([duckdb.Value(x, dbtype) for x in row]))
+    return parameters
+
+
+class TestUDFNullFiltering(object):
+    @pytest.mark.parametrize(
+        'table_data',
+        get_table_data(),
+    )
+    @pytest.mark.parametrize(
+        'test_type',
+        get_types(),
+    )
+    @pytest.mark.parametrize('udf_type', ['arrow', 'native'])
+    def test_null_filtering(self, duckdb_cursor, table_data: dict, test_type: Candidate, udf_type):
+        null_count = sum([1 for x in list(zip(*table_data.values())) if any([y == None for y in x])])
+        row_count = len(table_data)
+        table_data = {
+            key: [None if not x else test_type.variant_one if x == 'x' else test_type.variant_two for x in value]
+            for key, value in table_data.items()
+        }
+
+        tuples = list(zip(*table_data.values()))
+        query = construct_query(tuples)
+        parameters = construct_parameters(tuples, test_type.type)
+        rel = duckdb_cursor.sql(query + " t(a, b, c)", params=parameters)
+        rel.to_table('tbl')
+        rel.show()
+
+        def my_func(*args):
+            if udf_type == 'arrow':
+                my_func.count += len(args[0])
+            else:
+                my_func.count += 1
+            return args[0]
+
+        def create_parameters(table_data, dbtype):
+            return ", ".join(f'{key}::{dbtype}' for key in list(table_data.keys()))
+
+        my_func.count = 0
+        duckdb_cursor.create_function('test', my_func, None, test_type.type, type=udf_type)
+        query = f"select test({create_parameters(table_data, test_type.type)}) from tbl"
+        result = duckdb_cursor.sql(query).fetchall()
+
+        expected_output = [
+            (t[0],) if not any(x == None for x in t) else (None,) for t in list(zip(*table_data.values()))
+        ]
+        assert result == expected_output
+        assert len(result) == row_count
+        # Only the non-null tuples should have been seen by the UDF
+        assert my_func.count == row_count - null_count
+
+    @pytest.mark.parametrize(
+        'table_data',
+        [
+            [1, 2, 3, 4],
+            [1, 2, None, 4],
+        ],
+    )
+    def test_nulls_from_default_null_handling_native(self, duckdb_cursor, table_data):
+        def returns_null(x):
+            return None
+
+        df = pd.DataFrame({'a': table_data})
+        duckdb_cursor.execute("create table tbl as select * from df")
+        duckdb_cursor.create_function('test', returns_null, [str], int, type='native')
+        with pytest.raises(duckdb.InvalidInputException, match='The UDF is not expected to return NULL values'):
+            result = duckdb_cursor.sql("select test(a::VARCHAR) from tbl").fetchall()
+
+    @pytest.mark.parametrize(
+        'table_data',
+        [
+            [1, 2, 3, 4],
+            [1, 2, None, 4],
+        ],
+    )
+    def test_nulls_from_default_null_handling_arrow(self, duckdb_cursor, table_data):
+        def returns_null(x):
+            l = x.to_pylist()
+            return pa.array([None for _ in l], type=pa.int64())
+
+        df = pd.DataFrame({'a': table_data})
+        duckdb_cursor.execute("create table tbl as select * from df")
+        duckdb_cursor.create_function('test', returns_null, [str], int, type='arrow')
+        with pytest.raises(duckdb.InvalidInputException, match='The UDF is not expected to return NULL values'):
+            result = duckdb_cursor.sql("select test(a::VARCHAR) from tbl").fetchall()
+            print(result)