Please rename the option name to something more specific.

Changed this option to parallel_threads.

Do we need to when we already have the previous option?

Removed this since parallel_threads is sufficient.

Please copy&past the header from another file.

Copied RocksDB header for work_queue.h.

Are these source code copied from somewhere else? If not I suggest you keep the comment format similar to other code: "//" for each line.

The code was copied from facebook/zstd.

Can you put this information in a code comment? Maybe after line 27.

Added. Also add similar information in work_queue_test.cc.

Was the class copied from somewhere? If it is from a proven library then fine. If not, we probably should at least add some unit tests. It's also worth thinking whether we can simplify it but still satisfy our performance requirement.

Added a unit test for work_queue.h, since there is an out-of-box unit test from facebook/zstd. Currently all methods in the class are used except waitUntilFinished, so maybe we could keep the class WorkQueue as it was in zstd repo.

Do we need to keep it a subclass of BlockBasedTableBuilder, or can we define a separate class for it? It feels that the relationship between the two classes are quite loose.

ParallelCompressionRep was designed as a helper class for parallel compression in block based table only. ParallelCompressionRep will only be used by BlockBasedTableBuilder, so I made it a private inner class so that it's not visible to other code, including those who used block_based_table_builder.h.

Can you keep the coding convention? You can run "make format" to reformat it. I tried it and it also work with Ubuntu subsystem on Windows. Let me know if you want me to run it for you and give you a patch for formatting.

I've checked the make format results in latest commits. Sorry for the inconvenience.

Can you put this information in a code comment? Maybe after line 27.

If my understanding is correct, the function is a function used by compression threads. I think we should try to think about a better name. Right now, it's a little bit hard for me to imaging that it is a long running function that keeps taking work item from a queue and process them until it is signaled to finish. Maybe including something like "thread". Or keeping the convention of some parts of the code, prefix "Bg" in the function name.

Rename WriteBlocks to BGWorkCompression.

Similar to WriteBlocks(). If my understand is correct, this is the function used by the block writing thread. Can you try to think about a better name? Something like including "thread" in the function name or something like that.

Rename WriteRawBlocks to BGWorkWriteRawBlock. Get rid of "s" suffix to keep aligned with original WriteRawBlock.

I only see a unit test for work queue, but I didn't see unit test that builds SST files with parallel compression. Maybe I missed something. Also it is preferred to have at least one unit test that covers the whole flow: from setting the option, generating SST files, and read them back and check the values are correct.

Sorry I missed unit tests in former versions.
I leveraged existing RandomizedHarnessTest in table/table_test.cc as basic unit tests. They now also check cases where parallel compression is enabled.
I also enabled DBBasicTestWithParallelIO in db/db_basic_test.cc and DBTest2::CompressionOptions in db/db_test2.cc to check parallel compression cases.
These tests should all cover the whole flow.

Here is a problem: without parallel compression, inside this function call, rep_.data_begin_offset is updated, so that in BlockBasedTableBuilder::Add() we can determine the size of the file reaches limit so the file can terminate. But now, we don't know it until the background threads finished compressing the blocks.

I don't know a good way to solve the problem. Can we estimate the size and terminate the file in BlockBasedTableBuilder::Add()?

Also, can compression queue can be pointer that points to the object in the pool instead?

Here is a problem: without parallel compression, inside this function call, rep_.data_begin_offset is updated, so that in BlockBasedTableBuilder::Add() we can determine the size of the file reaches limit so the file can terminate. But now, we don't know it until the background threads finished compressing the blocks.

I don't know a good way to solve the problem. Can we estimate the size and terminate the file in BlockBasedTableBuilder::Add()?

rep_.data_begin_offset will only be increased in kBuffered state, where parallel compression code path is not involved. I think that variable tracks raw size instead of compressed size, in a synchronized and single-threaded way. Only when the state is changed to kUnbuffered, code related to parallel compression is executed. As a result, code related to rep_.data_begin_offset should work well with parallel compression enabled.

There is such a problem in ProcessKeyValueCompaction in compaction_job.cc though, where current_output_file_size is updated by builder's FileSize(). However, maximum blocks inflight is bounded by number of compression threads (I'll explain that in another comment). As a result, the file size with parallel compression is bounded by original_file_size + compressed_block_size * number_of_compression_threads. I think it's acceptable in most cases.

Also, can compression queue can be pointer that points to the object in the pool instead?

I make all WorkQueue's point to pointers in the pool. Now block_rep_pool_, compress_queue_ and write_queue_ contain references (pointers) to data in block_rep_buf_.

Add SST size estimation based on historical_compression_ratio * bytes_under_compression.

We follow Google C++ Style and class member's naming convention is keys_ptr_: https://google.github.io/styleguide/cppguide.html#Variable_Names

Fixed.

