diff --git a/docs/reports/0.2.4-native-memory.md b/docs/reports/0.2.4-native-memory.md
new file mode 100644
index 0000000..ec38d7b
--- /dev/null
+++ b/docs/reports/0.2.4-native-memory.md
@@ -0,0 +1,128 @@
+# Standalone native memory ownership fix for 0.2.4
+
+## Root causes
+
+The Python projection did not own newly allocated databases returned by the
+native factories. Their destructors never released layers, cached features or
+file backing. Numeric column lookup also requested row zero through
+`tryGetFeature`, allocating a feature-pointer vector and a geometry-pointer
+vector for every row. Bulk columns now use the table base, field offset and
+row stride directly.
+
+`Table` and its cached column accessor retained each other. That cycle kept
+one complete backing allocation alive per read/write cycle until cyclic GC.
+Build-only references also retained native builders after finalization.
+
+## Lifetime design
+
+The buffer loader acquires an independent memoryview and `Py_buffer` in the
+same C call that constructs the native database. Its cookie holds that export
+and, on successful shared-memory loading, the mapping owner. The real native
+destructor first destroys its layers and features; its callback then releases
+the export and closes the mapping. Failed loads release the export and use one factory error;
+high-level shared-memory loaders close mappings they opened on failure.
+
+Factory results are owning SWIG proxies. Borrowed layers, features, references
+and chunks retain their parents. Chunk arrays set their actual chunk as base
+with `PyArray_SetBaseObject` in C; the owner argument is supplied internally by
+SWIG and cannot be overridden by a caller. Numeric column arrays retain their
+layer through the array interface. Raw arrays, slices, rows, tables and column
+accessors keep backing alive until their last reference is released.
+
+Tables strongly cache their column accessors and arrays. These projections
+share a backing-state object that never references a table, accessor or view.
+The state and write callbacks retain their RecordEngine, which maintains one
+registry of weak state and table references with direct lookup by name.
+Registrations survive while a state is retained, so string views still follow
+remaps after the table object is released. Write callbacks receive the
+backing state as an argument. Snapshot remapping updates shared state and
+replaces the table's accessor cache; retained checked accessors reject the old generation, including
+fields first accessed after remapping. There is no ownership cycle.
+
+Builders and push dispatch caches are discarded when their build phase ends.
+Mutable-build states contain native builders and row counts, without engine
+references; they are separate from the weak registry of public tables.
+`ObjectEngine` holds no duplicate backing reference and relies on reference
+counting.
+
+`RecordEngine.close()` is idempotent and drops engine-owned references. It does not
+invalidate a caller's `FdbViewOwner`; callers use `fdb.invalidate` when a lease
+ends. Engine-dependent writes after closure raise `RuntimeError` with
+`RecordEngine is closed.` Direct writable array access remains valid.
+Retained checked views still enforce owner validity. Buffer resize or
+mapping close raises `BufferError` while views pin its export. These contracts
+are documented in the [Python binding README](../../python/README.md).
+
+Unused unsafe borrowing interfaces are excluded from the Python projection.
+Invalid standalone envelope, descriptor, table and legacy string-table bounds
+are checked in Core before constructing layers. Optional column-section
+handling retains its existing behavior.
+
+## Memory measurements
+
+Measurements use macOS ARM64, CPython 3.10.17 and NumPy 2.2.6. The published
+0.2.3 wheel and a locally built 0.2.4 wheel run the same scripts. Each large case
+has 15,498,593 F32 rows, six complete write or read cycles, and cyclic GC
+disabled throughout. Closing and deleting the last reference releases native
+objects immediately; no `gc.collect()` is used in these measurements.
+
+| Path | Published 0.2.3 | 0.2.4 |
+| --- | ---: | ---: |
+| truncate, fill, save, close, delete | 361.18 MiB/cycle | 0.00 MiB/cycle |
+| file load, column copy, close, delete | 301.50 MiB/cycle | 0.00 MiB/cycle |
+
+Values are mean increments over cycles 2–6, measured using macOS physical
+footprint and rounded to 0.1 MiB by the script. All six 0.2.4 write samples
+are 257.5 MiB; all six read samples are 257.1 MiB. There is no observed
+post-warm-up growth at this resolution. First-cycle allocator
+warm-up is allowed; stable reuse is the criterion, rather than an immediate
+return to the import baseline. No cyclic collection or allocator relief occurs
+between samples.
+
+Automated memory regressions use 1,000,000 rows and ten cycles with GC disabled.
+After two warm-up cycles, all remaining samples must stay within 16 MiB above
+that baseline. The 0.2.3 native cache and backing leaks alone retain about
+16–20 MiB per cycle at this row count. Separate weak-reference and mapping assertions check
+immediate release independently of allocator RSS behavior.
+
+## Conversion timings
+
+The same benchmark takes the best of five runs using the same Python/NumPy
+versions and data. Timing is reported here; regression tests use deterministic
+cache-identity and weak-reference assertions rather than timing thresholds.
+
+| Operation | Published 0.2.3 | 0.2.4 |
+| --- | ---: | ---: |
+| `StringColumn.to_pylist`, 200,000 rows | 747.6 ms | 723.3 ms |
+| `chunk.as_array`, 200,000 calls | 71.7 ms | 70.1 ms |
+| `table.column.value`, 100,000 reads | 66.4 ms | 54.2 ms |
+| `table[i].value`, 100,000 reads | 321.4 ms | 318.8 ms |
+
+Column accessors and their arrays remain cached per table. Each measured path
+has latency at or below the published version in this run; repeated runs
+vary by a few percent in either direction.
+
+## Verification scope
+
+The regression suite covers caller-owned bytes, bytearray and mmap exports,
+shared mappings, invalid inputs, retained accessors/rows/columns, shared checked
+owners, and immediate builder release. Shutdown subprocesses cover daemon
+readers, module destructors and cyclic destructors; mapping closure is asserted
+on native destruction and on the cyclic case. Native ASan/UBSan tests revoke
+backing access with `PROT_NONE` before destruction and check release callbacks
+exactly once. Tests write file fixtures only in their assigned temporary or
+build directory.
+
+The source suite passes 403 Python tests on CPython 3.13; installed wheels
+each pass all 403 tests on CPython 3.10 and 3.13. Compileall and local
+sdist/wheel builds also pass, together with
+45 native ASan/UBSan tests, 21 Rust unit/integration tests plus one compile-fail
+doctest, 58 TypeScript/WASM tests and 117-symbol native and WASM ABI checks.
+The three shutdown scripts each return zero twice in both Python runtimes.
+The six retained-view/release scenarios and chained writes pass on CPython
+3.10. Release policy tests pass 105 cases and 98 subtests.
+Windows, Linux, macOS x86_64 and free-threaded Python have
+not been executed for this fix. Registry artifacts and the complete release
+matrix remain subject to the [publishing process](../releases/publishing.md).
+
+Raw logs, measurements and build archives are kept outside the source repository.
diff --git a/fastcarto/fastdb/include/fastdb.h b/fastcarto/fastdb/include/fastdb.h
index 7d0d1c7..83c77a6 100644
--- a/fastcarto/fastdb/include/fastdb.h
+++ b/fastcarto/fastdb/include/fastdb.h
@@ -224,7 +224,7 @@ namespace wx
         void setTableBufferMaterialized(bool enabled);
         size_t tableBufferBytes();
         void setField(unsigned ix, const FastVectorDbFeatureRef* ref);
-        FastVectorDbFeatureRef* createFeatureRef(u32 ix=-1);
+        FastVectorDbFeatureRef* createFeatureRef(u32 ix=0xFFFFFFFF);
         void freeFeatureRef(FastVectorDbFeatureRef* ref);
         void addFeatureEnd();
         // List column build API
@@ -370,6 +370,9 @@ namespace wx
         const char*             getFieldDefn(unsigned ix, FieldTypeEnum &ft, double &vmin, double &vmax);
         size_t                  getFieldOffset(unsigned ix);
         size_t                  getFeatureByteSize();
+        // Borrowed base of the row table. Rows are separated by getFeatureByteSize().
+        // The address remains valid only while the database and its backing buffer live.
+        void*                   getTableDataAddress();
         void                    getExtent(double &minx, double &miny, double &maxx, double &maxy);
         u32                     getFeatureCount();
         void                    rewind();
diff --git a/fastcarto/fastdb/src/FastVectorDb.cpp b/fastcarto/fastdb/src/FastVectorDb.cpp
index bf519e8..8c18277 100644
--- a/fastcarto/fastdb/src/FastVectorDb.cpp
+++ b/fastcarto/fastdb/src/FastVectorDb.cpp
@@ -1,6 +1,7 @@
 #include "FastVectorDb_p.h"
 #include "FastVectorDbLayer_p.h"
 #include <cstring>
+#include <limits>
 #include <stdlib.h>
 
 #ifdef _WIN32
@@ -24,15 +25,61 @@ namespace wx
             std::memcpy(&value, source, sizeof(value));
             return value;
         }
+
+        bool valid_layer_bounds(const u8* layer, const layer_header_t& header)
+        {
+            const size_t tail = size_t(header.total_size) - sizeof(layer_header_t);
+            if (header.field_count > tail / sizeof(field_desc_ex_t))
+                return false;
+            const size_t descriptors = size_t(header.field_count) * sizeof(field_desc_ex_t);
+            const u8* data = layer + sizeof(layer_header_t) + descriptors;
+            const size_t size = tail - descriptors;
+            if (header.offset_table > size)
+                return false;
+            if (header.field_count) {
+                const auto last = load_unaligned<field_desc_ex_t>(
+                    layer + sizeof(layer_header_t) + descriptors - sizeof(field_desc_ex_t));
+                if (last.offset > std::numeric_limits<u64>::max() - last.size)
+                    return false;
+                const u64 stride = last.offset + last.size;
+                if (stride && header.feature_count > (size - header.offset_table) / stride)
+                    return false;
+                for (u16 i = 0; i < header.field_count; ++i) {
+                    const auto field = load_unaligned<field_desc_ex_t>(
+                        layer + sizeof(layer_header_t) + size_t(i) * sizeof(field_desc_ex_t));
+                    if (field.offset > stride || field.size > stride - field.offset)
+                        return false;
+                }
+            }
+            // Legacy string tables are read during construction, before the
+            // bounded optional column sections. Check their terminators here.
+            for (unsigned width : {1u, 2u}) {
+                u64 offset = width == 1 ? header.offset_strings : header.offset_wstrings;
+                if (offset > size || size - offset < sizeof(u32))
+                    return false;
+                const u32 count = load_unaligned<u32>(data + offset);
+                offset += sizeof(u32);
+                if (count > (size - offset) / width)
+                    return false;
+                for (u32 i = 0; i < count; ++i) {
+                    for (;;) {
+                        if (size - offset < width)
+                            return false;
+                        const bool terminated = width == 1 ? data[offset] == 0 :
+                            load_unaligned<u16>(data + offset) == 0;
+                        offset += width;
+                        if (terminated) break;
+                    }
+                }
+            }
+            return true;
+        }
     }
 
     FastVectorDb::Impl::Impl(void *pdata, size_t size, fnFreeDbBuffer fnFreeBuffer, void *cookie)
         : m_pdata(pdata), m_size(size), m_fnFreeBuffer(fnFreeBuffer), m_cookie(cookie)
     {
         u8 *ptr = (u8 *)pdata;
-        bool check_mask = strcmp((const char *)ptr, "FASTVectorDB0.1") == 0;
-        assert(check_mask);
-        m_mask_check_ok=check_mask;
         ptr += 16;
         const u32 count = load_unaligned<u32>(ptr);
         ptr += sizeof(u32);
@@ -113,17 +160,32 @@ namespace wx
 
     FastVectorDb *FastVectorDb::load(void *pdata, size_t size, fnFreeDbBuffer fnFreeBuffer, void *cookie)
     {
-        FastVectorDb::Impl *impl = new FastVectorDb::Impl(pdata, size, fnFreeBuffer, cookie);
-        if(impl->m_mask_check_ok)
-        {
-            auto db = new FastVectorDb(impl);
-            return db;
+        // Validate the standalone envelope once before reading its headers.
+        // The backing callback also releases invalid inputs.
+        bool valid = pdata && size >= 20 &&
+            std::memcmp(pdata, "FASTVectorDB0.1", sizeof("FASTVectorDB0.1")) == 0;
+        if (valid) {
+            const u8* ptr = static_cast<const u8*>(pdata) + 20;
+            size_t remaining = size - 20;
+            const u32 count = load_unaligned<u32>(static_cast<const u8*>(pdata) + 16);
+            for (u32 i = 0; i < count; ++i) {
+                if (remaining < sizeof(layer_header_t)) { valid = false; break; }
+                const auto header = load_unaligned<layer_header_t>(ptr);
+                if (header.total_size < sizeof(layer_header_t) || header.total_size > remaining ||
+                    !valid_layer_bounds(ptr, header)) {
+                    valid = false;
+                    break;
+                }
+                ptr += header.total_size;
+                remaining -= header.total_size;
+            }
         }
-        else
+        if (!valid)
         {
-            delete impl;
+            if (fnFreeBuffer) fnFreeBuffer(pdata, size, cookie);
             return nullptr;
         }
+        return new FastVectorDb(new Impl(pdata, size, fnFreeBuffer, cookie));
     }
 
     FastVectorDbFeature* FastVectorDb::tryGetFeature(wx::FastVectorDbFeatureRef* ref)
diff --git a/fastcarto/fastdb/src/FastVectorDbLayer.cpp b/fastcarto/fastdb/src/FastVectorDbLayer.cpp
index 5262421..2cd934a 100644
--- a/fastcarto/fastdb/src/FastVectorDbLayer.cpp
+++ b/fastcarto/fastdb/src/FastVectorDbLayer.cpp
@@ -727,6 +727,10 @@ namespace wx
     {
         return m_table_line_size;
     }
+    void* FastVectorDbLayer::Impl::getTableDataAddress()
+    {
+        return m_table_data_ptr0;
+    }
     ///////////////////////////////////////////
 
     FastVectorDbLayer::FastVectorDbLayer(Impl *_impl) : impl(_impl) {}
@@ -842,6 +846,10 @@ namespace wx
     {
         return impl->getFeatureByteSize();
     }
+    void* FastVectorDbLayer::getTableDataAddress()
+    {
+        return impl->getTableDataAddress();
+    }
 
     // ---- List column read helpers ----
     chunk_data_t FastVectorDbLayer::Impl::getFieldAsListView_internal(u32 ifeature, u32 ix)
diff --git a/fastcarto/fastdb/src/FastVectorDbLayer_p.h b/fastcarto/fastdb/src/FastVectorDbLayer_p.h
index 76d635d..f89bab8 100644
--- a/fastcarto/fastdb/src/FastVectorDbLayer_p.h
+++ b/fastcarto/fastdb/src/FastVectorDbLayer_p.h
@@ -65,6 +65,7 @@ namespace wx
 
         size_t          getFieldOffset(unsigned ix);
         size_t          getFeatureByteSize();
+        void*           getTableDataAddress();
     private:
         void            move_next_geometry_ptr();
         size_t          get_geometry_like_size(const u8* pdata);
diff --git a/fastcarto/fastdb/src/FastVectorDb_p.h b/fastcarto/fastdb/src/FastVectorDb_p.h
index 15817c7..200a010 100644
--- a/fastcarto/fastdb/src/FastVectorDb_p.h
+++ b/fastcarto/fastdb/src/FastVectorDb_p.h
@@ -20,7 +20,6 @@ namespace wx{
         size_t  m_size;
         fnFreeDbBuffer m_fnFreeBuffer;
         void* m_cookie;
-        bool    m_mask_check_ok;
         friend class FastVectorDb;
     };
 }
diff --git a/fastcarto/fastdb/swig/fastdb4py.i b/fastcarto/fastdb/swig/fastdb4py.i
index f6f8b9d..21913f1 100644
--- a/fastcarto/fastdb/swig/fastdb4py.i
+++ b/fastcarto/fastdb/swig/fastdb4py.i
@@ -4,11 +4,75 @@
     #include "fastdb.h"
     #include "fastdb-geometry-utils.h"
     using namespace wx;
+
+    struct PythonDbBacking {
+        Py_buffer buffer;
+        PyObject* mapping;
+    };
+
+    static void release_python_db_backing(void*, size_t, void* cookie) {
+        auto* backing = static_cast<PythonDbBacking*>(cookie);
+        PyGILState_STATE state = PyGILState_Ensure();
+        PyBuffer_Release(&backing->buffer);
+        if (backing->mapping) {
+            PyObject* result = PyObject_CallMethod(backing->mapping, "close", NULL);
+            if (result)
+                Py_DECREF(result);
+            else
+                PyErr_WriteUnraisable(backing->mapping);
+            Py_DECREF(backing->mapping);
+        }
+        PyMem_Free(backing);
+        PyGILState_Release(state);
+    }
 %}
 
 // Tell SWIG to ignore fastdb_api when parsing headers
 #define fastdb_api
 
