Add parallel compression for multithreaded process #711
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Member
jonathonmcmurray
left a comment
jonathonmcmurray
left a comment
There was a problem hiding this comment.
a few comments that are largely unrelated to your actual change but while we're modifying this file may as well clean it up a bit
code/common/compress.q
Outdated
Comment on lines
190
to
191
| /-log to the table if the algo wasn't 0 | ||
| statstab,:$[not 0=algo;(filetoCompress;algo;(-21!filetoCompress)`compressedLength;sizeuncomp);(filetoCompress;algo;comprL;sizeuncomp)]]; |
Member
There was a problem hiding this comment.
I think we should do this logging in the compressmaxage function rather than the cleancompressed - I think we then can reduce the args to cleancompressed and skip calcing comprL.
code/common/compress.q
Outdated
| statstab,:$[not 0=algo;(filetoCompress;algo;(-21!filetoCompress)`compressedLength;sizeuncomp);(filetoCompress;algo;comprL;sizeuncomp)]]; | ||
| [$[not count -21!compressedFile; | ||
| [.lg.o[`compression; "Failed to compress file ",string[filetoCompress]];hdel compressedFile]; | ||
| [.lg.o[`compression;cmp,"compressed ","file ",string[compressedFile]," doesn't match original. Deleting new file"];hdel compressedFile]]]]]} |
Member
There was a problem hiding this comment.
the number of ] on this line is horrifying - I think we should be able to refactor this function a bit to avoid the [...] block syntax inside conditionals and make this a bit easier to follow
e.g. something like
if[()~ key compressedFile;
.lg.o[`compression; "No compressed file present for the following file - ",string[filetoCompress];
:();
];
if[not ((get compressedFile)~sf:get filetoCompress) & (count -21!compressedFile) or algo=0;
.lg.o[`compression;cmp,"compressed ","file ",string[compressedFile]," doesn't match original. Deleting new file"];
hdel compressedFile;
:();
];
// rest of code to rename file etc.
code/common/compress.q
Outdated
Comment on lines
133
to
138
| $[0= system"s"; | ||
| [.lg.o[`compression; "Single threaded process, compress applied sequentially"]; | ||
| {compress[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize];cleancompressed[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} each table]; | ||
| [.lg.o[`compression; "Multithreaded process, compress applied in parallel "]; | ||
| {compress[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} peach table; | ||
| {cleancompressed[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} each table;]]} |
Member
There was a problem hiding this comment.
I think it'd be cleaner if we made this into two functions that can be called depending on whether proc is single threaded or not
Suggested change
| $[0= system"s"; | |
| [.lg.o[`compression; "Single threaded process, compress applied sequentially"]; | |
| {compress[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize];cleancompressed[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} each table]; | |
| [.lg.o[`compression; "Multithreaded process, compress applied in parallel "]; | |
| {compress[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} peach table; | |
| {cleancompressed[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} each table;]]} | |
| $[0= system"s";singlethreadcompress;multithreadcompress]table; | |
| } | |
| singlethreadcompress:{[table] | |
| .lg.o[`compression; "Single threaded process, compress applied sequentially"]; | |
| {compress[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]; | |
| cleancompressed[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} each table; | |
| } | |
| multithreadcompress:{[table] | |
| .lg.o[`compression; "Multithreaded process, compress applied in parallel "]; | |
| {compress[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} peach table; | |
| {cleancompressed[x `fullpath;x `calgo;x `cblocksize;x `clevel; x `currentsize]} each table; | |
| } |
or something like this
clean the code and make it more modular depending on threads count
jonathonmcmurray
approved these changes
Aug 29, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Addition of parallel compression where possible in multithreaded processes.
Split the compression function into two components:
Performance Statistics
Compression:
Decompression
Verifications of changes
Context on data
Table 1 - 2025.08.07 Partition - Quote - 2.5m rows - 9 columns
Table 2 - 2025.08.08 Partition - Quote - 2.5m rows - 9 columns
Table 3 - 2025.08.18 Partition - Trade - 0.6m rows - 8 columns
Single Threaded Process Verification
ksomerville@homer:~/repo/TorQ$ q torq.q -proctype test -procname mytest -debug
Verify single threaded
q)\s
0i
View Current compression ratio - returns nothing so currently not compressed
q)-21!`:/home/ksomerville/testDB/hdb/2025.08.07/quote/bid
()
Modify Compression File to have compression stats of (2;16;9):
`:/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv
q)read0 `:/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv
Run the compression
q) .cmp.docompression[hdbpath:
:/home/ksomerville/testDB/hdb;csvpath::/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv]Verify Compression has been applied:
q)-21!`:/home/ksomerville/testDB/hdb/2025.08.07/quote/bid
Run Decompression
Change the algo in the compression config to 0:
read0 `:/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv
Run decompression:
q).cmp.docompression[hdbpath:
:/home/ksomerville/testDB/hdb;csvpath::/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv]Verify the file is no longer compressed:
q)-21!`:/home/ksomerville/testDB/hdb/2025.08.07/quote/bid
()
Multithreaded Process
Start the test processes with a max thread setting of 8:
ksomerville@homer:~/repo/TorQ$ q torq.q -proctype test -procname mytest -debug -s 8
Verify processes is multithreaded
q)\s
8i
Show that the files are uncompressed (not touched from above):
q)-21!`:/home/ksomerville/testDB/hdb/2025.08.07/quote/bid
()
Set the Compression algo back to (2;16;9)
q) read0 `:/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv
Run the Compression:
q).cmp.docompression[hdbpath:
:/home/ksomerville/testDB/hdb;csvpath::/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv]Verify compression applied:
q) -21!`:/home/ksomerville/testDB/hdb/2025.08.07/quote/bid
Run decompression on the data, change compression algo to 0:
q) read0 `:/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv
Run the Decompression:
q).cmp.docompression[hdbpath:
:/home/ksomerville/testDB/hdb;csvpath::/home/ksomerville/deploy/TorQApp/latest/appconfig/compressionconfig.csv]Verify the data isn'compressed:
q)-21!`:/home/ksomerville/testDB/hdb/2025.08.07/quote/bid
()