dbms: development.

dbms/include/DB/IO/CompressedReadBuffer.h

@@ -1,6 +1,7 @@
 #ifndef DBMS_COMMON_COMPRESSED_READBUFFER_H
 #define DBMS_COMMON_COMPRESSED_READBUFFER_H
 
+#include <vector>
 #include <algorithm>
 
 #include <quicklz/quicklz_level1.h>
@@ -21,7 +22,7 @@ private:
 
 	std::vector<char> compressed_buffer;
 	std::vector<char> decompressed_buffer;
-	char scratch[QLZ_SCRATCH_COMPRESS];
+	char scratch[QLZ_SCRATCH_DECOMPRESS];
 
 	size_t pos_in_buffer;
 
@@ -38,34 +39,54 @@ public:
 	{
 		in.readStrict(&compressed_buffer[0], QUICKLZ_HEADER_SIZE);
 
-		size_t size_compressed = qlz_size_compressed(internal_buffer);
-		size_t size_decompressed = qlz_size_decompressed(internal_buffer);
+		size_t size_compressed = qlz_size_compressed(&compressed_buffer[0]);
+		size_t size_decompressed = qlz_size_decompressed(&compressed_buffer[0]);
 
 		compressed_buffer.resize(size_compressed);
 		decompressed_buffer.resize(size_decompressed);
 
+		std::cerr << size_compressed << ", " << size_decompressed << std::endl;
+		
 		in.readStrict(&compressed_buffer[QUICKLZ_HEADER_SIZE], size_compressed - QUICKLZ_HEADER_SIZE);
 
+		std::cerr << "#" << std::endl;
+
+		qlz_decompress(&compressed_buffer[0], &decompressed_buffer[0], &scratch[0]);
+
+		std::cerr << "##" << std::endl;
+
 		pos_in_buffer = 0;
 	}
 
 	bool next()
 	{
+		std::cerr << "?" << std::endl;
+	
 		if (pos_in_buffer == decompressed_buffer.size())
 		{
+			std::cerr << "??" << std::endl;
+		
 			if (in.eof())
 				return false;
 
 			readCompressedChunk();
+
+			std::cerr << "!" << std::endl;
 		}
 	
 		size_t bytes_to_copy = std::min(decompressed_buffer.size() - pos_in_buffer,
 			static_cast<size_t>(DEFAULT_READ_BUFFER_SIZE));
 		std::memcpy(internal_buffer, &decompressed_buffer[pos_in_buffer], bytes_to_copy);
 
+		std::cerr << "!!" << std::endl;
+
+		pos_in_buffer += bytes_to_copy;
 		pos = internal_buffer;
 		working_buffer = Buffer(internal_buffer, internal_buffer + bytes_to_copy);
 
+	/*	std::cerr.write(internal_buffer, bytes_to_copy);
+		std::cerr << std::endl;*/
+
 		return true;
 	}
 };

dbms/src/IO/tests/compressed_buffer.cpp

@@ -0,0 +1,82 @@
+#include <string>
+
+#include <iostream>
+#include <fstream>
+
+#include <DB/Core/Types.h>
+#include <DB/IO/WriteBufferFromOStream.h>
+#include <DB/IO/ReadBufferFromIStream.h>
+#include <DB/IO/CompressedWriteBuffer.h>
+#include <DB/IO/CompressedReadBuffer.h>
+#include <DB/IO/WriteHelpers.h>
+#include <DB/IO/ReadHelpers.h>
+#include <DB/IO/CompressedInputStream.h>
+#include <DB/IO/CompressedOutputStream.h>
+
+
+int main(int argc, char ** argv)
+{
+	try
+	{
+		size_t n = 10000000;
+	
+		{
+			std::ofstream ostr("test1");
+			DB::WriteBufferFromOStream buf(ostr);
+			DB::CompressedWriteBuffer compressed_buf(buf);
+
+			for (size_t i = 0; i < n; ++i)
+				DB::writeIntText(i, compressed_buf);
+		}
+
+		{
+			std::ofstream ostr("test2");
+			DB::CompressedOutputStream compressed_ostr(ostr);
+			DB::WriteBufferFromOStream compressed_buf(compressed_ostr);
+
+			for (size_t i = 0; i < n; ++i)
+				DB::writeIntText(i, compressed_buf);
+		}
+
+		std::cerr << "Writing done." << std::endl;
+
+		/*{
+			std::ifstream istr("test1");
+			DB::ReadBufferFromIStream buf(istr);
+			DB::CompressedReadBuffer compressed_buf(buf);
+			std::string s;
+
+			for (size_t i = 0; i < n; ++i)
+			{
+				size_t x;
+				DB::readIntText(x, compressed_buf);
+				if (x != i)
+					throw DB::Exception("Failed!");
+			}
+		}*/
+
+		{
+			std::ifstream istr("test2");
+			DB::CompressedInputStream compressed_istr(istr);
+			DB::ReadBufferFromIStream compressed_buf(compressed_istr);
+			std::string s;
+
+			for (size_t i = 0; i < n; ++i)
+			{
+				size_t x;
+				DB::readIntText(x, compressed_buf);
+				if (x != i)
+					throw DB::Exception("Failed!");
+			}
+		}
+
+		std::cerr << "Reading done." << std::endl;
+	}
+	catch (const DB::Exception & e)
+	{
+		std::cerr << e.what() << ", " << e.message() << std::endl;
+		return 1;
+	}
+
+	return 0;
+}

dbms/src/Storages/tests/storage_log.cpp

@@ -39,7 +39,7 @@ int main(int argc, char ** argv)
 
 			vec1.resize(rows);
 			for (size_t i = 0; i < rows; ++i)
-				vec1[i] = 'z';
+				vec1[i] = i;
 
 			block.insert(column1);
 
@@ -51,7 +51,7 @@ int main(int argc, char ** argv)
 
 			vec2.resize(rows);
 			for (size_t i = 0; i < rows; ++i)
-				vec2[i] = 'x';
+				vec2[i] = i * 2;
 
 			block.insert(column2);