Throw a reasonable error when an array stream was already consumed (#306)

Related to #105

We shouldn't throw an InternalException when an Arrow stream was already
consumed.

Author: evertlammerts

src/duckdb_py/arrow/arrow_array_stream.cpp

@@ -71,7 +71,7 @@ unique_ptr<ArrowArrayStreamWrapper> PythonTableArrowArrayStreamFactory::Produce(
 		auto capsule = py::reinterpret_borrow<py::capsule>(arrow_obj_handle);
 		auto stream = capsule.get_pointer<struct ArrowArrayStream>();
 		if (!stream->release) {
-			throw InternalException("ArrowArrayStream was released by another thread/library");
+			throw InvalidInputException("This ArrowArrayStream has already been consumed and cannot be scanned again.");
 		}
 		res->arrow_array_stream = *stream;
 		stream->release = nullptr;
@@ -123,7 +123,7 @@ void PythonTableArrowArrayStreamFactory::GetSchemaInternal(py::handle arrow_obj_
 		auto capsule = py::reinterpret_borrow<py::capsule>(arrow_obj_handle);
 		auto stream = capsule.get_pointer<struct ArrowArrayStream>();
 		if (!stream->release) {
-			throw InternalException("ArrowArrayStream was released by another thread/library");
+			throw InvalidInputException("This ArrowArrayStream has already been consumed and cannot be scanned again.");
 		}
 		stream->get_schema(stream, &schema.arrow_schema);
 		return;

tests/fast/arrow/test_arrow_pycapsule.py

@@ -68,6 +68,16 @@ def test_automatic_reexecution(self, duckdb_cursor):
         assert len(res1) == 100
         assert res1 == res2
 
+    def test_pycapsule_rescan_error_type(self, duckdb_cursor):
+        """Issue #105: re-executing a relation backed by a consumed PyCapsule."""
+        pa = pytest.importorskip("pyarrow")
+        tbl = pa.table({"a": [1]})
+        capsule = tbl.__arrow_c_stream__()  # noqa: F841
+        rel = duckdb_cursor.sql("SELECT * FROM capsule")
+        rel.fetchall()  # consumes the capsule
+        with pytest.raises(duckdb.InvalidInputException):
+            rel.fetchall()  # re-execution should be InvalidInputException, not InternalException
+
     def test_consumer_interface_roundtrip(self, duckdb_cursor):
         def create_table():
             class MyTable: