Add a map_element primitive to Lane and teach the node macro to emit it

node-graph/libraries/core-types/src/node.rs

@@ -188,20 +188,48 @@ impl<'a> RecordLane<'a> {
 
 /// One lane of a materialized level, element-typed. In a kernel's element
 /// position the output frame is copied from this lane.
+///
+/// The subject is substitutable: [`map_element`](Self::map_element) keeps the
+/// lane's whole record and swaps what it is a record *of*, so a node that
+/// rewrites its element still carries every column it never mentions. `Attr`
+/// members of the kernel's return tuple remain overrides layered on top.
 #[derive(Debug)]
 pub struct Lane<'a, T> {
 	lane: RecordLane<'a>,
-	_element: PhantomData<T>,
+	/// `Some` once the subject was replaced; the carried record's own element
+	/// is then ignored in favour of this value.
+	subject: Option<T>,
 }
 
-// A view regardless of `T`: copying a lane copies no record.
-impl<T> Clone for Lane<'_, T> {
+// A lane that carries no substitute is still a pure view: copying it copies no
+// record. One holding a replaced subject owns it, so it clones with the value.
+impl<T: Clone> Clone for Lane<'_, T> {
 	fn clone(&self) -> Self {
-		*self
+		Self {
+			lane: self.lane,
+			subject: self.subject.clone(),
+		}
 	}
 }
 
