From e94128de85391e3f0bececf8313f0f139f7ea177 Mon Sep 17 00:00:00 2001
From: "R. Tyler Croy" <rtyler@brokenco.de>
Date: Fri, 12 May 2023 10:34:42 -0700
Subject: [PATCH] Switch checkpoints over to use the new Decoder API

---
 rust/src/checkpoints.rs | 21 ++++-----------------
 1 file changed, 4 insertions(+), 17 deletions(-)

diff --git a/rust/src/checkpoints.rs b/rust/src/checkpoints.rs
index 1fb1b694fc..534e00c400 100644
--- a/rust/src/checkpoints.rs
+++ b/rust/src/checkpoints.rs
@@ -14,7 +14,6 @@ use regex::Regex;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::convert::TryFrom;
-use std::io::Write;
 use std::iter::Iterator;
 use std::ops::Add;
 
@@ -393,23 +392,11 @@ fn parquet_bytes_from_state(state: &DeltaTableState) -> Result<bytes::Bytes, Che
     let mut decoder = ReaderBuilder::new(arrow_schema)
         .with_batch_size(CHECKPOINT_RECORD_BATCH_SIZE)
         .build_decoder()?;
+    let jsons: Vec<serde_json::Value> = jsons.map(|r| r.unwrap()).collect();
+    decoder.serialize(&jsons)?;
 
-    let mut buf = vec![];
-    for res in jsons {
-        let json = res?;
-        buf.write_all(serde_json::to_string(&json)?.as_bytes())?;
-    }
-    let mut consumed = 0;
-
-    loop {
-        let read_bytes = decoder.decode(&buf)?;
-        consumed += read_bytes;
-        if let Some(batch) = decoder.flush()? {
-            writer.write(&batch)?;
-        }
-        if consumed == buf.len() {
-            break;
-        }
+    while let Some(batch) = decoder.flush()? {
+        writer.write(&batch)?;
     }
 
     let _ = writer.close()?;