refactor: call compile driver and planner stages directly and host CompiledResourceType natively

crates/sqlbuild-cache/src/hosted_shapes/constants.rs

@@ -1,7 +1,7 @@
 //! The Python shape names of the natively hosted compiler shapes.
 
 /// The Python model type a natively hosted compiler shape is captured as.
-pub const HOSTED_SHAPE_TYPES: [(&str, &str); 490] = [
+pub const HOSTED_SHAPE_TYPES: [(&str, &str); 491] = [
     (
         "sqlbuild._native:AlignedInterval",
         "sqlbuild.cursor_algebra.models:AlignedInterval",
@@ -446,6 +446,10 @@ pub const HOSTED_SHAPE_TYPES: [(&str, &str); 490] = [
         "sqlbuild._native:DeclarationVisibilityIndex",
         "sqlbuild.compiler.scopes.models:DeclarationVisibilityIndex",
     ),
+    (
+        "sqlbuild._native:CompiledResourceType",
+        "sqlbuild.compiler.compile.types:CompiledResourceType",
+    ),
     (
         "sqlbuild._native:DefaultsConfig",
         "sqlbuild.spec.contracts.models:DefaultsConfig",

crates/sqlbuild-python/src/bindings/_helpers/analysis/explicit_references.rs

@@ -77,7 +77,7 @@ impl<'py> PythonSqlOwner<'py> {
 /// Return P005, P007 and P008 errors and unmatched Python-SQL relations; raise P001.
 #[pyfunction]
 #[pyo3(signature = (*, project, discovered_inputs, enforce_explicit))]
-fn explicit_reference_report<'py>(
+pub(crate) fn explicit_reference_report<'py>(
     py: Python<'py>,
     project: &Bound<'py, PyAny>,
     discovered_inputs: &Bound<'py, PyAny>,

crates/sqlbuild-python/src/bindings/_helpers/analysis_session/stage_functions.rs

@@ -20,6 +20,9 @@ use crate::bindings::_helpers::analysis_session::session::{
     infer_expression_source_shapes, prove_dynamic_column_contracts,
 };
 use crate::bindings::_helpers::boundary::keyword_request::Request;
+use crate::bindings::_helpers::compile_driver::stage_work::{
+    EXPRESSION_SHAPE_BATCHES, EXPRESSION_SHAPES_INFERRED, count_stage,
+};
 use crate::bindings::_helpers::records::fields::field as native_field;
 use crate::bindings::_helpers::records::ids::RecordId;
 use crate::bindings::_helpers::records::registry;
@@ -105,9 +108,16 @@ fn infer_native_expression_source_shapes<'py>(
     request: Option<&Bound<'py, PyDict>>,
 ) -> PyResult<Bound<'py, PyTuple>> {
     let request = Request::new("infer_native_expression_source_shapes", request)?;
-    let py: Python<'py> = request.py();
-    let expressions: Bound<'py, PyAny> = request.get("expressions")?;
-    let profile: Bound<'py, PyAny> = request.get("profile")?;
+    expression_source_shapes(&request.get("expressions")?, &request.get("profile")?)
+}
+
+/// One shape per expression, cached on `profile`'s binding catalog; a failure raises.
+pub(crate) fn expression_source_shapes<'py>(
+    expressions: &Bound<'py, PyAny>,
+    profile: &Bound<'py, PyAny>,
+) -> PyResult<Bound<'py, PyTuple>> {
+    let py: Python<'py> = profile.py();
+    let profile: Bound<'py, PyAny> = profile.clone();
     let dialect: String = aliased_dialect(py, profile_dialect(&profile)?)?;
     let catalog: Bound<'py, PyAny> = profile_catalog(&profile, &dialect)?;
     let cache: Bound<'py, PyAny> = native_field(&catalog, "expression_shapes")?;
@@ -119,6 +129,8 @@ fn infer_native_expression_source_shapes<'py>(
         }
     }
     if !pending.is_empty() {
+        count_stage(py, EXPRESSION_SHAPE_BATCHES, 1)?;
+        count_stage(py, EXPRESSION_SHAPES_INFERRED, pending.len() as u64)?;
         let (rules, adapter_rules) = nullability_rules(&profile)?;
         let native: PyRef<'py, ProjectCatalog> = native_field(&catalog, "native")?.extract()?;
         let pending_names: Vec<String> = pending
@@ -169,10 +181,20 @@ fn prove_native_dynamic_contracts<'py>(
     request: Option<&Bound<'py, PyDict>>,
 ) -> PyResult<Bound<'py, PyTuple>> {
     let request = Request::new("prove_native_dynamic_contracts", request)?;