+// These factories transfer a newly allocated object to their caller. Layers,
+// cached features and builder-created refs remain owned by their parent.
+%newobject wx::FastVectorDb::load(const char *filename);
+%newobject wx::FastVectorDb::load_python_buffer;
+%rename(load_xbuffer) wx::FastVectorDb::load_python_buffer;
+%newobject wx::FastVectorDbFeatureRef::make_ref;
+
+%define BORROWED_RESULT(method)
+%pythonappend method %{
+    if val is not None:
+        val._fdb_parent = self
+%}
+%enddef
+BORROWED_RESULT(wx::FastVectorDbBuild::createLayerBegin)
+BORROWED_RESULT(wx::FastVectorDbLayerBuild::createFeatureRef)
+BORROWED_RESULT(wx::FastVectorDb::getLayer)
+BORROWED_RESULT(wx::FastVectorDb::tryGetFeature)
+BORROWED_RESULT(wx::FastVectorDb::buffer)
+BORROWED_RESULT(wx::FastVectorDbLayer::tryGetFeatureAt)
+BORROWED_RESULT(wx::FastVectorDbLayer::getGeometryLikeChunk)
+BORROWED_RESULT(wx::FastVectorDbLayer::getFieldAsStringView)
+BORROWED_RESULT(wx::FastVectorDbLayer::getStringColumnOffsets)
+BORROWED_RESULT(wx::FastVectorDbLayer::getStringColumnData)
+BORROWED_RESULT(wx::FastVectorDbLayer::getFieldAsFeatureRef)
+BORROWED_RESULT(wx::FastVectorDbFeature::layer)
+BORROWED_RESULT(wx::FastVectorDbFeature::getGeometryLikeChunk)
+BORROWED_RESULT(wx::FastVectorDbFeature::getFieldAsStringView)
+BORROWED_RESULT(wx::FastVectorDbFeature::getFieldAsListView)
+BORROWED_RESULT(wx::FastVectorDbFeature::getFieldAsFeatureRef)
+BORROWED_RESULT(wx::FastVectorDbFeature::getFieldListRefAt)
+BORROWED_RESULT(wx::MemoryStream::data)
+
+%define DATABASE_RESULT(method)
+%exception method {
+    $action
+    if (PyErr_Occurred()) SWIG_fail;
+    if (!result)
+        SWIG_exception_fail(SWIG_ValueError, "Invalid FastDB database buffer.");
+}
+%enddef
+DATABASE_RESULT(wx::FastVectorDb::load)
+DATABASE_RESULT(wx::FastVectorDb::load_python_buffer)
+
 // Release the GIL for pure C++ operations that do not touch Python objects.
 // This benefits both standard Python (better multi-threaded throughput) and
 // free-threaded Python 3.13+ (PEP 703).
@@ -32,7 +96,33 @@
     // When built on free-threaded Python, SWIG automatically declares Py_MOD_GIL_NOT_USED.
 %}
 
