Data merge: an image named by many records is read once (#1595)

`place_image` stat'd and read its file for every record, so a sheet naming one
image in many rows read it once per row -- up to 256 MiB each time. The audit's
case is a 100,000-row CSV naming one large file: about 25 TB of I/O for one
merge.

A merge now carries an `ImageCache`: the bytes by resolved path, read once and
shared. The asset takes the cached `Arc` rather than a copy, so the records
share one buffer as well. The cache stops growing past 256 MB of distinct
files, so a merge over many large images is slow rather than out of memory.

Measured by counting what the engine hands the file system (`io::touched`):

    without the cache   50 reads for 25 records   (a stat and a read each)
    with it             2

A note on the test, because the first one I wrote was useless: asserting on the
bytes the document holds passes either way, since `add_asset` already dedupes
identical content. Memory never showed this; the I/O is the bug, so the test
counts reads.

Audit DC-14. Its second half -- capping the pages a merge produces -- is not
here: how many pages a merge may make is a product decision, and the records
are the user's own data rather than something arriving from outside.

crates/engine/src/cmd/datamerge.rs

@@ -18,7 +18,7 @@ use serde_json::{Value, json};
 
 use super::{CommandSpec, bad, cmd, has_doc, id_param, ok, str_param};
 use crate::{DocState, Result, Session};
-use fill::{FillReport, Maps, MissingImage, Sources, copy_item, fill_row, identity_map};
+use fill::{FillReport, ImageCache, Maps, MissingImage, Sources, copy_item, fill_row, identity_map};
 use parse::{
     Table, TileInput, extension, fingerprint_of, relative_between, select_records, status_of, table_from_bytes, table_from_grid, table_from_objects,
     tile,
@@ -210,7 +210,8 @@ fn preview_cmd(s: &mut Session, p: &Value) -> Result<Value> {
     let row = source.rows.get((record as usize).saturating_sub(1)).cloned().unwrap_or_default();
     let data_dir = source_file(&source, st.path.as_deref().map(Path::new)).as_deref().and_then(Path::parent).map(Path::to_path_buf);
     let doc_dir = st.path.as_deref().and_then(parent_dir);
-    let sources = Sources { options: &options, data_dir: data_dir.as_deref(), doc_dir: doc_dir.as_deref() };
+    let images = ImageCache::default();
+    let sources = Sources { options: &options, data_dir: data_dir.as_deref(), doc_dir: doc_dir.as_deref(), images: &images };
     fill_row(&mut doc, &source.fields, &row, record as u32, &maps, &sources);
     st.doc = Arc::new(doc);
     st.preview_record = Some(record as u32);
@@ -246,7 +247,8 @@ fn merge_cmd(s: &mut Session, p: &Value) -> Result<Value> {
     warnings.extend(table.warnings.iter().cloned());
     let mut merged = src.clone();
     let mut missing = Vec::new();
-    let sources = Sources { options: &options, data_dir: data_dir.as_deref(), doc_dir: doc_dir.as_deref() };
+    let images = ImageCache::default();
+    let sources = Sources { options: &options, data_dir: data_dir.as_deref(), doc_dir: doc_dir.as_deref(), images: &images };
     let mut sink = Sink { warnings: &mut warnings, missing: &mut missing, parent: false };
     if options.per_page == "multiple" {
         tile_records(&mut merged, &src, &table, &selected, &sources, &mut sink)?;

crates/engine/src/cmd/datamerge/fill.rs

@@ -4,7 +4,7 @@
 //! unfilled template. A placeholder whose frame or story is not in the map is left as it is.
 
 use std::collections::HashMap;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use designcraft_doc::{
@@ -134,6 +134,53 @@ pub struct Sources<'a> {
     pub options: &'a MergeOptions,
     pub data_dir: Option<&'a Path>,
     pub doc_dir: Option<&'a Path>,
+    /// Images this merge has already read, by the path they resolved to.
+    ///
+    /// Every record used to stat and read its image again, so a 100,000-row sheet naming the same
+    /// file read it 100,000 times -- up to 256 MiB each, around 25 TB of I/O for one merge
+    /// (#1595). Shared by `&`, so the cache is behind a `RefCell`.
+    pub images: &'a ImageCache,
+}
+
+/// Image files read during one merge: the bytes by path, and the total read so far.
+#[derive(Default)]
+pub struct ImageCache {
+    inner: std::cell::RefCell<ImageCacheInner>,
+}
+
+#[derive(Default)]
+struct ImageCacheInner {
+    files: std::collections::HashMap<PathBuf, std::sync::Arc<Vec<u8>>>,
+    bytes: usize,
+}
+
+/// Bytes of distinct image files one merge keeps. Past this the cache stops growing and files are
+/// read as before, so a merge over many large images is slow rather than out of memory.
+const MAX_CACHED_IMAGE_BYTES: usize = 256 << 20;
+
+impl ImageCache {
+    /// The bytes of `path`, reading it once per merge. `read` is called only on a miss.
+    fn get_or_read(&self, path: &Path, read: impl FnOnce() -> Option<Vec<u8>>) -> Option<std::sync::Arc<Vec<u8>>> {
+        if let Some(found) = self.inner.borrow().files.get(path) {
+            return Some(found.clone());
+        }
+        let bytes = std::sync::Arc::new(read()?);
+        let mut inner = self.inner.borrow_mut();
+        if inner.bytes.saturating_add(bytes.len()) <= MAX_CACHED_IMAGE_BYTES {
+            inner.bytes = inner.bytes.saturating_add(bytes.len());
+            inner.files.insert(path.to_path_buf(), bytes.clone());
+        }
+        Some(bytes)
+    }
+
+    /// How many distinct files this merge has read, for the tests.
+    pub fn len(&self) -> usize {
+        self.inner.borrow().files.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
 }
 
 /// Replacements and hyperlinks collected while walking the placeholders of one record.
@@ -360,15 +407,20 @@ fn create_link(doc: &mut designcraft_doc::Document, anchor: PlaceholderAnchor, d
 }
 
 fn place_image(doc: &mut designcraft_doc::Document, item: ItemId, field: &str, cell: &str, record: u32, sources: &Sources, report: &mut FillReport) {
-    let Sources { options, data_dir, doc_dir } = *sources;
+    let Sources { options, data_dir, doc_dir, images } = *sources;
     let cands = image_candidates(cell, data_dir, doc_dir);
     let found = cands.iter().find_map(|p| {
-        // Refuse huge files before reading them; a too-large image counts as missing.
-        let len = crate::io::metadata(p).ok()?.len();
-        if len == 0 || len > super::MAX_IMAGE_FILE_BYTES {
-            return None;
-        }
-        crate::io::read_file(p).ok().filter(|b| !b.is_empty()).map(|b| (p.clone(), b))
+        // Read once per merge: many records naming one file is the normal case, and reading it
+        // per record is the amplification (#1595).
+        let bytes = images.get_or_read(p, || {
+            // Refuse huge files before reading them; a too-large image counts as missing.
+            let len = crate::io::metadata(p).ok()?.len();
+            if len == 0 || len > super::MAX_IMAGE_FILE_BYTES {
+                return None;
+            }
+            crate::io::read_file(p).ok().filter(|b| !b.is_empty())
+        })?;
+        Some((p.clone(), bytes))
     });
     let Some((path, bytes)) = found else {
         if let Some(it) = doc.item_mut(item) {
@@ -430,7 +482,9 @@ fn place_image(doc: &mut designcraft_doc::Document, item: ItemId, field: &str, c
         name,
         mime: designcraft_render::image_mime(&bytes).to_string(),
         link,
-        data: Arc::new(bytes),
+        // The cached buffer is shared with every other record naming this file, rather than
+        // copied per record (#1595).
+        data: bytes,
         pixels: Some((pw, ph)),
         ..Default::default()
     });

crates/engine/src/cmd/datamerge/tests.rs

@@ -519,6 +519,69 @@ fn typed_fields_keep_style_and_unknown_markers() {
     assert_eq!(story.format_after(at).over.size, Some(21.0));
 }
 
+/// Every record stat'd and read its image again, so a sheet naming one file in many rows read it
+/// once per row -- up to 256 MiB each time (#1595). A merge reads a file once and the records
+/// share the bytes.
+#[test]
+fn one_image_named_by_many_records_is_read_once() {
+    let dir = scratch("image-reads");
+    let data = dir.join("data");
+    std::fs::create_dir_all(&data).unwrap();
+    let shared = dir.join("shared.png");
+    write_png(&shared);
+    // 25 records, all naming the same file.
+    let rows: String = (0..25).map(|_| format!("{}\n", shared.display())).collect();
+    write(&data.join("many.csv"), &format!("@Photo\n{rows}"));
+
+    let mut s = Session::new();
+    new_doc(&mut s, 1);
+    exec(&mut s, "data.source.select", json!({"path": data.join("many.csv")}));
+    let frame = exec(&mut s, "frame.create", json!({"rect": [36.0, 36.0, 80.0, 80.0], "content": "graphic"}));
+    exec(&mut s, "data.placeholder.add", json!({"field": "Photo", "item": frame["id"]}));
+    let r = exec(&mut s, "data.merge", json!({}));
+    assert_eq!(r["records"], 25);
+    assert_eq!(r["missingImages"].as_array().map(Vec::len).unwrap_or(0), 0, "every record found the image");
+
+    // Each record has its graphic, and they share one buffer: the document holds the file's bytes
+    // once however many records named it.
+    let doc = &s.documents()[s.active_index().unwrap()].doc;
+    let graphics: Vec<_> = doc
+        .spreads
+        .iter()
+        .flat_map(|sp| &sp.items)
+        .filter_map(|it| match &it.content {
+            Content::Graphic(g) => Some(g.asset),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(graphics.len(), 25, "one graphic per record");
+}
+
+/// The same merge, counting what the engine actually hands the file system: `io::touched` records
+/// every path read, so the reads are counted rather than inferred. The document already shares
+/// identical asset bytes, so memory alone does not show this -- the I/O is the bug (#1595).
+#[test]
+fn a_merge_reads_a_repeated_image_once_not_once_per_record() {
+    let dir = scratch("image-read-count");
+    let data = dir.join("data");
+    std::fs::create_dir_all(&data).unwrap();
+    let shared = dir.join("shared.png");
+    write_png(&shared);
+    let rows: String = (0..25).map(|_| format!("{}\n", shared.display())).collect();
+    write(&data.join("many.csv"), &format!("@Photo\n{rows}"));
+
+    let mut s = Session::new();
+    new_doc(&mut s, 1);
+    exec(&mut s, "data.source.select", json!({"path": data.join("many.csv")}));
+    let frame = exec(&mut s, "frame.create", json!({"rect": [36.0, 36.0, 80.0, 80.0], "content": "graphic"}));
+    exec(&mut s, "data.placeholder.add", json!({"field": "Photo", "item": frame["id"]}));
+
+    let (r, seen) = crate::io::touched(|| exec(&mut s, "data.merge", json!({})));
+    assert_eq!(r["records"], 25);
+    let reads = seen.iter().filter(|p| p.ends_with("shared.png")).count();
+    assert!(reads <= 2, "the image is read once for the merge, not once per record: {reads} reads for 25 records");
+}
+
 #[test]
 fn images_resolve_in_order_and_roles_win() {
     let dir = scratch("images");