fix: refresh only a definition the view's schema can represent

The reloaded definition is now re-planned against the current source schema
under the lock, and its planned output must be exactly the view's physical
schema -- otherwise the refresh is refused rather than certifying rows the
stored table cannot represent. Planning also re-derives the inputs and
re-applies the immutability check, so a definition written by another
producer gets the same scrutiny as one written here.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Wyatt Alt
2026-08-15 01:09:12 -07:00
co-authored by Claude Fable 5
parent de9812c294
commit 2d8bc496e7
@@ -151,7 +151,48 @@ pub(crate) async fn execute_refresh(
let source_version = source_ds.version().version;
let source_ts = source_ds.manifest.timestamp_nanos;
// Re-plan the persisted definition against the current source schema and
// require its planned output to be exactly the view's physical schema: a
// definition the stored table cannot represent must not be certified.
let source_schema = Arc::new(ArrowSchema::from(source_ds.schema()));
let projections: Vec<(String, String)> = definition
.projections
.iter()
.map(|p| (p.output.clone(), p.expression.clone()))
.collect();
validate_inputs(&source_ds, definition)?;
let (replanned, mut planned_fields) = super::plan(
source_schema,
&definition.source_table,
&projections,
definition.filter.as_deref(),
definition.limit,
)?;
planned_fields.push(arrow_schema::Field::new(
SOURCE_ROW_ID_COLUMN,
arrow_schema::DataType::UInt64,
false,
));
let physical = ArrowSchema::from(view_ds.schema());
let planned_shape: Vec<_> = planned_fields
.iter()
.map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable()))
.collect();
let physical_shape: Vec<_> = physical
.fields()
.iter()
.map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable()))
.collect();
if planned_shape != physical_shape {
return Err(Error::Schema {
message: format!(
"the stored definition of view '{}' does not produce this \
view's schema; recreate the view",
view.name()
),
});
}
let definition = &replanned;
let metadata = &view_ds.schema().metadata;
let watermark: Option<u64> = metadata
@@ -1422,6 +1463,40 @@ mod tests {
assert_eq!(read(view.table(), "twice").await, vec![3, 6]);
}
/// A persisted definition that does not produce this view's schema must
/// not refresh at all, let alone be certified.
#[tokio::test]
async fn test_definition_view_schema_mismatch_is_refused() {
let (conn, _) = db_with_source(vec![1, 2]).await;
let view = doubled_view(&conn).await;
view.refresh().execute().await.unwrap();
let narrower = crate::materialized_view::MaterializedViewDefinition {
source_table: "src".into(),
projections: vec![crate::materialized_view::ViewProjection {
output: "x".into(),
expression: "x".into(),
}],
filter: None,
limit: None,
inputs: vec!["x".into()],
};
let mut metadata = HashMap::new();
metadata.insert(
crate::materialized_view::DEFINITION_META_KEY.to_string(),
crate::materialized_view::definition_to_metadata(&narrower).unwrap(),
);
view.table()
.as_native()
.unwrap()
.replace_schema_metadata(metadata)
.await
.unwrap();
let err = view.refresh().execute().await.unwrap_err();
assert!(matches!(err, Error::Schema { message } if message.contains("does not produce")),);
}
/// An output whose name needs quoting flows through as a projection
/// alias, never spliced into SQL text.
#[tokio::test]