compress large attachments on demand (#647)

- Improvement: Speed up oplog export by fetching/compressing big attachments on demand
- Improvement: Speed up oplog export by batch-fetcing small attachments
- Improvement: Speed up oplog import by batching writes of oplog ops
- Improvement: Tweak oplog export default block size and embed size limit
- Improvement: Add more messaging and progress during oplog import/export

2 files changed, 11 insertions, 28 deletions

diff --git a/src/zencore/compositebuffer.cpp b/src/zencore/compositebuffer.cpp
index f29f6e810..583ef19c6 100644
--- a/src/zencore/compositebuffer.cpp
+++ b/src/zencore/compositebuffer.cpp
@@ -155,7 +155,11 @@ CompositeBuffer::ViewOrCopyRange(Iterator& It, uint64_t Size, UniqueBuffer& Copy
 	// A hot path for this code is when we call CompressedBuffer::FromCompressed which
 	// is only interested in reading the header (first 64 bytes or so) and then throws
 	// away the materialized data.
-	MutableMemoryView WriteView;
+	if (CopyBuffer.GetSize() < Size)
+	{
+		CopyBuffer = UniqueBuffer::Alloc(Size);
+	}
+	MutableMemoryView WriteView	   = CopyBuffer.GetMutableView();
 	size_t			  SegmentCount = m_Segments.size();
 	ZEN_ASSERT(It.SegmentIndex < SegmentCount);
 	uint64_t SizeLeft = Size;
@@ -163,31 +167,10 @@ CompositeBuffer::ViewOrCopyRange(Iterator& It, uint64_t Size, UniqueBuffer& Copy
 	{
 		const SharedBuffer& Segment		= m_Segments[It.SegmentIndex];
 		size_t				SegmentSize = Segment.GetSize();
-		if (Size == SizeLeft && Size <= (SegmentSize - It.OffsetInSegment))
-		{
-			IoBuffer   SubSegment(Segment.AsIoBuffer(), It.OffsetInSegment, SizeLeft);
-			MemoryView View = SubSegment.GetView();
-			It.OffsetInSegment += SizeLeft;
-			ZEN_ASSERT_SLOW(It.OffsetInSegment <= SegmentSize);
-			if (It.OffsetInSegment == SegmentSize)
-			{
-				It.SegmentIndex++;
-				It.OffsetInSegment = 0;
-			}
-			return View;
-		}
-		if (WriteView.GetSize() == 0)
-		{
-			if (CopyBuffer.GetSize() < Size)
-			{
-				CopyBuffer = UniqueBuffer::Alloc(Size);
-			}
-			WriteView = CopyBuffer.GetMutableView();
-		}
-		size_t	   CopySize = zen::Min(SegmentSize - It.OffsetInSegment, SizeLeft);
-		IoBuffer   SubSegment(Segment.AsIoBuffer(), It.OffsetInSegment, CopySize);
-		MemoryView ReadView = SubSegment.GetView();
-		WriteView			= WriteView.CopyFrom(ReadView);
+		size_t				CopySize	= zen::Min(SegmentSize - It.OffsetInSegment, SizeLeft);
+		IoBuffer			SubSegment(Segment.AsIoBuffer(), It.OffsetInSegment, CopySize);
+		MemoryView			ReadView = SubSegment.GetView();
+		WriteView					 = WriteView.CopyFrom(ReadView);
 		It.OffsetInSegment += CopySize;
 		ZEN_ASSERT_SLOW(It.OffsetInSegment <= SegmentSize);
 		if (It.OffsetInSegment == SegmentSize)
diff --git a/src/zencore/include/zencore/stream.h b/src/zencore/include/zencore/stream.h
index a9d35ef1b..a28290041 100644
--- a/src/zencore/include/zencore/stream.h
+++ b/src/zencore/include/zencore/stream.h
@@ -34,8 +34,8 @@ public:
 
 	inline const uint8_t* Data() const { return m_Buffer.data(); }
 	inline const uint8_t* GetData() const { return m_Buffer.data(); }
-	inline uint64_t		  Size() const { return m_Buffer.size(); }
-	inline uint64_t		  GetSize() const { return m_Buffer.size(); }
+	inline uint64_t		  Size() const { return m_Offset; }
+	inline uint64_t		  GetSize() const { return m_Offset; }
 	void				  Reset();
 
 	inline MemoryView		 GetView() const { return MemoryView(m_Buffer.data(), m_Offset); }