Maybe I missed something but I didn't see maxSzie_ ever set in the queues used. Should we set something? Because the writer thread can be significantly faster than the background compression thread, without a limit to the queue size, we can end up with unlimited raw block in the queue, which consumes memory and make it harder to estimate file size, but does not help with anything.

The maximum size of queue, i.e. inflight blocks, was implicitly ensured because we have a determined number of BlockRep's, which is equal to number of compression threads. Each time we want to emit a block to compression, we have to fetch a BlockRep from its pool. I make this more explicit by adding setMaxSize calls in ParallelCompressionRep constructor.

Change setMaxSize to initialization in initialization list to keep the same convention as BlockBasedTableBuilder::Rep.

Is it possible to make it std::unique_ptr so that we don't have to do the cleaning up with delete?

Fixed.

Can you explain why the data structure is a WorkQueue and not a normal vector or deque?

The BlockRep pool will be pushed by writer thread and popped by block building thread concurrently. This is to reuse memory and keep a determined number of inflight compression payloads. As a result, the pool has to be thread-safe. More comments are added around BlockRep definition for this.

I don't think we need ptr in those variable names. It's clear they are pointers because their types are pointers.

Fixed.

Rather than delete them, can we make them unique_ptr?

Fixed.

I'm a little bit confused here. We moved block_rep.slot_ptr, but we still seem to continue using and cleaning up this pointer. It seems to be contradicting. Do we need std::move() here?

WorkQueue's push method only accepts r-value by design. We have to use std::move to wrap the pointer, but the data referenced by pointer is not moved, only pointer value, i.e. a scalar, is "moved". Actually, the pointer value is just copied into the queue.

slot_ptr is now made a unique_ptr, and unique_ptr::get() returns a r-value itself, so std::move is not necessary. But std::move for block_rep's is still needed, because it's a l-value.

This std::move() is confusing to me too. I'm not sure about the behavior we want for block_rep after the move.
My understanding is that, we want to reuse those allocated memory for keys, strings, first_key_in_next_block_ptr, etc. If that is the case, can we be more explicit here? If block_rep_pool keeps holding the ownership to all those objects pointed by those pointers, I don't think we should do std::move() here.

Either way, please add comments somewhere to explain the ownership for those objects.

When block_rep is a struct, std::move will only move its value (including pointers), but not the data it references. Now block_rep is a pointer, std::move will only move its value as well. I've added comments for object ownership around BlockRep definition and WorkQueue variable definitions.

Moving a pointer is confusing too. My vote would be to move away from this move.

Modified WorkQueue design so it copies elements instead of moves them. As long as we avoid passing large elements directly to WorkQueue in future, this should be fine. We can always pass large objects by pointers.
This behavior is more similar to STL queue, and maybe less misleading.

More demostrations about SST file inflation added.

Ideally we need a unit test that validates the data in the database. That is the keys are expected and not lost during the compaction process.

Added consistency check between data written and data in the database. This should benefit original DBTest2::CompressionOptions test as well.

This is a public header so ideally mentioning of internal function like BlockBasedTableBuilder::EstimatedFileSize() is not recommended. Imagine the readers of header files under include/rocksdb/ are RocksDB users who don't read source code. I think the mentioning of internal functions here can be removed.

Rewrote the option demonstration without code details.

Moving a pointer is confusing too. My vote would be to move away from this move.

We follow Google C++ Style, which bans C style casting: https://google.github.io/styleguide/cppguide.html#Casting try to use a C++ style casting instead.

Fixed.

We follow Google C++ Style, which bans C style casting: https://google.github.io/styleguide/cppguide.html#Casting try to use a C++ style casting instead.

Fixed.

I think if r->status is OK, we should exit the loop and avoid calling WriteRawBlock() again. I believe inside WriteRawBlock() we only assert status OK and in release mode we would just override the status, which can cause problem. I think it's safer to just exit the loop if r->status.

Ideally the failure case is tested.

Sorry for this late comment. I should have noticed it earlier. But this is serious and we have to fix it before commit the feature.

Sure. Did you mean exiting the loop when r->status is NOT OK?

Yes. Sorry for the typo.

Also, there seems to be a data race between the main thread and the block writer thread with rep_->status().

Those variables added to estimate file size might also have a data race.

For variables to estimate file size,

raw_bytes_submitted will only be updated and accessed in block building thread, should be safe;

raw_bytes_compressed and curr_compression_ratio will be updated in writer thread, and accessed in block building thread. Their updates should be single-threaded. There might be the case where r->offset is already updated, but raw_bytes_compressed and curr_compression_ratio are not updated yet, causing an estimation error of 1-compressed-block-size. Shall we make these updates atomic, or you think it's acceptable?

I noticed that, in orignal BlockBasedTableBuilder, when compression is aborted, rep_->status is set to Status::Corruption, but WriteRawBlock is still called. Is WriteRawBlock meant to write uncompressed data here, or we should return before WriteRawBlock is called?