You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2020/06/18 21:41:54 UTC
[impala] 03/03: IMPALA-2515: support parquet decimal with extra
padding
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 2c76ff5e6e49a56688ad1d9b4d1a8b4731fb5271
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Tue Jun 16 12:23:52 2020 -0700
IMPALA-2515: support parquet decimal with extra padding
This adds support for reading Parquet files where the
DECIMAL is encoded as a FIXED_LEN_BYTE_ARRAY field
with extra padding. This requires loosening file
validation and fixing up the decoding so that it
no longer assumes that the in-memory value is at
least as large as the encoded representation.
The decimal decoding logic was reworked so that we
could add the extra condition handling without regressing
performance of the decoding logic in the common case. In
the end I was able to significantly speed up the decoding
logic. The bottleneck, revealed by perf record while running
the below benchmark, was CPU stalls on the bitshift used for
sign extension instruction waiting on loading the result of
ByteSwap(). I worked around this by doing the sign-extension
before the ByteSwap(),
Perf:
Ran a microbenchmark to check that scanning perf didn't regress
as a result of the change. The query scans a DECIMAL column
that is mostly plain-encoded, so to maximally stress the
FIXED_LEN_BYTE_ARRAY decoding performance.
set mt_dop=1;
set num_nodes=1;
select min(l_extendedprice)
from tpch_parquet.lineitem
The SCAN time in the summary averaged out to 94ms before
the change and is reduced to 74ms after the change. The
actual speedup of the DECIMAL decoding is greater - it
went from ~20% of time in to ~6% of time as measured by
perf.
Testing:
Added a couple of parquet files that were generated with a
hacked version of Impala to have extra padding. Sanity-checked
that hacked tables returned the same results on Hive. The
tests failed before this code change.
Ran exhaustive tests with the hacked version of Impala
(so that all decimal tables got extra padding).
Change-Id: I2700652eab8ba7f23ffa75800a1712d310d4e1ec
Reviewed-on: http://gerrit.cloudera.org:8080/16090
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/parquet/parquet-metadata-utils.cc | 6 +--
be/src/util/decimal-util.h | 39 +++++++++------
testdata/data/README | 22 +++++++++
.../decimal_padded_fixed_len_byte_array.parquet | Bin 0 -> 1865 bytes
.../decimal_padded_fixed_len_byte_array2.parquet | Bin 0 -> 78804 bytes
.../queries/QueryTest/parquet-decimal-formats.test | 55 +++++++++++++++++++++
tests/query_test/test_scanners.py | 2 +
7 files changed, 105 insertions(+), 19 deletions(-)
diff --git a/be/src/exec/parquet/parquet-metadata-utils.cc b/be/src/exec/parquet/parquet-metadata-utils.cc
index 7ec32b6..09373f3 100644
--- a/be/src/exec/parquet/parquet-metadata-utils.cc
+++ b/be/src/exec/parquet/parquet-metadata-utils.cc
@@ -338,10 +338,10 @@ Status ParquetMetadataUtils::ValidateColumn(const char* filename,
}
int expected_len = ParquetPlainEncoder::DecimalSize(slot_desc->type());
- if (schema_element.type_length != expected_len) {
+ if (schema_element.type_length < expected_len) {
return Status(Substitute("File '$0' column '$1' has an invalid type length. "
- "Expecting: $2 len in file: $3", filename, schema_element.name, expected_len,
- schema_element.type_length));
+ "Expecting: len >= $2 in file: $3", filename, schema_element.name,
+ expected_len, schema_element.type_length));
}
}
if (!schema_element.__isset.scale) {
diff --git a/be/src/util/decimal-util.h b/be/src/util/decimal-util.h
index f505ecc..3517cf3 100644
--- a/be/src/util/decimal-util.h
+++ b/be/src/util/decimal-util.h
@@ -123,25 +123,32 @@ class DecimalUtil {
#endif
}
- template<typename T>
+ template <typename T>
static inline void DecodeFromFixedLenByteArray(
- const uint8_t* buffer, int fixed_len_size, T* v) {
+ const uint8_t* RESTRICT buffer, int fixed_len_size, T* RESTRICT v) {
DCHECK_GT(fixed_len_size, 0);
- DCHECK_LE(fixed_len_size, sizeof(T));
- *v = 0;
- // We need to sign extend val. For example, if the original value was
- // -1, the original bytes were -1,-1,-1,-1. If we only wrote out 1 byte, after
- // the encode step above, val would contain (-1, 0, 0, 0). We need to sign
- // extend the remaining 3 bytes to get the original value.
- // We do this by filling in the most significant bytes and (arithmetic) bit
- // shifting down.
- int bytes_to_fill = sizeof(T) - fixed_len_size;
-#if __BYTE_ORDER == __LITTLE_ENDIAN
- BitUtil::ByteSwap(reinterpret_cast<int8_t*>(v) + bytes_to_fill, buffer, fixed_len_size);
-#else
- memcpy(v, buffer, fixed_len_size);
+#if __BYTE_ORDER != __LITTLE_ENDIAN
+ static_assert(false, "Byte order must be little-endian");
#endif
- v->value() >>= (bytes_to_fill * 8);
+ if (LIKELY(sizeof(T) >= fixed_len_size)) {
+ // We need to sign extend val. For example, if the original value was
+ // -1, the original bytes were -1,-1,-1,-1. If we only wrote out 1 byte, ByteSwap()
+ // only fills in the first byte of 'v' - the least significant byte. We initialize
+ // the value to either 0 or -1 to ensure that the result is correctly sign-extended.
+
+ // GCC can optimize this code to an instruction pair of movsbq and sarq with no
+ // branches.
+ int8_t most_significant_byte = reinterpret_cast<const int8_t*>(buffer)[0];
+ v->value() = most_significant_byte >= 0 ? 0 : -1;
+ BitUtil::ByteSwap(reinterpret_cast<int8_t*>(v), buffer, fixed_len_size);
+ } else {
+ // If the destination 'v' is smaller than the input, discard the upper bytes.
+ // These bytes should only contain the sign-extended value if they were encoded
+ // correctly, so are not required.
+ int bytes_to_discard = fixed_len_size - sizeof(T);
+ BitUtil::ByteSwap(reinterpret_cast<int8_t*>(v),
+ buffer + bytes_to_discard, sizeof(T));
+ }
}
// Used to skip checking overflow in multiply of decimal values.
diff --git a/testdata/data/README b/testdata/data/README
index 0dee509..adfb84c 100644
--- a/testdata/data/README
+++ b/testdata/data/README
@@ -188,6 +188,28 @@ decimal_stored_as_int64.parquet:
Parquet file generated by Spark 2.3.1 that contains decimals stored as int64.
Impala needs to be able to read such values (IMPALA-5542)
+decimal_padded_fixed_len_byte_array.parquet:
+Parquet file generated by a hacked Impala where decimals are encoded as
+FIXED_LEN_BYTE_ARRAY with an extra byte of padding (IMPALA-2515). The
+data is the same as functional.decimal_tbl.
+
+decimal_padded_fixed_len_byte_array2.parquet:
+Parquet file generated by a hacked Impala where decimals are encoded as
+FIXED_LEN_BYTE_ARRAY with an extra byte of padding (IMPALA-2515).
+The Impala was hacked to limit dictionaries to 2000 entries, and
+PARQUET_PAGE_ROW_COUNT_LIMIT was set to 1234. The resulted in
+a file with two dictionary encoded pages and multiple plain
+encoded pages. The values are distributed across the full
+range of DECIMAL(10, 0). The file was created as follows:
+
+ create table d(d decimal(10, 0)) stored as parquet;
+ set num_nodes=1;
+ set PARQUET_PAGE_ROW_COUNT_LIMIT=1234;
+ insert into d
+ select distinct cast((1000000000 - (o_orderkey * 110503)) % 1000000000 as decimal(9, 0))
+ from tpch_parquet.orders where o_orderkey < 50000;
+
+
primitive_type_widening.parquet:
Parquet file that contains two rows with the following schema:
- int32 tinyint_col1
diff --git a/testdata/data/decimal_padded_fixed_len_byte_array.parquet b/testdata/data/decimal_padded_fixed_len_byte_array.parquet
new file mode 100644
index 0000000..274fee4
Binary files /dev/null and b/testdata/data/decimal_padded_fixed_len_byte_array.parquet differ
diff --git a/testdata/data/decimal_padded_fixed_len_byte_array2.parquet b/testdata/data/decimal_padded_fixed_len_byte_array2.parquet
new file mode 100644
index 0000000..59f662a
Binary files /dev/null and b/testdata/data/decimal_padded_fixed_len_byte_array2.parquet differ
diff --git a/testdata/workloads/functional-query/queries/QueryTest/parquet-decimal-formats.test b/testdata/workloads/functional-query/queries/QueryTest/parquet-decimal-formats.test
index b57c9e3..c7cd27b 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/parquet-decimal-formats.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/parquet-decimal-formats.test
@@ -41,3 +41,58 @@ select score from decimal_stored_as_int64
---- TYPES
DECIMAL
====
+---- QUERY
+select * from decimal_padded_fixed_len_byte_array
+---- RESULTS
+1234,2222,1.2345678900,0.12345678900000000000000000000000000000,12345.78900,1
+2345,111,12.3456789000,0.12345678900000000000000000000000000000,3.14100,1
+12345,333,123.4567890000,0.12345678900000000000000000000000000000,11.22000,1
+12345,333,1234.5678900000,0.12345678900000000000000000000000000000,0.10000,1
+132842,333,12345.6789000000,0.12345678900000000000000000000000000000,0.77889,1
+---- TYPES
+DECIMAL,DECIMAL,DECIMAL,DECIMAL,DECIMAL,DECIMAL
+====
+---- QUERY
+select count(distinct d), avg(d)
+from decimal_padded_fixed_len_byte_array2
+---- RESULTS
+12503,-296330482.059986
+---- TYPES
+BIGINT,DECIMAL
+====
+---- QUERY
+select d from decimal_padded_fixed_len_byte_array2
+order by d
+limit 10
+---- RESULTS
+-999987594
+-999877091
+-999766588
+-999656085
+-999545582
+-999435079
+-999382929
+-999324576
+-999272426
+-999161923
+---- TYPES
+DECIMAL
+====
+---- QUERY
+select d from decimal_padded_fixed_len_byte_array2
+order by d desc
+limit 10
+---- RESULTS
+999889497
+999778994
+999668491
+999557988
+999447485
+999336982
+999226479
+996463904
+996353401
+996242898
+---- TYPES
+DECIMAL
+====
diff --git a/tests/query_test/test_scanners.py b/tests/query_test/test_scanners.py
index 95cd782..02302d0 100644
--- a/tests/query_test/test_scanners.py
+++ b/tests/query_test/test_scanners.py
@@ -806,6 +806,8 @@ class TestParquet(ImpalaTestSuite):
create_table_from_parquet(self.client, unique_database, 'decimal_stored_as_int32')
create_table_from_parquet(self.client, unique_database, 'decimal_stored_as_int64')
+ create_table_from_parquet(self.client, unique_database, 'decimal_padded_fixed_len_byte_array')
+ create_table_from_parquet(self.client, unique_database, 'decimal_padded_fixed_len_byte_array2')
self.run_test_case('QueryTest/parquet-decimal-formats', vector, unique_database)