-    let py: Python<'py> = request.py();
-    let session: Bound<'py, PyAny> = request.get("session")?;
-    let tables: Bound<'py, PyAny> = request.get("tables")?;
-    let models: Bound<'py, PyAny> = request.get("models")?;
+    dynamic_contract_proofs(
+        &request.get("session")?,
+        &request.get("tables")?,
+        &request.get("models")?,
+    )
+}
+
+/// The proof row per `(sql, families)` model from `session`, or from `tables` without one.
+pub(crate) fn dynamic_contract_proofs<'py>(
+    session: &Bound<'py, PyAny>,
+    tables: &Bound<'py, PyAny>,
+    models: &Bound<'py, PyAny>,
+) -> PyResult<Bound<'py, PyTuple>> {
+    let py: Python<'py> = models.py();
     if !models.is_truthy()? {
         return Ok(PyTuple::empty(py));
     }
@@ -182,7 +204,7 @@ fn prove_native_dynamic_contracts<'py>(
         rows.push((sql, family_rows(&families)?));
     }
     let contracts: Vec<ContractRow> = if session.is_none() {
-        let by_table: Bound<'py, PyAny> = native_field(&tables, "dynamic_families_by_table")?;
+        let by_table: Bound<'py, PyAny> = native_field(tables, "dynamic_families_by_table")?;
         let mut families: Vec<(String, Vec<FamilyRow>)> = Vec::new();
         for item in by_table.call_method0("items")?.try_iter()? {
             let (name, declared): (String, Bound<'py, PyAny>) = item?.extract()?;
@@ -191,13 +213,13 @@ fn prove_native_dynamic_contracts<'py>(
         prove_dynamic_column_contracts(
             py,
             (
-                dialect_or_generic(&native_field(&tables, "dialect")?)?,
-                shape_rows(&native_field(&tables, "column_types_by_table")?)?,
+                dialect_or_generic(&native_field(tables, "dialect")?)?,
+                shape_rows(&native_field(tables, "column_types_by_table")?)?,
                 shape_rows(&native_field(
-                    &tables,
+                    tables,
                     "authoritative_column_types_by_table",
                 )?)?,
-                shape_rows(&native_field(&tables, "column_nullability_by_table")?)?,
+                shape_rows(&native_field(tables, "column_nullability_by_table")?)?,
                 families,
                 rows,
             ),

crates/sqlbuild-python/src/bindings/_helpers/build_command/defer_clone.rs

@@ -13,6 +13,7 @@ use sqlbuild_core::build_command::errors::BuildCommandError;
 
 use crate::bindings::_helpers::build_command::invocation::BuildContext;
 use crate::bindings::_helpers::build_command::policies::build_command_error;
+use crate::bindings::_helpers::compile_driver::project::build_compiled_project;
 use crate::bindings::_helpers::compile_driver::python_calls::{call_with, imported, keywords};
 use crate::bindings::_helpers::native_host::executor::ExecutorHost;
 use crate::bindings::_helpers::native_host::host;
@@ -586,30 +587,32 @@ pub(crate) fn run_defer_clone_boundary_prephase(
     }
     let adapter: Bound<'_, PyAny> = native_field(invocation, "adapter")?;
     let project_dir: Bound<'_, PyAny> = native_field(invocation, "effective_project_dir")?;
-    let project: Bound<'_, PyAny> = native(
+    let project: Bound<'_, PyAny> = build_compiled_project(
         py,
-        NativeAttr::BuildCompiledProject,
-        &[
-            ("discovered_inputs", discovered.clone()),
-            ("adapter", adapter.clone()),
-            ("selected_target", native_field(request, "selected_target")?),
-            (
-                "no_sql_validation",
-                native_field(request, "no_sql_validation")?,
-            ),
-            (
-                "analysis_selection",
-                call_with(
-                    &registry::class(py, RecordId::CompileAnalysisSelection)?,
-                    &[("no_cache", native_field(request, "no_cache")?)],
-                )?,
-            ),
-            ("cli_vars", native_field(request, "cli_vars")?),
-            (
-                "external_sql_reference_resolver",
-                external_resolver(&project_dir, &discovered)?,
-            ),
-        ],
+        Some(keywords(
+            py,
+            &[
+                ("discovered_inputs", discovered.clone()),
+                ("adapter", adapter.clone()),
+                ("selected_target", native_field(request, "selected_target")?),
+                (
+                    "no_sql_validation",
+                    native_field(request, "no_sql_validation")?,
+                ),
+                (
+                    "analysis_selection",
+                    call_with(
+                        &registry::class(py, RecordId::CompileAnalysisSelection)?,
+                        &[("no_cache", native_field(request, "no_cache")?)],
+                    )?,
+                ),
+                ("cli_vars", native_field(request, "cli_vars")?),
+                (
+                    "external_sql_reference_resolver",
+                    external_resolver(&project_dir, &discovered)?,
+                ),
+            ],
+        )?),
     )?;
     let scope: Bound<'_, PyAny> = native(
         py,

crates/sqlbuild-python/src/bindings/_helpers/build_command/invocation.rs

@@ -10,6 +10,8 @@ use sqlbuild_core::build_command::models::DurationLimit;
 use crate::bindings::_helpers::build_command::policies::build_command_error;
 use crate::bindings::_helpers::build_command::support::{boolean, or_none, text};
 use crate::bindings::_helpers::command_runtime::adapters::resolve_cli_adapter;
+use crate::bindings::_helpers::compile_driver::discovery::discover_project_inputs;
+use crate::bindings::_helpers::compile_driver::python_calls::keywords;
 use crate::bindings::_helpers::compile_driver::python_calls::{call_with, imported};
 use crate::bindings::_helpers::presentation_text::terminal::stdout_supports_color;
 use crate::bindings::_helpers::records::fields::field as native_field;
@@ -105,12 +107,15 @@ pub(crate) fn resolve_build_context<'py>(
     let project_dir: Bound<'py, PyAny> = or_none(native_field(request, "project_dir")?, || {
         imported(py, "pathlib", "Path")?.call_method0("cwd")
     })?;
-    let discovered: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::DiscoverProjectInputs)?,
-        &[
-            ("project_dir", project_dir.clone()),
-            ("extract_output_column_locations", boolean(py, false)),
-        ],
+    let discovered: Bound<'py, PyAny> = discover_project_inputs(
+        py,
+        Some(keywords(
+            py,
+            &[
+                ("project_dir", project_dir.clone()),
+                ("extract_output_column_locations", boolean(py, false)),
+            ],
+        )?),
     )?;
     let selected_target: Bound<'py, PyAny> = native_field(request, "selected_target")?;
     let target_name: Bound<'py, PyAny> = call_with(

