From 909320d959c2dfcfb56a7899f136311fb4a4f669 Mon Sep 17 00:00:00 2001 From: Evert Lammerts Date: Fri, 16 Oct 2026 13:53:22 +0000 Subject: [PATCH] Keep a view's file-like source alive through its relation --- src/include/duckdb_python/pyrelation.hpp | 2 + src/pyrelation.cpp | 118 ++++++- src/python_replacement_scan.cpp | 13 +- .../api/test_view_over_file_like_object.py | 300 ++++++++++++++++++ 4 files changed, 409 insertions(+), 24 deletions(-) create mode 100644 tests/fast/api/test_view_over_file_like_object.py diff --git a/src/include/duckdb_python/pyrelation.hpp b/src/include/duckdb_python/pyrelation.hpp index 7324b17e..6b086120 100644 --- a/src/include/duckdb_python/pyrelation.hpp +++ b/src/include/duckdb_python/pyrelation.hpp @@ -252,6 +252,7 @@ struct DuckDBPyRelation { shared_ptr GetSharedRelation() const { return rel; } + unique_ptr OwningSubquery() const; bool ContainsColumnByName(const string &name) const; @@ -259,6 +260,7 @@ struct DuckDBPyRelation { std::unique_ptr DeriveRelation(shared_ptr new_rel); private: + void CreateOwningView(const string &view_name, bool replace, bool temporary); string ToStringInternal(const BoxRendererConfig &config); string GenerateExpressionList(const string &function_name, const string &aggregated_columns, const string &groups = "", const string &function_parameter = "", diff --git a/src/pyrelation.cpp b/src/pyrelation.cpp index 75728b1c..eec7ba85 100644 --- a/src/pyrelation.cpp +++ b/src/pyrelation.cpp @@ -24,6 +24,9 @@ #include "duckdb/main/relation/table_relation.hpp" #include "duckdb/main/relation/explain_relation.hpp" #include "duckdb/main/relation/update_relation.hpp" +#include "duckdb/parser/expression/subquery_expression.hpp" +#include "duckdb/parser/parsed_expression_iterator.hpp" +#include "duckdb/parser/tableref/subqueryref.hpp" #include "duckdb_python/map.hpp" #include "duckdb_python/expression/pyexpression.hpp" #include "duckdb_python/arrow/arrow_export_utils.hpp" @@ -232,6 +235,100 @@ Relation &DuckDBPyRelation::GetRel() { return *rel; } +static shared_ptr OwningDependency(shared_ptr rel) { + auto dependency = make_shared_ptr(); + // The native relation carries the Python data the scan reads. Owning the wrapper instead would + // own its connection, and an open result holding that closes a cycle the garbage + // collector cannot see: connection to result to wrapper to connection. + dependency->AddDependency("source", RelationDependencyItem::Create(std::move(rel))); + return dependency; +} + +unique_ptr DuckDBPyRelation::OwningSubquery() const { + auto select = make_uniq(); + select->node = rel->GetQueryNode(); + auto subquery = make_uniq(std::move(select)); + subquery->external_dependency = OwningDependency(rel); + return std::move(subquery); +} + +static bool CopyKeepsDependency(const TableRef &ref) { + switch (ref.type) { + case TableReferenceType::BASE_TABLE: + case TableReferenceType::SUBQUERY: + case TableReferenceType::TABLE_FUNCTION: + case TableReferenceType::EXPRESSION_LIST: + case TableReferenceType::COLUMN_DATA: + case TableReferenceType::SHOW_REF: + return true; + default: + return false; + } +} + +static void FindCarrier(QueryNode &node, optional_ptr &carrier); + +static void FindCarrier(ParsedExpression &expr, optional_ptr &carrier) { + if (carrier) { + return; + } + if (expr.GetExpressionClass() == ExpressionClass::SUBQUERY) { + FindCarrier(*expr.Cast().SubqueryMutable()->node, carrier); + } + ParsedExpressionIterator::EnumerateChildren(expr, [&](ParsedExpression &child) { FindCarrier(child, carrier); }); +} + +//! The first TableRef, anywhere in the node and its expressions' subqueries, whose copy keeps a dependency +static void FindCarrier(QueryNode &node, optional_ptr &carrier) { + if (carrier) { + return; + } + ParsedExpressionIterator::EnumerateQueryNodeChildren( + node, + [&](unique_ptr &expr) { + if (expr) { + FindCarrier(*expr, carrier); + } + }, + [&](TableRef &ref) { + if (!carrier && CopyKeepsDependency(ref)) { + carrier = &ref; + } + }); +} + +//! The relation a view is made from: a view keeps only the query node, and the engine carries a dependency into a +//! view only on a TableRef, so one TableRef of the relation's own query node owns the native relation. +class OwningViewSource : public Relation { +public: + explicit OwningViewSource(shared_ptr source_p) + : Relation(source_p->context, source_p->type), source(std::move(source_p)) { + } + + const vector &Columns() override { + return source->Columns(); + } + + unique_ptr GetQueryNode() override { + auto node = source->GetQueryNode(); + optional_ptr carrier; + FindCarrier(*node, carrier); + // One is enough, as the native relation also owns whatever that TableRef carried; a query that reads no + // table has nothing to own. + if (carrier) { + carrier->external_dependency = OwningDependency(source); + } + return node; + } + + string ToString(idx_t depth) override { + return source->ToString(depth); + } + +private: + shared_ptr source; +}; + struct DescribeAggregateInfo { explicit DescribeAggregateInfo(string name_p, bool numeric_only = false) : name(std::move(name_p)), numeric_only(numeric_only) { @@ -1366,14 +1463,18 @@ void DuckDBPyRelation::ToCSV(const string &filename, const nb::object &sep, cons PyExecuteRelation(write_csv); } -// should this return a rel with the new view? -std::unique_ptr DuckDBPyRelation::CreateView(const string &view_name, bool replace) { +void DuckDBPyRelation::CreateOwningView(const string &view_name, bool replace, bool temporary) { try { - rel->CreateView(Identifier(view_name), replace); + make_shared_ptr(rel)->CreateView(Identifier(view_name), replace, temporary); } catch (std::exception &error) { engine::NameRefusal(*rel->context->GetContext(), error); throw; } +} + +// should this return a rel with the new view? +std::unique_ptr DuckDBPyRelation::CreateView(const string &view_name, bool replace) { + CreateOwningView(view_name, replace, /*temporary=*/false); return DeriveRelation(rel); } @@ -1425,12 +1526,7 @@ static void BindNameToRelation(QueryNode &node, const Identifier &name, Relation } std::unique_ptr DuckDBPyRelation::Query(const string &view_name, const string &sql_query) { - try { - rel->CreateView(Identifier(view_name), /*replace=*/true, /*temporary=*/true); - } catch (std::exception &error) { - engine::NameRefusal(*rel->context->GetContext(), error); - throw; - } + CreateOwningView(view_name, /*replace=*/true, /*temporary=*/true); auto &context = *rel->context->GetContext(); Parser parser(context); parser.ParseQuery(sql_query); @@ -1450,9 +1546,7 @@ std::unique_ptr DuckDBPyRelation::Query(const string &view_nam make_shared_ptr(rel->context->GetContext(), std::move(select_statement), "query_relation_" + StringUtil::GenerateRandomName(16), query_text); // File-like objects named in the source's node stay registered only while the source lives. - auto source_dependency = make_shared_ptr(); - source_dependency->AddDependency("source", RelationDependencyItem::Create(rel)); - query_relation->AddExternalDependency(std::move(source_dependency)); + query_relation->AddExternalDependency(OwningDependency(rel)); return DeriveRelation(std::move(query_relation)); } { diff --git a/src/python_replacement_scan.cpp b/src/python_replacement_scan.cpp index 0a488109..de9ddbcf 100644 --- a/src/python_replacement_scan.cpp +++ b/src/python_replacement_scan.cpp @@ -12,7 +12,6 @@ #include "duckdb/parser/expression/function_expression.hpp" #include "duckdb/common/typedefs.hpp" #include "duckdb_python/pandas/pandas_scan.hpp" -#include "duckdb/parser/tableref/subqueryref.hpp" #include "duckdb_python/pyrelation.hpp" #include "duckdb_python/python_dependency.hpp" #include @@ -131,17 +130,7 @@ unique_ptr PythonReplacementScan::TryReplacementObject(const nb::objec "created by another Connection and can therefore not be used by this Connection.", name); } - // create a subquery from the underlying relation object - auto select = make_uniq(); - select->node = pyrel->GetRel().GetQueryNode(); - auto subquery = make_uniq(std::move(select)); - auto dependency = make_uniq(); - // The native relation carries the Python data the scan reads. Owning the wrapper instead would - // own its connection, and an open result holding that closes a cycle the garbage - // collector cannot see: connection to result to wrapper to connection. - dependency->AddDependency("replacement_cache", RelationDependencyItem::Create(pyrel->GetSharedRelation())); - subquery->external_dependency = std::move(dependency); - return std::move(subquery); + return pyrel->OwningSubquery(); } else if (PolarsDataFrame::IsDataFrame(entry)) { // Polars DataFrames always go through one-time .to_arrow() materialization. // Polars's __arrow_c_stream__() serializes from its internal layout on every call, diff --git a/tests/fast/api/test_view_over_file_like_object.py b/tests/fast/api/test_view_over_file_like_object.py new file mode 100644 index 00000000..8a26cfba --- /dev/null +++ b/tests/fast/api/test_view_over_file_like_object.py @@ -0,0 +1,300 @@ +"""A view keeps what its relation reads, such as the store entry of a file like object, for as long as it exists.""" + +import gc +import io +import weakref + +import pytest + +import duckdb + +pytest.importorskip("fsspec") + +CSV = "a,b\n1,x\n2,y\n" +JSON = '[{"a": 1, "b": "x"}, {"a": 2, "b": "y"}]' +ROWS = [(1, "x"), (2, "y")] + + +@pytest.fixture(scope="module") +def parquet_data(tmp_path_factory): + path = tmp_path_factory.mktemp("parquet") / "t.parquet" + with duckdb.connect() as con: + con.execute(f"copy (select * from (values (1, 'x'), (2, 'y')) t(a, b)) to '{path}'") + return path.read_bytes() + + +def object_store_files(): + from duckdb.filesystem import ModifiedMemoryFileSystem + + return set(ModifiedMemoryFileSystem.store) + + +def read_view(con): + return con.sql("select a, b from v order by a").fetchall() + + +def read_csv_text(con, parquet_data): + return con.read_csv(io.StringIO(CSV)) + + +def read_csv_bytes(con, parquet_data): + return con.read_csv(io.BytesIO(CSV.encode())) + + +def read_json_text(con, parquet_data): + # Schema detection waits on worker threads that need the GIL this thread holds while binding. + con.execute("set threads = 1") + return con.read_json(io.StringIO(JSON)) + + +def read_parquet_bytes(con, parquet_data): + return con.read_parquet(io.BytesIO(parquet_data)) + + +def read_parquet_list(con, parquet_data): + return con.read_parquet([io.BytesIO(parquet_data)]) + + +READERS = [ + pytest.param(read_csv_text, id="read_csv StringIO"), + pytest.param(read_csv_bytes, id="read_csv BytesIO"), + pytest.param(read_json_text, id="read_json StringIO"), + pytest.param(read_parquet_bytes, id="read_parquet BytesIO"), + pytest.param(read_parquet_list, id="read_parquet list"), +] + + +class TestViewOutlivesItsRelation: + @pytest.mark.parametrize("read", READERS) + def test_to_view(self, duckdb_cursor, parquet_data, read): + read(duckdb_cursor, parquet_data).to_view("v") + gc.collect() + assert read_view(duckdb_cursor) == ROWS + + @pytest.mark.parametrize("replace", [pytest.param(True, id="replace"), pytest.param(False, id="no replace")]) + def test_create_view(self, duckdb_cursor, replace): + duckdb_cursor.read_csv(io.StringIO(CSV)).create_view("v", replace=replace) + gc.collect() + assert read_view(duckdb_cursor) == ROWS + + def test_rolled_back_drop(self, duckdb_cursor): + duckdb_cursor.read_csv(io.StringIO(CSV)).to_view("v") + gc.collect() + duckdb_cursor.execute("begin") + duckdb_cursor.execute("drop view v") + duckdb_cursor.execute("rollback") + gc.collect() + assert read_view(duckdb_cursor) == ROWS + + def test_query_temporary_view(self, duckdb_cursor): + duckdb_cursor.read_csv(io.StringIO(CSV)).query("v", "select count(*) from v") + gc.collect() + assert read_view(duckdb_cursor) == ROWS + + def test_query_temporary_view_of_a_statement(self, duckdb_cursor): + duckdb_cursor.read_csv(io.StringIO(CSV)).query("v", "create table t as select * from v") + gc.collect() + assert duckdb_cursor.sql("select a, b from t order by a").fetchall() == ROWS + assert read_view(duckdb_cursor) == ROWS + + def test_registered_relation(self, duckdb_cursor): + duckdb_cursor.register("v", duckdb_cursor.read_csv(io.StringIO(CSV))) + gc.collect() + assert read_view(duckdb_cursor) == ROWS + + +class TestViewKeepsItsShape: + def test_empty_view_name(self, duckdb_cursor): + duckdb_cursor.read_csv(io.StringIO(CSV)).to_view("") + assert duckdb_cursor.sql("select view_name from duckdb_views() where not internal").fetchall() == [("",)] + + def test_duplicate_column_names(self, duckdb_cursor): + duckdb_cursor.sql("select 1 as a, 2 as a").to_view("v") + assert [row[1] for row in duckdb_cursor.sql("pragma table_info('v')").fetchall()] == ["a", "a"] + rel = duckdb_cursor.sql("select * from v") + assert rel.columns == ["a", "a_1"] + assert rel.fetchall() == [(1, 2)] + + +def drop_view(con): + con.execute("drop view v") + + +def replace_view(con): + con.sql("select 3 as a, 'z' as b").to_view("v") + + +def query_again(con): + con.sql("select 3 as a, 'z' as b").query("v", "select count(*) from v") + + +def close(con): + con.close() + + +RELEASES = [ + pytest.param(drop_view, id="dropped"), + pytest.param(replace_view, id="replaced"), + pytest.param(close, id="connection closed"), +] + + +def assert_store_entry_lives_with_the_view(make, release): + before = object_store_files() + con = duckdb.connect() + make(con) + gc.collect() + assert read_view(con) == ROWS + assert len(object_store_files() - before) == 1 + release(con) + gc.collect() + assert object_store_files() - before == set() + con.close() + + +class TestStoreEntryLivesWithTheView: + @pytest.mark.parametrize("release", RELEASES) + @pytest.mark.parametrize("read", READERS) + def test_view(self, parquet_data, read, release): + assert_store_entry_lives_with_the_view(lambda con: read(con, parquet_data).to_view("v"), release) + + @pytest.mark.parametrize("release", RELEASES) + def test_view_created_without_replace(self, release): + assert_store_entry_lives_with_the_view( + lambda con: con.read_csv(io.StringIO(CSV)).create_view("v", replace=False), release + ) + + @pytest.mark.parametrize( + "release", + [ + pytest.param(drop_view, id="dropped"), + pytest.param(query_again, id="next query"), + pytest.param(close, id="connection closed"), + ], + ) + def test_temporary_view(self, release): + assert_store_entry_lives_with_the_view( + lambda con: con.read_csv(io.StringIO(CSV)).query("v", "select count(*) from v"), release + ) + + def test_view_over_a_join(self): + before = object_store_files() + con = duckdb.connect() + left = con.read_csv(io.StringIO(CSV)) + right = con.read_csv(io.StringIO("a,c\n1,p\n2,q\n")) + left.join(right, "a").to_view("v") + del left, right + gc.collect() + assert con.sql("select a, b, c from v order by a").fetchall() == [(1, "x", "p"), (2, "y", "q")] + assert len(object_store_files() - before) == 2 + con.execute("drop view v") + gc.collect() + assert object_store_files() - before == set() + con.close() + + def test_view_over_a_union(self): + before = object_store_files() + con = duckdb.connect() + con.read_csv(io.StringIO(CSV)).union(con.read_csv(io.StringIO("a,b\n3,z\n"))).to_view("v") + gc.collect() + assert read_view(con) == [*ROWS, (3, "z")] + assert len(object_store_files() - before) == 2 + con.execute("drop view v") + gc.collect() + assert object_store_files() - before == set() + con.close() + + def test_view_that_reads_its_source_only_through_the_query_name(self): + before = object_store_files() + con = duckdb.connect() + con.read_csv(io.StringIO(CSV)).query("x", "select (select count(*) from x) as n").to_view("v") + con.execute("drop view x") + gc.collect() + assert con.sql("select n from v").fetchall() == [(2,)] + assert len(object_store_files() - before) == 1 + con.execute("drop view v") + gc.collect() + assert object_store_files() - before == set() + con.close() + + @pytest.mark.parametrize( + ("sql", "expected"), + [ + pytest.param("with x as (select 0) select (select sum(a) from ({source})) as n", 3, id="select list"), + pytest.param("with x as (select 0) select 3 as n where (select sum(a) from ({source})) = 3", 3, id="where"), + pytest.param( + "with x as (select 0) select case when true then (select sum(a) from ({source})) end as n", + 3, + id="case", + ), + pytest.param( + "with x as (select 0) select (select sum(a) from ({source})) = any(select 3) as n", + True, + id="any operand", + ), + ], + ) + def test_view_that_reads_its_source_only_through_a_subquery_expression(self, sql, expected): + before = object_store_files() + con = duckdb.connect() + source = con.read_csv(io.StringIO(CSV)) + query = source.query("x", sql.format(source=source.sql_query())) + del source + con.execute("drop view x") + query.to_view("v") + del query + gc.collect() + assert con.sql("select n from v").fetchall() == [(expected,)] + assert len(object_store_files() - before) == 1 + con.execute("drop view v") + gc.collect() + assert object_store_files() - before == set() + con.close() + + def test_rolled_back_replacement(self): + before = object_store_files() + con = duckdb.connect() + con.read_csv(io.StringIO(CSV)).to_view("v") + con.execute("begin") + con.read_csv(io.StringIO("a,b\n3,z\n")).create_view("v", replace=True) + gc.collect() + assert len(object_store_files() - before) == 2 + con.execute("rollback") + gc.collect() + assert read_view(con) == ROWS + assert len(object_store_files() - before) == 1 + con.close() + + @pytest.mark.xfail( + strict=True, + raises=SystemError, + reason="pre existing error path: a failing call on an unnamed file like relation raises SystemError " + "(error return without exception set) and leaves its store entry, as to_table and insert_into do", + ) + def test_failed_create_view(self, duckdb_cursor): + duckdb_cursor.execute("create view v as select 1 as a") + before = object_store_files() + with pytest.raises(duckdb.CatalogException, match="already exists"): + duckdb_cursor.read_csv(io.StringIO(CSV)).create_view("v", replace=False) + gc.collect() + assert object_store_files() - before == set() + + +class TestDataFrameLivesWithTheView: + @pytest.mark.parametrize( + "release", [pytest.param(drop_view, id="dropped"), pytest.param(close, id="connection closed")] + ) + def test_released(self, release): + pd = pytest.importorskip("pandas") + con = duckdb.connect() + df = pd.DataFrame({"a": [1, 2]}) + frame = weakref.ref(df) + con.from_df(df).to_view("v") + del df + gc.collect() + assert frame() is not None + assert con.sql("select a from v order by a").fetchall() == [(1,), (2,)] + release(con) + gc.collect() + assert frame() is None + con.close()