-//%array_class(point2_t, LineString);
+%extend wx::FastVectorDb {
+    static wx::FastVectorDb* load_python_buffer(PyObject* pdata, PyObject* mapping = Py_None) {
+        // Independent view: callers may release their view after this call.
+        PyObject* view = PyMemoryView_FromObject(pdata);
+        if (!view) return NULL;
+        auto* backing = static_cast<PythonDbBacking*>(PyMem_Calloc(1, sizeof(PythonDbBacking)));
+        if (!backing) {
+            Py_DECREF(view);
+            PyErr_NoMemory();
+            return NULL;
+        }
+        int status = PyObject_GetBuffer(view, &backing->buffer, PyBUF_SIMPLE);
+        Py_DECREF(view);
+        if (status != 0) {
+            PyMem_Free(backing);
+            return NULL;
+        }
+        wx::FastVectorDb* db = wx::FastVectorDb::load(
+            backing->buffer.buf, (size_t)backing->buffer.len,
+            release_python_db_backing, backing);
+        if (db && mapping != Py_None) {
+            Py_INCREF(mapping);
+            backing->mapping = mapping;
+        }
+        return db;
+    }
+}
 
 %typemap(in) const wchar_t * (wchar_t *temp = NULL) {
     if ($input == Py_None) {
@@ -101,26 +191,19 @@
 }
 
 %typemap(default) (const u8* data,size_t size) {};
-// 在你的 .i 文件中对目标函数应用此类型映射
+
+// Copying stream writes borrow their input only for this C call. Retained
+// database buffers use load_python_buffer's native lifetime cookie instead.
 %typemap(in) (void* pdata, size_t size) (Py_buffer view) {
-    // 检查输入对象是否支持缓冲区接口
-    if (PyObject_CheckBuffer($input) == 0) {
-        SWIG_exception_fail(SWIG_TypeError, "Expected a bufferable object (e.g., bytes, bytearray)");
-    }
-    // 获取缓冲区视图，PyBUF_SIMPLE 请求一个简单的连续字节缓冲区
-    if (PyObject_GetBuffer($input, &view, PyBUF_SIMPLE) != 0) {
-        SWIG_exception_fail(SWIG_TypeError, "Failed to get buffer from object.");
+    memset(&view, 0, sizeof(view));
+    if (PyObject_GetBuffer($input, &view, PyBUF_SIMPLE) < 0) {
+        SWIG_fail;
     }
-    // 将获取到的缓冲区地址和长度赋给参数
-    $1 = view.buf;  // void* pdata 指向数据
-    $2 = view.len;  // size_t size 设置数据长度
+    $1 = view.buf;
+    $2 = (size_t)view.len;
 }
-
-// 定义 freearg 类型映射，在函数调用后释放缓冲区
 %typemap(freearg) (void* pdata, size_t size) {
-    if (view$argnum.obj) {
-        PyBuffer_Release(&view$argnum);
-    }
+    if (view$argnum.obj) PyBuffer_Release(&view$argnum);
 }
 
 // Typemap for set_field_list_numeric: accepts any buffer object; nbytes = byte length of buffer
@@ -191,6 +274,13 @@
 %ignore wx::FastVectorDbFeature::FastVectorDbFeature();
 %ignore wx::FastVectorDbFeature::~FastVectorDbFeature();
 %ignore wx::FastVectorDb::load(void *pdata, size_t size, fnFreeDbBuffer fnFreeBuffer, void *cookie);
+%ignore wx::FastVectorDb::load_xbuffer;
+%ignore wx::FixedBufferWriteStream::FixedBufferWriteStream;
+%nodefaultctor wx::FixedBufferWriteStream;
+%ignore wx::aabbox_t::minEdge;
+%ignore wx::aabbox_t::maxEdge;
+%ignore wx::GeometryUtils::get_point_at;
+%ignore wx::GeometryUtils::part_data;
 %ignore wx::FastVectorDbBuild::postToBuffer;
 %ignore wx::FastVectorDbLayerBuild::setNumericColumnBulk;
 %nodefaultctor FastVectorDbLayerBuild;
@@ -228,6 +318,7 @@
 %rename(get_field_defn)        getFieldDefn_p;
 %rename(get_extent)            getExtent_p;
 %rename(get_feature_count)     getFeatureCount;       
+%rename(get_table_data_address) getTableDataAddress;
 %rename(get_geometry_like_chunk) getGeometryLikeChunk;       
 %rename(get_field_as_float)     getFieldAsFloat;       
 %rename(get_field_as_int)       getFieldAsInt;       
@@ -261,11 +352,10 @@
 %rename(get_field_list_ref_at)  getFieldListRefAt;
 
 %extend wx::chunk_data_t {
-    PyObject *as_array(PyObject* npType) {
+    PyObject *as_array(PyObject* npType, PyObject** PYTHON_SELF) {
         int typenum;
         PyArray_Descr* descr = PyArray_DescrFromTypeObject(npType);
         if (descr == NULL) { 
-            PyErr_Clear(); // 或进行其他错误处理
             return NULL;
         } else { 
             typenum = descr->type_num;
@@ -281,21 +371,22 @@
 
         Py_DECREF(descr);
         
-        npy_intp num_elements = $self->size / item_size;
-        
-        if ($self->size % item_size != 0) {
+        if (item_size == 0 || $self->size % item_size != 0) {
             PyErr_SetString(PyExc_ValueError, "Buffer size does not match an integer number of elements for the given data type.");
             return NULL;
         }
+        npy_intp num_elements = $self->size / item_size;
         npy_intp dims[1] = {num_elements}; 
         PyObject *array = PyArray_SimpleNewFromData(1, dims, typenum, (void*)($self->pdata));
         if (!array) {
             PyErr_SetString(PyExc_RuntimeError, "Failed to create NumPy array");
             return NULL;
         }
-        PyArray_ENABLEFLAGS((PyArrayObject*)array, NPY_ARRAY_C_CONTIGUOUS);
-        PyArray_ENABLEFLAGS((PyArrayObject*)array, NPY_ARRAY_WRITEABLE);
-        PyArray_CLEARFLAGS((PyArrayObject*)array, NPY_ARRAY_OWNDATA);
+        Py_INCREF(*PYTHON_SELF);
+        if (PyArray_SetBaseObject((PyArrayObject*)array, *PYTHON_SELF) < 0) {
+            Py_DECREF(array);
+            return NULL;
+        }
         return array;
     }
     
@@ -333,6 +424,7 @@
 
 %pythoncode %{
     import numpy as np
+
 %}
 %extend wx::FastVectorDbLayer {
    %pythoncode %{
@@ -379,7 +471,7 @@
 
                     if(not tystr):
                         return None
-                    ptr = _fastdb4py.get_swig_ptr_as_long(table.tryGetFeature(0).get_address())
+                    ptr = _fastdb4py.get_swig_ptr_as_long(table.get_table_data_address())
                     offset = table.get_field_offset(index)
                     ptr+=offset
                     return __column_np_interface__(
diff --git a/python/README.md b/python/README.md
index c79060f..c8ba15f 100644
--- a/python/README.md
+++ b/python/README.md
@@ -129,6 +129,27 @@ print(table.column.x.mean())
 
 ## Backed View Lifetimes
 
+`RecordEngine.close()` releases the engine's own references and may be called
+repeatedly. Tables retain their engine, and cached column accessors and views
+share backing state that does not reference the table. Their last reference
+releases the native database immediately, without waiting for cyclic garbage
+collection. `ObjectEngine` releases its references when the engine itself is
+no longer referenced; it has no `close()` method. Closing a RecordEngine does
+not invalidate a caller's `FdbViewOwner`; the caller ends a lease with
+`fdb.invalidate(owner_or_view)`.
+
+After `RecordEngine.close()`, writes that require the engine, including
+`Table.fill()` and string-column fills, raise `RuntimeError: RecordEngine is
+closed.` Direct writes to retained writable NumPy views still update the
+memory those views retain.
+
+`core.WxDatabase.load_xbuffer(buffer)` pins a contiguous buffer export until
+the native database is destroyed. This includes borrowed `bytes`, `bytearray`,
+NumPy and `mmap` storage. Resizing or closing an exported buffer raises
+`BufferError` while a table, row, or column still keeps the database alive.
+Shared-memory mappings close after native layers and the buffer export are
+released.
+
 FastDB Python values have two lifetime modes. A plain `@feature` instance is owned Python data: its fields live in `__dict__`, it can be kept indefinitely, and assigning `point.x = 1.0` is ordinary Python object mutation. A row returned from a mapped `Table` is a backed feature view: schema fields read from the native table row, scalar numeric writes go back to native storage when the view is writeable, and the view is tied to a `FdbViewOwner`.
 
 Standalone FastDB tables use a trusted unchecked owner by default, so existing high-performance workflows keep raw NumPy column access:
diff --git a/python/fastdb4py/column_view.py b/python/fastdb4py/column_view.py
index f7b44cd..211bf38 100644
--- a/python/fastdb4py/column_view.py
+++ b/python/fastdb4py/column_view.py
@@ -7,20 +7,20 @@
 from .view_owner import FdbViewOwner, FdbViewWriteError
 
 if TYPE_CHECKING:
-    from .orm.table import Table
+    from .orm.table import _TableState
 
 
 class NumericColumnView:
     """Owner-checked view over a numeric native column."""
 
-    def __init__(self, table: 'Table', field_index: int, field_name: str, array: np.ndarray):
-        self._table = table
+    def __init__(self, state: '_TableState', field_index: int, field_name: str, array: np.ndarray):
+        self._state = state
         self._field_index = field_index
         self._field_name = field_name
         self._array = array
-        self._owner: FdbViewOwner = table._fdb_owner
-        self._writeable = table._fdb_writeable
-        self._owner_generation = table._fdb_owner.generation
+        self._owner: FdbViewOwner = state._fdb_owner
+        self._writeable = state._fdb_writeable
+        self._owner_generation = state._fdb_owner.generation
 
     def _checked_array(self) -> np.ndarray:
         self._owner.assert_alive(self._owner_generation)
diff --git a/python/fastdb4py/object_engine.py b/python/fastdb4py/object_engine.py
index 4eec18e..dc8a9e4 100644
--- a/python/fastdb4py/object_engine.py
+++ b/python/fastdb4py/object_engine.py
@@ -132,7 +132,6 @@ class ObjectEngine:
     def __init__(self):
         self._db_build: Optional[core.WxDatabaseBuild] = None
         self._db: Optional[core.WxDatabase] = None
-        self._buffer: Optional[bytes] = None
         self._layers: Dict[Type, LayerState] = {}
         self._layer_order: List[Type] = []
         self._built = False
@@ -170,10 +169,10 @@ def truncate(cls, layouts) -> 'ObjectEngine':
         engine._db_build.post(mem)
         buf = mem.data().as_array(np.uint8).tobytes()
         engine._db = core.WxDatabase.load_xbuffer(buf)
-        engine._db._buffer = buf
-        engine._buffer = buf
         engine._built = True
         engine._db_build = None
+        for state in engine._layers.values():
+            state.build = None
         return engine
 
     def push(self, obj: Any) -> None:
@@ -237,10 +236,10 @@ def combine(self):
         self._db_build.post(mem)
         buf = mem.data().as_array(np.uint8).tobytes()
         self._db = core.WxDatabase.load_xbuffer(buf)
-        self._db._buffer = buf
-        self._buffer = buf
         self._built = True
         self._db_build = None
+        for state in self._layers.values():
+            state.build = None
         self._pending.clear()
         self._pushed_ids.clear()
         self._pending_counts.clear()
@@ -393,9 +392,9 @@ def share(self, name: str):
         if not self._built:
             raise RuntimeError("Call combine() before sharing")
         import multiprocessing.shared_memory as shm_mod
-        data = self._buffer
-        seg = shm_mod.SharedMemory(name=name, create=True, size=len(data))
-        seg.buf[:len(data)] = data
+        chunk = self._db.buffer()
+        seg = shm_mod.SharedMemory(name=name, create=True, size=chunk.size)
+        chunk.copy_to_buffer(seg.buf)
         seg.close()
 
     @classmethod
@@ -408,8 +407,6 @@ def load(cls, name: str) -> 'ObjectEngine':
 
         orm = cls()
         orm._db = core.WxDatabase.load_xbuffer(buf)
-        orm._db._buffer = buf
-        orm._buffer = buf
         orm._built = True
 
         for i in range(orm._db.get_layer_count()):
diff --git a/python/fastdb4py/orm/table.py b/python/fastdb4py/orm/table.py
index 0792941..01ac91b 100644
--- a/python/fastdb4py/orm/table.py
+++ b/python/fastdb4py/orm/table.py
@@ -70,7 +70,7 @@ def _normalize_bool_fill_ndarray(field_name: str, values: np.ndarray) -> np.ndar
     )
 
 
-def _create_column_accessor(feature_type: Type[T], table) -> T:
+def _create_column_accessor(feature_type: Type[T], state) -> T:
     """
     Create a column accessor that provides numpy array access with proper type hints.
 
@@ -81,7 +81,7 @@ def _create_column_accessor(feature_type: Type[T], table) -> T:
         schema = get_schema(feature_type)
         ColumnAccessorClass = schema.column_accessor_class
         if ColumnAccessorClass is not None:
-            return ColumnAccessorClass(table, feature_type)
+            return ColumnAccessorClass(state)
 
         # Get original annotations from feature_type
         original_annotations = getattr(feature_type, '__annotations__', {}).copy()
@@ -96,10 +96,10 @@ class ColumnAccessor:
             # Set the new annotations
             __annotations__ = original_annotations
 
-            def __init__(self, table, feature_type):
-                # Don't call parent __init__ to avoid initializing cache
-                # Just set internal references
-                object.__setattr__(self, '_table', table)
+            def __init__(self, state):
+                object.__setattr__(self, '_state', state)
+                object.__setattr__(self, '_fdb_owner', state._fdb_owner)
+                object.__setattr__(self, '_owner_generation', state._fdb_owner.generation)
                 object.__setattr__(self, '_field_index_map', _field_index_map)
                 object.__setattr__(self, '_name_cache', {})
                 object.__setattr__(self, '_cache_lock', Lock())
@@ -108,8 +108,9 @@ def __getattr__(self, name: str) -> np.ndarray:
                 """Override to return numpy array instead of single value"""
                 # Hot path: name → array directly (1× getattribute + 1× dict.get).
                 # Cached arrays are stable: fixed-scale tables never reallocate columns.
-                table = object.__getattribute__(self, '_table')
-                table._assert_alive()
+                owner = object.__getattribute__(self, '_fdb_owner')
+                if owner.checked:
+                    owner.assert_alive(object.__getattribute__(self, '_owner_generation'))
                 name_cache = object.__getattribute__(self, '_name_cache')
                 arr = name_cache.get(name)
                 if arr is not None:
@@ -127,12 +128,13 @@ def __getattr__(self, name: str) -> np.ndarray:
                     if idx is None:
                         raise AttributeError(f'Field "{name}" not found in the table.')
 
-                    table_origin = table._origin
+                    state = object.__getattribute__(self, '_state')
+                    table_origin = state._origin
                     fd = schema.fields[idx]
                     if fd.field_type in {OriginFieldType.str, OriginFieldType.wstr}:
-                        arr = StringColumn(table, idx, name, wide=fd.field_type == OriginFieldType.wstr)
+                        arr = StringColumn(state, idx, name, wide=fd.field_type == OriginFieldType.wstr)
                     elif fd.field_type == OriginFieldType.bytes:
-                        arr = BytesColumn(table, idx, name)
+                        arr = BytesColumn(state, idx, name)
                     else:
                         column = table_origin.get_column(idx)
                         if column is None:
@@ -140,8 +142,8 @@ def __getattr__(self, name: str) -> np.ndarray:
                                 f'Field "{name}" does not expose a fixed native column.',
                             )
                         arr = column.as_nparray()
-                        if table._fdb_owner.checked or not table._fdb_writeable:
-                            arr = NumericColumnView(table, idx, name, arr)
+                        if state._fdb_owner.checked or not state._fdb_writeable:
+                            arr = NumericColumnView(state, idx, name, arr)
                     name_cache[name] = arr
                     return arr
 
@@ -157,21 +159,40 @@ def __setattr__(self, name: str, value):
 
         # Store the class in ClassSchema (replaces _column_accessor_cache WeakKeyDict).
         schema.column_accessor_class = ColumnAccessor
-        return ColumnAccessor(table, feature_type)
+        return ColumnAccessor(state)
+
+
+class _TableState:
+    """Backing shared by table projections, with no references to projections."""
 
-class Table(Generic[T]):
     def __init__(self):
         self._fc: np.ndarray = np.zeros(1, dtype=np.int64)
-        self._column: T | None = None
         self._feature_type: Type[T] | None = None
         self._db: core.WxDatabase | core.WxDatabaseBuild = None
         self._origin: core.WxLayerTable | core.WxLayerTableBuild | None = None
+        self._engine = None
+        self._table_name = ''
         self._fixed_fill_handler = None
         self._read_lock = RLock()
         self._fdb_owner: FdbViewOwner = trusted_view_owner(writeable=True)
-        self._fdb_checked = False
         self._fdb_writeable = True
 
+    def __len__(self) -> int:
+        self._assert_alive()
+        return self._origin.get_feature_count()
+
+    def _assert_alive(self, generation: int | None = None) -> None:
+        self._fdb_owner.assert_alive(generation)
+
+    def _assert_engine_open(self) -> None:
+        if self._engine is not None:
+            self._engine._assert_open()
+
+    def _remap(self, origin, db) -> None:
+        self._db = db
+        self._origin = origin
+        self.feature_count = origin.get_feature_count()
+
     @property
     def feature_count(self) -> int:
         return int(self._fc[0])
@@ -180,41 +201,110 @@ def feature_count(self) -> int:
     def feature_count(self, value: int) -> None:
         self._fc[0] = value
 
+
+class Table(Generic[T]):
+    def __init__(self):
+        self._state = _TableState()
+        self._column: T | None = None
+
+    @staticmethod
+    def _from_state(state: _TableState) -> 'Table[T]':
+        table = Table[T]()
+        table._state = state
+        table._reset_columns()
+        return table
+
+    def _reset_columns(self) -> None:
+        self._column = (
+            _create_column_accessor(self._feature_type, self._state)
+            if self.fixed and self._feature_type is not None else None
+        )
+
+    @property
+    def _db(self):
+        return self._state._db
+
+    @property
+    def _origin(self):
+        return self._state._origin
+
+    @_origin.setter
+    def _origin(self, value):
+        self._state._origin = value
+
+    @property
+    def _feature_type(self):
+        return self._state._feature_type
+
+    @_feature_type.setter
+    def _feature_type(self, value):
+        self._state._feature_type = value
+
+    @property
+    def _read_lock(self):
+        return self._state._read_lock
+
+    @_read_lock.setter
+    def _read_lock(self, value):
+        self._state._read_lock = value
+
+    @property
+    def _fdb_owner(self):
+        return self._state._fdb_owner
+
+    @property
+    def _fdb_writeable(self):
+        return self._state._fdb_writeable
+
+    @property
+    def _fixed_fill_handler(self):
+        return self._state._fixed_fill_handler
+
+    @property
+    def feature_count(self) -> int:
+        return self._state.feature_count
+
+    @feature_count.setter
+    def feature_count(self, value: int) -> None:
+        self._state.feature_count = value
+
     def __len__(self) -> int:
         self._assert_alive()
         return self._origin.get_feature_count()
 
     def __getitem__(self, index: int) -> T:
-        with self._read_lock:
-            self._assert_alive()
-            count = self._origin.get_feature_count()
+        state = self._state
+        with state._read_lock:
+            state._assert_alive()
+            count = state._origin.get_feature_count()
             if index < 0:
                 index = count + index
             if index < 0 or index >= count:
                 raise IndexError(f'Feature index {index} out of range [0, {count}).')
 
             return bind_feature(
-                self._feature_type,
-                self._db,
-                self._origin,
+                state._feature_type,
+                state._db,
+                state._origin,
                 index,
-                owner=self._fdb_owner,
-                writeable=self._fdb_writeable,
+                owner=state._fdb_owner,
+                writeable=state._fdb_writeable,
                 table=self,
             )
 
     def __iter__(self) -> Generator[T, None, None]:
         self._assert_alive()
         for i in range(self._origin.get_feature_count()):
-            with self._read_lock:
-                self._assert_alive()
+            state = self._state
+            with state._read_lock:
+                state._assert_alive()
                 item = bind_feature(
-                    self._feature_type,
-                    self._db,
-                    self._origin,
+                    state._feature_type,
+                    state._db,
+                    state._origin,
                     i,
-                    owner=self._fdb_owner,
-                    writeable=self._fdb_writeable,
+                    owner=state._fdb_owner,
+                    writeable=state._fdb_writeable,
                     table=self,
                 )
             yield item
@@ -287,7 +377,7 @@ def map_from(
         writeable: bool | None = None,
     ) -> 'Table[T]':
         table = Table[T]()
-        table._db = db
+        table._state._db = db
         table._origin = origin
         table._feature_type = feature_type
         table._set_view_owner(owner, writeable=writeable)
@@ -295,8 +385,6 @@ def map_from(
         # Get feature count if the fastdb table has fixed scale
         if table.fixed:
             table.feature_count = origin.get_feature_count()
-            # Create column accessor that pretends to be T but returns numpy arrays
-            table._column = _create_column_accessor(feature_type, table) if feature_type is not None else None
 
         return table
 
@@ -305,17 +393,8 @@ def _remap(
         origin: core.WxLayerTable | core.WxLayerTableBuild,
         db: core.WxDatabase | core.WxDatabaseBuild,
     ) -> None:
-        self._db = db
-        self._origin = origin
-        self._fdb_owner.bump_generation()
-        if self.fixed:
-            self.feature_count = origin.get_feature_count()
-            self._column = (
-                _create_column_accessor(self._feature_type, self)
-                if self._feature_type is not None else None
-            )
-        else:
-            self._column = None
+        self._state._remap(origin, db)
+        self._reset_columns()
 
     def rewind(self):
         self._assert_alive()
@@ -328,6 +407,7 @@ def fill(self, **col_arrays) -> None:
         ``tbl.column.<field>`` is stale and must not be reused. Always
         re-fetch ``tbl.column.<field>`` after fill().
         """
+        self._state._assert_engine_open()
         if not self.fixed:
             raise RuntimeError('fill() only supports fixed-scale tables.')
         self._assert_writeable()
@@ -338,7 +418,7 @@ def fill(self, **col_arrays) -> None:
 
         expected = len(self)
         writes = {}
-        col = self._column
+        col = self.column
         schema = get_schema(self._feature_type)
         for field_name, values in col_arrays.items():
             field_def = schema.get(field_name)
@@ -375,7 +455,7 @@ def fill(self, **col_arrays) -> None:
                 )
             writes[field_name] = arr
 
-        self._fixed_fill_handler(writes)
+        self._fixed_fill_handler(self._state, writes)
 
     def _set_view_owner(
         self,
@@ -385,17 +465,12 @@ def _set_view_owner(
     ) -> None:
         if owner is None:
             owner = trusted_view_owner(writeable=True if writeable is None else writeable)
-        self._fdb_owner = owner
-        self._fdb_checked = owner.checked
-        self._fdb_writeable = owner.writeable if writeable is None else writeable
-        if self.fixed:
-            self._column = (
-                _create_column_accessor(self._feature_type, self)
-                if self._feature_type is not None else None
-            )
+        self._state._fdb_owner = owner
+        self._state._fdb_writeable = owner.writeable if writeable is None else writeable
+        self._reset_columns()
 
     def _assert_alive(self, generation: int | None = None) -> None:
-        self._fdb_owner.assert_alive(generation)
+        self._state._fdb_owner.assert_alive(generation)
 
     def _assert_writeable(self, generation: int | None = None) -> None:
         self._fdb_owner.assert_alive(generation)
diff --git a/python/fastdb4py/record_engine.py b/python/fastdb4py/record_engine.py
index e6e9a18..cb0bb42 100644
--- a/python/fastdb4py/record_engine.py
+++ b/python/fastdb4py/record_engine.py
@@ -6,6 +6,7 @@
 from pathlib import Path
 from typing import List, TypeVar, Type
 from multiprocessing import shared_memory
+from weakref import ref
 
 from . import core
 from .registry import (
@@ -16,7 +17,7 @@
     raw_payload_storage_diagnostics,
 )
 from .layout import Layout
-from .orm.table import Table
+from .orm.table import Table, _TableState
 from .type import OriginFieldType
 from .view_owner import FdbViewOwner
 from .push_compiler import (
@@ -28,6 +29,14 @@
 T = TypeVar('T')
 
 
+def _load_shared_database(shm: shared_memory.SharedMemory):
+    try:
+        return core.WxDatabase.load_xbuffer(shm.buf, shm)
+    except BaseException:
+        shm.close()
+        raise
+
+
 def _post_build_to_bytearray(build: object) -> bytearray:
     size = build.byte_length()
     buffer = bytearray(size)
@@ -95,10 +104,28 @@ def _reject_unsupported_schema(schema: LayerSchema, cls_name: str) -> None:
     _reject_non_native_lists(schema, cls_name)
 
 
+class _TableRegistration:
+    """Weak projections of one backing state, retained until that state dies."""
+
+    def __init__(self, engine, key, table: Table):
+        engine_ref = ref(engine)
+
+        def forget_state(state_ref):
+            parent = engine_ref()
+            if parent is not None:
+                entry = parent._tables.get(key)
+                if entry is not None and entry.state is state_ref:
+                    del parent._tables[key]
+
+        self.state = ref(table._state, forget_state)
+        self.table = ref(table)
+
+
 class RecordEngine:
     def __init__(self):
         self._shm: shared_memory.SharedMemory | None = None
-        self._table_map: dict[str, Table] = {}
+        self._tables: dict[tuple[str, int], _TableRegistration] = {}
+        self._build_states: dict[str, _TableState] = {}
         self._table_feature_types: dict[str, Type] = {}
         self._origin: core.WxDatabase | core.WxDatabaseBuild | None = None
         self._is_mutable: bool = False
@@ -111,6 +138,11 @@ def __init__(self):
         self._push_batch_fn: dict = {}
         self._push_dispatch: dict = {}
         self._bool_push_schemas: dict = {}
+        self._closed = False
+
+    def _assert_open(self) -> None:
+        if self._closed:
+            raise RuntimeError('RecordEngine is closed.')
 
     @property
     def fixed(self) -> bool:
@@ -155,7 +187,7 @@ def _prepare_truncate(
 
             table_name = layout.name or ft_cls.__name__
             engine._table_feature_types[table_name] = ft_cls
-            existing = engine._table_map.get(table_name)
+            existing = engine._fixed_layer_builds.get(table_name)
             if existing is not None:
                 warnings.warn(
                     f'Table "{table_name}" already exists, truncate '
@@ -163,22 +195,17 @@ def _prepare_truncate(
                     UserWarning,
                 )
             else:
-                new_table = Table.map_from(
-                    ft_cls,
-                    _get_default_table_build(engine._fixed_build, table_name),
-                    engine._fixed_build,
-                )
+                layer = _get_default_table_build(engine._fixed_build, table_name)
                 field_ids = {}
                 for fd in schema.fields:
                     field_ids[fd.name] = len(field_ids)
                     if fd.field_type == OriginFieldType.list:
-                        new_table._origin.add_list_field(fd.name, fd.cpp_type)
+                        layer.add_list_field(fd.name, fd.cpp_type)
                     else:
-                        new_table._origin.add_field(fd.name, fd.field_type.value)
+                        layer.add_field(fd.name, fd.field_type.value)
                 if not materialize_table_buffer:
-                    new_table._origin._set_table_buffer_materialized(False)
-                engine._table_map[table_name] = new_table
-                engine._fixed_layer_builds[table_name] = new_table._origin
+                    layer._set_table_buffer_materialized(False)
+                engine._fixed_layer_builds[table_name] = layer
                 engine._fixed_table_fields[table_name] = field_ids
 
             engine._fixed_build.truncate(table_name, layout.capacity)
@@ -202,6 +229,7 @@ def truncate(layouts: List[Layout]) -> 'RecordEngine':
 
     def push(self, feature, table_name: str = '') -> None:
         """Push a single @feature instance to the database."""
+        self._assert_open()
         cls = feature.__class__
         # Fast path: batchable buffer
         buf = self._push_buf.get(cls)
@@ -242,25 +270,7 @@ def _push_full(self, feature, table_name: str = '') -> None:
         cache = normalize_bool_cache(feature.__dict__, schema)
 
         feat_table_name = table_name if table_name else feature_type.__name__
-        t_obj: Table = self._table_map.get(feat_table_name)
-        if t_obj is None:
-            new_table = Table.map_from(
-                feature_type,
-                _get_default_table_build(
-                    self._origin,
-                    feat_table_name,
-                    raw_payload=bool(schema.bytes_plan),
-                ),
-                self._origin,
-            )
-            for fd in schema.fields:
-                if fd.field_type == OriginFieldType.list:
-                    new_table._origin.add_list_field(fd.name, fd.cpp_type)
-                else:
-                    new_table._origin.add_field(fd.name, fd.field_type.value)
-            self._table_map[feat_table_name] = new_table
-            self._table_feature_types[feat_table_name] = feature_type
-            t_obj = new_table
+        t_obj = self._get_build_state(feature_type, feat_table_name, schema)
 
         # Ensure compiled push_fn exists
         if schema.push_fn is None:
@@ -294,6 +304,7 @@ def _push_full(self, feature, table_name: str = '') -> None:
 
     def push_many(self, features: list, table_name: str = '') -> None:
         """Push multiple @feature instances of the same type efficiently."""
+        self._assert_open()
         if not self._is_mutable or not features:
             return
         feature_type = features[0].__class__
@@ -320,25 +331,7 @@ def push_many(self, features: list, table_name: str = '') -> None:
                 for feat in features
             ]
         feat_table_name = table_name if table_name else feature_type.__name__
-        t_obj: Table = self._table_map.get(feat_table_name)
-        if t_obj is None:
-            new_table = Table.map_from(
-                feature_type,
-                _get_default_table_build(
-                    self._origin,
-                    feat_table_name,
-                    raw_payload=bool(schema.bytes_plan),
-                ),
-                self._origin,
-            )
-            for fd in schema.fields:
-                if fd.field_type == OriginFieldType.list:
-                    new_table._origin.add_list_field(fd.name, fd.cpp_type)
-                else:
-                    new_table._origin.add_field(fd.name, fd.field_type.value)
-            self._table_map[feat_table_name] = new_table
-            self._table_feature_types[feat_table_name] = feature_type
-            t_obj = new_table
+        t_obj = self._get_build_state(feature_type, feat_table_name, schema)
 
         t_origin = t_obj._origin
         fc = t_obj.feature_count
@@ -352,6 +345,22 @@ def push_many(self, features: list, table_name: str = '') -> None:
                 fc += 1
         t_obj.feature_count = fc
 
+    def _get_build_state(self, feature_type: Type, name: str, schema: LayerSchema) -> _TableState:
+        state = self._build_states.get(name)
+        if state is None:
+            state = _TableState()
+            state._db = self._origin
+            state._origin = _get_default_table_build(self._origin, name, raw_payload=bool(schema.bytes_plan))
+            state._feature_type = feature_type
+            for fd in schema.fields:
+                if fd.field_type == OriginFieldType.list:
+                    state._origin.add_list_field(fd.name, fd.cpp_type)
+                else:
+                    state._origin.add_field(fd.name, fd.field_type.value)
+            self._build_states[name] = state
+            self._table_feature_types[name] = feature_type
+        return state
+
     def _flush_push_batches(self):
         """Flush all pending batch push buffers to C++."""
         for cls, buf in self._push_buf.items():
@@ -365,6 +374,7 @@ def _flush_push_batches(self):
 
     def combine(self):
         """Combine memory from all tables into a single continuous block."""
+        self._assert_open()
         if self._origin is None:
             warnings.warn('Database is empty, cannot combine.', UserWarning)
             return
@@ -376,12 +386,15 @@ def combine(self):
 
         buffer = _post_build_to_bytearray(self._origin)
         self._origin = core.WxDatabase.load_xbuffer(buffer)
-        self._origin._buffer = buffer
 
-        if self._shm:
-            self._shm.close()
-            self._shm = None
-        self._table_map = {}
+        self._shm = None
+        self._tables.clear()
+        self._build_states.clear()
+        self._is_mutable = False
+        self._push_buf.clear()
+        self._push_batch_fn.clear()
+        self._push_dispatch.clear()
+        self._bool_push_schemas.clear()
 
     def table(
         self,
@@ -392,28 +405,33 @@ def table(
         writeable: bool | None = None,
     ) -> Table:
         """Get a Table for the given @feature class."""
+        self._assert_open()
         if not is_feature(feature_type):
             raise TypeError(
                 f'{feature_type.__name__} is not a @feature class. '
                 f'Use @feature decorator.'
             )
         table_name = name or feature_type.__name__
-        cached = self._table_map.get(table_name)
-        if cached is not None:
-            self._table_feature_types[table_name] = feature_type
-            if owner is not None or writeable is not None:
-                tbl = Table.map_from(
-                    feature_type,
-                    cached._origin,
-                    cached._db,
-                    owner=owner,
-                    writeable=writeable,
-                )
-                self._attach_owner_fixed_fill_handler(tbl, table_name)
-                return tbl
-            self._attach_fixed_fill_handler(cached, table_name)
-            return cached
+        alias = owner is not None or writeable is not None
+        if not alias:
+            entry = self._tables.get((table_name, 0))
+            if entry is not None:
+                state = entry.state()
+                if state is not None:
+                    cached = entry.table()
+                    if cached is None:
+                        cached = Table._from_state(state)
+                        entry.table = ref(cached)
+                    return cached
         db = self._origin
+        if isinstance(db, core.WxDatabaseBuild):
+            state = self._build_states.get(table_name)
+            if state is None:
+                raise KeyError(f'Table "{table_name}" not found')
+            tbl = Table.map_from(feature_type, state._origin, db, owner=owner, writeable=writeable)
+            tbl._state._fc = state._fc
+            self._register_table(tbl, table_name, alias)
+            return tbl
         fallback_names = list(self._table_feature_types)
         for i in range(db.get_layer_count()):
             layer = db.get_layer(i)
@@ -427,46 +445,38 @@ def table(
                     writeable=writeable,
                 )
                 self._table_feature_types[table_name] = feature_type
-                if owner is not None or writeable is not None:
-                    self._attach_owner_fixed_fill_handler(tbl, table_name)
-                    return tbl
-                self._attach_fixed_fill_handler(tbl, table_name)
-                self._table_map[table_name] = tbl
+                self._register_table(tbl, table_name, alias)
                 return tbl
         raise KeyError(f'Table "{table_name}" not found')
 
-    def _attach_owner_fixed_fill_handler(self, table: Table, table_name: str) -> None:
-        if table.fixed and self._fixed_direct_fill:
-            self._attach_fixed_fill_handler(table, table_name)
-            return
-        if table.fixed and self._fixed_build is not None and self._shm is None:
-            table._fixed_fill_handler = (
-                lambda writes, _table=table: self._fill_mapped_table_and_drop_fixed_build(_table, writes)
-            )
-            return
-        self._attach_fixed_fill_handler(table, table_name)
-
-    def _attach_fixed_fill_handler(self, table: Table, table_name: str) -> None:
-        if table.fixed and self._fixed_direct_fill:
-            table._fixed_fill_handler = (
-                lambda writes, _table=table: self._fill_mapped_table_in_place(_table, writes)
-            )
-        elif table.fixed and self._fixed_build is not None and self._shm is None:
-            table._fixed_fill_handler = (
-                lambda writes: self._fill_fixed_table(table_name, writes)
-            )
+    def _register_table(self, table: Table, name: str, alias: bool) -> None:
+        key = (name, id(table._state) if alias else 0)
+        self._tables[key] = _TableRegistration(self, key, table)
+        table._state._engine = self
+        self._attach_fixed_fill_handler(table._state, name, alias)
+
+    def _attach_fixed_fill_handler(self, state: _TableState, table_name: str, alias: bool = False) -> None:
+        state._table_name = table_name
+        fixed = isinstance(state._origin, core.WxLayerTable)
+        if fixed and self._fixed_direct_fill:
+            state._fixed_fill_handler = self._fill_mapped_table_in_place
+        elif fixed and self._fixed_build is not None and self._shm is None:
+            state._fixed_fill_handler = self._fill_mapped_table_and_drop_fixed_build if alias else self._fill_fixed_table
         else:
-            table._fixed_fill_handler = None
+            state._fixed_fill_handler = None
 
     def _invalidate_fixed_writes(self) -> None:
         self._fixed_build = None
         self._fixed_layer_builds = {}
         self._fixed_table_fields = {}
         self._fixed_direct_fill = False
-        for table_name, table in self._table_map.items():
-            self._attach_fixed_fill_handler(table, table_name)
+        for (table_name, alias_id), entry in list(self._tables.items()):
+            state = entry.state()
+            if state is not None:
+                self._attach_fixed_fill_handler(state, table_name, bool(alias_id))
 
-    def _fill_fixed_table(self, table_name: str, writes: dict[str, object]) -> None:
+    def _fill_fixed_table(self, state: _TableState, writes: dict[str, object]) -> None:
+        table_name = state._table_name
         layer_build = self._fixed_layer_builds[table_name]
         field_ids = self._fixed_table_fields[table_name]
         try:
@@ -520,26 +530,38 @@ def _publish_fixed_snapshot_to_buffer(self, buffer: object, *, direct_fill: bool
         self._publish_fixed_buffer(writable[:size], direct_fill=direct_fill)
 
     def _publish_fixed_buffer(self, buffer: object, *, direct_fill: bool) -> None:
+        previous_db = self._origin
         self._fixed_direct_fill = direct_fill
         self._origin = core.WxDatabase.load_xbuffer(buffer)
-        self._origin._buffer = buffer
 
-        for table_name, table in list(self._table_map.items()):
+        remaps = []
+        for (table_name, alias_id), entry in list(self._tables.items()):
+            state = entry.state()
+            if state is None or state._db is not previous_db:
+                continue
             new_layer = self._find_layer(table_name)
             if new_layer is None:
                 continue
-            table._remap(new_layer, self._origin)
-            self._attach_fixed_fill_handler(table, table_name)
+            remaps.append((state, entry.table(), new_layer, table_name, bool(alias_id)))
+        for owner in {state._fdb_owner for state, *_ in remaps}:
+            owner.bump_generation()
+        for state, table, layer, table_name, alias in remaps:
+            if table is None:
+                state._remap(layer, self._origin)
+            else:
+                table._remap(layer, self._origin)
+            self._attach_fixed_fill_handler(state, table_name, alias)
 
-    def _fill_mapped_table_in_place(self, table: Table, writes: dict[str, object]) -> None:
+    def _fill_mapped_table_in_place(self, state: _TableState, writes: dict[str, object]) -> None:
+        schema = get_schema(state._feature_type)
         for field_name, payload in writes.items():
             if isinstance(payload, _StringSequencePayload) or isinstance(payload, tuple):
                 raise RuntimeError('direct fixed backing writes do not support variable-size string payloads.')
-            column = getattr(table.column, field_name)
+            column = state._origin.get_column(schema.field_index_map[field_name]).as_nparray()
             column[:] = payload
 
-    def _fill_mapped_table_and_drop_fixed_build(self, table: Table, writes: dict[str, object]) -> None:
-        self._fill_mapped_table_in_place(table, writes)
+    def _fill_mapped_table_and_drop_fixed_build(self, state: _TableState, writes: dict[str, object]) -> None:
+        self._fill_mapped_table_in_place(state, writes)
         self._invalidate_fixed_writes()
 
     # ------------------------------------------------------------------
@@ -560,13 +582,14 @@ def load(name: str, from_file: bool = False) -> 'RecordEngine':
             name = _normalize_shm_name(name)
             try:
                 engine._shm = shared_memory.SharedMemory(name=name)
-                engine._origin = core.WxDatabase.load_xbuffer(engine._shm.buf)
+                engine._origin = _load_shared_database(engine._shm)
             except FileNotFoundError:
                 raise FileNotFoundError(f"Database '{name}' not found in shared memory.")
         return engine
 
     def share(self, shm_name: str, close_after: bool = False):
         """Share the database in shared memory."""
+        self._assert_open()
         if self._origin is None:
             raise RuntimeError('Database is empty, cannot share.')
         if isinstance(self._origin, core.WxDatabaseBuild):
@@ -575,17 +598,17 @@ def share(self, shm_name: str, close_after: bool = False):
         shm_name = _normalize_shm_name(shm_name)
         chunk = self._origin.buffer()
         self._shm = shared_memory.SharedMemory(create=True, size=chunk.size, name=shm_name)
-        dest = np.ndarray(chunk.size, dtype=np.uint8, buffer=self._shm.buf)
-        dest[:] = chunk.as_array(np.uint8)
-        self._origin._buffer = None
-        self._origin = core.WxDatabase.load_xbuffer(self._shm.buf)
+        chunk.copy_to_buffer(self._shm.buf)
+        self._origin = _load_shared_database(self._shm)
         self._invalidate_fixed_writes()
+        self._tables.clear()
 
         if close_after and platform.system() != 'Windows':
             self.close()
 
     def save(self, path: str):
         """Save the database to a file."""
+        self._assert_open()
         if self._origin is None:
             raise RuntimeError('Database is empty, cannot save.')
         if isinstance(self._origin, core.WxDatabaseBuild):
@@ -596,18 +619,28 @@ def save(self, path: str):
                 f.write(chunk.to_bytes())
 
     def close(self):
-        """Close the database and release resources."""
-        if self._shm:
-            self._shm.close()
-            self._shm = None
-            self._origin = None
+        """Release engine references; existing views retain their native backing."""
+        self._closed = True
+        self._tables.clear()
+        self._build_states.clear()
+        self._table_feature_types.clear()
+        self._origin = None
+        self._shm = None
+        self._fixed_build = None
+        self._fixed_layer_builds.clear()
+        self._fixed_table_fields.clear()
+        self._fixed_direct_fill = False
+        self._is_mutable = False
+        self._push_buf.clear()
+        self._push_batch_fn.clear()
+        self._push_dispatch.clear()
+        self._bool_push_schemas.clear()
 
     def unlink(self):
         """Unlink the shared memory database."""
         if self._shm:
             self._shm.unlink()
-            self._shm = None
-            self._origin = None
+        self.close()
 
     def __len__(self):
         if self._origin is None:
diff --git a/python/fastdb4py/serializer.py b/python/fastdb4py/serializer.py
index 5322c84..83b310f 100644
--- a/python/fastdb4py/serializer.py
+++ b/python/fastdb4py/serializer.py
@@ -294,6 +294,7 @@ def loads_shm(shm_name: str, length: int, offset: int, root_type: type):
         from multiprocessing import shared_memory as _shm_mod
 
         shm = _shm_mod.SharedMemory(name=shm_name, create=False)
+        db = None
         try:
             if offset + length > shm.size:
                 raise ValueError(
@@ -301,13 +302,9 @@ def loads_shm(shm_name: str, length: int, offset: int, root_type: type):
                     f"exceeds shared memory size ({shm.size})"
                 )
 
-            # Create a memoryview slice and hand it to the C++ loader.
-            # load_xbuffer's SWIG typemap acquires/releases its own Py_buffer,
-            # so we can release *buf* immediately after the call.  The C++ db
-            # holds a raw pointer into the still-alive mmap (owned by *shm*).
-            buf = shm.buf[offset:offset + length]
-            db = core.WxDatabase.load_xbuffer(buf)
-            buf.release()
+            # Native destruction releases the export before closing the mapping.
+            with shm.buf[offset:offset + length] as buf:
+                db = core.WxDatabase.load_xbuffer(buf, shm)
 
             if db.get_layer_count() == 0:
                 return None
@@ -335,7 +332,8 @@ def loads_shm(shm_name: str, length: int, offset: int, root_type: type):
 
             return result
         finally:
-            shm.close()
+            if db is None:
+                shm.close()
 
 # --- Internal Helpers ---
 
diff --git a/python/fastdb4py/string_column.py b/python/fastdb4py/string_column.py
index 2166d36..5abecbc 100644
--- a/python/fastdb4py/string_column.py
+++ b/python/fastdb4py/string_column.py
@@ -7,7 +7,7 @@
 from .view_owner import FdbViewOwner, FdbViewWriteError
 
 if TYPE_CHECKING:
-    from .orm.table import Table
+    from .orm.table import _TableState
 
 _UINT32_MAX = np.iinfo(np.uint32).max
 
@@ -69,14 +69,14 @@ def _normalize_string_values(
 
 
 class StringColumn:
-    def __init__(self, table: 'Table', field_index: int, field_name: str, *, wide: bool = False):
-        self._table = table
+    def __init__(self, state: '_TableState', field_index: int, field_name: str, *, wide: bool = False):
+        self._state = state
         self._field_index = field_index
         self._field_name = field_name
         self._wide = wide
-        self._owner: FdbViewOwner = table._fdb_owner
-        self._writeable = table._fdb_writeable
-        self._owner_generation = table._fdb_owner.generation
+        self._owner: FdbViewOwner = state._fdb_owner
+        self._writeable = state._fdb_writeable
+        self._owner_generation = state._fdb_owner.generation
 
     def _assert_alive(self) -> None:
         self._owner.assert_alive(self._owner_generation)
@@ -89,18 +89,18 @@ def _assert_writeable(self) -> None:
 
     def __len__(self) -> int:
         self._assert_alive()
-        return len(self._table)
+        return len(self._state)
 
     def _offsets(self) -> np.ndarray | None:
         self._assert_alive()
-        chunk = self._table._origin.get_string_column_offsets(self._field_index)
+        chunk = self._state._origin.get_string_column_offsets(self._field_index)
         if chunk is None or chunk.size == 0:
             return None
         return chunk.as_array(np.uint32)
 
     def _data(self) -> np.ndarray | None:
         self._assert_alive()
-        chunk = self._table._origin.get_string_column_data(self._field_index)
+        chunk = self._state._origin.get_string_column_data(self._field_index)
         if chunk is None or chunk.size == 0:
             return np.empty(0, dtype=np.uint8)
         return chunk.as_array(np.uint8)
@@ -120,8 +120,8 @@ def get(self, index: int) -> str:
             end = int(offsets[index + 1])
             return bytes(data[start:end]).decode('utf-8')
 
-        with self._table._read_lock:
-            feat = self._table._origin.tryGetFeature(index)
+        with self._state._read_lock:
+            feat = self._state._origin.tryGetFeature(index)
             if self._wide:
                 return feat.get_field_as_wstring(self._field_index) or ''
             raw = feat.get_field_as_string_view(self._field_index)
@@ -145,6 +145,7 @@ def to_owned(self) -> list[str]:
         return self.to_pylist()
 
     def fill(self, strings: Iterable[Optional[str]]) -> None:
+        self._state._assert_engine_open()
         self._assert_writeable()
         payload = _normalize_string_values(strings, len(self))
         self._dispatch_string_payload(payload)
@@ -158,9 +159,9 @@ def _normalize_fill_values(
         return self._validate_utf8_payload(offsets_arr, data_arr, expected_len)
 
     def _dispatch_string_payload(self, payload: _StringSequencePayload) -> None:
-        fill_handler = self._table._fixed_fill_handler
+        fill_handler = self._state._fixed_fill_handler
         if fill_handler is not None:
-            fill_handler({self._field_name: payload})
+            fill_handler(self._state, {self._field_name: payload})
             return
 
         raise RuntimeError(
@@ -194,6 +195,7 @@ def _validate_utf8_payload(
         return offsets_arr, data_arr
 
     def fill_utf8(self, offsets: np.ndarray, data: np.ndarray) -> None:
+        self._state._assert_engine_open()
         self._assert_writeable()
         offsets_arr, data_arr = self._validate_utf8_payload(offsets, data, len(self))
         self._dispatch_utf8_payload(offsets_arr, data_arr)
@@ -203,9 +205,9 @@ def _dispatch_utf8_payload(
         offsets_arr: np.ndarray,
         data_arr: np.ndarray,
     ) -> None:
-        fill_handler = self._table._fixed_fill_handler
+        fill_handler = self._state._fixed_fill_handler
         if fill_handler is not None:
-            fill_handler({self._field_name: (offsets_arr, data_arr)})
+            fill_handler(self._state, {self._field_name: (offsets_arr, data_arr)})
             return
 
         raise RuntimeError(
@@ -215,19 +217,19 @@ def _dispatch_utf8_payload(
 
 
 class BytesColumn:
-    def __init__(self, table: 'Table', field_index: int, field_name: str):
-        self._table = table
+    def __init__(self, state: '_TableState', field_index: int, field_name: str):
+        self._state = state
         self._field_index = field_index
         self._field_name = field_name
-        self._owner: FdbViewOwner = table._fdb_owner
-        self._owner_generation = table._fdb_owner.generation
+        self._owner: FdbViewOwner = state._fdb_owner
+        self._owner_generation = state._fdb_owner.generation
 
     def _assert_alive(self) -> None:
         self._owner.assert_alive(self._owner_generation)
 
     def __len__(self) -> int:
         self._assert_alive()
-        return len(self._table)
+        return len(self._state)
 
     def get(self, index: int) -> bytes:
         self._assert_alive()
@@ -237,8 +239,8 @@ def get(self, index: int) -> bytes:
         if index < 0 or index >= length:
             raise IndexError(f'Bytes index {index} out of range [0, {length}).')
 
-        with self._table._read_lock:
-            feat = self._table._origin.tryGetFeature(index)
+        with self._state._read_lock:
+            feat = self._state._origin.tryGetFeature(index)
             raw = feat.get_geometry_like_chunk()
             if raw is None:
                 return b''
diff --git a/tests/cpp/CMakeLists.txt b/tests/cpp/CMakeLists.txt
index 518df91..593dace 100644
--- a/tests/cpp/CMakeLists.txt
+++ b/tests/cpp/CMakeLists.txt
@@ -85,6 +85,27 @@ add_test(
     COMMAND fastdb_legacy_descriptor_determinism
 )
 
+add_executable(fastdb_legacy_lifetime test_legacy_lifetime.cpp)
+target_include_directories(
+    fastdb_legacy_lifetime
+    PRIVATE
+        ${CMAKE_CURRENT_SOURCE_DIR}/../../fastcarto/fastdb/include
+        ${CMAKE_CURRENT_SOURCE_DIR}/../../fastcarto/fastdb/src
+        ${CMAKE_CURRENT_SOURCE_DIR}/payload
+)
+target_link_libraries(fastdb_legacy_lifetime PRIVATE fastdb)
+target_compile_options(
+    fastdb_legacy_lifetime
+    PRIVATE
+        $<$<COMPILE_LANG_AND_ID:CXX,MSVC>:/Zc:preprocessor>
+)
+fastdb_apply_sanitizers(fastdb_legacy_lifetime)
+add_test(
+    NAME legacy.lifetime
+    COMMAND fastdb_legacy_lifetime
+        "$<TARGET_FILE_DIR:fastdb_legacy_lifetime>/legacy_lifetime.fdb"
+)
+
 set(FASTDB_PAYLOAD_FUZZ_MATCHING_SPEC
     ${CMAKE_CURRENT_SOURCE_DIR}/../golden/payload/v1/binary/spec/component-list-composition.source.json
 )
diff --git a/tests/cpp/test_legacy_lifetime.cpp b/tests/cpp/test_legacy_lifetime.cpp
new file mode 100644
index 0000000..4471299
--- /dev/null
+++ b/tests/cpp/test_legacy_lifetime.cpp
@@ -0,0 +1,191 @@
+#include "TestSupport.hpp"
+#include "fastdb.h"
+#include "FastVectorDbLayerBuild_p.h"
+
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <fstream>
+#include <memory>
+#include <vector>
+
+#ifdef _WIN32
+#include <windows.h>
+#else
+#include <sys/mman.h>
+#endif
+
+namespace {
+
+void* map_buffer(size_t size) {
+#ifdef _WIN32
+    return VirtualAlloc(nullptr, size, MEM_COMMIT | MEM_RESERVE,
+                        PAGE_READWRITE);
+#else
+    void* data = mmap(nullptr, size, PROT_READ | PROT_WRITE,
+                      MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
+    return data == MAP_FAILED ? nullptr : data;
+#endif
+}
+
+bool protect_buffer(void* data, size_t size, bool readable) {
+#ifdef _WIN32
+    DWORD previous = 0;
+    return VirtualProtect(data, size, readable ? PAGE_READWRITE : PAGE_NOACCESS,
+                          &previous) != 0;
+#else
+    return mprotect(data, size, readable ? PROT_READ | PROT_WRITE : PROT_NONE) == 0;
+#endif
+}
+
+bool unmap_buffer(void* data, size_t size) {
+#ifdef _WIN32
+    (void)size;
+    return VirtualFree(data, 0, MEM_RELEASE) != 0;
+#else
+    return munmap(data, size) == 0;
+#endif
+}
+
+std::vector<wx::u8> build_database() {
+    wx::FastVectorDbBuild build;
+    build.begin("{}");
+    build.createLayerBegin("values");
+    build.addField("tag", wx::ftU8);
+    build.addField("value", wx::ftF64);
+    for (int row = 0; row < 2; ++row) {
+        build.addFeatureBegin();
+        wx::point2_t point{double(row), double(row)};
+        build.setGeometry(&point, sizeof(point), wx::ginPoint2);
+        build.setField(0, row + 1);
+        build.setField(1, 2.5 + row);
+        build.addFeatureEnd();
+    }
+    build.createLayerEnd();
+    std::vector<wx::u8> bytes(build.byteLength());
+    if (build.postToBuffer(bytes.data(), bytes.size()) != bytes.size())
+        return {};
+    return bytes;
+}
+
+struct CallbackState {
+    void* expected_data = nullptr;
+    size_t expected_size = 0;
+    int calls = 0;
+    bool matched = false;
+    bool restored = false;
+    bool unmapped = false;
+};
+
+void release_buffer(void* data, size_t size, void* cookie) {
+    auto* state = static_cast<CallbackState*>(cookie);
+    state->matched = data == state->expected_data &&
+                     size == state->expected_size;
+    ++state->calls;
+    state->restored = protect_buffer(data, size, true);
+    state->unmapped = unmap_buffer(data, size);
+}
+
+int check_database(wx::FastVectorDb* db) {
+    require(db != nullptr);
+    require(db->getLayerCount() == 1);
+    auto* layer = db->getLayer(0);
+    require(layer != nullptr);
+    require(layer == db->getLayer(0));
+    require(layer->getFeatureCount() == 2);
+    require(layer->getFieldOffset(0) == 0);
+    require(layer->getFieldOffset(1) == sizeof(wx::u8));
+    require(layer->getFeatureByteSize() >= sizeof(wx::u8) + sizeof(double));
+
+    auto* base = static_cast<const wx::u8*>(layer->getTableDataAddress());
+    require(base != nullptr);
+    const size_t stride = layer->getFeatureByteSize();
+    const size_t value_offset = layer->getFieldOffset(1);
+    for (int row = 0; row < 2; ++row) {
+        require(base[row * stride] == row + 1);
+        double value = 0;
+        std::memcpy(&value, base + row * stride + value_offset,
+                    sizeof(value));
+        require(value == 2.5 + row);
+        auto* feature = layer->tryGetFeatureAt(row);
+        require(feature != nullptr);
+        require(feature == layer->tryGetFeatureAt(row));
+        require(feature->getFieldAsFloat(1) == value);
+    }
+    require(layer->tryGetFeatureAt(2) == nullptr);
+    return EXIT_SUCCESS;
+}
+
+}  // namespace
+
+int main(int argc, char** argv) {
+    require(argc == 2);
+    // A complete layer header does not imply its descriptors or string counts
+    // exist. Reject these envelopes before the layer constructor reads them.
+    std::vector<wx::u8> truncated(20 + sizeof(wx::layer_header_t), 0);
+    std::memcpy(truncated.data(), "FASTVectorDB0.1", sizeof("FASTVectorDB0.1"));
+    const wx::u32 layer_count = 1;
+    std::memcpy(truncated.data() + 16, &layer_count, sizeof(layer_count));
+    wx::layer_header_t header{};
+    header.total_size = sizeof(header);
+    for (wx::u16 fields : {wx::u16(1), wx::u16(0)}) {
+        header.field_count = fields;
+        std::memcpy(truncated.data() + 20, &header, sizeof(header));
+        require(wx::FastVectorDb::load(truncated.data(), truncated.size(), nullptr, nullptr) == nullptr);
+    }
+    void* invalid = map_buffer(truncated.size());
+    require(invalid != nullptr);
+    std::memcpy(invalid, truncated.data(), truncated.size());
+    CallbackState rejected{invalid, truncated.size()};
+    require(wx::FastVectorDb::load(invalid, truncated.size(), release_buffer, &rejected) == nullptr);
+    require(rejected.calls == 1 && rejected.restored && rejected.unmapped);
+
+    auto bytes = build_database();
+    require(!bytes.empty());
+
+    // Revoking reads after caching children catches destructor access to backing.
+    void* borrowed = map_buffer(bytes.size());
+    require(borrowed != nullptr);
+    std::memcpy(borrowed, bytes.data(), bytes.size());
+    {
+        std::unique_ptr<wx::FastVectorDb> db(
+            wx::FastVectorDb::load(borrowed, bytes.size(), nullptr, nullptr));
+        require(check_database(db.get()) == EXIT_SUCCESS);
+        require(protect_buffer(borrowed, bytes.size(), false));
+    }
+    require(protect_buffer(borrowed, bytes.size(), true));
+    require(static_cast<const char*>(borrowed)[0] == 'F');
+    require(unmap_buffer(borrowed, bytes.size()));
+
+    // The owned callback restores and releases a protected mapping exactly once.
+    void* owned = map_buffer(bytes.size());
+    require(owned != nullptr);
+    std::memcpy(owned, bytes.data(), bytes.size());
+    CallbackState state{owned, bytes.size()};
+    {
+        std::unique_ptr<wx::FastVectorDb> db(wx::FastVectorDb::load(
+            owned, bytes.size(), release_buffer, &state));
+        require(check_database(db.get()) == EXIT_SUCCESS);
+        require(state.calls == 0);
+        require(protect_buffer(owned, bytes.size(), false));
+    }
+    require(state.calls == 1);
+    require(state.matched);
+    require(state.restored);
+    require(state.unmapped);
+
+    // CTest supplies a path in the build directory, outside the launch cwd.
+    const char* filename = argv[1];
+    {
+        std::ofstream file(filename, std::ios::binary);
+        require(file.good());
+        file.write(reinterpret_cast<const char*>(bytes.data()), bytes.size());
+        require(file.good());
+    }
+    {
+        std::unique_ptr<wx::FastVectorDb> db(wx::FastVectorDb::load(filename));
+        require(check_database(db.get()) == EXIT_SUCCESS);
+    }
+    require(std::remove(filename) == 0);
+    return EXIT_SUCCESS;
+}
diff --git a/tests/python/test_native_lifetime.py b/tests/python/test_native_lifetime.py
new file mode 100644
index 0000000..9d5773b
--- /dev/null
+++ b/tests/python/test_native_lifetime.py
@@ -0,0 +1,342 @@
+"""Native ownership, borrowed view safety, and bounded long-running retention."""
+import gc
+import json
+from pathlib import Path
+import subprocess
+import sys
+import weakref
+
+import numpy as np
+import pytest
+
+import fastdb4py as fdb
+from fastdb4py import core
+
+
+@fdb.feature
+class NativeValue:
+    value: fdb.F32
+
+
+def _engine(kind):
+    engine_type = getattr(fdb, kind)
+    engine = engine_type.truncate([fdb.Layout(NativeValue, 3)])
+    engine.table(NativeValue).column.value[:] = [1.25, 2.5, 3.75]
+    return engine
+
+
+@pytest.fixture(autouse=True)
+def _without_gc():
+    enabled = gc.isenabled()
+    gc.disable()
+    try:
+        yield
+    finally:
+        if enabled:
+            gc.enable()
+
+
+@pytest.mark.parametrize('kind', ['RecordEngine', 'ObjectEngine'])
+@pytest.mark.parametrize('view_kind', ['table', 'row', 'column', 'slice', 'asarray', 'reuse'])
+def test_trusted_views_keep_native_database_after_engine_close(kind, view_kind):
+    engine = _engine(kind)
+    table = engine.table(NativeValue)
+    native = table._db
+    assert native.thisown
+    reference = weakref.ref(native)
+    if view_kind == 'table':
+        view = table
+    elif view_kind == 'row':
+        view = table[1]
+    elif view_kind == 'reuse':
+        view = next(table.iter_reuse())
+    elif view_kind == 'slice':
+        view = table.column.value[1:]
+    elif view_kind == 'asarray':
+        view = np.asarray(table.column.value)[1:]
+    else:
+        view = table.column.value
+    if kind == 'RecordEngine':
+        engine.close()
+        engine.close()
+    del native, table, engine
+    assert reference() is not None
+    if view_kind == 'table':
+        assert view[1].value == 2.5
+    elif view_kind in ('row', 'reuse'):
+        assert view.value == (1.25 if view_kind == 'reuse' else 2.5)
+    else:
+        expected = [2.5, 3.75] if view_kind in ('slice', 'asarray') else [1.25, 2.5, 3.75]
+        np.testing.assert_array_equal(view, expected)
+        view[-1] = 9
+        assert view[-1] == 9
+    del view
+    assert reference() is None
+
+
+@pytest.mark.parametrize('kind', ['RecordEngine', 'ObjectEngine'])
+def test_checked_views_survive_engine_close_until_caller_invalidates_owner(kind):
+    engine = _engine(kind)
+    owner = fdb.FdbViewOwner(checked=True, writeable=True)
+    table = engine.table(NativeValue, owner=owner)
+    row = table[0]
+    column = table.column.value
+    detached = column.to_numpy()
+    if kind == 'RecordEngine':
+        engine.close()
+    del engine
+    assert row.value == 1.25
+    assert column[0] == 1.25
+    fdb.invalidate(owner)
+    for read in (lambda: len(table), lambda: row.value, lambda: column[0]):
+        with pytest.raises(fdb.FdbViewInvalidatedError):
+            read()
+    np.testing.assert_array_equal(detached, [1.25, 2.5, 3.75])
+
+
+def test_file_loaded_database_and_its_raw_column_release_after_last_reference(tmp_path):
+    engine = _engine('RecordEngine')
+    path = tmp_path / 'value.fdb'
+    engine.save(str(path))
+    engine.close()
+    loaded = fdb.RecordEngine.load(str(path), from_file=True)
+    native = loaded._origin
+    reference = weakref.ref(native)
+    assert native.thisown
+    column = loaded.table(NativeValue).column.value
+    loaded.close()
+    assert loaded._origin is None
+    del native, loaded
+    np.testing.assert_array_equal(column, [1.25, 2.5, 3.75])
+    assert reference() is not None
+    del column
+    assert reference() is None
+
+
+def test_borrowed_buffer_remains_exported_and_alive_until_native_database_release():
+    engine = _engine('RecordEngine')
+    backing = bytearray(engine._origin.buffer().to_bytes())
+    engine.close()
+    native = core.WxDatabase.load_xbuffer(backing)
+    assert native.thisown
+    reference = weakref.ref(native)
+    layer = native.get_layer(0)
+    column = layer.get_column(0).as_nparray()
+    with pytest.raises(BufferError):
+        backing.extend(b'x')
+    del native, layer
+    np.testing.assert_array_equal(column, [1.25, 2.5, 3.75])
+    assert reference() is not None
+    del column
+    assert reference() is None
+    backing.extend(b'x')
+
+
+def test_raw_chunk_array_keeps_database_alive():
+    engine = _engine('RecordEngine')
+    expected = engine._origin.buffer().to_bytes()
+    reference = weakref.ref(engine._origin)
+    array = engine._origin.buffer().as_array(np.uint8)
+    engine.close()
+    del engine
+    assert reference() is not None
+    assert array.tobytes() == expected
+    del array
+    assert reference() is None
+
+
+def test_stream_copy_input_and_borrowed_array_lifetime():
+    stream = core.WxMemoryStream()
+    stream.write(b'copied input')
+    reference = weakref.ref(stream)
+    array = stream.data().as_array(np.uint8)
+    del stream
+    assert reference() is not None
+    assert array.tobytes() == b'copied input'
+    del array
+    assert reference() is None
+
+
+def test_shared_mapping_closes_after_last_raw_view():
+    import secrets
+
+    engine = _engine('RecordEngine')
+    old_table = engine.table(NativeValue)
+    engine.share('fdb_lifetime_' + secrets.token_hex(8))
+    with pytest.raises(RuntimeError, match='read-only fixed tables'):
+        old_table.fill(value=[4, 5, 6])
+    mapping = engine._shm._mmap
+    column = engine.table(NativeValue).column.value
+    engine.unlink()
+    del engine
+    assert not mapping.closed
+    np.testing.assert_array_equal(column, [1.25, 2.5, 3.75])
+    del column
+    assert mapping.closed
+
+
+@pytest.mark.skipif(sys.platform == 'win32', reason='close_after releases the publisher on POSIX only')
+def test_shared_close_after_publish_releases_mapping(monkeypatch):
+    import secrets
+    from multiprocessing import shared_memory
+    import fastdb4py.record_engine as record_module
+
+    mappings = []
+    load_shared = record_module._load_shared_database
+
+    def capture_mapping(shm):
+        mappings.append(shm._mmap)
+        return load_shared(shm)
+
+    monkeypatch.setattr(record_module, '_load_shared_database', capture_mapping)
+
+    name = 'fdb_close_' + secrets.token_hex(8)
+    engine = _engine('RecordEngine')
+    engine.share(name, close_after=True)
+    assert mappings[0].closed
+    engine.close()
+    loaded = fdb.RecordEngine.load(name)
+    np.testing.assert_array_equal(loaded.table(NativeValue).column.value, [1.25, 2.5, 3.75])
+    loaded.close()
+    assert mappings[1].closed
+    segment = shared_memory.SharedMemory(name=name)
+    segment.unlink()
+    segment.close()
+
+
+def test_object_get_row_survives_engine_release():
+    engine = _engine('ObjectEngine')
+    row = engine.get(NativeValue, 1)
+    reference = weakref.ref(engine._db)
+    del engine
+    assert row.value == 2.5
+    assert reference() is not None
+    del row
+    assert reference() is None
+
+
+def test_factory_reference_ownership_matches_native_parent():
+    standalone = core.WxFeatureRef.make_ref(0, 1)
+    assert standalone.thisown
+    build = core.WxDatabaseBuild()
+    build.begin('')
+    layer = build.create_layer_begin('values')
+    layer.add_field('value', core.ftF32)
+    layer.add_feature_begin()
+    layer.set_field(0, 2.5)
+    layer.add_feature_end()
+    borrowed = layer.create_feature_ref()
+    assert not borrowed.thisown
+    reference = weakref.ref(build)
+    del build, layer
+    assert reference() is not None
+    assert borrowed.ilayer == 0
+    del borrowed
+    assert reference() is None
+
+
+@pytest.mark.parametrize('kind', ['RecordEngine', 'ObjectEngine'])
+def test_retained_column_accessor_keeps_native_backing_without_gc(kind):
+    engine = _engine(kind)
+    native = engine._origin if kind == 'RecordEngine' else engine._db
+    reference = weakref.ref(native)
+    accessor = engine.table(NativeValue).column
+    if kind == 'RecordEngine':
+        engine.close()
+    del engine, native
+    np.testing.assert_array_equal(accessor.value, [1.25, 2.5, 3.75])
+    assert reference() is not None
+    del accessor
+    assert reference() is None
+
+
+def test_table_retains_engine_without_engine_retaining_table():
+    engine = _engine('RecordEngine')
+    table = engine.table(NativeValue)
+    engine_ref = weakref.ref(engine)
+    native_ref = weakref.ref(engine._origin)
+    del engine
+    assert engine_ref() is not None
+    assert table[1].value == 2.5
+    del table
+    assert engine_ref() is None
+    assert native_ref() is None
+
+
+def test_record_combine_releases_builder_and_push_dispatch_without_gc():
+    engine = fdb.RecordEngine.create()
+    engine.push(NativeValue(value=1.25))
+    builder_ref = weakref.ref(engine._origin)
+    engine.combine()
+    assert builder_ref() is None
+    assert not engine._push_buf
+    assert not engine._push_batch_fn
+    assert not engine._push_dispatch
+    engine.close()
+
+
+def test_object_combine_releases_layer_and_database_builders_without_gc():
+    engine = fdb.ObjectEngine.create()
+    state = engine._ensure_layer(NativeValue)
+    builder_ref = weakref.ref(engine._db_build)
+    layer_ref = weakref.ref(state.build)
+    engine.push(NativeValue(value=1.25))
+    engine.combine()
+    assert state.build is None
+    assert builder_ref() is None
+    assert layer_ref() is None
+    del engine
+
+
+def test_snapshot_remap_updates_checked_alias_and_invalidates_its_old_views():
+    engine = _engine('RecordEngine')
+    table = engine.table(NativeValue)
+    owner = fdb.FdbViewOwner(checked=True, writeable=True)
+    alias = engine.table(NativeValue, owner=owner)
+    old_row = alias[0]
+    old_column = alias.column.value
+    raw = table.column.value
+    table.fill(value=[4, 5, 6])
+    np.testing.assert_array_equal(alias.column.value.to_numpy(), [4, 5, 6])
+    np.testing.assert_array_equal(raw, [1.25, 2.5, 3.75])
+    with pytest.raises(fdb.FdbViewInvalidatedError):
+        _ = old_row.value
+    with pytest.raises(fdb.FdbViewInvalidatedError):
+        _ = old_column[0]
+    engine.close()
+
+
+def test_empty_numeric_column():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(NativeValue, 0)])
+    assert engine.table(NativeValue).column.value.shape == (0,)
+    engine.close()
+
+
+def test_bulk_columns_do_not_request_per_row_native_features(monkeypatch):
+    def reject_row_lookup(*args):
+        raise AssertionError('bulk columns must not populate the feature cache')
+
+    monkeypatch.setattr(core.WxLayerTable, 'tryGetFeature', reject_row_lookup)
+    engine = fdb.RecordEngine.truncate([fdb.Layout(NativeValue, 1_000_000)])
+    table = engine.table(NativeValue)
+    table.fill(value=np.full(1_000_000, 7, dtype=np.float32))
+    assert table.column.value[0] == 7
+    assert table.column.value[-1] == 7
+    engine.close()
+
+
+@pytest.mark.parametrize('mode', ['write', 'read'])
+def test_million_row_cycles_have_bounded_retained_memory(tmp_path, mode):
+    probe = Path(__file__).resolve().parents[2] / 'tools' / 'probe_standalone_memory.py'
+    result = subprocess.run(
+        [sys.executable, str(probe), '--mode', mode, '--rows', '1000000',
+         '--cycles', '10', '--path', str(tmp_path / 'frame.fdb')],
+        capture_output=True, text=True, check=True,
+    )
+    report = json.loads(result.stdout.splitlines()[-1])
+    # Allow two allocator warm-up cycles, then at most 16 MiB growth above
+    # that baseline across eight cycles (2 MiB/cycle). The old leak retains 16-20 MiB per
+    # cycle at this row count, so this tolerates allocator noise while rejecting
+    # linear retention by at least an order of magnitude.
+    samples = report['retained_bytes'][2:]
+    assert max(samples) - report['retained_bytes'][1] < 16 * 1024**2, result.stdout
diff --git a/tests/python/test_native_shutdown.py b/tests/python/test_native_shutdown.py
new file mode 100644
index 0000000..cc7c3dd
--- /dev/null
+++ b/tests/python/test_native_shutdown.py
@@ -0,0 +1,214 @@
+"""Borrowed buffers survive finalization and release through native destruction."""
+import gc
+import mmap
+from multiprocessing import shared_memory
+import subprocess
+import sys
+import textwrap
+import weakref
+
+import numpy as np
+import pytest
+
+import fastdb4py as fdb
+from fastdb4py import core
+
+
+@fdb.feature
+class ShutdownValue:
+    value: fdb.F32
+
+
+def _bytes():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(ShutdownValue, 4)])
+    engine.table(ShutdownValue).fill(value=np.arange(4, dtype=np.float32))
+    data = engine._origin.buffer().to_bytes()
+    engine.close()
+    return data
+
+
+@pytest.mark.parametrize('kind', ['bytes', 'bytearray', 'mmap', 'shared_memory'])
+def test_native_buffer_pin_and_mapping_release_without_cyclic_gc(kind):
+    enabled = gc.isenabled()
+    gc.disable()
+    data = _bytes()
+    mapping = None
+    segment = None
+    if kind == 'mmap':
+        mapping = mmap.mmap(-1, len(data))
+        mapping[:] = data
+        backing = mapping
+    elif kind == 'shared_memory':
+        segment = shared_memory.SharedMemory(create=True, size=len(data))
+        segment.buf[:len(data)] = data
+        mapping = segment._mmap
+        backing = segment.buf
+    else:
+        backing = bytes(data) if kind == 'bytes' else bytearray(data)
+    try:
+        native = core.WxDatabase.load_xbuffer(backing, segment)
+        reference = weakref.ref(native)
+        column = native.get_layer(0).get_column(0).as_nparray()
+        if kind == 'bytearray':
+            with pytest.raises(BufferError):
+                backing.extend(b'x')
+        elif kind in ('mmap', 'shared_memory'):
+            with pytest.raises(BufferError):
+                mapping.close()
+        del native
+        assert reference() is not None
+        np.testing.assert_array_equal(column, [0, 1, 2, 3])
+        del column
+        assert reference() is None
+        if kind == 'shared_memory':
+            assert mapping.closed
+        elif kind == 'mmap':
+            mapping.close()
+            assert mapping.closed
+        elif kind == 'bytearray':
+            backing.extend(b'x')
+    finally:
+        if segment is not None:
+            segment.unlink()
+            segment.close()
+        elif mapping is not None:
+            mapping.close()
+        if enabled:
+            gc.enable()
+
+
+@pytest.mark.parametrize('data', [b'', b'wrong database', b'FASTVectorDB0.1\0' + b'\0' * 3])
+def test_invalid_buffers_have_one_error_and_release_their_export(data):
+    backing = bytearray(data)
+    with pytest.raises(ValueError, match=r'^Invalid FastDB database buffer\.$'):
+        core.WxDatabase.load_xbuffer(backing)
+    backing.extend(b'x')
+
+
+def test_array_base_always_retains_the_actual_chunk():
+    native = core.WxDatabase.load_xbuffer(_bytes())
+    chunk = native.buffer()
+    array = chunk.as_array(np.uint8)
+    assert array.base is chunk
+    with pytest.raises(TypeError):
+        chunk.as_array(np.uint8, None)
+    reference = weakref.ref(native)
+    del native, chunk
+    assert reference() is not None
+    assert array.tobytes().startswith(b'FASTVectorDB0.1')
+    del array
+    assert reference() is None
+
+
+def test_failed_shared_load_closes_its_mapping():
+    from fastdb4py.record_engine import _load_shared_database
+
+    segment = shared_memory.SharedMemory(create=True, size=64)
+    mapping = segment._mmap
+    try:
+        with pytest.raises(ValueError, match=r'^Invalid FastDB database buffer\.$'):
+            _load_shared_database(segment)
+        assert mapping.closed
+    finally:
+        segment.unlink()
+        segment.close()
+
+
+def test_invalid_file_uses_the_same_native_factory_error(tmp_path):
+    path = tmp_path / 'invalid.fdb'
+    path.write_bytes(b'not a database')
+    with pytest.raises(ValueError, match=r'^Invalid FastDB database buffer\.$'):
+        fdb.RecordEngine.load(str(path), from_file=True)
+
+
+def test_close_does_not_invalidate_an_owner_shared_by_two_engines():
+    owner = fdb.FdbViewOwner(checked=True)
+    first = fdb.RecordEngine.truncate([fdb.Layout(ShutdownValue, 4)])
+    second = fdb.RecordEngine.truncate([fdb.Layout(ShutdownValue, 4)])
+    left = first.table(ShutdownValue, owner=owner)
+    right = second.table(ShutdownValue, owner=owner)
+    first.close()
+    assert owner.alive
+    assert left[0].value == right[0].value == 0
+    fdb.invalidate(owner)
+    with pytest.raises(fdb.FdbViewInvalidatedError):
+        _ = right[0].value
+    second.close()
+
+
+_SHUTDOWN_SETUP = '''
+import gc, secrets, threading, time
+import numpy as np
+import fastdb4py as fdb
+@fdb.feature
+class V:
+    value: fdb.F32
+name = 'fdb_shutdown_' + secrets.token_hex(8)
+publisher = fdb.RecordEngine.truncate([fdb.Layout(V, 2000000)])
+publisher.table(V).fill(value=np.ones(2000000, dtype=np.float32))
+publisher.share(name)
+'''
+
+_SHUTDOWN_CASES = {
+    'daemon': '''
+reader = fdb.RecordEngine.load(name)
+column = reader.table(V).column.value
+publisher.unlink()
+ready = threading.Event()
+def worker():
+    ready.set()
+    while True:
+        float(column.sum())
+threading.Thread(target=worker, daemon=True).start()
+ready.wait()
+time.sleep(0.1)
+print('daemon reader exits safely', flush=True)
+''',
+    'module_del': '''
+import os
+class Reporter:
+    # Late teardown cannot import, so read and report through C-level calls only.
+    def __init__(self, column):
+        self.column = column
+        self.one = np.float32(1).tobytes()
+        self.write = os.write
+    def __del__(self):
+        assert self.column.tobytes().count(self.one) == 2000000
+        self.write(1, b'module destructor read passed\\n')
+reader = fdb.RecordEngine.load(name)
+reporter = Reporter(reader.table(V).column.value)
+publisher.unlink()
+''',
+    'gc_cycle': '''
+class Service:
+    def __init__(self):
+        self.engine = fdb.RecordEngine.load(name)
+        self.table = self.engine.table(V)
+        self.cycle = self
+    def __del__(self):
+        assert self.table[3].value == 1
+        print('cycle destructor read passed', flush=True)
+service = Service()
+mapping = service.engine._shm._mmap
+del service
+gc.collect()
+assert mapping.closed
+publisher.unlink()
+print('cycle mapping closed', flush=True)
+''',
+}
+
+
+@pytest.mark.parametrize('case', _SHUTDOWN_CASES)
+def test_shared_memory_shutdown_subprocess(case, tmp_path):
+    result = subprocess.run(
+        [sys.executable, '-X', 'faulthandler', '-c',
+         textwrap.dedent(_SHUTDOWN_SETUP + _SHUTDOWN_CASES[case])],
+        cwd=tmp_path, capture_output=True, text=True, timeout=30,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert 'Exception ignored' not in result.stderr, result.stderr
+    if case == 'gc_cycle':
+        assert 'cycle mapping closed' in result.stdout
+    elif case == 'module_del':
+        assert 'module destructor read passed' in result.stdout
diff --git a/tests/python/test_object_engine_lifetime.py b/tests/python/test_object_engine_lifetime.py
new file mode 100644
index 0000000..010103a
--- /dev/null
+++ b/tests/python/test_object_engine_lifetime.py
@@ -0,0 +1,37 @@
+"""Reference-counted lifetime of standalone ObjectEngine views."""
+
+import weakref
+
+import numpy as np
+
+import fastdb4py as fdb
+
+
+@fdb.feature
+class ObjectLifetimeValue:
+    value: fdb.F32
+
+
+def test_raw_column_retains_native_database_after_engine_is_released():
+    engine = fdb.ObjectEngine.create()
+    engine.push(ObjectLifetimeValue(value=1.25))
+    engine.push(ObjectLifetimeValue(value=2.5))
+    engine.combine()
+
+    assert engine._db_build is None
+    assert all(state.build is None for state in engine._layers.values())
+
+    table = engine.table(ObjectLifetimeValue)
+    column = table.column.value
+    native_ref = weakref.ref(table._db)
+    engine_ref = weakref.ref(engine)
+    del table, engine
+
+    assert engine_ref() is None
+    assert native_ref() is not None
+    np.testing.assert_array_equal(column, [1.25, 2.5])
+    column[1] = 3.75
+    assert column[1] == 3.75
+
+    del column
+    assert native_ref() is None
diff --git a/tests/python/test_push.py b/tests/python/test_push.py
index eddc589..bfb7824 100644
--- a/tests/python/test_push.py
+++ b/tests/python/test_push.py
@@ -45,7 +45,6 @@ def test_push_scalar_fields():
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     layer = rdb.get_layer(0)
     feat = layer.tryGetFeature(0)
     assert abs(feat.get_field_as_float(0) - 42.0) < 1e-9
@@ -67,7 +66,6 @@ class Named:
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     layer = rdb.get_layer(0)
     feat = layer.tryGetFeature(0)
 def test_push_list_field():
@@ -85,7 +83,6 @@ class WithList:
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     layer = rdb.get_layer(0)
     feat = layer.tryGetFeature(0)
     arr = feat.get_field_as_list_view(0).as_array(np.float64)
@@ -102,7 +99,6 @@ def test_push_default_values():
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     layer = rdb.get_layer(0)
     feat = layer.tryGetFeature(0)
     assert feat.get_field_as_float(0) == 0.0
@@ -124,7 +120,6 @@ class WithEmptyList:
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     layer = rdb.get_layer(0)
     feat = layer.tryGetFeature(0)
     arr = feat.get_field_as_list_view(0).as_array(np.float64)
@@ -144,7 +139,6 @@ class WithNumpyList:
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     layer = rdb.get_layer(0)
     feat = layer.tryGetFeature(0)
     arr = feat.get_field_as_list_view(0).as_array(np.float64)
@@ -176,7 +170,6 @@ class Mixed:
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     layer = rdb.get_layer(0)
     feat = layer.tryGetFeature(0)
     
@@ -231,4 +224,4 @@ def test_push_ref_list_via_orm():
     tag0 = orm.get(Tag, 0, mode='copy')
     tag1 = orm.get(Tag, 1, mode='copy')
     assert tag0.name == "python"
-    assert tag1.name == "fastdb"
\ No newline at end of file
+    assert tag1.name == "fastdb"
diff --git a/tests/python/test_reader.py b/tests/python/test_reader.py
index ea307b6..76c9940 100644
--- a/tests/python/test_reader.py
+++ b/tests/python/test_reader.py
@@ -41,7 +41,6 @@ def _build_db_with_points():
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     return rdb
 
 
@@ -188,7 +187,6 @@ class WithBytes:
         db.post(mem)
         buf = mem.data().as_array(np.uint8).tobytes()
         rdb = core.WxDatabase.load_xbuffer(buf)
-        rdb._buffer = buf
 
         layer = rdb.get_layer(0)
         obj = copy_feature(WithBytes, layer, 0)
@@ -199,7 +197,6 @@ class WithBytes:
 def test_varlen_string_column_reader_exposes_raw_buffers():
     buf, offsets, data = _build_varlen_string_db_bytes(["a", "bé", "中"])
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
 
     layer = rdb.get_layer(0)
     np.testing.assert_array_equal(
@@ -220,7 +217,6 @@ def test_varlen_string_column_reader_rejects_truncated_section_payload():
     corrupted_buf = _corrupt_first_varlen_string_byte_count(valid_buf, data.nbytes + 8)
 
     rdb = core.WxDatabase.load_xbuffer(corrupted_buf)
-    rdb._buffer = corrupted_buf
 
     layer = rdb.get_layer(0)
     assert layer.get_string_column_offsets(0).size == 0
@@ -242,7 +238,6 @@ def test_truncate_layer_build_set_numeric_column_bulk_round_trips():
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     out = rdb.get_layer(0).get_column(0).as_nparray()
     np.testing.assert_allclose(out, np.array([1.0, 2.5, 3.5], dtype=np.float64))
 
@@ -278,7 +273,6 @@ def test_truncate_layer_build_set_numeric_column_bulk_silently_ignores_wrong_siz
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     out = rdb.get_layer(0).get_column(0).as_nparray()
     np.testing.assert_allclose(out, np.array([1.0, 2.5, 3.5], dtype=np.float64))
 
@@ -297,7 +291,6 @@ def test_truncate_layer_build_set_numeric_column_bulk_zero_rows_round_trips():
     db.post(mem)
     buf = mem.data().as_array(np.uint8).tobytes()
     rdb = core.WxDatabase.load_xbuffer(buf)
-    rdb._buffer = buf
     out_layer = rdb.get_layer(0)
     assert out_layer.get_feature_count() == 0
     assert out_layer.get_field_count() == 1
diff --git a/tests/python/test_record_engine.py b/tests/python/test_record_engine.py
index 825666d..090cd15 100644
--- a/tests/python/test_record_engine.py
+++ b/tests/python/test_record_engine.py
@@ -260,7 +260,7 @@ def test_record_engine_push_many_bool_fields_reject_before_table_creation():
             CEBoolPoint(active='maybe'),
         ])
 
-    assert CEBoolPoint.__name__ not in engine._table_map
+    assert CEBoolPoint.__name__ not in engine._table_feature_types
     assert CEBoolPoint.__name__ not in engine._table_feature_types
 
 
@@ -447,7 +447,7 @@ def test_table_fill_coerces_float32_into_f64_columns():
 def test_table_fill_rejects_non_fixed_tables():
     engine = RecordEngine.create()
     engine.push(CEPoint(x=1.0, y=2.0))
-    tbl = engine._table_map[CEPoint.__name__]
+    tbl = engine.table(CEPoint)
 
     with pytest.raises(RuntimeError, match='fixed-scale tables'):
         tbl.fill(x=np.array([1.0], dtype=np.float64))
diff --git a/tests/python/test_string_column.py b/tests/python/test_string_column.py
index d3311ef..03117ad 100644
--- a/tests/python/test_string_column.py
+++ b/tests/python/test_string_column.py
@@ -95,15 +95,15 @@ def test_normalize_string_values_wraps_raw_sequence_payload():
 
 def test_string_column_fill_routes_sequence_payload_to_fixed_handler(monkeypatch):
     engine = RecordEngine.truncate([Layout(CEStringPoint, 2)])
-    tbl = engine.table(CEStringPoint)
     captured = []
     original = engine._fill_fixed_table
 
-    def capture_fill(table_name, writes):
-        captured.append((table_name, writes))
-        return original(table_name, writes)
+    def capture_fill(state, writes):
+        captured.append((state._table_name, writes))
+        return original(state, writes)
 
     monkeypatch.setattr(engine, '_fill_fixed_table', capture_fill)
+    tbl = engine.table(CEStringPoint)
 
     tbl.column.name.fill(["hi", None])
 
@@ -118,16 +118,16 @@ def capture_fill(table_name, writes):
 
 def test_string_column_fill_utf8_uses_unified_fixed_writer(monkeypatch):
     engine = RecordEngine.truncate([Layout(CEStringPoint, 2)])
-    tbl = engine.table(CEStringPoint)
     captured = []
 
     original = engine._fill_fixed_table
 
-    def capture_fill(table_name, writes):
-        captured.append((table_name, writes))
-        return original(table_name, writes)
+    def capture_fill(state, writes):
+        captured.append((state._table_name, writes))
+        return original(state, writes)
 
     monkeypatch.setattr(engine, '_fill_fixed_table', capture_fill)
+    tbl = engine.table(CEStringPoint)
 
     offsets, data = _pack_utf8(["hi", "中"])
     tbl.column.name.fill_utf8(offsets, data)
diff --git a/tests/python/test_table_ownership_cache.py b/tests/python/test_table_ownership_cache.py
new file mode 100644
index 0000000..a53626f
--- /dev/null
+++ b/tests/python/test_table_ownership_cache.py
@@ -0,0 +1,166 @@
+"""Cached table projections retain parents without requiring cyclic GC."""
+import gc
+import weakref
+
+import numpy as np
+import pytest
+
+import fastdb4py as fdb
+
+
+@fdb.feature
+class CachedValue:
+    value: fdb.F32
+    name: fdb.STR
+
+
+@pytest.fixture(autouse=True)
+def without_gc():
+    enabled = gc.isenabled()
+    gc.disable()
+    try:
+        yield
+    finally:
+        if enabled:
+            gc.enable()
+
+
+@pytest.mark.parametrize('checked', [False, True])
+def test_accessor_and_columns_are_cached_without_retaining_table(checked):
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    owner = fdb.FdbViewOwner(checked=checked, writeable=True)
+    table = engine.table(CachedValue, owner=owner)
+    accessor = table.column
+    numeric, strings = accessor.value, accessor.name
+    assert table.column is accessor
+    assert table.column.value is numeric
+    assert table.column.name is strings
+    assert weakref.ref(table.column)() is accessor
+    table_ref = weakref.ref(table)
+    engine_ref = weakref.ref(engine)
+    native_ref = weakref.ref(table._db)
+    engine.close()
+    del engine, table
+    assert table_ref() is None
+    assert engine_ref() is not None
+    assert len(numeric) == len(strings) == 3
+    del accessor, numeric, strings
+    assert engine_ref() is None
+    assert native_ref() is None
+
+
+def test_table_retains_engine_for_chained_fill_and_releases_immediately():
+    table = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]).table(CachedValue)
+    table.fill(value=[1, 2, 3], name=['a', 'b', 'c'])
+    np.testing.assert_array_equal(table.column.value, [1, 2, 3])
+    assert table.column.name.to_pylist() == ['a', 'b', 'c']
+    native_ref = weakref.ref(table._db)
+    del table
+    assert native_ref() is None
+
+
+def test_table_strongly_caches_accessor_between_independent_reads():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    table = engine.table(CachedValue)
+    cached = weakref.ref(table.column)
+    assert cached() is not None
+    assert table.column is cached()
+    column = table.column.value
+    assert table.column.value is column
+    table_ref = weakref.ref(table)
+    del table
+    assert table_ref() is None
+    assert cached() is None
+
+
+def test_one_weak_registry_keeps_default_table_identity_and_releases_aliases():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    table = engine.table(CachedValue)
+    assert engine.table(CachedValue) is table
+    owner = fdb.FdbViewOwner(checked=True, writeable=True)
+    alias = engine.table(CachedValue, owner=owner)
+    assert len(engine._tables) == 2
+    table_ref, alias_ref = weakref.ref(table), weakref.ref(alias)
+    del table, alias
+    assert table_ref() is alias_ref() is None
+    assert not engine._tables
+    engine_ref = weakref.ref(engine)
+    del engine
+    assert engine_ref() is None
+
+
+def test_remap_replaces_cached_columns_and_releases_old_database():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    table = engine.table(CachedValue)
+    accessor = table.column
+    column = accessor.value
+    previous = weakref.ref(table._db)
+    table.fill(value=[4, 5, 6], name=['x', 'y', 'z'])
+    assert table.column is not accessor
+    assert table.column.value is table.column.value
+    np.testing.assert_array_equal(table.column.value, [4, 5, 6])
+    np.testing.assert_array_equal(column, [0, 0, 0])
+    assert previous() is not None
+    del accessor, column
+    assert previous() is None
+    current = weakref.ref(table._db)
+    del table, engine
+    assert current() is None
+
+
+def test_checked_accessor_cannot_create_columns_on_stale_backing_after_remap():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    owner = fdb.FdbViewOwner(checked=True, writeable=True)
+    table = engine.table(CachedValue, owner=owner)
+    accessor = table.column
+    engine.table(CachedValue).fill(value=[1, 2, 3], name=['a', 'b', 'c'])
+    for field in ('value', 'name'):
+        with pytest.raises(fdb.FdbViewInvalidatedError, match='stale after owner remap'):
+            getattr(accessor, field)
+    np.testing.assert_array_equal(table.column.value.to_numpy(), [1, 2, 3])
+    assert table.column.name.to_pylist() == ['a', 'b', 'c']
+
+
+def test_retained_string_column_follows_snapshot_without_retaining_table():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    table = engine.table(CachedValue)
+    column = table.column.name
+    table_ref = weakref.ref(table)
+    del table
+    assert table_ref() is None
+    column.fill(['a', 'b', 'c'])
+    assert column.to_pylist() == ['a', 'b', 'c']
+    engine.table(CachedValue).fill(name=['x', 'y', 'z'])
+    assert column.to_pylist() == ['x', 'y', 'z']
+    native_ref = weakref.ref(engine._origin)
+    del engine
+    assert column.to_pylist() == ['x', 'y', 'z']
+    del column
+    assert native_ref() is None
+
+
+def test_remap_refreshes_aliases_that_share_a_checked_owner():
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    owner = fdb.FdbViewOwner(checked=True, writeable=True)
+    aliases = [engine.table(CachedValue, owner=owner) for _ in range(2)]
+    engine.table(CachedValue).fill(value=[1, 2, 3])
+    for table in aliases:
+        np.testing.assert_array_equal(table.column.value.to_numpy(), [1, 2, 3])
+
+
+@pytest.mark.parametrize('method', ['table', 'string', 'utf8'])
+def test_engine_writes_after_close_fail_clearly_but_raw_array_writes_work(method):
+    engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)])
+    table = engine.table(CachedValue)
+    strings = table.column.name
+    array = table.column.value
+    engine.close()
+    with pytest.raises(RuntimeError, match='RecordEngine is closed'):
+        if method == 'table':
+            table.fill(value=[1, 2, 3])
+        elif method == 'string':
+            strings.fill(['a', 'b', 'c'])
+        else:
+            strings.fill_utf8(*fdb.pack_utf8_column(['a', 'b', 'c']))
+    array[:] = [7, 8, 9]
+    np.testing.assert_array_equal(table.column.value, [7, 8, 9])
diff --git a/tools/probe_standalone_memory.py b/tools/probe_standalone_memory.py
new file mode 100644
index 0000000..5b07a44
--- /dev/null
+++ b/tools/probe_standalone_memory.py
@@ -0,0 +1,109 @@
+"""Measure current process memory after complete standalone database cycles.
+
+Runs in its own process so allocator warm-up and the import baseline are explicit.
+"""
+import argparse
+import ctypes
+import json
+import os
+from pathlib import Path
+import sys
+
+import numpy as np
+import fastdb4py as fdb
+
+# ctypes libraries cache function objects that reference their library. Build
+# these once so the monitor itself does not add cycles with GC disabled.
+if sys.platform == 'darwin':
+    _proc_pidinfo = ctypes.CDLL('/usr/lib/libproc.dylib').proc_pidinfo
+    _proc_pidinfo.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_uint64, ctypes.c_void_p, ctypes.c_int]
+elif sys.platform == 'win32':
+    from ctypes import wintypes
+
+    class Counters(ctypes.Structure):
+        _fields_ = [('cb', wintypes.DWORD), ('faults', wintypes.DWORD)] + [
+            (name, ctypes.c_size_t) for name in (
+                'peak_working', 'working', 'peak_paged', 'paged',
+                'peak_nonpaged', 'nonpaged', 'pagefile', 'peak_pagefile', 'private',
+            )
+        ]
+
+    _kernel = ctypes.WinDLL('kernel32', use_last_error=True)
+    _kernel.GetCurrentProcess.restype = wintypes.HANDLE
+    _psapi = ctypes.WinDLL('psapi', use_last_error=True)
+    _psapi.GetProcessMemoryInfo.argtypes = [wintypes.HANDLE, ctypes.c_void_p, wintypes.DWORD]
+
+
+@fdb.feature
+class MemoryValue:
+    value: fdb.F32
+
+
+def resident_bytes():
+    if sys.platform == 'darwin':
+        info = ctypes.create_string_buffer(96)
+        count = _proc_pidinfo(os.getpid(), 4, 0, info, len(info))
+        if count != len(info):
+            raise OSError('proc_pidinfo(PROC_PIDTASKINFO) failed')
+        return ctypes.c_uint64.from_buffer(info, 8).value
+    if sys.platform == 'win32':
+        counters = Counters()
+        counters.cb = ctypes.sizeof(counters)
+        if not _psapi.GetProcessMemoryInfo(_kernel.GetCurrentProcess(), ctypes.byref(counters), counters.cb):
+            raise ctypes.WinError(ctypes.get_last_error())
+        return counters.private
+    return int(Path('/proc/self/statm').read_text().split()[1]) * os.sysconf('SC_PAGE_SIZE')
+
+
+def run(path, rows, cycles, mode):
+    import gc
+    gc.disable()
+    values = np.random.default_rng(0).random(rows, dtype=np.float32)
+    if mode == 'read':
+        engine = fdb.RecordEngine.truncate([fdb.Layout(MemoryValue, rows)])
+        engine.table(MemoryValue).fill(value=values)
+        engine.save(str(path))
+        engine.close()
+        del engine
+    results = []
+    for iteration in range(cycles):
+        steps = [('start', resident_bytes())]
+        if mode == 'write':
+            engine = fdb.RecordEngine.truncate([fdb.Layout(MemoryValue, rows)])
+            steps.append(('truncate', resident_bytes()))
+            table = engine.table(MemoryValue)
+            table.fill(value=values)
+            steps.append(('fill', resident_bytes()))
+            engine.save(str(path))
+            steps.append(('save', resident_bytes()))
+            del table
+        else:
+            engine = fdb.RecordEngine.load(str(path), from_file=True)
+            steps.append(('load', resident_bytes()))
+            copied = np.array(engine.table(MemoryValue).column.value, copy=True)
+            np.testing.assert_array_equal(copied, values)
+            steps.append(('column_copy', resident_bytes()))
+            del copied
+        owns = engine._origin.thisown
+        engine.close()
+        steps.append(('close', resident_bytes()))
+        del engine
+        steps.append(('released', resident_bytes()))
+        entry = {'iteration': iteration + 1, 'thisown': owns, 'bytes': dict(steps)}
+        entry['net_bytes'] = steps[-1][1] - steps[0][1]
+        results.append(entry)
+        print(json.dumps(entry), flush=True)
+    summary = {'mode': mode, 'rows': rows, 'cycles': cycles,
+               'retained_bytes': [r['bytes']['released'] for r in results]}
+    print(json.dumps(summary), flush=True)
+    return summary
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--path', type=Path, required=True)
+    parser.add_argument('--rows', type=int, default=1_000_000)
+    parser.add_argument('--cycles', type=int, default=8)
+    parser.add_argument('--mode', choices=('write', 'read'), required=True)
+    args = parser.parse_args()
+    run(args.path, args.rows, args.cycles, args.mode)