fix: ensure the checkpoint decoder is regularly flushed

For checkpoint buffers that cannot fit into the batch size the
checkpoint will be written with an insufficient number of bytes.

Unfortunately our tests didn't catch this and it only manifested on
tables with very large transaction logs

Signed-off-by: R. Tyler Croy <[email protected]>

crates/core/src/protocol/checkpoints.rs

@@ -356,11 +356,12 @@ fn parquet_bytes_from_state(
     for j in jsons {
         let buf = serde_json::to_string(&j?).unwrap();
         let _ = decoder.decode(buf.as_bytes())?;
+
+        while let Some(batch) = decoder.flush()? {
+            writer.write(&batch)?;
+        }
         total_actions += 1;
     }
-    while let Some(batch) = decoder.flush()? {
-        writer.write(&batch)?;
-    }
 
     let _ = writer.close()?;
     debug!("Finished writing checkpoint parquet buffer.");