thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-24 08:32:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

dbms: fixed error with multiple INSERTs in StripeLog [#METR-19901].

Browse Source
This commit is contained in:
Alexey Milovidov 2016-02-02 23:33:03 +03:00
parent 06f346bfd5
commit 038a6278d5
5 changed files with 64 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dbms/include/DB/DataStreams/NativeBlockOutputStream.h
View File

@ -14,6 +14,7 @@ class CompressedWriteBuffer;
* Предназначено для взаимодействия между серверами.
*
* Может быть указан поток для записи индекса. Индекс содержит смещения до каждого кусочка каждого столбца.
* Если делается append в уже существующий файл, и нужно записать индекс, то укажите initial_size_of_file.
*/
class NativeBlockOutputStream : public IBlockOutputStream
{
@ -23,7 +24,7 @@ public:
*/
NativeBlockOutputStream(
WriteBuffer & ostr_, UInt64 client_revision_ = 0,
WriteBuffer * index_ostr_ = nullptr);
WriteBuffer * index_ostr_ = nullptr, size_t initial_size_of_file_ = 0);
void write(const Block & block) override;
void flush() override { ostr.next(); }
@ -37,6 +38,7 @@ private:
UInt64 client_revision;
WriteBuffer * index_ostr;
size_t initial_size_of_file; /// Начальный размер файла с данными, если делается append. Используется для индекса.
/// Если требуется записывать индекс, то ostr обязан быть CompressedWriteBuffer.
CompressedWriteBuffer * ostr_concrete = nullptr;
};

6
dbms/src/DataStreams/NativeBlockOutputStream.cpp
View File

@ -19,9 +19,9 @@ namespace DB
NativeBlockOutputStream::NativeBlockOutputStream(
WriteBuffer & ostr_, UInt64 client_revision_,
WriteBuffer * index_ostr_)
WriteBuffer * index_ostr_, size_t initial_size_of_file_)
: ostr(ostr_), client_revision(client_revision_),
index_ostr(index_ostr_)
index_ostr(index_ostr_), initial_size_of_file(initial_size_of_file_)
{
if (index_ostr)
{
@ -112,7 +112,7 @@ void NativeBlockOutputStream::write(const Block & block)
if (index_ostr)
{
ostr_concrete->next(); /// Заканчиваем сжатый блок.
mark.offset_in_compressed_file = ostr_concrete->getCompressedBytes();
mark.offset_in_compressed_file = initial_size_of_file + ostr_concrete->getCompressedBytes();
mark.offset_in_decompressed_block = ostr_concrete->getRemainingBytes();
}

6
dbms/src/Storages/StorageStripeLog.cpp
View File

@ -118,11 +118,11 @@ class StripeLogBlockOutputStream : public IBlockOutputStream
public:
StripeLogBlockOutputStream(StorageStripeLog & storage_)
: storage(storage_), lock(storage.rwlock),
data_out_compressed(storage.full_path() + "data.bin"),
data_out_compressed(storage.full_path() + "data.bin", DBMS_DEFAULT_BUFFER_SIZE, O_WRONLY | O_APPEND | O_CREAT),
data_out(data_out_compressed, CompressionMethod::LZ4, storage.max_compress_block_size),
index_out_compressed(storage.full_path() + "index.mrk", INDEX_BUFFER_SIZE),
index_out_compressed(storage.full_path() + "index.mrk", INDEX_BUFFER_SIZE, O_WRONLY | O_APPEND | O_CREAT),
index_out(index_out_compressed),
block_out(data_out, 0, &index_out)
block_out(data_out, 0, &index_out, Poco::File(storage.full_path() + "data.bin").getSize())
{
}

18
dbms/tests/queries/0_stateless/00299_stripe_log_multiple_inserts.reference Normal file
View File

@ -0,0 +1,18 @@
0
0
1
0
1
2
0
0
1
0
1
2
0
0
1
0
1
2

37
dbms/tests/queries/0_stateless/00299_stripe_log_multiple_inserts.sql Normal file
View File

@ -0,0 +1,37 @@
DROP TABLE IF EXISTS test.log;
CREATE TABLE test.log (x UInt8) ENGINE = StripeLog;
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (0);
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (1);
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (2);
SELECT * FROM test.log ORDER BY x;
DROP TABLE test.log;
CREATE TABLE test.log (x UInt8) ENGINE = TinyLog;
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (0);
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (1);
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (2);
SELECT * FROM test.log ORDER BY x;
DROP TABLE test.log;
CREATE TABLE test.log (x UInt8) ENGINE = Log;
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (0);
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (1);
SELECT * FROM test.log ORDER BY x;
INSERT INTO test.log VALUES (2);
SELECT * FROM test.log ORDER BY x;
DROP TABLE test.log;