dbms: Compression: fixed.

dbms/src/CompressedInputStream.cpp

@@ -59,11 +59,17 @@ int DecompressingStreamBuf::readFromDevice(char * buffer, std::streamsize length
 			pos_in_buffer = 0;
 
 			if (!p_istr->good())
+			{
+				p_istr = 0;
 				return bytes_processed;
+			}
 		}
 
-		size_t bytes_to_copy = std::min(uncompressed_buffer.size() - pos_in_buffer, static_cast<size_t>(length));
-		memcpy(buffer, &uncompressed_buffer[0], bytes_to_copy);
+		size_t bytes_to_copy = std::min(
+			uncompressed_buffer.size() - pos_in_buffer,
+			static_cast<size_t>(length) - bytes_processed);
+		
+		memcpy(buffer + bytes_processed, &uncompressed_buffer[pos_in_buffer], bytes_to_copy);
 		pos_in_buffer += bytes_to_copy;
 		bytes_processed += bytes_to_copy;
 	}

dbms/src/CompressedOutputStream.cpp

@@ -57,16 +57,21 @@ int CompressingStreamBuf::writeToDevice(const char * buffer, std::streamsize len
 
 	while (bytes_processed < static_cast<size_t>(length))
 	{
-		size_t bytes_to_copy = std::min(uncompressed_buffer.size() - pos_in_buffer, static_cast<size_t>(length));
-		memcpy(&uncompressed_buffer[0], buffer, bytes_to_copy);
+		size_t bytes_to_copy = std::min(
+			uncompressed_buffer.size() - pos_in_buffer,
+			static_cast<size_t>(length) - bytes_processed);
+		memcpy(&uncompressed_buffer[pos_in_buffer], buffer + bytes_processed, bytes_to_copy);
 		pos_in_buffer += bytes_to_copy;
 		bytes_processed += bytes_to_copy;
-		
+
 		if (pos_in_buffer == uncompressed_buffer.size())
 			writeCompressedChunk();
 
 		if (!p_ostr->good())
+		{
+			p_ostr = 0;
 			return bytes_processed;
+		}
 	}
 
 	return static_cast<int>(length);

dbms/src/tests/compression.cpp

@@ -0,0 +1,66 @@
+#include <iostream>
+#include <fstream>
+
+#include <Poco/Stopwatch.h>
+#include <Poco/Exception.h>
+#include <Poco/NumberFormatter.h>
+
+#include <DB/CompressedOutputStream.h>
+#include <DB/CompressedInputStream.h>
+
+
+int main(int argc, char ** argv)
+{
+	try
+	{
+		int n = 1024 * 1024 * 10;
+		std::string str = "123456789";
+		Poco::Stopwatch stopwatch;
+
+		std::ofstream out_f("test");
+		DB::CompressedOutputStream compressor(out_f);
+		
+		{
+			stopwatch.restart();
+
+			for (int i = 0; i < n; ++i)
+				compressor << str << std::endl;
+
+			compressor.close();
+			out_f.close();
+
+			stopwatch.stop();
+			std::cout << "Compressing: " << static_cast<double>(stopwatch.elapsed()) / 1000000 << std::endl;
+		}
+
+		std::ifstream in_f("test");
+		DB::CompressedInputStream decompressor(in_f);
+
+		int i = 0;
+		std::string read_str;
+
+		{
+			stopwatch.restart();
+
+			while (std::getline(decompressor, read_str))
+			{
+				++i;
+				/*if (read_str != str)
+					throw Poco::Exception(read_str);*/
+			}
+
+			stopwatch.stop();
+			std::cout << "Decompressing: " << static_cast<double>(stopwatch.elapsed()) / 1000000 << std::endl;
+		}
+		
+		if (i != n)
+			throw Poco::Exception(Poco::NumberFormatter::format(i));
+	}
+	catch (Poco::Exception & e)
+	{
+		std::cerr << e.message() << std::endl;
+		return 1;
+	}
+	
+	return 0;
+}