From 0e25532e340d1c58ea09dce9ce547933ee698804 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Wed, 21 Oct 2026 16:34:09 +0100
Subject: [PATCH 1/4] refactor(native): read compiled audits through a counted
 native accessor

`audit_reads` returns each audit attribute the planner reads from the
native audit view, built exactly as its Python attribute is, and
counts any other object in `compiled_view_fallbacks` before reading it
by attribute. Audit plan entries, audit attachment, relation scopes
and source deferral read audits through it. The fallback test now also
plans the waffle shop fixture, whose sources, seeds, functions and
audits are all read natively.
---
 .../_helpers/compiled_views/audit_reads.rs    | 232 ++++++++++++++++++
 .../bindings/_helpers/compiled_views/mod.rs   |   1 +
 .../_helpers/planner_entries/relations.rs     |   3 +-
 .../planner_strategy/audit_entries.rs         |  65 ++---
 .../_helpers/planner_strategy/audits.rs       |  13 +-
 .../planner_warehouse/source_deferral.rs      |  10 +-
 .../sqlbuild/cli/commands/main/_test_types.py |   1 +
 .../src/sqlbuild/cli/commands/main/helpers.py |  16 +-
 .../commands/main/test_compiled_view_reads.py |  13 +-
 9 files changed, 307 insertions(+), 47 deletions(-)
 create mode 100644 crates/sqlbuild-python/src/bindings/_helpers/compiled_views/audit_reads.rs

diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/audit_reads.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/audit_reads.rs
new file mode 100644
index 0000000000..e418fbc0cd
--- /dev/null
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/audit_reads.rs
@@ -0,0 +1,232 @@
+//! A compiled audit's attributes the planner reads, from its native view.
+
+use pyo3::IntoPyObjectExt;
+use pyo3::prelude::{Bound, PyAny, PyResult, Python};
+
+use crate::bindings::_helpers::compile_driver::stage_work::{COMPILED_VIEW_FALLBACKS, count_stage};
+use crate::bindings::_helpers::compiled_views::audit_views::NativeCompiledAudit;
+use crate::bindings::_helpers::records::fields::field as native_field;
+
+/// The native view behind an audit object; any other object is a Python-built record, counted.
+#[inline]
+fn native_audit<'a>(audit: &'a Bound<'_, PyAny>) -> PyResult<Option<&'a NativeCompiledAudit>> {
+    if let Ok(view) = audit.cast::<NativeCompiledAudit>() {
+        return Ok(Some(view.get()));
+    }
+    count_stage(audit.py(), COMPILED_VIEW_FALLBACKS, 1)?;
+    Ok(None)
+}
+
+/// The compiled audit's `always_run`, as its Python attribute reads it.
+pub(crate) fn always_run<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .always_run(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "always_run")
+}
+
+/// The compiled audit's `attached_column_name`, as its Python attribute reads it.
+pub(crate) fn attached_column_name<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .attached_column_name(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "attached_column_name")
+}
+
+/// The compiled audit's `attached_target_kind`, as its Python attribute reads it.
+pub(crate) fn attached_target_kind<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .attached_target_kind(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "attached_target_kind")
+}
+
+/// The compiled audit's `attached_target_name`, as its Python attribute reads it.
+pub(crate) fn attached_target_name<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .attached_target_name(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "attached_target_name")
+}
+
+/// The compiled audit's `definition_name`, as its Python attribute reads it.
+pub(crate) fn definition_name<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .definition_name(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "definition_name")
+}
+
+/// The compiled audit's `description`, as its Python attribute reads it.
+pub(crate) fn description<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .description(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "description")
+}
+
+/// The compiled audit's `evaluation_mode`, as its Python attribute reads it.
+pub(crate) fn evaluation_mode<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .evaluation_mode(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "evaluation_mode")
+}
+
+/// The compiled audit's `evidence_limit`, as its Python attribute reads it.
+pub(crate) fn evidence_limit<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .evidence_limit(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "evidence_limit")
+}
+
+/// The compiled audit's `evidence_sql`, as its Python attribute reads it.
+pub(crate) fn evidence_sql<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .evidence_sql(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "evidence_sql")
+}
+
+/// The compiled audit's `key`, as its Python attribute reads it.
+pub(crate) fn key<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "key")
+}
+
+/// The compiled audit's `measurement_contract`, as its Python attribute reads it.
+pub(crate) fn measurement_contract<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .measurement_contract(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "measurement_contract")
+}
+
+/// The compiled audit's `meta`, as its Python attribute reads it.
+pub(crate) fn meta<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view.meta(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "meta")
+}
+
+/// The compiled audit's `minimum_samples`, as its Python attribute reads it.
+pub(crate) fn minimum_samples<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .minimum_samples(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "minimum_samples")
+}
+
+/// The compiled audit's `name`, as its Python attribute reads it.
+pub(crate) fn name<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "name")
+}
+
+/// The compiled audit's `references`, as its Python attribute reads it.
+pub(crate) fn references<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .references(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "references")
+}
+
+/// The compiled audit's `run_scope`, as its Python attribute reads it.
+pub(crate) fn run_scope<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .run_scope(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "run_scope")
+}
+
+/// The compiled audit's `scope_deps`, as its Python attribute reads it.
+pub(crate) fn scope_deps<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .scope_deps(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "scope_deps")
+}
+
+/// The compiled audit's `severity`, as its Python attribute reads it.
+pub(crate) fn severity<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .severity(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "severity")
+}
+
+/// The compiled audit's `sql_body`, as its Python attribute reads it.
+pub(crate) fn sql_body<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .sql_body(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "sql_body")
+}
+
+/// The compiled audit's `thresholds`, as its Python attribute reads it.
+pub(crate) fn thresholds<'py>(audit: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = audit.py();
+    if let Some(view) = native_audit(audit)? {
+        return view
+            .thresholds(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(audit, "thresholds")
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
index ecdb130732..f2c6a9d5a7 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
@@ -1,4 +1,5 @@
 //! Python views of the compiled project: model, input, resource and project views and their records.
+pub(crate) mod audit_reads;
 pub(crate) mod audit_views;
 pub(crate) mod compiled_model_analysis;
 pub(crate) mod compiled_model_view;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/relations.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/relations.rs
index 1e4797cbac..dfe63632bf 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/relations.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/relations.rs
@@ -2,6 +2,7 @@
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
+use crate::bindings::_helpers::compiled_views::audit_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
 use crate::bindings::_helpers::planner_entries::models::{
     PlannerRelationsContext, RelationsRequest,
@@ -35,7 +36,7 @@ fn selected_audit_relation_keys<'py>(
     for audit in project_reads::audits(project)?.try_iter()? {
         let audit = audit?;
         if audit_is_selected(&audit, selected_keys)? {
-            for dependency in native_field(&audit, "scope_deps")?.try_iter()? {
+            for dependency in audit_reads::scope_deps(&audit)?.try_iter()? {
                 keys.add(dependency?)?;
             }
         }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audit_entries.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audit_entries.rs
index f4dcdde7be..cab7a9818a 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audit_entries.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audit_entries.rs
@@ -3,6 +3,7 @@
 use crate::bindings::_helpers::attachments::reference_markers::assert_no_unresolved_sql_markers;
 use crate::bindings::_helpers::auditing::render::render_audit_sql;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
+use crate::bindings::_helpers::compiled_views::audit_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
 use crate::bindings::_helpers::graph::views::project_lineage_views;
 use crate::bindings::_helpers::planner_phases::objects::object;
@@ -27,6 +28,9 @@ use crate::bindings::_helpers::records::fields::field as native_field;
 use crate::bindings::_helpers::records::fields::mapping_get;
 use crate::bindings::_helpers::records::ids::RecordId;
 
+/// One attribute read of a compiled audit.
+type AuditRead<'py> = fn(&Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>>;
+
 /// Render and validate one audit query with the supplied adapter callbacks.
 fn render_query<'py>(
     request: &AuditPlanning<'py>,
@@ -56,13 +60,11 @@ fn render_query<'py>(
 
 /// Resolve an audit enum, applying the default only to absent values.
 fn audit_enum<'py>(
-    audit: &Bound<'py, PyAny>,
-    attribute: &str,
+    value: Bound<'py, PyAny>,
     enum_name: &str,
     default: &str,
 ) -> PyResult<Bound<'py, PyAny>> {
-    let py = audit.py();
-    let value = native_field(audit, attribute)?;
+    let py = value.py();
     imported(py, AUDIT_TYPES_MODULE, enum_name)?.call1((if value.is_none() {
         PyString::new(py, default).into_any()
     } else {
@@ -98,13 +100,13 @@ fn plan_audit<'py>(
 ) -> PyResult<Bound<'py, PyAny>> {
     let py = audit.py();
     let audit = audit.clone();
-    let name: String = native_field(&audit, "name")?.extract()?;
+    let name: String = audit_reads::name(&audit)?.extract()?;
     let resolved = render_query(
         request,
-        native_field(&audit, "sql_body")?,
+        audit_reads::sql_body(&audit)?,
         &format!("audit '{name}' planned SQL"),
     )?;
-    let evidence = native_field(&audit, "evidence_sql")?;
+    let evidence = audit_reads::evidence_sql(&audit)?;
     let evidence_resolved = if evidence.is_none() {
         py.None().into_bound(py)
     } else {
@@ -124,32 +126,33 @@ fn plan_audit<'py>(
     } else {
         py.None().into_bound(py)
     };
-    let requested = audit_enum(&audit, "run_scope", "AuditRunScope", "final")?;
-    let severity = audit_enum(&audit, "severity", "AuditSeverity", "warn")?;
+    let requested = audit_enum(audit_reads::run_scope(&audit)?, "AuditRunScope", "final")?;
+    let severity = audit_enum(audit_reads::severity(&audit)?, "AuditSeverity", "warn")?;
     let effective = resolve_effective_run_scope(&requested, &materialization)?;
-    let mut target_kind = native_field(&audit, "attached_target_kind")?;
+    let mut target_kind = audit_reads::attached_target_kind(&audit)?;
     if target_kind.is_none() && (kind_text == MODEL_ATTACHMENT || kind_text == SOURCE_ATTACHMENT) {
         target_kind =
             imported(py, COMPILE_TYPES_MODULE, "AttachedAuditTargetKind")?.call1((kind_text,))?;
     }
     let values = PyDict::new(py);
-    for field in [
-        "key",
-        "name",
-        "definition_name",
-        "evaluation_mode",
-        "thresholds",
-        "minimum_samples",
-        "evidence_limit",
-        "scope_deps",
-        "attached_column_name",
-        "always_run",
-        "description",
-        "meta",
-    ] {
-        values.set_item(field, native_field(&audit, field)?)?;
+    let reads: [(&str, AuditRead<'py>); 12] = [
+        ("key", audit_reads::key),
+        ("name", audit_reads::name),
+        ("definition_name", audit_reads::definition_name),
+        ("evaluation_mode", audit_reads::evaluation_mode),
+        ("thresholds", audit_reads::thresholds),
+        ("minimum_samples", audit_reads::minimum_samples),
+        ("evidence_limit", audit_reads::evidence_limit),
+        ("scope_deps", audit_reads::scope_deps),
+        ("attached_column_name", audit_reads::attached_column_name),
+        ("always_run", audit_reads::always_run),
+        ("description", audit_reads::description),
+        ("meta", audit_reads::meta),
+    ];
+    for (field, read) in reads {
+        values.set_item(field, read(&audit)?)?;
     }
-    let measurement = native_field(&audit, "measurement_contract")?;
+    let measurement = audit_reads::measurement_contract(&audit)?;
     for field in ["value_column", "sample_count_column", "sample_unit"] {
         values.set_item(
             field,
@@ -162,7 +165,7 @@ fn plan_audit<'py>(
     }
     for (field, value) in [
         ("resolved_sql", resolved),
-        ("unresolved_sql", native_field(&audit, "sql_body")?),
+        ("unresolved_sql", audit_reads::sql_body(&audit)?),
         ("attachment_kind", kind),
         ("severity", severity),
         ("requested_run_scope", requested),
@@ -170,7 +173,7 @@ fn plan_audit<'py>(
         ("evidence_resolved_sql", evidence_resolved),
         (
             "evidence_unresolved_sql",
-            native_field(&audit, "evidence_sql")?,
+            audit_reads::evidence_sql(&audit)?,
         ),
         ("attached_target_kind", target_kind),
         ("attached_target_name", target),
@@ -197,8 +200,8 @@ pub(crate) fn audit_is_selected(
     selected_keys: &Bound<'_, PyAny>,
 ) -> PyResult<bool> {
     let py = audit.py();
-    let kind = native_field(audit, "attached_target_kind")?;
-    let name = native_field(audit, "attached_target_name")?;
+    let kind = audit_reads::attached_target_kind(audit)?;
+    let name = audit_reads::attached_target_name(audit)?;
     if !kind.is_none() && !name.is_none() {
         let target_value = imported(py, COMPILE_TYPES_MODULE, "AttachedAuditTargetKind")?
             .call1((kind,))?
@@ -211,7 +214,7 @@ pub(crate) fn audit_is_selected(
         )?;
         return selected_keys.contains(key);
     }
-    for dependency in native_field(audit, "scope_deps")?.try_iter()? {
+    for dependency in audit_reads::scope_deps(audit)?.try_iter()? {
         if selected_keys.contains(dependency?)? {
             return Ok(true);
         }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs
index c5aa2da64b..e2bafd04a4 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs
@@ -10,6 +10,7 @@ use sqlbuild_analysis::planner_strategy::main::singular_attachment::singular_att
 use sqlbuild_analysis::planner_strategy::models::{AuditAttachmentKind, AuditRunScope};
 
 use crate::bindings::_helpers::boundary::python_calls::imported;
+use crate::bindings::_helpers::compiled_views::audit_reads;
 use crate::bindings::_helpers::graph::algorithms::closure;
 use crate::bindings::_helpers::planner_strategy::values::{input_error, optional_string, text};
 use crate::bindings::_helpers::records::construct::{enum_from_value, new_record_with};
@@ -39,7 +40,7 @@ fn object_key<'py>(py: Python<'py>, kind: &str, name: &str) -> PyResult<Bound<'p
 /// The names of an audit's references of one kind, in order.
 fn reference_names(audit: &Bound<'_, PyAny>, kind: &str) -> PyResult<Vec<String>> {
     let mut names: Vec<String> = Vec::new();
-    for reference in native_field(audit, "references")?.try_iter()? {
+    for reference in audit_reads::references(audit)?.try_iter()? {
         let reference: Bound<'_, PyAny> = reference?;
         if text(&native_field(&reference, "ref_kind")?)? == kind {
             names.push(text(&native_field(&reference, "ref_name")?)?);
@@ -65,7 +66,7 @@ pub(crate) fn resolve_attachment_kind<'py>(
     downstream_deps: &Bound<'py, PyAny>,
 ) -> PyResult<Bound<'py, PyTuple>> {
     let py: Python<'py> = audit.py();
-    let attached_kind: Bound<'py, PyAny> = native_field(audit, "attached_target_kind")?;
+    let attached_kind: Bound<'py, PyAny> = audit_reads::attached_target_kind(audit)?;
     let (kind, name): (String, Option<String>) = if attached_kind.is_none() {
         let reaches = |from: &str, to: &str| -> PyResult<bool> {
             let start: Bound<'py, PyAny> = object_key(py, MODEL_RESOURCE_TYPE, from)?;
@@ -82,7 +83,7 @@ pub(crate) fn resolve_attachment_kind<'py>(
         )?;
         (attachment_value(kind).to_owned(), name)
     } else {
-        let name: Bound<'py, PyAny> = native_field(audit, "attached_target_name")?;
+        let name: Bound<'py, PyAny> = audit_reads::attached_target_name(audit)?;
         if name.is_none() {
             return Err(input_error(
                 py,
@@ -117,8 +118,8 @@ pub(crate) fn reads_outside_target_lineage(
     upstream_deps: &Bound<'_, PyAny>,
 ) -> PyResult<bool> {
     let py: Python<'_> = audit.py();
-    let kind: Bound<'_, PyAny> = native_field(audit, "attached_target_kind")?;
-    let name: Bound<'_, PyAny> = native_field(audit, "attached_target_name")?;
+    let kind: Bound<'_, PyAny> = audit_reads::attached_target_kind(audit)?;
+    let name: Bound<'_, PyAny> = audit_reads::attached_target_name(audit)?;
     if kind.is_none() || name.is_none() {
         return Ok(false);
     }
@@ -126,7 +127,7 @@ pub(crate) fn reads_outside_target_lineage(
         text(&imported(py, COMPILE_TYPES_MODULE, "AttachedAuditTargetKind")?.call1((kind,))?)?;
     let target: Bound<'_, PyAny> = object_key(py, &kind, &text(&name)?)?;
     let lineage: Bound<'_, PySet> = closure(vec![target], upstream_deps, (true, None))?;
-    for dependency in native_field(audit, "scope_deps")?.try_iter()? {
+    for dependency in audit_reads::scope_deps(audit)?.try_iter()? {
         if !lineage.contains(dependency?)? {
             return Ok(true);
         }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
index 502d7bf9e9..e76159ea9e 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
@@ -4,6 +4,7 @@ use sqlbuild_host::entries::models::UserCodeHost;
 
 use crate::bindings::_helpers::authored_values::project_vars::render_project_var_text;
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
+use crate::bindings::_helpers::compiled_views::audit_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
 use crate::bindings::_helpers::contracts::spec_resolution::resolve_target_config;
@@ -107,7 +108,7 @@ fn selected_managed_refs(
     }
     for audit in project_reads::audits(project)?.try_iter()? {
         let audit = audit?;
-        let deps = native_field(&audit, "scope_deps")?;
+        let deps = audit_reads::scope_deps(&audit)?;
         if deps.is_truthy()? {
             let mut any = false;
             for dep in deps.try_iter()? {
@@ -120,17 +121,14 @@ fn selected_managed_refs(
                 continue;
             }
         }
-        for reference in native_field(&audit, "references")?.try_iter()? {
+        for reference in audit_reads::references(&audit)?.try_iter()? {
             let reference = reference?;
             let name: String = native_field(&reference, "ref_name")?.extract()?;
             if native_field(&reference, "ref_kind")?.eq("source")? && managed.contains(&name) {
                 names.insert(name);
             }
         }
-        names.extend(sql_source_reads(
-            &native_field(&audit, "sql_body")?,
-            managed,
-        )?);
+        names.extend(sql_source_reads(&audit_reads::sql_body(&audit)?, managed)?);
     }
     Ok(names)
 }
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
index 91579426fe..f522f1c502 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
@@ -1283,6 +1283,7 @@ class CompiledViewFallbacksTestCase:
     """One command run over a natively compiled project."""
 
     description: str
+    fixture: str | None
     command: tuple[str, ...]
     expected_exit_code: int
     expected_fallbacks: int
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
index 81ad095b0b..2102c91504 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
@@ -3,6 +3,7 @@
 import _thread
 import json
 import os
+import shutil
 import sys
 import tempfile
 import threading
@@ -36,6 +37,8 @@
 
 STAGE_CAPTURE_SUFFIX: str = ".json"
 
+E2E_FIXTURES_DIR: Path = Path(__file__).resolve().parents[6] / "e2e" / "fixtures"
+
 
 def staging_directories(target_dir: Path) -> list[str]:
     """Return hidden entries left in target_dir, such as abandoned artifact staging."""
@@ -1011,9 +1014,14 @@ def typed_config_reads(*, project_dir: Path, extra_header: str) -> int:
     return native_module.compile_stage_work_info(reset=True).get("typed_config_reads", 0)
 
 
-def compiled_view_fallbacks(*, project_dir: Path, command: tuple[str, ...]) -> tuple[int, int]:
+def compiled_view_fallbacks(
+    *, project_dir: Path, fixture: str | None, command: tuple[str, ...]
+) -> tuple[int, int]:
     """The exit code of `command` over a native project and its reads that missed a native view."""
 
+    if fixture is not None:
+        shutil.copytree(E2E_FIXTURES_DIR / fixture, project_dir)
+        return _counted_fallbacks(project_dir=project_dir, command=command)
     project_dir.mkdir(parents=True)
     (project_dir / "sqlbuild_project.toml").write_text(
         'name = "orders"\nadapter = "duckdb"\ndefault_target = "dev"\n\n'
@@ -1038,6 +1046,12 @@ def compiled_view_fallbacks(*, project_dir: Path, command: tuple[str, ...]) -> t
         'SELECT order_id FROM __ref("daily_orders")\n',
         encoding="utf-8",
     )
+    return _counted_fallbacks(project_dir=project_dir, command=command)
+
+
+def _counted_fallbacks(*, project_dir: Path, command: tuple[str, ...]) -> tuple[int, int]:
+    """The exit code of `command` and how many of its reads missed a native view."""
+
     _ = native_module.compile_stage_work_info(reset=True)
     exit_code: int = main(["--project-dir", str(project_dir), *command])
     fallbacks: int = native_module.compile_stage_work_info(reset=True).get(
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py b/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py
index 8764dd4332..42139fd0c9 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py
@@ -1,4 +1,4 @@
-"""Compile and plan read a native project's models through its native views."""
+"""Compile and plan read a native project's resources through their native views."""
 
 from __future__ import annotations
 
@@ -17,12 +17,21 @@
     [
         CompiledViewFallbacksTestCase(
             description="compiling reads every model natively",
+            fixture=None,
             command=("compile", "--json", "--no-cache"),
             expected_exit_code=0,
             expected_fallbacks=0,
         ),
         CompiledViewFallbacksTestCase(
             description="planning reads every model natively",
+            fixture=None,
+            command=("plan", "--json", "--no-cache"),
+            expected_exit_code=0,
+            expected_fallbacks=0,
+        ),
+        CompiledViewFallbacksTestCase(
+            description="planning reads every source, seed, function and audit natively",
+            fixture="waffle_shop",
             command=("plan", "--json", "--no-cache"),
             expected_exit_code=0,
             expected_fallbacks=0,
@@ -36,7 +45,7 @@ def test_given_native_project_when_running_then_no_model_read_falls_back_to_pyth
     capsys: pytest.CaptureFixture[str],
 ) -> None:
     exit_code, fallbacks = compiled_view_fallbacks(
-        project_dir=tmp_path / "project", command=test_case.command
+        project_dir=tmp_path / "project", fixture=test_case.fixture, command=test_case.command
     )
     _ = capsys.readouterr()
 

From ae29b415d8f18ce10bf9670f335f2d388cecd2af Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Wed, 21 Oct 2026 18:15:59 +0100
Subject: [PATCH 2/4] refactor(native): read compiled functions through a
 counted native accessor

`function_reads` returns each function attribute the planner reads
from the native function view, built exactly as its Python attribute
is, and counts any other object in `compiled_view_fallbacks`. Function
plan entries, identity hashes, relation locations and clone plans read
functions through it.
---
 .../_helpers/compiled_views/function_reads.rs | 181 ++++++++++++++++++
 .../bindings/_helpers/compiled_views/mod.rs   |   1 +
 .../_helpers/planner_changes/detect.rs        |   7 +-
 .../_helpers/planner_entries/model_entries.rs |   3 +-
 .../_helpers/planner_entries/plan_output.rs   |  41 ++--
 .../_helpers/planner_identity/identities.rs   |   7 +-
 .../_helpers/planner_identity/metadata.rs     |  45 +++--
 .../planner_resolution/orchestration.rs       |   5 +-
 .../planner_warehouse/source_deferral.rs      |   5 +-
 9 files changed, 250 insertions(+), 45 deletions(-)
 create mode 100644 crates/sqlbuild-python/src/bindings/_helpers/compiled_views/function_reads.rs

diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/function_reads.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/function_reads.rs
new file mode 100644
index 0000000000..6792c24350
--- /dev/null
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/function_reads.rs
@@ -0,0 +1,181 @@
+//! A compiled function's attributes the planner reads, from its native view.
+
+use pyo3::IntoPyObjectExt;
+use pyo3::prelude::{Bound, PyAny, PyResult, Python};
+
+use crate::bindings::_helpers::compile_driver::stage_work::{COMPILED_VIEW_FALLBACKS, count_stage};
+use crate::bindings::_helpers::compiled_views::function_views::NativeCompiledFunction;
+use crate::bindings::_helpers::records::fields::field as native_field;
+
+/// The native view behind a function object; any other object is a Python-built record, counted.
+#[inline]
+fn native_function<'a>(
+    function: &'a Bound<'_, PyAny>,
+) -> PyResult<Option<&'a NativeCompiledFunction>> {
+    if let Ok(view) = function.cast::<NativeCompiledFunction>() {
+        return Ok(Some(view.get()));
+    }
+    count_stage(function.py(), COMPILED_VIEW_FALLBACKS, 1)?;
+    Ok(None)
+}
+
+/// The compiled function's `arguments`, as its Python attribute reads it.
+pub(crate) fn arguments<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .arguments(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "arguments")
+}
+
+/// The compiled function's `body_sql`, as its Python attribute reads it.
+pub(crate) fn body_sql<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .body_sql(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "body_sql")
+}
+
+/// The compiled function's `deps`, as its Python attribute reads it.
+pub(crate) fn deps<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view.deps(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "deps")
+}
+
+/// The compiled function's `destination`, as its Python attribute reads it.
+pub(crate) fn destination<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .destination(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "destination")
+}
+
+/// The compiled function's `entry_point`, as its Python attribute reads it.
+pub(crate) fn entry_point<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .entry_point(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "entry_point")
+}
+
+/// The compiled function's `fingerprint_destination`, as its Python attribute reads it.
+pub(crate) fn fingerprint_destination<'py>(
+    function: &Bound<'py, PyAny>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .fingerprint_destination(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "fingerprint_destination")
+}
+
+/// The compiled function's `key`, as its Python attribute reads it.
+pub(crate) fn key<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "key")
+}
+
+/// The compiled function's `language`, as its Python attribute reads it.
+pub(crate) fn language<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .language(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "language")
+}
+
+/// The compiled function's `name`, as its Python attribute reads it.
+pub(crate) fn name<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "name")
+}
+
+/// The compiled function's `packages`, as its Python attribute reads it.
+pub(crate) fn packages<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .packages(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "packages")
+}
+
+/// The compiled function's `relative_path`, as its Python attribute reads it.
+pub(crate) fn relative_path<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .relative_path(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "relative_path")
+}
+
+/// The compiled function's `return_columns`, as its Python attribute reads it.
+pub(crate) fn return_columns<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .return_columns(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "return_columns")
+}
+
+/// The compiled function's `returns`, as its Python attribute reads it.
+pub(crate) fn returns<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .returns(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "returns")
+}
+
+/// The compiled function's `runtime_version`, as its Python attribute reads it.
+pub(crate) fn runtime_version<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .runtime_version(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "runtime_version")
+}
+
+/// The compiled function's `source_file_path`, as its Python attribute reads it.
+pub(crate) fn source_file_path<'py>(function: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = function.py();
+    if let Some(view) = native_function(function)? {
+        return view
+            .source_file_path(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(function, "source_file_path")
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
index f2c6a9d5a7..dfe7ec5113 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
@@ -3,6 +3,7 @@ pub(crate) mod audit_reads;
 pub(crate) mod audit_views;
 pub(crate) mod compiled_model_analysis;
 pub(crate) mod compiled_model_view;
+pub(crate) mod function_reads;
 pub(crate) mod function_views;
 pub(crate) mod getters;
 pub(crate) mod inputs_view;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_changes/detect.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_changes/detect.rs
index 3aeb87e01c..3da15bed9c 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_changes/detect.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_changes/detect.rs
@@ -2,6 +2,7 @@
 
 use sqlbuild_host::entries::models::{ExecutorHost, UserCodeHost};
 
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
 use crate::bindings::_helpers::contracts::spec_resolution::matching_dynamic_families;
@@ -110,13 +111,13 @@ pub(crate) fn detect_changes<'py>(
     let dialect: Option<String> = native_field(snapshot, "column_dialect")?.extract()?;
     for function in project_reads::functions(project)?.try_iter()? {
         let function: Bound<'py, PyAny> = function?;
-        let name: Bound<'py, PyAny> = native_field(&function, "name")?;
+        let name: Bound<'py, PyAny> = function_reads::name(&function)?;
         let fingerprint_sql: String = compiled_function_fingerprint_sql(&function)?;
         let mut arguments: Vec<(&str, Bound<'py, PyAny>)> = vec![(
             "fingerprint_sql",
             PyString::new(py, &fingerprint_sql).into_any(),
         )];
-        if selected.contains(native_field(&function, "key")?)? {
+        if selected.contains(function_reads::key(&function)?)? {
             let reason: &str = if full_refresh {
                 "full_refresh"
             } else {
@@ -129,7 +130,7 @@ pub(crate) fn detect_changes<'py>(
                         &name.extract::<String>()?,
                         (
                             &fingerprint_sql,
-                            &native_field(&function, "language")?
+                            &function_reads::language(&function)?
                                 .str()?
                                 .extract::<String>()?,
                             dialect.as_deref(),
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/model_entries.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/model_entries.rs
index ddba156a81..81ee6bd601 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/model_entries.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/model_entries.rs
@@ -2,6 +2,7 @@
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
 use crate::bindings::_helpers::discovery::guidance::microbatch_guidance;
@@ -429,7 +430,7 @@ pub(crate) fn plan_entries<'py>(
     let functions_by_name = PyDict::new(py);
     for function in project_reads::functions(&project)?.try_iter()? {
         let function = function?;
-        functions_by_name.set_item(native_field(&function, "name")?, function)?;
+        functions_by_name.set_item(function_reads::name(&function)?, function)?;
     }
     let settings = native_field(&project, "settings")?;
     let models_by_name = native_field(&scope, "models_by_name")?;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs
index 8aebad4969..92b1b0745a 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs
@@ -2,6 +2,7 @@
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
 use crate::bindings::_helpers::planner_entries::model_entries::scope_overlaps;
 use crate::bindings::_helpers::planner_entries::models::PlanOutputInputs;
@@ -369,26 +370,34 @@ pub(super) fn function_entry<'py>(
             ],
         )?),
     )?;
-    let field = |name: &str| native_field(function, name);
     planner_model(
         py,
         RecordId::FunctionPlanEntry,
         &[
-            ("key", field("key")?),
-            ("name", field("name")?),
-            ("relative_path", field("relative_path")?),
-            ("destination", field("destination")?),
-            ("arguments", field("arguments")?),
-            ("returns", field("returns")?),
+            ("key", function_reads::key(function)?),
+            ("name", function_reads::name(function)?),
+            ("relative_path", function_reads::relative_path(function)?),
+            ("destination", function_reads::destination(function)?),
+            ("arguments", function_reads::arguments(function)?),
+            ("returns", function_reads::returns(function)?),
             ("body_sql", body_sql),
             ("fingerprint_query_sql", fingerprint_query_sql),
-            ("fingerprint_destination", field("fingerprint_destination")?),
-            ("return_columns", field("return_columns")?),
-            ("language", field("language")?),
-            ("source_file_path", field("source_file_path")?),
-            ("runtime_version", field("runtime_version")?),
-            ("entry_point", field("entry_point")?),
-            ("packages", field("packages")?),
+            (
+                "fingerprint_destination",
+                function_reads::fingerprint_destination(function)?,
+            ),
+            ("return_columns", function_reads::return_columns(function)?),
+            ("language", function_reads::language(function)?),
+            (
+                "source_file_path",
+                function_reads::source_file_path(function)?,
+            ),
+            (
+                "runtime_version",
+                function_reads::runtime_version(function)?,
+            ),
+            ("entry_point", function_reads::entry_point(function)?),
+            ("packages", function_reads::packages(function)?),
             ("previous_query_sql", previous_query_sql),
             ("reason", reason),
         ],
@@ -407,10 +416,10 @@ fn function_entries<'py>(
     let mut entries: Vec<Bound<'py, PyAny>> = Vec::new();
     for function in project_reads::functions(project)?.try_iter()? {
         let function = function?;
-        if !selected_keys.contains(native_field(&function, "key")?)? {
+        if !selected_keys.contains(function_reads::key(&function)?)? {
             continue;
         }
-        let name = native_field(&function, "name")?;
+        let name = function_reads::name(&function)?;
         let change = functions.get_item(&name)?;
         let previous_query_sql = if previous.contains(&name)? {
             native_field(&previous.get_item(&name)?, "definition")?
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs
index 0e56c7adad..a5f0095bf2 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs
@@ -16,6 +16,7 @@ use sqlbuild_analysis::planner_identity::models::{GraphKey, IdentityGraph, Ident
 use sqlbuild_host::entries::models::UserCodeHost;
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::planner_identity::hashing::{graph_key, model_query_hash};
@@ -103,14 +104,14 @@ pub(crate) fn direct_model_version_identities<'py>(
     let mut graph = IdentityGraph::default();
     for function in functions.try_iter()? {
         let function: Bound<'py, PyAny> = function?;
-        let name: String = native_field(&function, "name")?.extract()?;
+        let name: String = function_reads::name(&function)?.extract()?;
         let Some(function_hash) = function_hashes.get_item(&name)? else {
             continue;
         };
         graph.nodes.insert(
-            graph_key(&native_field(&function, "key")?)?,
+            graph_key(&function_reads::key(&function)?)?,
             IdentityNode {
-                upstream_keys: graph_keys(&native_field(&function, "deps")?)?,
+                upstream_keys: graph_keys(&function_reads::deps(&function)?)?,
                 local_hash: Some(function_hash.extract()?),
             },
         );
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs
index 3be7c2a31b..f89a059c5f 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs
@@ -15,6 +15,7 @@ use sqlbuild_analysis::planner_identity::main::stable_identity_hash::stable_iden
 use sqlbuild_analysis::planner_identity::models::FunctionFingerprintFields;
 
 use crate::bindings::_helpers::boundary::python_calls::{imported, keywords, within};
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::discovery::hook_entries::hook_entries;
 use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
@@ -282,14 +283,14 @@ pub(crate) fn function_local_hashes<'py>(
     let hashes: Bound<'py, PyDict> = PyDict::new(py);
     for function in functions.try_iter()? {
         let function: Bound<'py, PyAny> = function?;
-        let name: String = native_field(&function, "name")?.extract()?;
+        let name: String = function_reads::name(&function)?.extract()?;
         let language: Bound<'py, PyAny> =
-            native_field(&native_field(&function, "language")?, "value")?;
+            native_field(&function_reads::language(&function)?, "value")?;
         let body_hash: String = function_definition_hash(
             py,
             &name,
             (
-                &native_field(&function, "body_sql")?.extract::<String>()?,
+                &function_reads::body_sql(&function)?.extract::<String>()?,
                 &language.extract::<String>()?,
                 dialect,
             ),
@@ -297,18 +298,21 @@ pub(crate) fn function_local_hashes<'py>(
         let payload: Bound<'py, PyDict> = PyDict::new(py);
         payload.set_item(
             "arguments",
-            name_type_pairs(&native_field(&function, "arguments")?)?,
+            name_type_pairs(&function_reads::arguments(&function)?)?,
         )?;
-        payload.set_item("returns", native_field(&function, "returns")?)?;
+        payload.set_item("returns", function_reads::returns(&function)?)?;
         payload.set_item(
             "return_columns",
-            name_type_pairs(&native_field(&function, "return_columns")?)?,
+            name_type_pairs(&function_reads::return_columns(&function)?)?,
         )?;
         payload.set_item("body_sql", body_hash)?;
         payload.set_item("language", language)?;
-        for field in ["runtime_version", "entry_point", "packages"] {
-            payload.set_item(field, native_field(&function, field)?)?;
-        }
+        payload.set_item(
+            "runtime_version",
+            function_reads::runtime_version(&function)?,
+        )?;
+        payload.set_item("entry_point", function_reads::entry_point(&function)?)?;
+        payload.set_item("packages", function_reads::packages(&function)?)?;
         hashes.set_item(
             name,
             stable_identity_hash(&sorted_dumps(payload.as_any(), SPACED, true)?),
@@ -331,7 +335,12 @@ fn name_type_pairs<'py>(items: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyList
 
 /// `f"{value or ''}"` of an optional attribute.
 fn text_or_empty(owner: &Bound<'_, PyAny>, name: &str) -> PyResult<String> {
-    let value: Bound<'_, PyAny> = match native_field(owner, name) {
+    text_of(native_field(owner, name))
+}
+
+/// `f"{value or ''}"` of an attribute read, empty when the read failed.
+fn text_of(read: PyResult<Bound<'_, PyAny>>) -> PyResult<String> {
+    let value: Bound<'_, PyAny> = match read {
         Ok(value) => value,
         Err(_) => return Ok(String::new()),
     };
@@ -355,17 +364,17 @@ fn rendered_pairs(items: &Bound<'_, PyAny>) -> PyResult<Vec<(String, String)>> {
 /// Stable function definition text used for change fingerprints.
 pub(crate) fn compiled_function_fingerprint_sql(function: &Bound<'_, PyAny>) -> PyResult<String> {
     Ok(function_fingerprint_sql(&FunctionFingerprintFields {
-        language: native_field(function, "language")?.str()?.extract()?,
-        arguments: rendered_pairs(&native_field(function, "arguments")?)?,
-        returns: native_field(function, "returns")?.str()?.extract()?,
-        return_columns: rendered_pairs(&native_field(function, "return_columns")?)?,
-        runtime_version: text_or_empty(function, "runtime_version")?,
-        entry_point: text_or_empty(function, "entry_point")?,
-        packages: native_field(function, "packages")?
+        language: function_reads::language(function)?.str()?.extract()?,
+        arguments: rendered_pairs(&function_reads::arguments(function)?)?,
+        returns: function_reads::returns(function)?.str()?.extract()?,
+        return_columns: rendered_pairs(&function_reads::return_columns(function)?)?,
+        runtime_version: text_of(function_reads::runtime_version(function))?,
+        entry_point: text_of(function_reads::entry_point(function))?,
+        packages: function_reads::packages(function)?
             .try_iter()?
             .map(|package| package?.extract::<String>())
             .collect::<PyResult<_>>()?,
-        body_sql: native_field(function, "body_sql")?.extract()?,
+        body_sql: function_reads::body_sql(function)?.extract()?,
     }))
 }
 
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/orchestration.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/orchestration.rs
index 4f4a0953e9..8ed2dd220c 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/orchestration.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/orchestration.rs
@@ -4,6 +4,7 @@ use crate::bindings::_helpers::attachments::cursor_intrinsics::resolve_cursor_in
 use crate::bindings::_helpers::attachments::reference_markers::assert_no_unresolved_sql_markers;
 use crate::bindings::_helpers::boundary::python_calls::{KeywordArguments, call_with};
 use crate::bindings::_helpers::compile_attachment::cursor_roles::resolve_cursor_input_roles;
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::cursor::scalars::sentinel_to_token;
 use crate::bindings::_helpers::model_config::typed_config::{
@@ -316,9 +317,9 @@ pub(crate) fn resolve_function_sql<'py>(
         function_locations: args.value("function_locations")?,
     };
     let function = args.value("function")?;
-    let name: String = native_field(&function, "name")?.extract()?;
+    let name: String = function_reads::name(&function)?.extract()?;
     let sql: String = resolution.unbounded(
-        native_field(&function, "body_sql")?,
+        function_reads::body_sql(&function)?,
         &format!("Function '{name}'"),
     )?;
     Ok(PyString::new(py, &sql).into_any())
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
index e76159ea9e..fa898fac1d 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
@@ -5,6 +5,7 @@ use sqlbuild_host::entries::models::UserCodeHost;
 use crate::bindings::_helpers::authored_values::project_vars::render_project_var_text;
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::compiled_views::audit_reads;
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
 use crate::bindings::_helpers::contracts::spec_resolution::resolve_target_config;
@@ -99,9 +100,9 @@ fn selected_managed_refs(
     }
     for function in project_reads::functions(project)?.try_iter()? {
         let function = function?;
-        if selected.contains(native_field(&function, "key")?)? {
+        if selected.contains(function_reads::key(&function)?)? {
             names.extend(sql_source_reads(
-                &native_field(&function, "body_sql")?,
+                &function_reads::body_sql(&function)?,
                 managed,
             )?);
         }

From 8977e2c9f84ccd0d2307565e09b5a446ce855bf2 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Wed, 21 Oct 2026 19:10:53 +0100
Subject: [PATCH 3/4] refactor(native): read compiled seeds, sources and shared
 resource fields natively

`seed_reads` and `source_reads` return the seed and source attributes
the planner reads from their native views, and `resource_reads` reads
a name, key, dependency keys, destination, schema entry or references
from whichever native view a resource is. Each counts any other object
in `compiled_view_fallbacks`. Seed, source, scenario, clone, command,
identity, scope and warehouse planning read through them; seed plan
entries keep reading their own attributes.
---
 .../bindings/_helpers/compiled_views/mod.rs   |   3 +
 .../_helpers/compiled_views/resource_reads.rs | 143 ++++++++++++++++++
 .../_helpers/compiled_views/seed_reads.rs     |  78 ++++++++++
 .../_helpers/compiled_views/source_reads.rs   |  58 +++++++
 .../_helpers/planner_entries/clone_plans.rs   |  45 +++---
 .../_helpers/planner_entries/command_plans.rs |  26 ++--
 .../_helpers/planner_entries/plan_output.rs   |  13 +-
 .../_helpers/planner_identity/graph.rs        |   5 +-
 .../_helpers/planner_identity/identities.rs   |   3 +-
 .../_helpers/planner_identity/metadata.rs     |   9 +-
 .../_helpers/planner_phases/buildability.rs   |   5 +-
 .../_helpers/planner_phases/objects.rs        |   3 +-
 .../_helpers/planner_resolution/locations.rs  |   6 +-
 .../_helpers/planner_scenario/entries.rs      |   9 +-
 .../_helpers/planner_scenario/graph.rs        |   6 +-
 .../_helpers/planner_scenario/relations.rs    |   8 +-
 .../bindings/_helpers/planner_scope/scope.rs  |   5 +-
 .../_helpers/planner_scope/sources.rs         |   5 +-
 .../bindings/_helpers/planner_scope/values.rs |   3 +-
 .../_helpers/planner_strategy/audits.rs       |   2 +-
 .../planner_strategy/fixture_context.rs       |  16 +-
 .../_helpers/planner_strategy/lineage.rs      |   3 +-
 .../planner_strategy/model_entries.rs         |   3 +-
 .../_helpers/planner_warehouse/cursors.rs     |  10 +-
 .../_helpers/planner_warehouse/snapshot.rs    |  24 +--
 .../planner_warehouse/source_columns.rs       |   3 +-
 .../planner_warehouse/source_deferral.rs      |  13 +-
 .../planner_warehouse/source_freshness.rs     |   5 +-
 .../planner_warehouse/source_tables.rs        |  13 +-
 .../_helpers/planner_warehouse/watermarks.rs  |   9 +-
 30 files changed, 423 insertions(+), 111 deletions(-)
 create mode 100644 crates/sqlbuild-python/src/bindings/_helpers/compiled_views/resource_reads.rs
 create mode 100644 crates/sqlbuild-python/src/bindings/_helpers/compiled_views/seed_reads.rs
 create mode 100644 crates/sqlbuild-python/src/bindings/_helpers/compiled_views/source_reads.rs

diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
index dfe7ec5113..8bff8fd13e 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/mod.rs
@@ -15,5 +15,8 @@ pub(crate) mod project_view;
 pub(crate) mod records;
 pub(crate) mod registration;
 pub(crate) mod relation_keys;
+pub(crate) mod resource_reads;
+pub(crate) mod seed_reads;
 pub(crate) mod seed_views;
+pub(crate) mod source_reads;
 pub(crate) mod source_views;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/resource_reads.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/resource_reads.rs
new file mode 100644
index 0000000000..fef6bdadc4
--- /dev/null
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/resource_reads.rs
@@ -0,0 +1,143 @@
+//! A compiled resource's attributes read alike for models, functions, seeds and sources.
+
+use pyo3::IntoPyObjectExt;
+use pyo3::prelude::{Bound, PyAny, PyResult, Python};
+
+use crate::bindings::_helpers::compile_driver::stage_work::{COMPILED_VIEW_FALLBACKS, count_stage};
+use crate::bindings::_helpers::compiled_views::compiled_model_view::NativeCompiledModel;
+use crate::bindings::_helpers::compiled_views::function_views::NativeCompiledFunction;
+use crate::bindings::_helpers::compiled_views::seed_views::NativeCompiledSeed;
+use crate::bindings::_helpers::compiled_views::source_views::NativeCompiledSource;
+use crate::bindings::_helpers::records::fields::field as native_field;
+
+/// A resource's `deps` from whichever native view it is; any other object is read by attribute, counted.
+pub(crate) fn deps<'py>(resource: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = resource.py();
+    if let Ok(view) = resource.cast::<NativeCompiledModel>() {
+        let view: &NativeCompiledModel = view.get();
+        return view.deps(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledFunction>() {
+        let view: &NativeCompiledFunction = view.get();
+        return view.deps(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSeed>() {
+        let view: &NativeCompiledSeed = view.get();
+        return view.deps(py).into_bound_py_any(py);
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSource>() {
+        let view: &NativeCompiledSource = view.get();
+        return view.deps(py).into_bound_py_any(py);
+    }
+    count_stage(py, COMPILED_VIEW_FALLBACKS, 1)?;
+    native_field(resource, "deps")
+}
+
+/// A resource's `destination` from whichever native view it is; any other object is read by attribute, counted.
+pub(crate) fn destination<'py>(resource: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = resource.py();
+    if let Ok(view) = resource.cast::<NativeCompiledModel>() {
+        let view: &NativeCompiledModel = view.get();
+        return view
+            .destination(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledFunction>() {
+        let view: &NativeCompiledFunction = view.get();
+        return view
+            .destination(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSeed>() {
+        let view: &NativeCompiledSeed = view.get();
+        return view
+            .destination(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    count_stage(py, COMPILED_VIEW_FALLBACKS, 1)?;
+    native_field(resource, "destination")
+}
+
+/// A resource's `key` from whichever native view it is; any other object is read by attribute, counted.
+pub(crate) fn key<'py>(resource: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = resource.py();
+    if let Ok(view) = resource.cast::<NativeCompiledModel>() {
+        let view: &NativeCompiledModel = view.get();
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledFunction>() {
+        let view: &NativeCompiledFunction = view.get();
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSeed>() {
+        let view: &NativeCompiledSeed = view.get();
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSource>() {
+        let view: &NativeCompiledSource = view.get();
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    count_stage(py, COMPILED_VIEW_FALLBACKS, 1)?;
+    native_field(resource, "key")
+}
+
+/// A resource's `name` from whichever native view it is; any other object is read by attribute, counted.
+pub(crate) fn name<'py>(resource: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = resource.py();
+    if let Ok(view) = resource.cast::<NativeCompiledModel>() {
+        let view: &NativeCompiledModel = view.get();
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledFunction>() {
+        let view: &NativeCompiledFunction = view.get();
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSeed>() {
+        let view: &NativeCompiledSeed = view.get();
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSource>() {
+        let view: &NativeCompiledSource = view.get();
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    count_stage(py, COMPILED_VIEW_FALLBACKS, 1)?;
+    native_field(resource, "name")
+}
+
+/// A resource's `references` from whichever native view it is; any other object is read by attribute, counted.
+pub(crate) fn references<'py>(resource: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = resource.py();
+    if let Ok(view) = resource.cast::<NativeCompiledModel>() {
+        let view: &NativeCompiledModel = view.get();
+        return view
+            .references(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledFunction>() {
+        let view: &NativeCompiledFunction = view.get();
+        return view
+            .references(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    count_stage(py, COMPILED_VIEW_FALLBACKS, 1)?;
+    native_field(resource, "references")
+}
+
+/// A resource's `schema_entry` from whichever native view it is; any other object is read by attribute, counted.
+pub(crate) fn schema_entry<'py>(resource: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = resource.py();
+    if let Ok(view) = resource.cast::<NativeCompiledModel>() {
+        let view: &NativeCompiledModel = view.get();
+        return view
+            .schema_entry(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    if let Ok(view) = resource.cast::<NativeCompiledSeed>() {
+        let view: &NativeCompiledSeed = view.get();
+        return view
+            .schema_entry(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    count_stage(py, COMPILED_VIEW_FALLBACKS, 1)?;
+    native_field(resource, "schema_entry")
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/seed_reads.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/seed_reads.rs
new file mode 100644
index 0000000000..ae6bb086aa
--- /dev/null
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/seed_reads.rs
@@ -0,0 +1,78 @@
+//! A compiled seed's attributes the planner reads, from its native view.
+
+use pyo3::IntoPyObjectExt;
+use pyo3::prelude::{Bound, PyAny, PyResult, Python};
+
+use crate::bindings::_helpers::compile_driver::stage_work::{COMPILED_VIEW_FALLBACKS, count_stage};
+use crate::bindings::_helpers::compiled_views::seed_views::NativeCompiledSeed;
+use crate::bindings::_helpers::records::fields::field as native_field;
+
+/// The native view behind a seed object; any other object is a Python-built record, counted.
+#[inline]
+fn native_seed<'a>(seed: &'a Bound<'_, PyAny>) -> PyResult<Option<&'a NativeCompiledSeed>> {
+    if let Ok(view) = seed.cast::<NativeCompiledSeed>() {
+        return Ok(Some(view.get()));
+    }
+    count_stage(seed.py(), COMPILED_VIEW_FALLBACKS, 1)?;
+    Ok(None)
+}
+
+/// The compiled seed's `destination`, as its Python attribute reads it.
+pub(crate) fn destination<'py>(seed: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = seed.py();
+    if let Some(view) = native_seed(seed)? {
+        return view
+            .destination(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(seed, "destination")
+}
+
+/// The compiled seed's `external`, as its Python attribute reads it.
+pub(crate) fn external<'py>(seed: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = seed.py();
+    if let Some(view) = native_seed(seed)? {
+        return view.is_external().into_bound_py_any(py);
+    }
+    native_field(seed, "external")
+}
+
+/// The compiled seed's `key`, as its Python attribute reads it.
+pub(crate) fn key<'py>(seed: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = seed.py();
+    if let Some(view) = native_seed(seed)? {
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(seed, "key")
+}
+
+/// The compiled seed's `name`, as its Python attribute reads it.
+pub(crate) fn name<'py>(seed: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = seed.py();
+    if let Some(view) = native_seed(seed)? {
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(seed, "name")
+}
+
+/// The compiled seed's `schema_entry`, as its Python attribute reads it.
+pub(crate) fn schema_entry<'py>(seed: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = seed.py();
+    if let Some(view) = native_seed(seed)? {
+        return view
+            .schema_entry(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(seed, "schema_entry")
+}
+
+/// The compiled seed's `seed_file`, as its Python attribute reads it.
+pub(crate) fn seed_file<'py>(seed: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = seed.py();
+    if let Some(view) = native_seed(seed)? {
+        return view
+            .seed_file(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(seed, "seed_file")
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/source_reads.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/source_reads.rs
new file mode 100644
index 0000000000..00f81044fb
--- /dev/null
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_views/source_reads.rs
@@ -0,0 +1,58 @@
+//! A compiled source's attributes the planner reads, from its native view.
+
+use pyo3::IntoPyObjectExt;
+use pyo3::prelude::{Bound, PyAny, PyResult, Python};
+
+use crate::bindings::_helpers::compile_driver::stage_work::{COMPILED_VIEW_FALLBACKS, count_stage};
+use crate::bindings::_helpers::compiled_views::source_views::NativeCompiledSource;
+use crate::bindings::_helpers::records::fields::field as native_field;
+
+/// The native view behind a source object; any other object is a Python-built record, counted.
+#[inline]
+fn native_source<'a>(source: &'a Bound<'_, PyAny>) -> PyResult<Option<&'a NativeCompiledSource>> {
+    if let Ok(view) = source.cast::<NativeCompiledSource>() {
+        return Ok(Some(view.get()));
+    }
+    count_stage(source.py(), COMPILED_VIEW_FALLBACKS, 1)?;
+    Ok(None)
+}
+
+/// The compiled source's `key`, as its Python attribute reads it.
+pub(crate) fn key<'py>(source: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = source.py();
+    if let Some(view) = native_source(source)? {
+        return view.key(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(source, "key")
+}
+
+/// The compiled source's `name`, as its Python attribute reads it.
+pub(crate) fn name<'py>(source: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = source.py();
+    if let Some(view) = native_source(source)? {
+        return view.name(py).and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(source, "name")
+}
+
+/// The compiled source's `source_entry`, as its Python attribute reads it.
+pub(crate) fn source_entry<'py>(source: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = source.py();
+    if let Some(view) = native_source(source)? {
+        return view
+            .source_entry(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(source, "source_entry")
+}
+
+/// The compiled source's `source_file`, as its Python attribute reads it.
+pub(crate) fn source_file<'py>(source: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    let py: Python<'py> = source.py();
+    if let Some(view) = native_source(source)? {
+        return view
+            .source_file(py)
+            .and_then(|value| value.into_bound_py_any(py));
+    }
+    native_field(source, "source_file")
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs
index 5150cc3d73..d7daa964c3 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs
@@ -2,8 +2,11 @@
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::graph::project_graph::NativeProjectGraph;
 use crate::bindings::_helpers::model_config::typed_config::{config_node_of, model_values};
 use crate::bindings::_helpers::planner_entries::models::ModelPlanContext;
@@ -137,7 +140,7 @@ fn resolved_source_entries<'py>(
     }
     for function in project_reads::functions(project)?.try_iter()? {
         let function = function?;
-        if selected.contains(native_field(&function, "key")?)? {
+        if selected.contains(function_reads::key(&function)?)? {
             readers.push(function);
         }
     }
@@ -152,8 +155,8 @@ fn resolved_source_entries<'py>(
     let mut entries: Vec<Bound<'py, PyAny>> = Vec::new();
     for source in native_field(source_project, "sources")?.try_iter()? {
         let source = source?;
-        if read.contains(native_field(&source, "name")?)? {
-            entries.push(native_field(&source, "source_entry")?);
+        if read.contains(source_reads::name(&source)?)? {
+            entries.push(source_reads::source_entry(&source)?);
         }
     }
     tuple_of(py, entries)
@@ -175,8 +178,8 @@ fn clone_resolution<'py>(
     for source in native_field(source_project, "sources")?.try_iter()? {
         let source = source?;
         source_map.set_item(
-            native_field(&source, "name")?,
-            native_field(&source, "source_entry")?,
+            source_reads::name(&source)?,
+            source_reads::source_entry(&source)?,
         )?;
     }
     let source_warehouse_columns = object(
@@ -247,7 +250,7 @@ fn clone_model_entries<'py>(
             let functions = PyDict::new(py);
             for function in project_reads::functions(project)?.try_iter()? {
                 let function = function?;
-                functions.set_item(native_field(&function, "name")?, function)?;
+                functions.set_item(function_reads::name(&function)?, function)?;
             }
             let context = Bound::new(
                 py,
@@ -308,8 +311,8 @@ fn clone_source_entries<'py>(
     let entries = PyDict::new(py);
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
-        let entry = native_field(&source, "source_entry")?;
-        if !selected.contains(native_field(&source, "key")?)?
+        let entry = source_reads::source_entry(&source)?;
+        if !selected.contains(source_reads::key(&source)?)?
             || !native_field(&entry, "managed")?.is_truthy()?
             || native_field(&entry, "loader")?.is_none()
             || !native_field(&entry, "expression")?.is_none()
@@ -341,13 +344,13 @@ fn clone_source_entries<'py>(
             }),
         )?;
         entries.set_item(
-            native_field(&source, "key")?,
+            source_reads::key(&source)?,
             planner_model(
                 py,
                 RecordId::CloneSourcePlanEntry,
                 &[
-                    ("key", native_field(&source, "key")?),
-                    ("name", native_field(&source, "name")?),
+                    ("key", source_reads::key(&source)?),
+                    ("name", source_reads::name(&source)?),
                     ("destination", destination),
                 ],
             )?,
@@ -366,7 +369,7 @@ fn clone_function_entries<'py>(
     let entries = PyDict::new(py);
     for function in project_reads::functions(project)?.try_iter()? {
         let function = function?;
-        if !selected.contains(native_field(&function, "key")?)? {
+        if !selected.contains(function_reads::key(&function)?)? {
             continue;
         }
         let body_sql = resolve_function_sql(
@@ -416,7 +419,7 @@ fn clone_function_entries<'py>(
                 ("packages", field("packages")?),
             ],
         )?;
-        entries.set_item(native_field(&function, "key")?, entry)?;
+        entries.set_item(function_reads::key(&function)?, entry)?;
     }
     in_order(&entries, &native_field(plan, "execution_order")?)
 }
@@ -427,17 +430,17 @@ fn seed_entry<'py>(seed: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
         py,
         RecordId::SeedPlanEntry,
         &[
-            ("key", native_field(seed, "key")?),
-            ("name", native_field(seed, "name")?),
-            ("destination", native_field(seed, "destination")?),
+            ("key", seed_reads::key(seed)?),
+            ("name", seed_reads::name(seed)?),
+            ("destination", seed_reads::destination(seed)?),
             (
                 "file_path",
-                native_field(&native_field(seed, "seed_file")?, "file_path")?,
+                native_field(&seed_reads::seed_file(seed)?, "file_path")?,
             ),
             ("columns", PyTuple::empty(py).into_any()),
             (
                 "csv_settings",
-                native_field(&native_field(seed, "schema_entry")?, "csv_settings")?,
+                native_field(&seed_reads::schema_entry(seed)?, "csv_settings")?,
             ),
         ],
     )
@@ -494,10 +497,10 @@ fn origin_source_entries<'py>(
     let mut order: Vec<Bound<'py, PyAny>> = Vec::new();
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
-        if selected_names.contains(native_field(&source, "name")?)? {
-            keys.push(native_field(&source, "key")?);
+        if selected_names.contains(source_reads::name(&source)?)? {
+            keys.push(source_reads::key(&source)?);
         }
-        order.push(native_field(&source, "key")?);
+        order.push(source_reads::key(&source)?);
     }
     clone_source_entries(
         project,
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/command_plans.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/command_plans.rs
index 759b7fa9a1..bea6d82ebe 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/command_plans.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/command_plans.rs
@@ -2,8 +2,10 @@
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
+use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
 use crate::bindings::_helpers::fingerprints::storage_io::read_latest_fingerprints;
 use crate::bindings::_helpers::planner_entries::models::RelationsRequest;
 use crate::bindings::_helpers::planner_entries::plan_output::{
@@ -203,7 +205,7 @@ fn called_function_keys<'py>(
     let functions: Bound<'py, PyDict> = PyDict::new(py);
     for function in project_reads::functions(project)?.try_iter()? {
         let function = function?;
-        functions.set_item(native_field(&function, "key")?, function)?;
+        functions.set_item(function_reads::key(&function)?, function)?;
     }
     let called: Bound<'py, PySet> = PySet::empty(py)?;
     let mut pending: Vec<Bound<'py, PyAny>> = direct;
@@ -215,7 +217,7 @@ fn called_function_keys<'py>(
         let Some(function) = functions.get_item(&key)? else {
             continue;
         };
-        for dependency in native_field(&function, "deps")?.try_iter()? {
+        for dependency in function_reads::deps(&function)?.try_iter()? {
             let dependency = dependency?;
             if functions.contains(&dependency)? {
                 pending.push(dependency);
@@ -274,7 +276,7 @@ pub(crate) fn build_test_command_plan<'py>(
     let mut functions: Vec<Bound<'py, PyAny>> = Vec::new();
     for function in project_reads::functions(&project)?.try_iter()? {
         let function = function?;
-        if !function_keys.contains(native_field(&function, "key")?)? {
+        if !function_keys.contains(function_reads::key(&function)?)? {
             continue;
         }
         let fingerprint_sql = object(py, build_compiled_function_fingerprint_sql(&function)?)?;
@@ -334,12 +336,12 @@ pub(crate) fn build_seed_command_plan<'py>(
     let mut entries: Vec<Bound<'py, PyAny>> = Vec::new();
     for seed in project_reads::seeds(project)?.try_iter()? {
         let seed = seed?;
-        if !selected_keys.contains(native_field(&seed, "key")?)? {
+        if !selected_keys.contains(seed_reads::key(&seed)?)? {
             continue;
         }
         let identity = object(py, build_seed_identity(&seed)?)?;
         let (version_hash, metadata_json) = (identity.get_item(0)?, identity.get_item(1)?);
-        let previous = mapping_get(&existing, native_field(&seed, "name")?)?;
+        let previous = mapping_get(&existing, seed_reads::name(&seed)?)?;
         let reason = if previous.is_none() {
             "FIRST_RUN"
         } else if native_field(&previous, "version_hash")?.ne(&version_hash)? {
@@ -351,17 +353,17 @@ pub(crate) fn build_seed_command_plan<'py>(
             py,
             RecordId::SeedPlanEntry,
             &[
-                ("key", native_field(&seed, "key")?),
-                ("name", native_field(&seed, "name")?),
-                ("destination", native_field(&seed, "destination")?),
+                ("key", seed_reads::key(&seed)?),
+                ("name", seed_reads::name(&seed)?),
+                ("destination", seed_reads::destination(&seed)?),
                 (
                     "file_path",
-                    native_field(&native_field(&seed, "seed_file")?, "file_path")?,
+                    native_field(&seed_reads::seed_file(&seed)?, "file_path")?,
                 ),
                 ("columns", object(py, extract_seed_columns(&seed)?)?),
                 (
                     "csv_settings",
-                    native_field(&native_field(&seed, "schema_entry")?, "csv_settings")?,
+                    native_field(&seed_reads::schema_entry(&seed)?, "csv_settings")?,
                 ),
                 ("fingerprint_definition", metadata_json.clone()),
                 ("fingerprint_version_hash", version_hash),
@@ -409,7 +411,7 @@ pub(crate) fn read_selected_seed_fingerprints<'py>(
         BTreeMap::new();
     for seed in seeds.try_iter()? {
         let seed = seed?;
-        let destination = native_field(&seed, "destination")?;
+        let destination = seed_reads::destination(&seed)?;
         let schema = native_field(&destination, "schema")?;
         if schema.is_none() {
             continue;
@@ -420,7 +422,7 @@ pub(crate) fn read_selected_seed_fingerprints<'py>(
             .entry((database.clone().unwrap_or_default(), schema.clone()))
             .or_insert_with(|| ((database, schema), BTreeSet::new()))
             .1
-            .insert(native_field(&seed, "name")?.extract()?);
+            .insert(seed_reads::name(&seed)?.extract()?);
     }
     let fingerprints = PyDict::new(py);
     let mut relations_by_target: Vec<Bound<'py, PyAny>> = Vec::new();
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs
index 92b1b0745a..de60957dc4 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs
@@ -4,6 +4,7 @@ use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
 use crate::bindings::_helpers::planner_entries::model_entries::scope_overlaps;
 use crate::bindings::_helpers::planner_entries::models::PlanOutputInputs;
 use crate::bindings::_helpers::planner_entries::values::{
@@ -108,25 +109,25 @@ fn seed_entries<'py>(
     let mut entries: Vec<Bound<'py, PyAny>> = Vec::new();
     for seed in project_reads::seeds(project)?.try_iter()? {
         let seed = seed?;
-        if !selected_keys.contains(native_field(&seed, "key")?)? {
+        if !selected_keys.contains(seed_reads::key(&seed)?)? {
             continue;
         }
-        let name = native_field(&seed, "name")?;
+        let name = seed_reads::name(&seed)?;
         entries.push(planner_model(
             py,
             RecordId::SeedPlanEntry,
             &[
-                ("key", native_field(&seed, "key")?),
+                ("key", seed_reads::key(&seed)?),
                 ("name", name.clone()),
-                ("destination", native_field(&seed, "destination")?),
+                ("destination", seed_reads::destination(&seed)?),
                 (
                     "file_path",
-                    native_field(&native_field(&seed, "seed_file")?, "file_path")?,
+                    native_field(&seed_reads::seed_file(&seed)?, "file_path")?,
                 ),
                 ("columns", object(py, extract_seed_columns(&seed)?)?),
                 (
                     "csv_settings",
-                    native_field(&native_field(&seed, "schema_entry")?, "csv_settings")?,
+                    native_field(&seed_reads::schema_entry(&seed)?, "csv_settings")?,
                 ),
                 (
                     "fingerprint_definition",
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/graph.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/graph.rs
index 40ac2bf9cf..f5c228d432 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/graph.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/graph.rs
@@ -21,6 +21,7 @@ use sqlbuild_analysis::planner_identity::models::{AvailableUpstreams, GraphKey,
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
 use crate::bindings::_helpers::graph::algorithms::{closure, inverted};
 use crate::bindings::_helpers::graph::keys::KeyCache;
 use crate::bindings::_helpers::planner_identity::hashing::graph_key;
@@ -100,10 +101,10 @@ pub(crate) fn build_execution_upstream_deps<'py>(
         for resource in collection.try_iter()? {
             let resource: Bound<'py, PyAny> = resource?;
             upstream.set_item(
-                native_field(&resource, "key")?,
+                resource_reads::key(&resource)?,
                 PyList::new(
                     py,
-                    native_field(&resource, "deps")?
+                    resource_reads::deps(&resource)?
                         .try_iter()?
                         .collect::<PyResult<Vec<_>>>()?,
                 )?,
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs
index a5f0095bf2..2d6a09c871 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/identities.rs
@@ -18,6 +18,7 @@ use sqlbuild_host::entries::models::UserCodeHost;
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::planner_identity::hashing::{graph_key, model_query_hash};
 use crate::bindings::_helpers::planner_identity::metadata::{
@@ -122,7 +123,7 @@ pub(crate) fn direct_model_version_identities<'py>(
         for seed in seeds.try_iter()? {
             let seed: Bound<'py, PyAny> = seed?;
             let (seed_hash, metadata_json): (String, String) = seed_identity(&seed)?;
-            let name: Bound<'py, PyAny> = native_field(&seed, "name")?;
+            let name: Bound<'py, PyAny> = seed_reads::name(&seed)?;
             seed_hashes.set_item(&name, &seed_hash)?;
             seed_metadata.set_item(&name, metadata_json)?;
         }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs
index f89a059c5f..f4d132463e 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs
@@ -17,6 +17,7 @@ use sqlbuild_analysis::planner_identity::models::FunctionFingerprintFields;
 use crate::bindings::_helpers::boundary::python_calls::{imported, keywords, within};
 use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
 use crate::bindings::_helpers::discovery::hook_entries::hook_entries;
 use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
 use crate::bindings::_helpers::planner_identity::hashing::function_definition_hash;
@@ -381,22 +382,22 @@ pub(crate) fn compiled_function_fingerprint_sql(function: &Bound<'_, PyAny>) ->
 /// `(identity_hash, metadata_json)` of one compiled seed from its content and load-affecting config.
 pub(crate) fn seed_identity(seed: &Bound<'_, PyAny>) -> PyResult<(String, String)> {
     let py: Python<'_> = seed.py();
-    let schema_entry: Bound<'_, PyAny> = native_field(seed, "schema_entry")?;
+    let schema_entry: Bound<'_, PyAny> = seed_reads::schema_entry(seed)?;
     let csv_settings: Bound<'_, PyAny> = native_field(&schema_entry, "csv_settings")?;
     let columns: Bound<'_, PyList> = PyList::empty(py);
     for column in native_field(&schema_entry, "columns")?.try_iter()? {
         columns.append(column_signature(&column?)?)?;
     }
-    let rows: Bound<'_, PyAny> = if native_field(seed, "external")?.is_truthy()? {
+    let rows: Bound<'_, PyAny> = if seed_reads::external(seed)?.is_truthy()? {
         PyList::empty(py).into_any()
     } else {
         seed_rows(
-            &native_field(&native_field(seed, "seed_file")?, "file_path")?,
+            &native_field(&seed_reads::seed_file(seed)?, "file_path")?,
             &csv_settings,
         )?
     };
     let payload: Bound<'_, PyDict> = PyDict::new(py);
-    payload.set_item("name", native_field(seed, "name")?)?;
+    payload.set_item("name", seed_reads::name(seed)?)?;
     payload.set_item(
         "csv_settings",
         imported(py, DATACLASSES_MODULE, "asdict")?.call1((&csv_settings,))?,
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/buildability.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/buildability.rs
index fb58fa50d4..4cbc9d2036 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/buildability.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/buildability.rs
@@ -2,6 +2,7 @@
 
 use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
 use crate::bindings::_helpers::planner_identity::graph::build_execution_edge_origins;
 use crate::bindings::_helpers::planner_identity::graph::check_buildability;
 use crate::bindings::_helpers::planner_identity::graph::missing_upstream_message;
@@ -31,8 +32,8 @@ pub(crate) fn check_selected_scope_buildability<'py>(
     let mut external_seeds: Vec<_> = Vec::new();
     for seed in project_reads::seeds(&project)?.try_iter()? {
         let seed = seed?;
-        if native_field(&seed, "external")?.is_truthy()? {
-            external_seeds.push(native_field(&seed, "key")?);
+        if seed_reads::external(&seed)?.is_truthy()? {
+            external_seeds.push(seed_reads::key(&seed)?);
         }
     }
     let missing = object(
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/objects.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/objects.rs
index acac698320..78e3c4b3f2 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/objects.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/objects.rs
@@ -4,6 +4,7 @@ use sqlbuild_host::entries::models::HostId;
 
 use crate::bindings::_helpers::boundary::python_calls::{call_with, keywords};
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
 use crate::bindings::_helpers::model_config::typed_config::{config_text, model_values};
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::records::construct::error_class;
@@ -153,7 +154,7 @@ pub(crate) fn by_name<'py>(
     let nodes = pyo3::types::PyDict::new(project.py());
     for node in project_reads::collection(project, field)?.try_iter()? {
         let node = node?;
-        pyo3::types::PyDictMethods::set_item(&nodes, native_field(&node, "name")?, node)?;
+        pyo3::types::PyDictMethods::set_item(&nodes, resource_reads::name(&node)?, node)?;
     }
     Ok(nodes)
 }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/locations.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/locations.rs
index 57b0a41787..6610d71335 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/locations.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/locations.rs
@@ -1,6 +1,6 @@
 //! Location maps retaining the original compiled Python destination objects.
 
-use crate::bindings::_helpers::records::fields::field as native_field;
+use crate::bindings::_helpers::compiled_views::resource_reads;
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyModule, PyModuleMethods, PyResult};
 use pyo3::types::{PyDict, PyDictMethods};
 use pyo3::{pyfunction, wrap_pyfunction};
@@ -10,8 +10,8 @@ fn locations<'py>(items: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyDict>> {
     for item in items.try_iter()? {
         let item = item?;
         result.set_item(
-            native_field(&item, "name")?,
-            native_field(&item, "destination")?,
+            resource_reads::name(&item)?,
+            resource_reads::destination(&item)?,
         )?;
     }
     Ok(result)
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/entries.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/entries.rs
index 561b359710..fc7a409b45 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/entries.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/entries.rs
@@ -3,6 +3,7 @@
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
 use crate::bindings::_helpers::planner_identity::metadata::build_compiled_function_fingerprint_sql;
 use crate::bindings::_helpers::planner_resolution::orchestration::resolve_function_sql;
 use crate::bindings::_helpers::planner_scenario::constants::SCENARIO_PLAN_UNKNOWN_SEED;
@@ -52,17 +53,17 @@ pub(super) fn seed_entries<'py>(
             py,
             RecordId::SeedPlanEntry,
             &[
-                ("key", native_field(&seed, "key")?),
-                ("name", native_field(&seed, "name")?),
+                ("key", seed_reads::key(&seed)?),
+                ("name", seed_reads::name(&seed)?),
                 ("destination", destination),
                 (
                     "file_path",
-                    native_field(&native_field(&seed, "seed_file")?, "file_path")?,
+                    native_field(&seed_reads::seed_file(&seed)?, "file_path")?,
                 ),
                 ("columns", extract_seed_columns(&seed)?.into_any()),
                 (
                     "csv_settings",
-                    native_field(&native_field(&seed, "schema_entry")?, "csv_settings")?,
+                    native_field(&seed_reads::schema_entry(&seed)?, "csv_settings")?,
                 ),
             ],
         )?);
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/graph.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/graph.rs
index 29039b9dd8..15a5ff5048 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/graph.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/graph.rs
@@ -2,6 +2,8 @@
 
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::graph::keys::{KeyCache, native_key};
 use crate::bindings::_helpers::planner_scenario::constants::SCENARIO_PLAN_GRAPH_VALIDATION;
 use crate::bindings::_helpers::planner_scenario::values::string;
@@ -47,12 +49,12 @@ pub(crate) fn plan_scenario_graph<'py>(
     for source in project_reads::sources(project)?.try_iter()? {
         input
             .project_sources
-            .insert(native_field(&source?, "name")?.extract()?);
+            .insert(source_reads::name(&source?)?.extract()?);
     }
     for seed in project_reads::seeds(project)?.try_iter()? {
         input
             .project_seeds
-            .insert(native_field(&seed?, "name")?.extract()?);
+            .insert(seed_reads::name(&seed?)?.extract()?);
     }
     let output = graph(&input);
     input.targets.sort();
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/relations.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/relations.rs
index 42d92879bf..5e55421365 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/relations.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_scenario/relations.rs
@@ -1,6 +1,8 @@
 //! Build scenario relation locations through adapter qualification callbacks.
 
 use crate::bindings::_helpers::boundary::python_calls::{KeywordArguments, call_with};
+use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::planner_scenario::constants::{REF, SOURCE};
 use crate::bindings::_helpers::planner_scenario::constants::{
     SCENARIO_PLAN_INTERNAL, SCENARIO_PLAN_MISSING_RELATION_TARGET,
@@ -63,11 +65,11 @@ pub(crate) fn build_scenario_relation_plan<'py>(
         )?;
     }
     let sources = PyDict::new(py);
-    for source in native_field(&args.project, "sources")?.try_iter()? {
+    for source in project_reads::sources(&args.project)?.try_iter()? {
         let source = source?;
         sources.set_item(
-            native_field(&source, "name")?,
-            native_field(&source, "source_entry")?,
+            source_reads::name(&source)?,
+            source_reads::source_entry(&source)?,
         )?;
     }
     let models = PyDict::new(py);
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/scope.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/scope.rs
index 3ea5d202b5..5bf88db5fb 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/scope.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/scope.rs
@@ -3,6 +3,7 @@
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::graph::keys::KeyCache;
 use crate::bindings::_helpers::graph::project_graph::NativeProjectGraph;
 use crate::bindings::_helpers::graph::views::lineage_graph_names;
@@ -37,8 +38,8 @@ pub(crate) fn managed_source_upstream_keys<'py>(
     let mut managed: HashSet<String> = HashSet::new();
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
-        if !native_field(&native_field(&source, "source_entry")?, "loader")?.is_none() {
-            managed.insert(native_field(&source, "name")?.extract()?);
+        if !native_field(&source_reads::source_entry(&source)?, "loader")?.is_none() {
+            managed.insert(source_reads::name(&source)?.extract()?);
         }
     }
     let selected = managed_sources(&keys(selected_keys)?, &edges(upstream_deps)?, &managed);
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/sources.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/sources.rs
index 3b789fa489..a1376c22a4 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/sources.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/sources.rs
@@ -3,6 +3,7 @@
 use sqlbuild_host::entries::models::ExecutorHost;
 
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::graph::algorithms::closure;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::planner_scope::loaders::intermediate_map;
@@ -34,11 +35,11 @@ pub(crate) fn build_source_load_map<'py>(
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
         if let Some(keys) = &relevant
-            && !keys.contains(native_field(&source, "key")?)?
+            && !keys.contains(source_reads::key(&source)?)?
         {
             continue;
         }
-        let entry = native_field(&source, "source_entry")?;
+        let entry = source_reads::source_entry(&source)?;
         result.set_item(native_field(&entry, "name")?, entry)?;
     }
     for upstream in [false, true] {
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/values.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/values.rs
index 8102e74d73..bc79b589ee 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/values.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_scope/values.rs
@@ -1,6 +1,7 @@
 //! Python function identity and compiled graph conversion.
 
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::graph::keys::native_key;
 use crate::bindings::_helpers::records::fields::field as native_field;
 use crate::bindings::_helpers::records::fields::mapping_items;
@@ -69,7 +70,7 @@ pub(crate) fn loader_graph(loaders: &Bound<'_, PyAny>) -> PyResult<LoaderGraph>
 pub(crate) fn project_graph(project: &Bound<'_, PyAny>) -> PyResult<LoaderGraph> {
     let mut graph = loader_graph(&native_field(project, "loader_functions")?)?;
     for source in project_reads::sources(project)?.try_iter()? {
-        let entry = native_field(&source?, "source_entry")?;
+        let entry = source_reads::source_entry(&source?)?;
         graph.sources.push((
             native_field(&entry, "name")?.extract()?,
             native_field(&entry, "loader")?.extract()?,
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs
index e2bafd04a4..78c6ff857a 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/audits.rs
@@ -90,7 +90,7 @@ pub(crate) fn resolve_attachment_kind<'py>(
                 ErrorId::PlannerInput,
                 &format!(
                     "audit '{}': attached audit is missing an attached target name",
-                    text(&native_field(audit, "name")?)?
+                    text(&audit_reads::name(audit)?)?
                 ),
                 None,
             ));
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/fixture_context.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/fixture_context.rs
index dc3f0b3088..e641f17398 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/fixture_context.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/fixture_context.rs
@@ -14,6 +14,9 @@ use sqlbuild_core::text::main::python_casefold::python_casefold;
 use crate::bindings::_helpers::analysis_session::fixture_inferences::FixtureColumnInferences;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::planner_strategy::constants::{
     MODEL_REFERENCE_KIND, SEED_REFERENCE_KIND, SOURCE_REFERENCE_KIND,
 };
@@ -68,10 +71,7 @@ pub(crate) fn planning_context(
             model_fixture_relation(&model)?,
         );
     }
-    for (entry, kind, source) in [
-        ("source_entry", SOURCE_REFERENCE_KIND, true),
-        ("schema_entry", SEED_REFERENCE_KIND, false),
-    ] {
+    for (kind, source) in [(SOURCE_REFERENCE_KIND, true), (SEED_REFERENCE_KIND, false)] {
         let resources: Bound<'_, PyAny> = if source {
             project_reads::sources(project)?
         } else {
@@ -79,7 +79,11 @@ pub(crate) fn planning_context(
         };
         for resource in resources.try_iter()? {
             let resource = resource?;
-            let entry = native_field(&resource, entry)?;
+            let entry: Bound<'_, PyAny> = if source {
+                source_reads::source_entry(&resource)?
+            } else {
+                seed_reads::schema_entry(&resource)?
+            };
             let columns: Vec<FixtureColumn> = columns(&native_field(&entry, "columns")?, false)?;
             let authoritative = if source {
                 native_field(&entry, "contract")?.eq("enforced")?
@@ -87,7 +91,7 @@ pub(crate) fn planning_context(
                 !columns.is_empty()
             };
             relations.insert(
-                (kind.to_owned(), native_field(&resource, "name")?.extract()?),
+                (kind.to_owned(), resource_reads::name(&resource)?.extract()?),
                 FixtureRelation {
                     columns,
                     authoritative,
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/lineage.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/lineage.rs
index cf58cc1fd9..c7193aa6ac 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/lineage.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/lineage.rs
@@ -14,6 +14,7 @@ use sqlbuild_analysis::planner_strategy::types::ReferenceGraph;
 
 use crate::bindings::_helpers::compile_attachment::cursor_roles::resolve_cursor_input_roles;
 use crate::bindings::_helpers::compiled_views::model_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
 use crate::bindings::_helpers::model_config::typed_config::{config_text_is, model_values};
 use crate::bindings::_helpers::planner_strategy::microbatches::{
     CAP_ACTIONS, microbatch_limit_config,
@@ -44,7 +45,7 @@ fn mapped_references<'py>(
 }
 
 fn references<'py>(owner: &Bound<'py, PyAny>) -> PyResult<Vec<Bound<'py, PyAny>>> {
-    native_field(owner, "references")?.try_iter()?.collect()
+    resource_reads::references(owner)?.try_iter()?.collect()
 }
 
 /// Whether a model's `incremental_mode` is microbatch.
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/model_entries.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/model_entries.rs
index 009cd9e0b6..7eb33c6dc5 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/model_entries.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_strategy/model_entries.rs
@@ -18,6 +18,7 @@ use sqlbuild_model_config::model_validation::constants::PHYSICAL_STORAGE_KEYS;
 
 use crate::bindings::_helpers::boundary::python_calls::call_with;
 use crate::bindings::_helpers::compiled_views::model_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
 use crate::bindings::_helpers::cursor::scalars::sentinel_to_token;
 use crate::bindings::_helpers::model_config::typed_config::{
     ConfigValues, config_cursor_bound, config_flag, config_integer_object, config_text_object,
@@ -102,7 +103,7 @@ fn schema_columns<'py>(
     typed_only: bool,
 ) -> PyResult<Bound<'py, PyTuple>> {
     let py: Python<'py> = model.py();
-    let entry: Bound<'py, PyAny> = native_field(model, "schema_entry")?;
+    let entry: Bound<'py, PyAny> = resource_reads::schema_entry(model)?;
     if entry.is_none() {
         return Ok(PyTuple::empty(py));
     }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/cursors.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/cursors.rs
index ff17cb0f39..0a82820a15 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/cursors.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/cursors.rs
@@ -4,6 +4,8 @@ use crate::bindings::_helpers::boundary::python_calls::{call_with, imported, key
 use crate::bindings::_helpers::compile_attachment::cursor_roles::resolve_cursor_input_roles;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::cursor::intervals::partition;
 use crate::bindings::_helpers::cursor::scalars;
 use crate::bindings::_helpers::cursor::scalars::compare_values;
@@ -197,8 +199,8 @@ fn runtime_producer_names(
     let mut loaded: HashSet<String> = HashSet::new();
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
-        if !native_field(&native_field(&source, "source_entry")?, "loader")?.is_none() {
-            loaded.insert(native_field(&source, "name")?.extract()?);
+        if !native_field(&source_reads::source_entry(&source)?, "loader")?.is_none() {
+            loaded.insert(source_reads::name(&source)?.extract()?);
         }
     }
     for key in selected.try_iter()? {
@@ -224,7 +226,7 @@ fn upstream_relation<'py>(
     let name_text: String = name.extract()?;
     let is_selected = selected.is_some_and(|names| names.contains(&name_text));
     let qualified = |node: Bound<'py, PyAny>| {
-        native_field(&native_field(&node, "destination")?, "qualified_name")
+        native_field(&resource_reads::destination(&node)?, "qualified_name")
     };
     if (kind.eq("ref")? || kind.eq("seed")?)
         && !deferred.is_none()
@@ -250,7 +252,7 @@ fn upstream_relation<'py>(
         return object(
             py,
             render_source_relation(
-                &native_field(&source, "source_entry")?,
+                &source_reads::source_entry(&source)?,
                 Some(adapter).filter(|value| !value.is_none()),
             )?,
         );
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/snapshot.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/snapshot.rs
index d208b15f31..8d1d95fb7a 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/snapshot.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/snapshot.rs
@@ -5,6 +5,9 @@ use crate::bindings::_helpers::boundary::python_calls::call_with;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::fingerprints::latest_rows::select_latest_fingerprints;
 use crate::bindings::_helpers::fingerprints::storage_io::read_latest_fingerprints;
 use crate::bindings::_helpers::graph::algorithms::transitive_closure_many;
@@ -90,7 +93,7 @@ pub(crate) fn destination_nodes<'py>(
 }
 
 fn selected_node(selected: &Bound<'_, PyAny>, node: &Bound<'_, PyAny>) -> PyResult<bool> {
-    Ok(selected.is_none() || selected.contains(native_field(node, "key")?)?)
+    Ok(selected.is_none() || selected.contains(resource_reads::key(node)?)?)
 }
 
 fn relevant_state_keys<'py>(
@@ -113,7 +116,7 @@ fn resolve_database<'py>(
         if !selected_node(selected, &node)? {
             continue;
         }
-        let database = native_field(&native_field(&node, "destination")?, "database")?;
+        let database = native_field(&resource_reads::destination(&node)?, "database")?;
         if !database.is_none() {
             return Ok(database);
         }
@@ -130,7 +133,7 @@ fn target_schemas(
         if !selected_node(selected, &node)? {
             continue;
         }
-        let schema = native_field(&native_field(&node, "destination")?, "schema")?;
+        let schema = native_field(&resource_reads::destination(&node)?, "schema")?;
         if !schema.is_none() {
             schemas.insert(schema.extract()?);
         }
@@ -196,7 +199,7 @@ fn metadata_name_filter(
         let seeds = by_name(project, "seeds")?;
         let sources = PyDict::new(py);
         for source in project_reads::sources(project)?.try_iter()? {
-            let entry = native_field(&source?, "source_entry")?;
+            let entry = source_reads::source_entry(&source?)?;
             sources.set_item(native_field(&entry, "name")?, entry)?;
         }
         for key in selected.try_iter()? {
@@ -209,8 +212,7 @@ fn metadata_name_filter(
                     &selected_names,
                 )?);
             } else if let Some(seed) = seeds.get_item(&name)? {
-                names
-                    .insert(native_field(&native_field(&seed, "destination")?, "name")?.extract()?);
+                names.insert(native_field(&seed_reads::destination(&seed)?, "name")?.extract()?);
             } else if let Some(entry) = sources.get_item(&name)? {
                 names.insert(source_table_name(&entry)?);
             }
@@ -279,8 +281,8 @@ fn prefetch_schemas<'py>(
     }
     for source in project_reads::sources(&request.project)?.try_iter()? {
         let source = source?;
-        let entry = native_field(&source, "source_entry")?;
-        if (!relevant.is_none() && !relevant.contains(native_field(&source, "key")?)?)
+        let entry = source_reads::source_entry(&source)?;
+        if (!relevant.is_none() && !relevant.contains(source_reads::key(&source)?)?)
             || !native_field(&entry, "expression")?.is_none()
             || native_field(&entry, "schema")?.is_none()
         {
@@ -380,8 +382,8 @@ fn gather_relations<'py>(
     let logical = PyDict::new(py);
     for node in destination_nodes(&request.project)? {
         logical.set_item(
-            location_identity(&native_field(&node, "destination")?)?,
-            native_field(&node, "name")?,
+            location_identity(&resource_reads::destination(&node)?)?,
+            resource_reads::name(&node)?,
         )?;
     }
     let result = PyDict::new(py);
@@ -431,7 +433,7 @@ fn gather_columns<'py>(
     let py = request.project.py();
     let mut node_names: HashSet<String> = HashSet::new();
     for node in destination_nodes(&request.project)? {
-        node_names.insert(native_field(&node, "name")?.extract()?);
+        node_names.insert(resource_reads::name(&node)?.extract()?);
     }
     let mut needed: Vec<(Bound<'py, PyAny>, Bound<'py, PyAny>)> = Vec::new();
     for (logical, relation) in relations.iter() {
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_columns.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_columns.rs
index 6f0bcca6b9..81f04ff66b 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_columns.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_columns.rs
@@ -2,6 +2,7 @@
 
 use crate::bindings::_helpers::boundary::python_calls::call_with;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::planner_phases::objects::{frozenset, text, tuple};
 use crate::bindings::_helpers::planner_warehouse::constants::{
@@ -22,7 +23,7 @@ fn entries_of<'py>(
         Some(entries) => entries.try_iter()?.collect(),
         None => project_reads::sources(project)?
             .try_iter()?
-            .map(|source| native_field(&source?, "source_entry"))
+            .map(|source| source_reads::source_entry(&source?))
             .collect(),
     }
 }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
index fa898fac1d..603769e400 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_deferral.rs
@@ -8,6 +8,7 @@ use crate::bindings::_helpers::compiled_views::audit_reads;
 use crate::bindings::_helpers::compiled_views::function_reads;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::contracts::spec_resolution::resolve_target_config;
 use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
 use crate::bindings::_helpers::native_host::host;
@@ -31,7 +32,7 @@ use std::collections::BTreeSet;
 fn managed_names(project: &Bound<'_, PyAny>) -> PyResult<BTreeSet<String>> {
     let mut names: BTreeSet<_> = BTreeSet::new();
     for source in project_reads::sources(project)?.try_iter()? {
-        let entry = native_field(&source?, "source_entry")?;
+        let entry = source_reads::source_entry(&source?)?;
         if !native_field(&entry, "loader")?.is_none() {
             names.insert(native_field(&entry, "name")?.extract()?);
         }
@@ -172,11 +173,9 @@ fn source_target_name<'py>(
 }
 
 fn raw_source_entry<'py>(source: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
-    let declared = native_field(source, "source_entry")?;
+    let declared = source_reads::source_entry(source)?;
     let name = native_field(&declared, "name")?;
-    for entry in
-        native_field(&native_field(source, "source_file")?, "source_entries")?.try_iter()?
-    {
+    for entry in native_field(&source_reads::source_file(source)?, "source_entries")?.try_iter()? {
         let entry = entry?;
         if native_field(&entry, "name")?.eq(&name)? {
             return Ok(entry);
@@ -312,7 +311,7 @@ pub(crate) fn build_source_read_map<'py>(
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
         let name: String =
-            native_field(&native_field(&source, "source_entry")?, "name")?.extract()?;
+            native_field(&source_reads::source_entry(&source)?, "name")?.extract()?;
         if !selected.contains(&name) {
             continue;
         }
@@ -345,7 +344,7 @@ pub(crate) fn with_declared_source_reads<'py>(
     }
     let result = source_map.copy()?;
     for source in project_reads::sources(project)?.try_iter()? {
-        let entry = native_field(&source?, "source_entry")?;
+        let entry = source_reads::source_entry(&source?)?;
         let name: String = native_field(&entry, "name")?.extract()?;
         if missing.contains(&name) {
             result.set_item(name, entry)?;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_freshness.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_freshness.rs
index 9d51a7f735..cafffee3cc 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_freshness.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_freshness.rs
@@ -3,6 +3,7 @@
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::{call_with, imported, within};
 use crate::bindings::_helpers::compiled_views::model_reads;
+use crate::bindings::_helpers::compiled_views::resource_reads;
 use crate::bindings::_helpers::freshness::planning::{
     build_direct_source_freshness_planning_result, build_direct_source_freshness_propagation_result,
 };
@@ -69,7 +70,7 @@ pub(crate) fn source_freshness_unknown_warnings<'py>(
 
 fn state_database<'py>(project: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
     for node in destination_nodes(project)? {
-        let database = native_field(&native_field(&node, "destination")?, "database")?;
+        let database = native_field(&resource_reads::destination(&node)?, "database")?;
         if !database.is_none() {
             return Ok(database);
         }
@@ -83,7 +84,7 @@ fn state_schemas(
 ) -> PyResult<BTreeSet<String>> {
     let mut schemas: BTreeSet<_> = BTreeSet::new();
     for node in destination_nodes(project)? {
-        let schema = native_field(&native_field(&node, "destination")?, "schema")?;
+        let schema = native_field(&resource_reads::destination(&node)?, "schema")?;
         if !schema.is_none() {
             schemas.insert(schema.extract()?);
         }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_tables.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_tables.rs
index d41bb47bc3..1970820ed6 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_tables.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/source_tables.rs
@@ -3,6 +3,7 @@
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::{call_with, imported, keywords};
 use crate::bindings::_helpers::compiled_views::project_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::discovery::locations::yaml_entry_line;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::planner_phases::objects::{planner_error, tuple};
@@ -71,13 +72,13 @@ fn read_relation<'py>(
 }
 
 fn declaration_location(source: &Bound<'_, PyAny>) -> PyResult<String> {
-    let file = native_field(source, "source_file")?;
+    let file = source_reads::source_file(source)?;
     let path: String = native_field(&file, "relative_path")?
         .call_method0("as_posix")?
         .extract()?;
     let line = yaml_entry_line(
         &native_field(&file, "contents")?.extract::<String>()?,
-        &native_field(source, "name")?.extract::<String>()?,
+        &source_reads::name(source)?.extract::<String>()?,
     );
     Ok(match line {
         None => path,
@@ -99,11 +100,11 @@ pub(crate) fn check_selected_source_tables_exist<'py>(
     let mut unlisted: Vec<(Bound<'py, PyAny>, Bound<'py, PyAny>, Vec<String>)> = Vec::new();
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
-        let name: String = native_field(&source, "name")?.extract()?;
+        let name: String = source_reads::name(&source)?.extract()?;
         let Some(source_readers) = readers.get(&name).filter(|readers| !readers.is_empty()) else {
             continue;
         };
-        let declared = native_field(&source, "source_entry")?;
+        let declared = source_reads::source_entry(&source)?;
         let entry = mapping_get_or(&read_map, &name, &declared)?;
         if !external_table(&declared)? || !external_table(&entry)? {
             continue;
@@ -153,7 +154,7 @@ pub(crate) fn check_selected_source_tables_exist<'py>(
         }
         let role = native_field(&probe, "role")?;
         missing.push(MissingSourceTable {
-            source_name: native_field(&source, "name")?.extract()?,
+            source_name: source_reads::name(&source)?.extract()?,
             relation: String::new(),
             read_by,
             unreadable: status.eq("missing_or_unreadable")?,
@@ -176,7 +177,7 @@ pub(crate) fn check_selected_source_tables_exist<'py>(
     let sources = PyDict::new(py);
     for source in project_reads::sources(project)?.try_iter()? {
         let source = source?;
-        sources.set_item(native_field(&source, "name")?, source)?;
+        sources.set_item(source_reads::name(&source)?, source)?;
     }
     for item in missing.iter_mut().take(10) {
         if let Some(source) = sources.get_item(&item.source_name)? {
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/watermarks.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/watermarks.rs
index 899d92932a..3d9c3a3441 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/watermarks.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_warehouse/watermarks.rs
@@ -1,6 +1,8 @@
 //! Validate reliable watermark column contracts before warehouse inspection.
 
 use crate::bindings::_helpers::compiled_views::model_reads;
+use crate::bindings::_helpers::compiled_views::seed_reads;
+use crate::bindings::_helpers::compiled_views::source_reads;
 use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
 use crate::bindings::_helpers::planner_phases::objects::object;
 use crate::bindings::_helpers::planner_phases::objects::{config_str, planner_error, text};
@@ -109,7 +111,7 @@ pub(crate) fn validate_watermark_column<'py>(
         let Some(source) = sources.get_item(&ref_name)? else {
             return Ok(());
         };
-        let entry = native_field(&source, "source_entry")?;
+        let entry = source_reads::source_entry(&source)?;
         if !native_field(&entry, "contract")?.eq("enforced")? {
             return Ok(());
         }
@@ -118,10 +120,7 @@ pub(crate) fn validate_watermark_column<'py>(
         let Some(seed) = seeds.get_item(&ref_name)? else {
             return Ok(());
         };
-        declared_columns(&native_field(
-            &native_field(&seed, "schema_entry")?,
-            "columns",
-        )?)?
+        declared_columns(&native_field(&seed_reads::schema_entry(&seed)?, "columns")?)?
     } else {
         return Ok(());
     };

From ca586f8e79380622251185264f1e4f4589971556 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Wed, 21 Oct 2026 21:35:06 +0100
Subject: [PATCH 4/4] test: give each compiled-view fallback case its own
 project writer

---
 .../sqlbuild/cli/commands/main/_test_types.py |  2 +-
 .../src/sqlbuild/cli/commands/main/helpers.py | 35 ++++++++++---------
 .../commands/main/test_compiled_view_reads.py | 16 ++++++---
 3 files changed, 31 insertions(+), 22 deletions(-)

diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
index f522f1c502..8eaebda0bf 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
@@ -1283,7 +1283,7 @@ class CompiledViewFallbacksTestCase:
     """One command run over a natively compiled project."""
 
     description: str
-    fixture: str | None
+    write_project: Callable[[Path], None]
     command: tuple[str, ...]
     expected_exit_code: int
     expected_fallbacks: int
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
index 2102c91504..d5b9897c3c 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
@@ -1015,13 +1015,28 @@ def typed_config_reads(*, project_dir: Path, extra_header: str) -> int:
 
 
 def compiled_view_fallbacks(
-    *, project_dir: Path, fixture: str | None, command: tuple[str, ...]
+    *, project_dir: Path, write_project: Callable[[Path], None], command: tuple[str, ...]
 ) -> tuple[int, int]:
     """The exit code of `command` over a native project and its reads that missed a native view."""
 
-    if fixture is not None:
-        shutil.copytree(E2E_FIXTURES_DIR / fixture, project_dir)
-        return _counted_fallbacks(project_dir=project_dir, command=command)
+    write_project(project_dir)
+    _ = native_module.compile_stage_work_info(reset=True)
+    exit_code: int = main(["--project-dir", str(project_dir), *command])
+    fallbacks: int = native_module.compile_stage_work_info(reset=True).get(
+        "compiled_view_fallbacks", 0
+    )
+    return exit_code, fallbacks
+
+
+def copy_waffle_shop(project_dir: Path) -> None:
+    """Copy the waffle shop fixture, with sources, seeds, functions and audits, to `project_dir`."""
+
+    _ = shutil.copytree(E2E_FIXTURES_DIR / "waffle_shop", project_dir)
+
+
+def write_orders_project(project_dir: Path) -> None:
+    """Write a three-model orders project to `project_dir`."""
+
     project_dir.mkdir(parents=True)
     (project_dir / "sqlbuild_project.toml").write_text(
         'name = "orders"\nadapter = "duckdb"\ndefault_target = "dev"\n\n'
@@ -1046,18 +1061,6 @@ def compiled_view_fallbacks(
         'SELECT order_id FROM __ref("daily_orders")\n',
         encoding="utf-8",
     )
-    return _counted_fallbacks(project_dir=project_dir, command=command)
-
-
-def _counted_fallbacks(*, project_dir: Path, command: tuple[str, ...]) -> tuple[int, int]:
-    """The exit code of `command` and how many of its reads missed a native view."""
-
-    _ = native_module.compile_stage_work_info(reset=True)
-    exit_code: int = main(["--project-dir", str(project_dir), *command])
-    fallbacks: int = native_module.compile_stage_work_info(reset=True).get(
-        "compiled_view_fallbacks", 0
-    )
-    return exit_code, fallbacks
 
 
 FORMAT_CACHE_UNFORMATTED_MODELS: dict[str, str] = {
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py b/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py
index 42139fd0c9..5c8450649f 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_compiled_view_reads.py
@@ -9,7 +9,11 @@
 from tests.integration.src.sqlbuild.cli.commands.main._test_types import (
     CompiledViewFallbacksTestCase,
 )
-from tests.integration.src.sqlbuild.cli.commands.main.helpers import compiled_view_fallbacks
+from tests.integration.src.sqlbuild.cli.commands.main.helpers import (
+    compiled_view_fallbacks,
+    copy_waffle_shop,
+    write_orders_project,
+)
 
 
 @pytest.mark.parametrize(
@@ -17,21 +21,21 @@
     [
         CompiledViewFallbacksTestCase(
             description="compiling reads every model natively",
-            fixture=None,
+            write_project=write_orders_project,
             command=("compile", "--json", "--no-cache"),
             expected_exit_code=0,
             expected_fallbacks=0,
         ),
         CompiledViewFallbacksTestCase(
             description="planning reads every model natively",
-            fixture=None,
+            write_project=write_orders_project,
             command=("plan", "--json", "--no-cache"),
             expected_exit_code=0,
             expected_fallbacks=0,
         ),
         CompiledViewFallbacksTestCase(
             description="planning reads every source, seed, function and audit natively",
-            fixture="waffle_shop",
+            write_project=copy_waffle_shop,
             command=("plan", "--json", "--no-cache"),
             expected_exit_code=0,
             expected_fallbacks=0,
@@ -45,7 +49,9 @@ def test_given_native_project_when_running_then_no_model_read_falls_back_to_pyth
     capsys: pytest.CaptureFixture[str],
 ) -> None:
     exit_code, fallbacks = compiled_view_fallbacks(
-        project_dir=tmp_path / "project", fixture=test_case.fixture, command=test_case.command
+        project_dir=tmp_path / "project",
+        write_project=test_case.write_project,
+        command=test_case.command,
     )
     _ = capsys.readouterr()