[Enhancement] Support some compress functions #47307
Open
+577
−2
Conversation
This file contains 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
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
zclllyybb
suggested changes
Jan 22, 2025
|
|
||
| Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments, | ||
| uint32_t result, size_t input_rows_count) const override { | ||
| // LOG(INFO) << "Executing FunctionCompress with " << input_rows_count |
There was a problem hiding this comment.
remove these commented lines
| col_data[idx] = '0', col_data[idx + 1] = 'x'; | ||
| for (int i = 0; i < 4; i++) { | ||
| unsigned char byte = (value >> (i * 8)) & 0xFF; | ||
| col_data[idx + 2 + i * 2] = "0123456789ABCDEF"[byte >> 4]; // 高4位 |
|
|
||
| auto st = compression_codec->compress(data, &compressed_str); | ||
|
|
||
| if (!st.ok()) { |
There was a problem hiding this comment.
add a comment about when will it fails
There was a problem hiding this comment.
add cases like regression-test/suites/query_p0/sql_functions/test_template_one_arg.groovy did
There was a problem hiding this comment.
we dont need modify this file anymore
| std::string func_name = "compress"; | ||
| InputTypeSet input_types = {TypeIndex::String}; | ||
|
|
||
| // 压缩多个不同的字符串 |
There was a problem hiding this comment.
dont use Chinese comment
| std::string uncompressed; | ||
| Slice data; | ||
| Slice uncompressed_slice; | ||
| for (int row = 0; row < input_rows_count; row++) { |
| illegal = 1; | ||
| } else { | ||
| if (data[0] != '0' || data[1] != 'x') { | ||
| LOG(INFO) << "illegal: " |
| if (x >= 'A' && x <= 'F') return true; | ||
| return false; | ||
| }; | ||
| auto trans = [](char x) { |
There was a problem hiding this comment.
just use from_chars and to_chars to replace your user implemented lambdas
| // Print the compressed string (after compression) | ||
| // LOG(INFO) << "Compressed string at row " << row << ": " | ||
| // << std::string(reinterpret_cast<const char*>(col_data.data())); | ||
| col_offset[row] = col_offset[row - 1] + 10 + compressed_str.size() * 2; |
There was a problem hiding this comment.
The first ten digits of the compress value are "0x" and eight digits long, followed by each digit split into two hexadecimal values
lzyy2024
force-pushed
the
CompressFunctions
branch
from
fa80a74 to
ca2b27e
Compare
zclllyybb
suggested changes
Jan 24, 2025
| const auto& str = arg_column.get_data_at(row); | ||
| data = Slice(str.data, str.size); | ||
|
|
||
| auto st = compression_codec->compress(data, &compressed_str); |
There was a problem hiding this comment.
when will compress fail?
| Slice data; | ||
| for (size_t row = 0; row < input_rows_count; row++) { | ||
| null_map[row] = false; | ||
| const auto& str = arg_column.get_data_at(row); |
There was a problem hiding this comment.
no need to use virtual function here
| \N \N | ||
|
|
||
| -- !const_not_nullable -- | ||
| 0x05000000789C73C92FCA2C060005B00202 0x446F726973 |
There was a problem hiding this comment.
carefully review your result!!!
| Slice data; | ||
| Slice uncompressed_slice; | ||
| for (size_t row = 0; row < input_rows_count; row++) { | ||
| auto check = [](char x) { |
There was a problem hiding this comment.
try to use std function firstly
| const auto& str = arg_column.get_data_at(row); | ||
| data = Slice(str.data, str.size); | ||
|
|
||
| int illegal = 0; |
| unsigned char* src = compressed_str.data(); | ||
| { | ||
| for (size_t i = 0; i < compressed_str.size(); i++) { | ||
| col_data[idx] = |
There was a problem hiding this comment.
so tricky here. try to improve code like it
zclllyybb
reviewed
Jan 24, 2025
| Slice data; | ||
| Slice uncompressed_slice; | ||
| for (size_t row = 0; row < input_rows_count; row++) { | ||
| std::function<bool(char)> check = [](char x) { |
zclllyybb
reviewed
Jan 25, 2025
be/src/util/block_compression.cpp
Outdated
| @@ -854,8 +854,13 @@ class ZlibBlockCompression : public BlockCompressionCodec { | |||
| Slice s(*output); | |||
|
|
|||
| auto zres = ::compress((Bytef*)s.data, &s.size, (Bytef*)input.data, input.size); | |||
| if (zres != Z_OK) { | |||
| return Status::InvalidArgument("Fail to do ZLib compress, error={}", zError(zres)); | |||
| if (zres == Z_MEM_ERROR) { | |||
There was a problem hiding this comment.
also change other same calls
There was a problem hiding this comment.
split them to another PR may be better
zclllyybb
reviewed
Jan 25, 2025
| implements UnaryExpression, ExplicitlyCastableSignature, PropagateNullable { | ||
|
|
||
| public static final List<FunctionSignature> SIGNATURES = ImmutableList.of( | ||
| FunctionSignature.ret(StringType.INSTANCE).args(StringType.INSTANCE)); |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
zclllyybb
reviewed
Jan 25, 2025
| implements UnaryExpression, ExplicitlyCastableSignature, AlwaysNullable { | ||
|
|
||
| public static final List<FunctionSignature> SIGNATURES = ImmutableList.of( | ||
| FunctionSignature.ret(StringType.INSTANCE).args(StringType.INSTANCE)); |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
zclllyybb
reviewed
Jan 25, 2025
|
|
||
| unsigned int length = 0; | ||
| for (size_t i = 2; i <= 9; i += 2) { | ||
| unsigned char byte = (hex_ctoi.at(data[i]) << 4) + hex_ctoi.at(data[i + 1]); |
There was a problem hiding this comment.
remove hex_ctoi and just use from_chars
zclllyybb
reviewed
Jan 25, 2025
| unsigned int length = 0; | ||
| for (size_t i = 2; i <= 9; i += 2) { | ||
| unsigned char byte = (hex_ctoi.at(data[i]) << 4) + hex_ctoi.at(data[i + 1]); | ||
| length += (byte << (8 * (i / 2 - 1))); //Little Endian : 0x01000000 -> 1 |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
zclllyybb
reviewed
Jan 25, 2025
| std::string uncompressed; | ||
| Slice data; | ||
| Slice uncompressed_slice; | ||
| for (size_t row = 0; row < input_rows_count; row++) { |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
zclllyybb
reviewed
Jan 25, 2025
| } | ||
| idx += 10; | ||
|
|
||
| col_data.resize(col_data.size() + 2 * compressed_str.size()); |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
zclllyybb
reviewed
Jan 25, 2025
| //Converts a hexadecimal readable string to a compressed byte stream | ||
| std::string s(((int)data.size - 10) / 2, ' '); // byte stream data.size >= 10 | ||
| for (size_t i = 10, j = 0; i < data.size; i += 2, j++) { | ||
| s[j] = (hex_ctoi.at(data[i]) << 4) + hex_ctoi.at(data[i + 1]); |
lzyy2024
force-pushed
the
CompressFunctions
branch
from
6ef40dd to
b8b3bad
Compare
|
run buildall |
|
run buildall |
TPC-H: Total hot run time: 32100 ms |
TPC-DS: Total hot run time: 184954 ms |
ClickBench: Total hot run time: 30.68 s |
|
run buildall |
|
run buildall |
TPC-H: Total hot run time: 32971 ms |
TPC-DS: Total hot run time: 191631 ms |
ClickBench: Total hot run time: 30.67 s |
|
run buildall |
TPC-H: Total hot run time: 32328 ms |
TPC-DS: Total hot run time: 185720 ms |
ClickBench: Total hot run time: 30.3 s |
|
run buildall |
lzyy2024
force-pushed
the
CompressFunctions
branch
from
0377751 to
c94f814
Compare
lzyy2024
force-pushed
the
CompressFunctions
branch
from
c94f814 to
9263dea
Compare
|
run buildall |
TPC-H: Total hot run time: 32060 ms |
TPC-DS: Total hot run time: 184635 ms |
ClickBench: Total hot run time: 31.04 s |
|
TeamCity be ut coverage result: |
zclllyybb
reviewed
Jan 27, 2025
| } | ||
|
|
||
| // first ten digits represent the length of the uncompressed string | ||
| col_data.resize(col_data.size() + 10); |
There was a problem hiding this comment.
L111 and L120's resize could be merged.
| if (data[0] != '0' || data[1] != 'x') { | ||
| illegal = true; | ||
| } | ||
| for (size_t i = 2; i <= 9; i += 2) { |
There was a problem hiding this comment.
why +=2 here? should be ++?
| } | ||
|
|
||
| { | ||
| std::string func_name = "uncompress"; |
There was a problem hiding this comment.
we should add some case of uncompressing a string which is making from a valid compressed string with minor modifications.
like, for valid '0x1204', try to uncompress '0x12F4' which in invalid to get the NULL result.
| } | ||
|
|
||
| uncompressed.resize(length); | ||
| uncompressed_slice = Slice(uncompressed); |
There was a problem hiding this comment.
remove the uncompressed, just resize col_data to add its length with length. so just point uncompressed_slice into col_data. then we can do decompress inplace and no need to do heavily memcpy(at L237)
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.
What problem does this PR solve?
Added the compress and uncompressed functions similar to mysql
Issue Number: close #45530
Related PR: #xxx
Problem Summary:
Release note
None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)