Function prepare_flat_mem_stream 

pub(crate) fn prepare_flat_mem_stream(
    context: Arc<StreamContext>,
    stream: impl Stream<Item = Result<RecordBatch>> + Send + 'static,
) -> BoxedRecordBatchStream

Aligns memtable JSON2 columns to query types before merge/dedup.

For append-only queries with TimeSeriesRowSelector::LastRow { after_merge: false }, first keeps all rows tied at each series’ maximum timestamp within each sorted PUT-only batch, avoiding JSON2 conversion for discarded rows. Final cross-source LastRow selection still runs. Batches containing deletes are not reduced. This optimization uses the query’s series row selector, not its merge mode; non-append tables retain all input rows for deduplication and last-non-null backfill.