-impl<T> Copy for Lane<'_, T> {}
+impl<T: Copy> Copy for Lane<'_, T> {}
+
+impl<'a, T> Lane<'a, T> {
+	/// The same source lane with a different subject. The record's columns are
+	/// carried unchanged; only the element becomes `element`.
+	pub fn map_element<U>(self, element: U) -> Lane<'a, U> {
+		Lane {
+			lane: self.lane,
+			subject: Some(element),
+		}
+	}
+
+	/// The substituted subject, or `None` where this lane still stands for its
+	/// source record's own element.
+	pub fn into_element(self) -> Option<T> {
+		self.subject
+	}
+}
 
 impl<'a, T> Deref for Lane<'a, T> {
 	type Target = RecordLane<'a>;
@@ -276,7 +304,7 @@ impl<'a, T> List<'a, T> {
 	pub fn lane(&self, index: usize) -> Lane<'a, T> {
 		Lane {
 			lane: self.batch.get(index),
-			_element: PhantomData,
+			subject: None,
 		}
 	}
 

node-graph/libraries/core-types/src/record/serve.rs

@@ -127,6 +127,23 @@ impl<'e, 'l> FrameClaim<'e, 'l> {
 		self.filled_fields = true;
 	}
 
+	/// Copies the source record's element bytes into the frame, for a gathered
+	/// lane whose subject [`map_element`](crate::node::Lane::map_element) never
+	/// replaced. Carrying the element is not a field write, so it leaves
+	/// `filled_fields` alone.
+	///
+	/// # Safety
+	/// As [`Self::carry`]: `src` must be a live record of the gather's source
+	/// layout and must not overlap this frame. The element slots agree because
+	/// an unsubstituted `Lane<T>` can only come from a level whose element is
+	/// already `T`.
+	pub unsafe fn carry_element(&mut self, src: Rec<'_>) {
+		let size = self.layout.element.size;
+		if size > 0 {
+			unsafe { apply_plan(src, self.dst(), &[(0, 0, size)]) };
+		}
+	}
+
 	/// Writes a field at its wiring-resolved offset.
 	///
 	/// # Safety

node-graph/node-macro/src/codegen.rs

@@ -822,9 +822,13 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 	// The record-io write set, resolved from the output item and carrier input.
 	let write_markers: Vec<&Type> = node.output.shape.attrs.iter().map(|attr| &attr.marker).collect();
 	let removes: Vec<&Type> = node.output.removes.iter().map(|attr| &attr.marker).collect();
-	// A gathered element rides the copy plan, never a write.
+	// A gathered element of a known type is a write, not a plan entry: the lane's
+	// subject is substitutable through `map_element`, so the frame takes the
+	// replacement where there is one and carries the source bytes where there is
+	// not. A gathered element that is still generic keeps the plan's byte carry
+	// and cannot be substituted.
 	let element_write: Option<&Type> = match &node.output.shape.element {
-		crate::codegen::ir::Element::Concrete(ty) if !gather_carrier => Some(ty),
+		crate::codegen::ir::Element::Concrete(ty) => Some(ty),
 		_ => None,
 	};
 	let carrier_read_ty: Option<Type> = node.inputs.first().filter(|input| input.subject).and_then(|input| match &input.shape.element {
@@ -2116,14 +2120,33 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 		// A droppable element parks in the arena and rides as a reference.
 		let element_store = element_write.map(|ty| {
 			let ty = &crate::codegen::classify::substitute_lifetimes(ty, "'_");
-			quote! {
-				if __frame.element::<#ty>(__element, #core_types::context::ExtractArena::arena(__input)).is_none() {
-					return #core_types::gpoll::Interrupt::from(#core_types::gpoll::GraphError {
-						kind: #core_types::gpoll::ErrorKind::ArenaExhausted,
-						trace: ::std::vec::Vec::new(),
-					})
-					.into();
-				}
+			let exhausted = quote! {
+				return #core_types::gpoll::Interrupt::from(#core_types::gpoll::GraphError {
+					kind: #core_types::gpoll::ErrorKind::ArenaExhausted,
+					trace: ::std::vec::Vec::new(),
+				})
+				.into();
+			};
+			match gather_carrier {
+				// A gathered lane writes the subject `map_element` substituted,
+				// and otherwise carries the source record's own element bytes.
+				true => quote! {
+					match #core_types::node::Lane::into_element(__element) {
+						::core::option::Option::Some(__subject) => {
+							if __frame.element::<#ty>(__subject, #core_types::context::ExtractArena::arena(__input)).is_none() {
+								#exhausted
+							}
+						}
+						// SAFETY: `__src_rec` is the gathered lane's own live record, whose
+						// element slot the output layout resolved against.
+						::core::option::Option::None => unsafe { __frame.carry_element(__src_rec) },
+					}
+				},
+				false => quote! {
+					if __frame.element::<#ty>(__element, #core_types::context::ExtractArena::arena(__input)).is_none() {
+						#exhausted
+					}
+				},
 			}
 		});
 		let attr_stores = attr_binders.iter().enumerate().map(|(index, binder)| {

node-graph/nodes/gcore/src/record.rs

@@ -300,7 +300,11 @@ fn mirror_extent(content: ListIn<'_, f64>, keep_original: ValueIn<'_, bool>, lev
 /// copies that lane's whole record, so undeclared attributes ride along and
 /// only the declared opacity is rewritten.
 #[node_macro::node(category("Test"), extent(reverse_lanes_extent))]
-fn reverse_lanes(ctx: impl Ctx + ExtractIndex + InjectIndex + Copy, content: IList<f64>, opacity: f64) -> Result<IList<(Lane<f64>, Attr<Opacity>)>, Interrupt> {
+fn reverse_lanes<'e>(
+	ctx: impl Ctx + core_types::context::ExtractArena<'e> + ExtractIndex + InjectIndex + Copy,
+	content: IList<f64>,
+	opacity: f64,
+) -> Result<IList<(Lane<f64>, Attr<Opacity>)>, Interrupt> {
 	let lane = ctx.innermost_index() as usize;
 	if lane >= content.len() {
 		return Err(GraphError::past_end().into());
@@ -315,6 +319,24 @@ fn reverse_lanes_extent(content: ListIn<'_, f64>, _opacity: ValueIn<'_, f64>, le
 	}
 }
 
+/// Gather-carrier kernel with a substituted subject: the lane's record carries
+/// exactly as it does for [`reverse_lanes`], but `map_element` replaces the
+/// element, so a node that rewrites what it produces still keeps every column
+/// it never declares.
+#[node_macro::node(category("Test"), extent(reverse_lanes_extent))]
+fn scale_lanes<'e>(
+	ctx: impl Ctx + core_types::context::ExtractArena<'e> + ExtractIndex + InjectIndex + Copy,
+	content: IList<f64>,
+	opacity: f64,
+) -> Result<IList<(Lane<f64>, Attr<Opacity>)>, Interrupt> {
+	let lane = ctx.innermost_index() as usize;
+	if lane >= content.len() {
+		return Err(GraphError::past_end().into());
+	}
+	let scaled = content.element_ref(lane) * 2.;
+	Ok((content.lane(lane).map_element(scaled), Attr(opacity)))
+}
+
 #[node_macro::node(category("Test"))]
 fn source_opacity(_: impl Ctx, _: (), element: f64, opacity: f64) -> (f64, Attr<Opacity>) {
 	(element, Attr(opacity))
@@ -531,6 +553,42 @@ mod tests {
 		}
 	}
 
+	/// Serves lanes carrying both a Transform no gather kernel declares and an
+	/// Opacity one does, so a carried column can be told apart from a written one.
+	struct LeveledCarriedSource {
+		layout: Layout,
+		rows: Vec<(f64, DAffine2, f64)>,
+	}
+
+	impl<C: ExtractIndex> Node<C> for LeveledCarriedSource {
+		fn serve<'e, 'l>(&self, input: &C, slot: FrameClaim<'e, 'l>) -> GPoll<Served<'e>>
+		where
+			C: ExtractArena<ArenaRef = &'e Arena>,
+		{
+			let (element, transform, opacity) = self.rows[input.innermost_index() as usize % self.rows.len()];
+			let mut frame = slot;
+			let arena = ExtractArena::arena(input);
+			if frame.element(element, arena).is_none() {
+				return GPoll::arena_exhausted();
+			}
+			write_attr_at::<Transform>(&mut frame, &self.layout, transform);
+			write_attr_at::<Opacity>(&mut frame, &self.layout, opacity);
+			// SAFETY: the writes above complete the record of this layout.
+			GPoll::Final(unsafe { frame.finish_served() })
+		}
+
+		fn extent_at<'x>(&self, _input: &C, _level: u8, _frames: &core_types::record::Frames<'x>) -> GPoll<Extent>
+		where
+			C: ExtractArena<ArenaRef = &'x Arena>,
+		{
+			GPoll::Final(Extent::Exactly(self.rows.len()))
+		}
+
+		fn layout(&self) -> &Layout {
+			&self.layout
+		}
+	}
+
 	/// A leveled source that keeps its count to itself: the extent is a lower
 	/// bound and lanes past the data answer the past-end signal.
 	struct DrainSourceNode {
@@ -1508,6 +1566,56 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn a_substituted_subject_keeps_the_lanes_other_columns() {
+		let arena = Arena::new(1 << 16).unwrap();
+		let generations = [];
+		let scope = scope_fixture(&generations, &arena);
+		let ctx = ContextImpl::root(&scope);
+
+		// Each lane carries a Transform the kernel never declares and an Opacity
+		// it does, so the carried columns and the written one are distinguishable.
+		let layout = Layout::default().with_writes(
+			1,
+			core_types::record::element_write::<f64>(),
+			&[core_types::record::FieldWrite::of::<Transform>(0), core_types::record::FieldWrite::of::<Opacity>(0)],
+		);
+		let frames = frames_for(&[&layout]);
+		let rows = [(1., 10., 0.1), (2., 30., 0.2), (3., 20., 0.3)];
+		let content = LeveledCarriedSource {
+			layout: layout.clone(),
+			rows: rows
+				.iter()
+				.map(|&(element, x, opacity)| (element, DAffine2::from_translation(glam::DVec2::new(x, 0.)), opacity))
+				.collect(),
+		};
+		let node = install(
+			ScaleLanesNode::new(RecordSource::new(content, &layout, &layout), ValueSource::new(0.25)),
+			scale_lanes_layout_meta(),
+			&[Some(&layout)],
+		);
+
+		let out = Node::<ContextImpl>::layout(&node).clone();
+		assert_eq!(out.depth, 1, "substituting the subject preserves the level's depth");
+		assert!(
+			out.offset_of(<Transform as AttributeMarker>::NAME, 0).is_some(),
+			"the undeclared attribute survives a substituted subject"
+		);
+
+		let head = ctx.index_head();
+		for (lane, &(element, x, carried_opacity)) in rows.iter().enumerate() {
+			let GPoll::Final(served) = core_types::record::capture(&node, &ctx.promoted(&head, lane as u64), &frames) else {
+				panic!("expected a final record");
+			};
+			assert_eq!(served.element::<f64>(), element * 2., "lane {lane} takes the substituted element, not the source's");
+			let transform: DAffine2 = served.attr::<Transform>();
+			assert_eq!(transform.translation.x, x, "lane {lane} still carries the undeclared transform");
+			let opacity: f64 = served.attr::<Opacity>();
+			assert_ne!(opacity, carried_opacity, "the declared write must not leave the carried opacity in place");
+			assert_eq!(opacity, 0.25, "lane {lane} takes the declared write over the carried column");
+		}
+	}
+
 	#[test]
 	fn batch_lanes_match_per_lane_eval() {
 		let arena = Arena::new(1 << 16).unwrap();