thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-20 08:40:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simpliest codec multiple works

Browse Source
This commit is contained in:
alesapin 2018-12-18 16:08:22 +03:00
parent 6fe01af829
commit fd49cc3c34
3 changed files with 90 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
dbms/src/Compression/CompressionCodecMultiple.cpp
View File

@ -2,10 +2,17 @@
#include <IO/CompressedStream.h>
#include <common/unaligned.h>
#include <Compression/CompressionFactory.h>
#include <IO/ReadHelpers.h>
#include <IO/WriteHelpers.h>
namespace DB
{
namespace ErrorCodes
{
extern const int UNKNOWN_CODEC;
extern const int CORRUPTED_DATA;
}
CompressionCodecMultiple::CompressionCodecMultiple(Codecs codecs)
: codecs(codecs)
@ -43,58 +50,119 @@ size_t CompressionCodecMultiple::getCompressedReserveSize(size_t uncompressed_si
size_t CompressionCodecMultiple::compress(char * source, size_t source_size, char * dest)
{
static constexpr size_t without_method_header_size = sizeof(UInt32) + sizeof(UInt32);
static constexpr size_t header_for_size_store = sizeof(UInt32) + sizeof(UInt32);
PODArray<char> compressed_buf;
PODArray<char> uncompressed_buf(source_size);
uncompressed_buf.insert(source, source + source_size);
PODArray<char> uncompressed_buf(source, source + source_size);
dest[0] = static_cast<char>(getMethodByte());
dest[1] = static_cast<char>(codecs.size());
std::cerr << "(compress) codecs size:" << codecs.size() << std::endl;
std::cerr << "(compress) desc:" << codec_desc << std::endl;
//std::cerr << "(compress) codecs size:" << codecs.size() << std::endl;
//std::cerr << "(compress) desc:" << codec_desc << std::endl;
//std::cerr << "(compress) SOURCE WRITE:\n";
//for (size_t i = 0 ; i < source_size; ++i) {
// std::cerr << std::hex << +source[i] << std::endl;
//}
//std::cerr << "(compress) UNCOMPRESSED BUF:\n";
//for(size_t i = 0; i < source_size; ++i)
// std::cerr << std::hex << +uncompressed_buf[i] << " ";
//std::cerr << std::dec << std::endl;
size_t codecs_byte_pos = 2;
for (size_t idx = 0; idx < codecs.size(); ++idx, ++codecs_byte_pos)
{
const auto codec = codecs[idx];
dest[codecs_byte_pos] = codec->getMethodByte();
compressed_buf.resize(without_method_header_size + codec->getCompressedReserveSize(source_size));
//String name;
//codec->getCodecDesc(name);
//std::cerr << std::hex << "(compress) Compression method byte:" << +codec->getMethodByte() << " name:" << name << std::endl;
compressed_buf.resize(header_for_size_store + codec->getCompressedReserveSize(source_size));
size_t size_compressed = without_method_header_size;
size_compressed += codec->compress(&uncompressed_buf[0], source_size, &compressed_buf[without_method_header_size]);
size_t size_compressed = header_for_size_store;
size_compressed += codec->compress(&uncompressed_buf[0], source_size, &compressed_buf[header_for_size_store]);
UInt32 compressed_size_32 = size_compressed;
UInt32 uncompressed_size_32 = source_size;
//std::cerr << std::dec << "(compress) Compressed size:" << compressed_size_32 << std::endl;
//std::cerr << std::dec << "(compress) Uncompressed size:" << uncompressed_size_32 << std::endl;
unalignedStore(&compressed_buf[0], compressed_size_32);
unalignedStore(&compressed_buf[4], uncompressed_size_32);
//std::cerr << "(compress) COMPRESSED BUF:\n";
//for(size_t i = header_for_size_store; i < size_compressed; ++i)
// std::cerr << std::hex << +compressed_buf[i] << " ";
//std::cerr << std::dec << std::endl;
uncompressed_buf.swap(compressed_buf);
source_size = size_compressed;
}
memcpy(&dest[2 + codecs.size()], &compressed_buf[0], source_size);
return source_size;
memcpy(&dest[2 + codecs.size()], &uncompressed_buf[0], source_size);
//std::cerr << std::dec;
//std::cerr << "(compress) WRITING BUF:" << 2 + codecs.size() + source_size << std::endl;
//for(size_t i = 0; i < 2 + codecs.size() + source_size; ++i)
// std::cerr << std::hex << +dest[i] << " ";
//std::cerr << std::endl;
return 2 + codecs.size() + source_size;
}
size_t CompressionCodecMultiple::decompress(char * source, size_t source_size, char * dest, size_t decompressed_size)
{
static constexpr size_t header_for_size_store = sizeof(UInt32) + sizeof(UInt32);
if (source[0] != getMethodByte())
throw Exception("Incorrect compression method for codec multiple, given " + toString(source[0]) + ", expected " + toString(getMethodByte()),
ErrorCodes::UNKNOWN_CODEC);
UInt8 compression_methods_size = source[1];
std::cerr << "(decompress) Methods size:" << +compression_methods_size << std::endl;
//std::cerr << std::dec;
//std::cerr << "(decompress) Source size:" << source_size << std::endl;
//std::cerr << "(decompress) decompressed size:" << decompressed_size << std::endl;
//std::cerr << "(decompress) Methods size:" << +compression_methods_size << std::endl;
//std::cerr << "READING BUF:" << std::dec << source_size << std::endl;
//for(size_t i = 0; i < source_size; ++i)
// std::cerr << std::hex << +source[i] << " ";
//std::cerr << std::endl;
PODArray<char> compressed_buf;
PODArray<char> uncompressed_buf;
compressed_buf.insert(&source[compression_methods_size + 2], &source[source_size - (compression_methods_size + 2)]);
/// Insert all data into compressed buf
compressed_buf.insert(&source[compression_methods_size + 2], &source[source_size]);
for (size_t idx = 0; idx < compression_methods_size; ++idx)
for (long idx = compression_methods_size - 1; idx >= 0; --idx)
{
UInt8 compression_method = source[idx + 2];
std::cerr << std::hex;
std::cerr << "(decompress) Compression method byte:" << +compression_method << std::endl;
const auto codec = CompressionCodecFactory::instance().get(compression_method);
codec->decompress(&compressed_buf[8], 0, uncompressed_buf.data(), 0);
//String name;
//codec->getCodecDesc(name);
//std::cerr << std::hex << "(decompress) Compression method byte:" << +compression_method << " name:" << name << std::endl;
UInt32 compressed_size = static_cast<UInt32>(compressed_buf[0]);
UInt32 uncompressed_size = static_cast<UInt32>(compressed_buf[4]);
if (idx == 0 && uncompressed_size != decompressed_size)
throw Exception("Wrong final decompressed size in codec Multiple, got " + toString(uncompressed_size) + ", expected " + toString(decompressed_size), ErrorCodes::CORRUPTED_DATA);
uncompressed_buf.resize(uncompressed_size);
//std::cerr << std::dec << "(decompress) expected compressed size:" << compressed_size << std::endl;
//std::cerr << std::dec << "(decompress) expected uncompressed size:" << uncompressed_size << std::endl;
//std::cerr << "COMPRESSED BUF:\n";
//for(size_t i = header_for_size_store; i < compressed_size; ++i)
// std::cerr<< std::hex << +compressed_buf[i] << " ";
//std::cerr << std::dec << std::endl;
codec->decompress(&compressed_buf[header_for_size_store], compressed_size - header_for_size_store, &uncompressed_buf[0], uncompressed_size);
//std::cerr << "FINAL UNCOMPRESSED:\n";
//for (size_t i = 0; i < uncompressed_size; ++i) {
// std::cerr << std::hex << +uncompressed_buf[i] << std::endl;
//}
uncompressed_buf.swap(compressed_buf);
}
//std::cerr << "FINISHED\n";
memcpy(dest, uncompressed_buf.data(), decompressed_size);
memcpy(dest, compressed_buf.data(), decompressed_size);
return decompressed_size;
}

4
dbms/src/Compression/CompressionCodecNone.cpp
View File

@ -22,7 +22,7 @@ size_t CompressionCodecNone::compress(char * source, size_t source_size, char *
return source_size;
}
size_t CompressionCodecNone::decompress(char *source, size_t /*source_size*/, char *dest, size_t size_decompressed)
size_t CompressionCodecNone::decompress(char * source, size_t /*source_size*/, char * dest, size_t size_decompressed)
{
memcpy(dest, source, size_decompressed);
return size_decompressed;
@ -35,4 +35,4 @@ void registerCodecNone(CompressionCodecFactory & factory)
});
}
}
}

10
dbms/tests/integration/test_odbc_interaction/configs/dictionaries/postgres_odbc_hashed_dictionary.xml
View File

@ -13,19 +13,15 @@
<max>5</max>
</lifetime>
<layout>
<hashed />
<complex_key_hashed />
</layout>
<structure>
<id>
<name>column1</name>
</id>
<attribute>
<key>
<name>column1</name>
<type>Int64</type>
<null_value>1</null_value>
</attribute>
</key>
<attribute>
<name>column2</name>