crates/sqlbuild-python/src/bindings/_helpers/build_command/lifecycle.rs

@@ -22,7 +22,6 @@ use crate::bindings::_helpers::records::fields::field as native_field;
 use crate::bindings::_helpers::records::ids::{NativeAttr, RecordId};
 use crate::bindings::_helpers::records::registry;
 
-const COMPILE_TYPES: &str = "sqlbuild.compiler.compile.types";
 const SUCCESS_STATUS: &str = "success";
 
 /// The cursor window and output settings the Python lifecycle runs with.
@@ -323,7 +322,7 @@ fn lifecycle_plan<'py>(
         &[("selection", selection), ("python_graph", graph.clone())],
     )?;
     let source_type: Bound<'py, PyAny> =
-        imported(py, COMPILE_TYPES, "CompiledResourceType")?.getattr("SOURCE")?;
+        registry::class(py, RecordId::CompiledResourceType)?.getattr("SOURCE")?;
     let key_class: Bound<'py, PyAny> = registry::class(py, RecordId::CompiledObjectKey)?;
     let mut keys: Vec<Bound<'py, PyAny>> = Vec::new();
     for name in native_field(&plan, "ingress_loader_names")?.try_iter()? {

crates/sqlbuild-python/src/bindings/_helpers/check/check_command.rs

@@ -36,18 +36,18 @@ use crate::bindings::_helpers::check::check_results::{
     check_result_text, results_json, write_results,
 };
 use crate::bindings::_helpers::command_runtime::invocation::resolve_reported_command_invocation;
-use crate::bindings::_helpers::compile_driver::python_calls::{call_with, imported};
+use crate::bindings::_helpers::compile_driver::python_calls::call_with;
 use crate::bindings::_helpers::compile_outputs::runtime_target::write_python_check_runtime_target;
 use crate::bindings::_helpers::graph::python_node_graph::python_node_graph;
 use crate::bindings::_helpers::graph::python_selection::python_node_names;
 use crate::bindings::_helpers::native_host::executor::ExecutorHost;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::native_host::user_code::UserCodeHost;
 use crate::bindings::_helpers::records::fields::field as native_field;
-use crate::bindings::_helpers::records::ids::{ErrorId, NativeAttr, RecordId};
+use crate::bindings::_helpers::records::ids::{ErrorId, RecordId};
 use crate::bindings::_helpers::records::registry;
+use crate::bindings::_helpers::result_output::event_writer::ExecutionEventWriter;
 
-const COMPILE_TYPES: &str = "sqlbuild.compiler.compile.types";
 const GRAPH_OPERATOR_CODE: &str = "C681";
 const GRAPH_OPERATOR_MESSAGE: &str = "sqb check selectors do not support graph operators; \
     select checks directly without '+', '~', or path-between expansion";
@@ -201,7 +201,7 @@ fn compile_pipeline<'py>(
     let loaders: Bound<'py, PyFrozenSet> = PyFrozenSet::new(py, loaders)?;
     let key_class: Bound<'py, PyAny> = registry::class(py, RecordId::CompiledObjectKey)?;
     let source_type: Bound<'py, PyAny> =
-        imported(py, COMPILE_TYPES, "CompiledResourceType")?.getattr("SOURCE")?;
+        registry::class(py, RecordId::CompiledResourceType)?.getattr("SOURCE")?;
     let mut keys: Vec<Bound<'py, PyAny>> = Vec::new();
     for reference in required_refs.try_iter()? {
         keys.push(sql_ref_key(py, &reference?)?);
@@ -572,7 +572,7 @@ fn execute<'py>(
     let connection: Bound<'py, PyAny> =
         adapter.call_method1("connect", (native_field(invocation, "connection_config")?,))?;
     let event_writer: Bound<'py, PyAny> =
-        registry::native(py, NativeAttr::ExecutionEventWriter)?.call0()?;
+        Bound::new(py, ExecutionEventWriter::new(py, None)?)?.into_any();
     let executed: PyResult<Vec<Bound<'py, PyAny>>> = (|| {
         let projector: Bound<'py, PyAny> = current_native_progress_projector(py)?;
         if !projector.is_none() {

crates/sqlbuild-python/src/bindings/_helpers/cli/progress_reporters.rs

@@ -48,7 +48,7 @@ impl ConnectionProgressReporter {
 impl ConnectionProgressReporter {
     #[new]
     #[pyo3(signature = (**options))]
-    fn new(py: Python<'_>, options: Option<Bound<'_, PyDict>>) -> PyResult<Self> {
+    pub(crate) fn new(py: Python<'_>, options: Option<Bound<'_, PyDict>>) -> PyResult<Self> {
         let arguments: KeywordArguments<'_> = KeywordArguments::read(
             py,
             "ConnectionProgressReporter",
@@ -133,7 +133,11 @@ impl PlanningProgressReporter {
 impl PlanningProgressReporter {
     #[new]
     #[pyo3(signature = (*, stream, use_color = false))]
-    fn new(py: Python<'_>, stream: &Bound<'_, PyAny>, use_color: bool) -> PyResult<Self> {
+    pub(crate) fn new(
+        py: Python<'_>,
+        stream: &Bound<'_, PyAny>,
+        use_color: bool,
+    ) -> PyResult<Self> {
         let status: Bound<'_, PyAny> = status_reporter(stream, use_color)?;
         Ok(Self {
             status: status.unbind(),

crates/sqlbuild-python/src/bindings/_helpers/clone/clone_command.rs

@@ -24,6 +24,8 @@ use crate::bindings::_helpers::clone::clone_execution::{
 use crate::bindings::_helpers::clone::clone_pipeline::{ClonePipeline, prepared_clone_pipeline};
 use crate::bindings::_helpers::clone::clone_progress::{clone_item_text, write_line};
 use crate::bindings::_helpers::command_runtime::adapters::resolve_cli_adapter;
+use crate::bindings::_helpers::compile_driver::discovery::discover_project_inputs;
+use crate::bindings::_helpers::compile_driver::python_calls::keywords;
 use crate::bindings::_helpers::compile_driver::python_calls::{call_with, within};
 use crate::bindings::_helpers::native_host::executor::ExecutorHost;
 use crate::bindings::_helpers::native_host::host;
@@ -132,9 +134,9 @@ fn invocation<'py>(py: Python<'py>, request: &Bound<'py, PyAny>) -> PyResult<Inv
     let stream: Bound<'py, PyAny> = py.import("sys")?.getattr("stdout")?;
     let progress: Bound<'py, PyAny> = progress_reporter(&stream, style)?;
     progress.call_method1("start", ("Discovering project...",))?;
-    let discovered: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::DiscoverProjectInputs)?,
-        &[("project_dir", project_dir.clone())],
+    let discovered: Bound<'py, PyAny> = discover_project_inputs(
+        py,
+        Some(keywords(py, &[("project_dir", project_dir.clone())])?),
     )?;
     progress.call_method1("complete", ("Discovered project.",))?;
     call_with(

crates/sqlbuild-python/src/bindings/_helpers/clone/clone_execution.rs

@@ -12,14 +12,13 @@ use crate::bindings::_helpers::clone::clone_command::Invocation;
 use crate::bindings::_helpers::clone::clone_progress::{
     CloneFingerprintReporter, CloneItemPrinter,
 };
-use crate::bindings::_helpers::compile_driver::python_calls::{call_with, imported, within};
+use crate::bindings::_helpers::compile_driver::python_calls::{call_with, within};
 use crate::bindings::_helpers::native_host::executor::ExecutorHost;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::records::fields::field as native_field;
 use crate::bindings::_helpers::records::ids::{NativeAttr, RecordId};
 use crate::bindings::_helpers::records::registry;
 
-const COMPILE_TYPES: &str = "sqlbuild.compiler.compile.types";
 const DESTINATION_ENTRY_GROUPS: [&str; 4] = [
     "destination_source_entries",
     "destination_model_entries",
@@ -125,7 +124,7 @@ fn dependency_locations<'py>(
     let location_class: Bound<'py, PyAny> =
         registry::class(py, RecordId::CompiledRelationLocation)?;
     let source_type: Bound<'py, PyAny> =
-        imported(py, COMPILE_TYPES, "CompiledResourceType")?.getattr("SOURCE")?;
+        registry::class(py, RecordId::CompiledResourceType)?.getattr("SOURCE")?;
     for pair in source_map.call_method0("items")?.try_iter()? {
         let (name, source): (Bound<'py, PyAny>, Bound<'py, PyAny>) = pair?.extract()?;
         let table: Bound<'py, PyAny> = native_field(&source, "table")?;