You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by cs...@apache.org on 2019/01/17 15:06:05 UTC
[impala] branch master updated (3338bae -> 8da44ce)
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from 3338bae IMPALA-8043: Fix BE test failures related to SystemV timezones.
new e1e2580 IMPALA-8078: fix test_corrupt_stats explain output
new 0906e08 IMPALA-7889: Write new logical types in Parquet
new 8da44ce IMPALA-6964: Track stats about column and page sizes in Parquet reader
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/exec/hdfs-scan-node-base.cc | 45 +++++
be/src/exec/hdfs-scan-node-base.h | 33 ++++
be/src/exec/parquet/hdfs-parquet-scanner.cc | 15 ++
be/src/exec/parquet/hdfs-parquet-scanner.h | 18 ++
be/src/exec/parquet/hdfs-parquet-table-writer.cc | 35 +---
be/src/exec/parquet/parquet-column-readers.cc | 11 ++
be/src/exec/parquet/parquet-metadata-utils.cc | 112 +++++++++++--
be/src/exec/parquet/parquet-metadata-utils.h | 6 +
be/src/util/runtime-profile.cc | 2 +-
.../queries/QueryTest/corrupt-stats.test | 21 ++-
.../queries/QueryTest/stats-extrapolation.test | 2 +-
tests/infra/test_utils.py | 17 ++
tests/query_test/test_insert_parquet.py | 183 ++++++++++++++++-----
tests/query_test/test_scanners.py | 101 +++++++++++-
tests/util/get_parquet_metadata.py | 17 ++
tests/util/parse_util.py | 51 ++++++
16 files changed, 569 insertions(+), 100 deletions(-)
[impala] 02/03: IMPALA-7889: Write new logical types in Parquet
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 0906e0817ce2301d7e20f355b334861f0232f16f
Author: Csaba Ringhofer <cs...@cloudera.com>
AuthorDate: Wed Nov 28 21:01:04 2018 +0100
IMPALA-7889: Write new logical types in Parquet
Fill the LogicalType field in Parquet schemas for columns
that have an associated logical type. ConvertedType still
has to be filled to remain compatible with older readers.
Testing:
- added new tests to check both logical and converted types
to test_insert_parquet.py
Change-Id: I6f377950845683ab9c6dea79f4c54db0359d0b91
Reviewed-on: http://gerrit.cloudera.org:8080/12004
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/parquet/hdfs-parquet-table-writer.cc | 35 +---
be/src/exec/parquet/parquet-metadata-utils.cc | 112 +++++++++++--
be/src/exec/parquet/parquet-metadata-utils.h | 6 +
.../queries/QueryTest/stats-extrapolation.test | 2 +-
tests/query_test/test_insert_parquet.py | 183 ++++++++++++++++-----
tests/util/get_parquet_metadata.py | 17 ++
6 files changed, 272 insertions(+), 83 deletions(-)
diff --git a/be/src/exec/parquet/hdfs-parquet-table-writer.cc b/be/src/exec/parquet/hdfs-parquet-table-writer.cc
index 80e340b..94f2b9e 100644
--- a/be/src/exec/parquet/hdfs-parquet-table-writer.cc
+++ b/be/src/exec/parquet/hdfs-parquet-table-writer.cc
@@ -22,6 +22,7 @@
#include "common/version.h"
#include "exec/hdfs-table-sink.h"
#include "exec/parquet/parquet-column-stats.inline.h"
+#include "exec/parquet/parquet-metadata-utils.h"
#include "exprs/scalar-expr-evaluator.h"
#include "exprs/scalar-expr.h"
#include "rpc/thrift-util.h"
@@ -969,35 +970,11 @@ Status HdfsParquetTableWriter::CreateSchema() {
file_metadata_.schema[0].name = "schema";
for (int i = 0; i < columns_.size(); ++i) {
- parquet::SchemaElement& node = file_metadata_.schema[i + 1];
- const ColumnType& type = output_expr_evals_[i]->root().type();
- node.name = table_desc_->col_descs()[i + num_clustering_cols].name();
- node.__set_type(ConvertInternalToParquetType(type.type));
- node.__set_repetition_type(parquet::FieldRepetitionType::OPTIONAL);
- if (type.type == TYPE_DECIMAL) {
- // This column is type decimal. Update the file metadata to include the
- // additional fields:
- // 1) converted_type: indicate this is really a decimal column.
- // 2) type_length: the number of bytes used per decimal value in the data
- // 3) precision/scale
- node.__set_converted_type(parquet::ConvertedType::DECIMAL);
- node.__set_type_length(
- ParquetPlainEncoder::DecimalSize(output_expr_evals_[i]->root().type()));
- node.__set_scale(output_expr_evals_[i]->root().type().scale);
- node.__set_precision(output_expr_evals_[i]->root().type().precision);
- } else if (type.type == TYPE_VARCHAR || type.type == TYPE_CHAR ||
- (type.type == TYPE_STRING &&
- state_->query_options().parquet_annotate_strings_utf8)) {
- node.__set_converted_type(parquet::ConvertedType::UTF8);
- } else if (type.type == TYPE_TINYINT) {
- node.__set_converted_type(parquet::ConvertedType::INT_8);
- } else if (type.type == TYPE_SMALLINT) {
- node.__set_converted_type(parquet::ConvertedType::INT_16);
- } else if (type.type == TYPE_INT) {
- node.__set_converted_type(parquet::ConvertedType::INT_32);
- } else if (type.type == TYPE_BIGINT) {
- node.__set_converted_type(parquet::ConvertedType::INT_64);
- }
+ parquet::SchemaElement& col_schema = file_metadata_.schema[i + 1];
+ const ColumnType& col_type = output_expr_evals_[i]->root().type();
+ col_schema.name = table_desc_->col_descs()[i + num_clustering_cols].name();
+ ParquetMetadataUtils::FillSchemaElement(col_type, state_->query_options(),
+ &col_schema);
}
return Status::OK();
diff --git a/be/src/exec/parquet/parquet-metadata-utils.cc b/be/src/exec/parquet/parquet-metadata-utils.cc
index 597b187..3e369af 100644
--- a/be/src/exec/parquet/parquet-metadata-utils.cc
+++ b/be/src/exec/parquet/parquet-metadata-utils.cc
@@ -85,8 +85,59 @@ bool IsSupportedType(PrimitiveType impala_type,
parquet::Type::type parquet_type = element.type;
return encodings->second.find(parquet_type) != encodings->second.end();
}
+
+/// Returns true if encoding 'e' is supported by Impala, false otherwise.
+static bool IsEncodingSupported(parquet::Encoding::type e) {
+ switch (e) {
+ case parquet::Encoding::PLAIN:
+ case parquet::Encoding::PLAIN_DICTIONARY:
+ case parquet::Encoding::BIT_PACKED:
+ case parquet::Encoding::RLE:
+ return true;
+ default:
+ return false;
+ }
}
+/// Sets logical and converted types in 'col_schema' to signed 'bitwidth' bit integer.
+void SetIntLogicalType(int bitwidth, parquet::SchemaElement* col_schema) {
+ parquet::IntType int_type;
+ int_type.__set_bitWidth(bitwidth);
+ int_type.__set_isSigned(true);
+ parquet::LogicalType logical_type;
+ logical_type.__set_INTEGER(int_type);
+ col_schema->__set_logicalType(logical_type);
+}
+
+/// Sets logical and converted types in 'col_schema' to UTF8.
+void SetUtf8ConvertedAndLogicalType(parquet::SchemaElement* col_schema) {
+ col_schema->__set_converted_type(parquet::ConvertedType::UTF8);
+ parquet::LogicalType logical_type;
+ logical_type.__set_STRING(parquet::StringType());
+ col_schema->__set_logicalType(logical_type);
+}
+
+/// Sets logical and converted types in 'col_schema' to DECIMAL.
+/// Precision and scale are set according to 'col_type'.
+void SetDecimalConvertedAndLogicalType(
+ const ColumnType& col_type, parquet::SchemaElement* col_schema) {
+ DCHECK_EQ(col_type.type, TYPE_DECIMAL);
+
+ col_schema->__set_type_length(ParquetPlainEncoder::DecimalSize(col_type));
+ col_schema->__set_scale(col_type.scale);
+ col_schema->__set_precision(col_type.precision);
+ col_schema->__set_converted_type(parquet::ConvertedType::DECIMAL);
+
+ parquet::DecimalType decimal_type;
+ decimal_type.__set_scale(col_type.scale);
+ decimal_type.__set_precision(col_type.precision);
+ parquet::LogicalType logical_type;
+ logical_type.__set_DECIMAL(decimal_type);
+ col_schema->__set_logicalType(logical_type);
+}
+
+} // anonymous namespace
+
// Needs to be in sync with the order of enum values declared in TParquetArrayResolution.
const std::vector<ParquetSchemaResolver::ArrayEncoding>
ParquetSchemaResolver::ORDERED_ARRAY_ENCODINGS[] =
@@ -145,18 +196,6 @@ Status ParquetMetadataUtils::ValidateOffsetInFile(const string& filename, int co
return Status::OK();;
}
-static bool IsEncodingSupported(parquet::Encoding::type e) {
- switch (e) {
- case parquet::Encoding::PLAIN:
- case parquet::Encoding::PLAIN_DICTIONARY:
- case parquet::Encoding::BIT_PACKED:
- case parquet::Encoding::RLE:
- return true;
- default:
- return false;
- }
-}
-
Status ParquetMetadataUtils::ValidateRowGroupColumn(
const parquet::FileMetaData& file_metadata, const char* filename, int row_group_idx,
int col_idx, const parquet::SchemaElement& schema_element, RuntimeState* state) {
@@ -275,6 +314,55 @@ Status ParquetMetadataUtils::ValidateColumn(const char* filename,
return Status::OK();
}
+void ParquetMetadataUtils::FillSchemaElement(const ColumnType& col_type,
+ const TQueryOptions& query_options, parquet::SchemaElement* col_schema) {
+ col_schema->__set_type(ConvertInternalToParquetType(col_type.type));
+ col_schema->__set_repetition_type(parquet::FieldRepetitionType::OPTIONAL);
+
+ switch (col_type.type) {
+ case TYPE_DECIMAL:
+ SetDecimalConvertedAndLogicalType(col_type, col_schema);
+ break;
+ case TYPE_VARCHAR:
+ case TYPE_CHAR:
+ SetUtf8ConvertedAndLogicalType(col_schema);
+ break;
+ case TYPE_STRING:
+ // By default STRING has no logical type, see IMPALA-5982.
+ // VARCHAR and CHAR are always set to UTF8.
+ if (query_options.parquet_annotate_strings_utf8) {
+ SetUtf8ConvertedAndLogicalType(col_schema);
+ }
+ break;
+ case TYPE_TINYINT:
+ col_schema->__set_converted_type(parquet::ConvertedType::INT_8);
+ SetIntLogicalType(8, col_schema);
+ break;
+ case TYPE_SMALLINT:
+ col_schema->__set_converted_type(parquet::ConvertedType::INT_16);
+ SetIntLogicalType(16, col_schema);
+ break;
+ case TYPE_INT:
+ col_schema->__set_converted_type(parquet::ConvertedType::INT_32);
+ SetIntLogicalType(32, col_schema);
+ break;
+ case TYPE_BIGINT:
+ col_schema->__set_converted_type(parquet::ConvertedType::INT_64);
+ SetIntLogicalType(64, col_schema);
+ break;
+ case TYPE_TIMESTAMP:
+ case TYPE_BOOLEAN:
+ case TYPE_FLOAT:
+ case TYPE_DOUBLE:
+ // boolean/float/double/INT96 encoded timestamp have no logical or converted types.
+ // INT64 encoded timestamp will have logical and converted types (IMPALA-5051).
+ break;
+ default:
+ DCHECK(false);
+ break;
+ }
+}
+
ParquetFileVersion::ParquetFileVersion(const string& created_by) {
string created_by_lower = created_by;
std::transform(created_by_lower.begin(), created_by_lower.end(),
diff --git a/be/src/exec/parquet/parquet-metadata-utils.h b/be/src/exec/parquet/parquet-metadata-utils.h
index f3a144d..e9e95f2 100644
--- a/be/src/exec/parquet/parquet-metadata-utils.h
+++ b/be/src/exec/parquet/parquet-metadata-utils.h
@@ -26,6 +26,7 @@
namespace impala {
class RuntimeState;
+class TQueryOptions;
class ParquetMetadataUtils {
public:
@@ -55,6 +56,11 @@ class ParquetMetadataUtils {
static Status ValidateColumn(const char* filename,
const parquet::SchemaElement& schema_element, const SlotDescriptor* slot_desc,
RuntimeState* state);
+
+ /// Sets type related fields in a SchemaElement based on the column's internal type
+ /// and query options.
+ static void FillSchemaElement(const ColumnType& col_type,
+ const TQueryOptions& query_options, parquet::SchemaElement* col_schema);
};
struct ParquetFileVersion {
diff --git a/testdata/workloads/functional-query/queries/QueryTest/stats-extrapolation.test b/testdata/workloads/functional-query/queries/QueryTest/stats-extrapolation.test
index 0c6deb4..1398988 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/stats-extrapolation.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/stats-extrapolation.test
@@ -33,7 +33,7 @@ show table stats alltypes
YEAR, MONTH, #ROWS, EXTRAP #ROWS, #FILES, SIZE, BYTES CACHED, CACHE REPLICATION, FORMAT, INCREMENTAL STATS, LOCATION
---- RESULTS
'2009','1',-1,308,1,regex:.*B,'NOT CACHED','NOT CACHED','PARQUET','false','$NAMENODE/test-warehouse/$DATABASE.db/alltypes/year=2009/month=1'
-'2009','2',-1,288,1,regex:.*B,'NOT CACHED','NOT CACHED','PARQUET','false','$NAMENODE/test-warehouse/$DATABASE.db/alltypes/year=2009/month=2'
+'2009','2',-1,289,1,regex:.*B,'NOT CACHED','NOT CACHED','PARQUET','false','$NAMENODE/test-warehouse/$DATABASE.db/alltypes/year=2009/month=2'
'2009','3',-1,308,1,regex:.*B,'NOT CACHED','NOT CACHED','PARQUET','false','$NAMENODE/test-warehouse/$DATABASE.db/alltypes/year=2009/month=3'
'2009','4',-1,302,1,regex:.*B,'NOT CACHED','NOT CACHED','PARQUET','false','$NAMENODE/test-warehouse/$DATABASE.db/alltypes/year=2009/month=4'
'2009','5',-1,308,1,regex:.*B,'NOT CACHED','NOT CACHED','PARQUET','false','$NAMENODE/test-warehouse/$DATABASE.db/alltypes/year=2009/month=5'
diff --git a/tests/query_test/test_insert_parquet.py b/tests/query_test/test_insert_parquet.py
index 6924d26..5790859 100644
--- a/tests/query_test/test_insert_parquet.py
+++ b/tests/query_test/test_insert_parquet.py
@@ -23,17 +23,18 @@ from collections import namedtuple
from datetime import datetime
from decimal import Decimal
from subprocess import check_call
-from parquet.ttypes import ColumnOrder, SortingColumn, TypeDefinedOrder
+from parquet.ttypes import ColumnOrder, SortingColumn, TypeDefinedOrder, ConvertedType
from tests.common.environ import impalad_basedir
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.parametrize import UniqueDatabase
-from tests.common.skip import SkipIfEC, SkipIfIsilon, SkipIfLocal, SkipIfS3, SkipIfABFS, \
- SkipIfADLS
+from tests.common.skip import (SkipIfEC, SkipIfIsilon, SkipIfLocal, SkipIfS3, SkipIfABFS,
+ SkipIfADLS)
from tests.common.test_dimensions import create_exec_option_dimension
from tests.common.test_vector import ImpalaTestDimension
from tests.util.filesystem_utils import get_fs_path
-from tests.util.get_parquet_metadata import get_parquet_metadata, decode_stats_value
+from tests.util.get_parquet_metadata import (decode_stats_value,
+ get_parquet_metadata_from_hdfs_folder)
PARQUET_CODECS = ['none', 'snappy', 'gzip']
@@ -246,14 +247,11 @@ class TestHdfsParquetTableWriter(ImpalaTestSuite):
self.execute_query(query)
# Download hdfs files and extract rowgroup metadata
+ file_metadata_list = get_parquet_metadata_from_hdfs_folder(hdfs_path, tmpdir.strpath)
row_groups = []
- check_call(['hdfs', 'dfs', '-get', hdfs_path, tmpdir.strpath])
- for root, subdirs, files in os.walk(tmpdir.strpath):
- for f in files:
- parquet_file = os.path.join(root, str(f))
- file_meta_data = get_parquet_metadata(parquet_file)
- row_groups.extend(file_meta_data.row_groups)
+ for file_metadata in file_metadata_list:
+ row_groups.extend(file_metadata.row_groups)
# Verify that the files have the sorted_columns set
expected = [SortingColumn(4, False, False), SortingColumn(0, False, False)]
@@ -279,21 +277,18 @@ class TestHdfsParquetTableWriter(ImpalaTestSuite):
self.execute_query(query)
# Download hdfs files and verify column orders
- check_call(['hdfs', 'dfs', '-get', hdfs_path, tmpdir.strpath])
+ file_metadata_list = get_parquet_metadata_from_hdfs_folder(hdfs_path, tmpdir.strpath)
expected_col_orders = [ColumnOrder(TYPE_ORDER=TypeDefinedOrder())] * 11
- for root, subdirs, files in os.walk(tmpdir.strpath):
- for f in files:
- parquet_file = os.path.join(root, str(f))
- file_meta_data = get_parquet_metadata(parquet_file)
- assert file_meta_data.column_orders == expected_col_orders
+ for file_metadata in file_metadata_list:
+ assert file_metadata.column_orders == expected_col_orders
- def test_read_write_logical_types(self, vector, unique_database, tmpdir):
+ def test_read_write_integer_logical_types(self, vector, unique_database, tmpdir):
"""IMPALA-5052: Read and write signed integer parquet logical types
This test creates a src_tbl like a parquet file. The parquet file was generated
to have columns with different signed integer logical types. The test verifies
- that parquet file written by the hdfs parquet table writer using the genererated
+ that parquet file written by the hdfs parquet table writer using the generated
file has the same column type metadata as the generated one."""
hdfs_path = (os.environ['DEFAULT_FS'] + "/test-warehouse/{0}.db/"
"signed_integer_logical_types.parquet").format(unique_database)
@@ -358,6 +353,115 @@ class TestHdfsParquetTableWriter(ImpalaTestSuite):
% dst_tbl)
assert result_src.data == result_dst.data
+ def _ctas_and_get_metadata(self, vector, unique_database, tmp_dir, source_table):
+ """CTAS 'source_table' into a Parquet table and returns its Parquet metadata."""
+ table_name = "test_hdfs_parquet_table_writer"
+ qualified_table_name = "{0}.{1}".format(unique_database, table_name)
+ hdfs_path = get_fs_path('/test-warehouse/{0}.db/{1}/'.format(unique_database,
+ table_name))
+
+ # Setting num_nodes = 1 ensures that the query is executed on the coordinator,
+ # resulting in a single parquet file being written.
+ query = ("create table {0} stored as parquet as select * from {1}").format(
+ qualified_table_name, source_table)
+ vector.get_value('exec_option')['num_nodes'] = 1
+ self.execute_query_expect_success(self.client, query,
+ vector.get_value('exec_option'))
+
+ file_metadata_list = get_parquet_metadata_from_hdfs_folder(hdfs_path, tmp_dir)
+ assert len(file_metadata_list) == 1
+ assert file_metadata_list[0] is not None
+ return file_metadata_list[0]
+
+ @staticmethod
+ def _get_schema(schemas, column_name):
+ """Searches 'schemas' for a schema with name 'column_name'. Asserts if non is found.
+ """
+ for schema in schemas:
+ if schema.name == column_name:
+ return schema
+ assert False, "schema element %s not found" % column_name
+
+ @staticmethod
+ def _check_only_one_member_var_is_set(obj, var_name):
+ """Checks that 'var_name' is the only member of 'obj' that is not None. Useful to
+ check Thrift unions."""
+ keys = [k for k, v in vars(obj).iteritems() if v is not None]
+ assert keys == [var_name]
+
+ def _check_no_logical_type(self, schemas, column_name):
+ """Checks that the schema with name 'column_name' has no logical or converted type."""
+ schema = self._get_schema(schemas, column_name)
+ assert schema.converted_type is None
+ assert schema.logicalType is None
+
+ def _check_int_logical_type(self, schemas, column_name, bit_width):
+ """Checks that the schema with name 'column_name' has logical and converted type that
+ describe a signed integer with 'bit_width' bits."""
+ schema = self._get_schema(schemas, column_name)
+ bit_width_to_converted_type_map = {
+ 8: ConvertedType.INT_8,
+ 16: ConvertedType.INT_16,
+ 32: ConvertedType.INT_32,
+ 64: ConvertedType.INT_64
+ }
+ assert schema.converted_type == bit_width_to_converted_type_map[bit_width]
+ assert schema.logicalType is not None
+ self._check_only_one_member_var_is_set(schema.logicalType, "INTEGER")
+ assert schema.logicalType.INTEGER is not None
+ assert schema.logicalType.INTEGER.bitWidth == bit_width
+ assert schema.logicalType.INTEGER.isSigned
+
+ def _check_decimal_logical_type(self, schemas, column_name, precision, scale):
+ """Checks that the schema with name 'column_name' has logical and converted type that
+ describe a decimal with given 'precision' and 'scale'."""
+ schema = self._get_schema(schemas, column_name)
+ assert schema.converted_type == ConvertedType.DECIMAL
+ assert schema.precision == precision
+ assert schema.scale == scale
+ assert schema.logicalType is not None
+ self._check_only_one_member_var_is_set(schema.logicalType, "DECIMAL")
+ assert schema.logicalType.DECIMAL.precision == precision
+ assert schema.logicalType.DECIMAL.scale == scale
+
+ def test_logical_types(self, vector, unique_database, tmpdir):
+ """Tests that the Parquet writers set logical type and converted type correctly
+ for all types except DECIMAL"""
+ source = "functional.alltypestiny"
+ file_metadata = \
+ self._ctas_and_get_metadata(vector, unique_database, tmpdir.strpath, source)
+ schemas = file_metadata.schema
+
+ self._check_int_logical_type(schemas, "tinyint_col", 8)
+ self._check_int_logical_type(schemas, "smallint_col", 16)
+ self._check_int_logical_type(schemas, "int_col", 32)
+ self._check_int_logical_type(schemas, "bigint_col", 64)
+
+ self._check_no_logical_type(schemas, "bool_col")
+ self._check_no_logical_type(schemas, "float_col")
+ self._check_no_logical_type(schemas, "double_col")
+
+ # By default STRING has no logical type, see IMPALA-5982.
+ self._check_no_logical_type(schemas, "string_col")
+
+ # Currently TIMESTAMP is written as INT96 and has no logical type.
+ # This test will break once INT64 becomes the default Parquet type for TIMESTAMP
+ # columns in the future (IMPALA-5049).
+ self._check_no_logical_type(schemas, "timestamp_col")
+
+ def test_decimal_logical_types(self, vector, unique_database, tmpdir):
+ """Tests that the Parquet writers set logical type and converted type correctly
+ for DECIMAL type."""
+ source = "functional.decimal_tiny"
+ file_metadata = \
+ self._ctas_and_get_metadata(vector, unique_database, tmpdir.strpath, source)
+ schemas = file_metadata.schema
+
+ self._check_decimal_logical_type(schemas, "c1", 10, 4)
+ self._check_decimal_logical_type(schemas, "c2", 15, 5)
+ self._check_decimal_logical_type(schemas, "c3", 1, 1)
+
+
@SkipIfIsilon.hive
@SkipIfLocal.hive
@SkipIfS3.hive
@@ -399,14 +503,14 @@ class TestHdfsParquetTableStatsWriter(ImpalaTestSuite):
assert len(decoded) == len(schemas)
return decoded
- def _get_row_group_stats_from_file(self, parquet_file):
- """Returns a list of statistics for each row group in file 'parquet_file'. The result
- is a two-dimensional list, containing stats by row group and column."""
- file_meta_data = get_parquet_metadata(parquet_file)
+ def _get_row_group_stats_from_file_metadata(self, file_metadata):
+ """Returns a list of statistics for each row group in Parquet file metadata
+ 'file_metadata'. The result is a two-dimensional list, containing stats by
+ row group and column."""
# We only support flat schemas, the additional element is the root element.
- schemas = file_meta_data.schema[1:]
+ schemas = file_metadata.schema[1:]
file_stats = []
- for row_group in file_meta_data.row_groups:
+ for row_group in file_metadata.row_groups:
num_columns = len(row_group.columns)
assert num_columns == len(schemas)
column_stats = [c.meta_data.statistics for c in row_group.columns]
@@ -421,12 +525,9 @@ class TestHdfsParquetTableStatsWriter(ImpalaTestSuite):
two-dimensional list, containing stats by row group and column."""
row_group_stats = []
- check_call(['hdfs', 'dfs', '-get', hdfs_path, tmp_dir])
-
- for root, subdirs, files in os.walk(tmp_dir):
- for f in files:
- parquet_file = os.path.join(root, str(f))
- row_group_stats.extend(self._get_row_group_stats_from_file(parquet_file))
+ file_metadata_list = get_parquet_metadata_from_hdfs_folder(hdfs_path, tmp_dir)
+ for file_metadata in file_metadata_list:
+ row_group_stats.extend(self._get_row_group_stats_from_file_metadata(file_metadata))
return row_group_stats
@@ -509,12 +610,12 @@ class TestHdfsParquetTableStatsWriter(ImpalaTestSuite):
"""
# Expected values for functional.decimal_tbl
expected_min_max_values = [
- ColumnStats('d1', 1234, 132842, 0),
- ColumnStats('d2', 111, 2222, 0),
- ColumnStats('d3', Decimal('1.23456789'), Decimal('12345.6789'), 0),
- ColumnStats('d4', Decimal('0.123456789'), Decimal('0.123456789'), 0),
- ColumnStats('d5', Decimal('0.1'), Decimal('12345.789'), 0),
- ColumnStats('d6', 1, 1, 0)
+ ColumnStats('d1', 1234, 132842, 0),
+ ColumnStats('d2', 111, 2222, 0),
+ ColumnStats('d3', Decimal('1.23456789'), Decimal('12345.6789'), 0),
+ ColumnStats('d4', Decimal('0.123456789'), Decimal('0.123456789'), 0),
+ ColumnStats('d5', Decimal('0.1'), Decimal('12345.789'), 0),
+ ColumnStats('d6', 1, 1, 0)
]
self._ctas_table_and_verify_stats(vector, unique_database, tmpdir.strpath,
@@ -670,11 +771,11 @@ class TestHdfsParquetTableStatsWriter(ImpalaTestSuite):
# Expected values for tpch_parquet.customer
expected_min_max_values = [
- ColumnStats('id', '8600000US00601', '8600000US999XX', 0),
- ColumnStats('zip', '00601', '999XX', 0),
- ColumnStats('description1', '\"00601 5-Digit ZCTA', '\"999XX 5-Digit ZCTA', 0),
- ColumnStats('description2', ' 006 3-Digit ZCTA\"', ' 999 3-Digit ZCTA\"', 0),
- ColumnStats('income', 0, 189570, 29),
+ ColumnStats('id', '8600000US00601', '8600000US999XX', 0),
+ ColumnStats('zip', '00601', '999XX', 0),
+ ColumnStats('description1', '\"00601 5-Digit ZCTA', '\"999XX 5-Digit ZCTA', 0),
+ ColumnStats('description2', ' 006 3-Digit ZCTA\"', ' 999 3-Digit ZCTA\"', 0),
+ ColumnStats('income', 0, 189570, 29),
]
self._ctas_table_and_verify_stats(vector, unique_database, tmpdir.strpath,
diff --git a/tests/util/get_parquet_metadata.py b/tests/util/get_parquet_metadata.py
index f6a0e59..56281f8 100644
--- a/tests/util/get_parquet_metadata.py
+++ b/tests/util/get_parquet_metadata.py
@@ -21,6 +21,7 @@ import struct
from datetime import date, datetime, time, timedelta
from decimal import Decimal
from parquet.ttypes import ColumnIndex, FileMetaData, OffsetIndex, PageHeader, Type
+from subprocess import check_call
from thrift.protocol import TCompactProtocol
from thrift.transport import TTransport
@@ -168,3 +169,19 @@ def get_parquet_metadata(filename):
# Return deserialized FileMetaData object
return read_serialized_object(FileMetaData, f, metadata_pos, metadata_len)
+
+
+def get_parquet_metadata_from_hdfs_folder(hdfs_path, tmp_dir):
+ """Returns a list with the FileMetaData of every file in 'hdfs_path' and its
+ subdirectories. The hdfs folder is copied into 'tmp_dir' before processing.
+ """
+ check_call(['hdfs', 'dfs', '-get', hdfs_path, tmp_dir])
+ result = []
+ for root, subdirs, files in os.walk(tmp_dir):
+ for f in files:
+ if not f.endswith('parq'):
+ continue
+ parquet_file = os.path.join(root, str(f))
+ file_meta_data = get_parquet_metadata(parquet_file)
+ result.append(file_meta_data)
+ return result
[impala] 01/03: IMPALA-8078: fix test_corrupt_stats explain output
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit e1e258039d2b0129cc2d3a0feadfb8e7284457cf
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Wed Jan 16 08:34:25 2019 -0800
IMPALA-8078: fix test_corrupt_stats explain output
Testing:
impala-py.test tests/metadata/test_compute_stats.py \
-k corrupt_stats \
--workload_exploration_strategy=functional-query:exhaustive
Change-Id: I0201f7d5363f793290179defcac3a56c6f47d1a9
Reviewed-on: http://gerrit.cloudera.org:8080/12227
Reviewed-by: Bharath Vissapragada <bh...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../queries/QueryTest/corrupt-stats.test | 21 ++++++++++++++-------
1 file changed, 14 insertions(+), 7 deletions(-)
diff --git a/testdata/workloads/functional-query/queries/QueryTest/corrupt-stats.test b/testdata/workloads/functional-query/queries/QueryTest/corrupt-stats.test
index 74707db..65409b5 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/corrupt-stats.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/corrupt-stats.test
@@ -66,7 +66,8 @@ explain select count(*) from corrupted where org = 1;
'| output: count(*)'
'|'
'00:SCAN HDFS [$DATABASE.corrupted]'
-' partitions=1/2 files=1 size=24B row-size=0B cardinality=0'
+' partitions=1/2 files=1 size=24B'
+' row-size=0B cardinality=0'
---- TYPES
STRING
====
@@ -92,7 +93,8 @@ explain select count(*) from corrupted;
'| output: count(*)'
'|'
'00:SCAN HDFS [$DATABASE.corrupted]'
-' partitions=2/2 files=2 size=48B row-size=0B cardinality=6'
+' partitions=2/2 files=2 size=48B'
+' row-size=0B cardinality=6'
---- TYPES
STRING
====
@@ -117,7 +119,8 @@ explain select count(*) from corrupted;
'| output: count(*)'
'|'
'00:SCAN HDFS [$DATABASE.corrupted]'
-' partitions=2/2 files=2 size=48B row-size=0B cardinality=6'
+' partitions=2/2 files=2 size=48B'
+' row-size=0B cardinality=6'
---- TYPES
STRING
====
@@ -139,7 +142,8 @@ explain select count(*) from corrupted where org = 2;
'| output: count(*)'
'|'
'00:SCAN HDFS [$DATABASE.corrupted]'
-' partitions=1/2 files=1 size=24B row-size=0B cardinality=6'
+' partitions=1/2 files=1 size=24B'
+' row-size=0B cardinality=6'
---- TYPES
STRING
====
@@ -165,7 +169,8 @@ explain select count(*) from corrupted_no_part;
'| output: count(*)'
'|'
'00:SCAN HDFS [$DATABASE.corrupted_no_part]'
-' partitions=1/1 files=1 size=6B row-size=0B cardinality=3'
+' partitions=1/1 files=1 size=6B'
+' row-size=0B cardinality=3'
---- TYPES
STRING
====
@@ -197,7 +202,8 @@ explain select count(*) from corrupted_no_part;
'| output: count(*)'
'|'
'00:SCAN HDFS [$DATABASE.corrupted_no_part]'
-' partitions=1/1 files=1 size=6B row-size=0B cardinality=0'
+' partitions=1/1 files=1 size=6B'
+' row-size=0B cardinality=0'
---- TYPES
STRING
====
@@ -219,7 +225,8 @@ explain select count(*) from corrupted_no_part;
'| output: count(*)'
'|'
'00:SCAN HDFS [$DATABASE.corrupted_no_part]'
-' partitions=1/1 files=1 size=6B row-size=0B cardinality=unavailable'
+' partitions=1/1 files=1 size=6B'
+' row-size=0B cardinality=unavailable'
---- TYPES
STRING
====
[impala] 03/03: IMPALA-6964: Track stats about column and page
sizes in Parquet reader
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 8da44ce16bb190dadab2ff3d22e5df726d1128e3
Author: stakiar <ta...@gmail.com>
AuthorDate: Fri Oct 19 07:48:16 2018 -0500
IMPALA-6964: Track stats about column and page sizes in Parquet reader
Adds the following new stats:
* ParquetCompressedPageSize - a summary (average, min, max) counter that
tracks the size of compressed pages read, if no compressed pages are
read then this counter is empty
* ParquetUncompressedPageSize - a summary counter that tracks the size
of uncompressed pages read, it is updated in two places: (1) when a
compressed page is de-compressed, and (2) when a page that is not
compressed is read
* ParquetCompressedDataReadPerColumn - a summary counter that tracks the
amount of compressed data read per column for a scan node
* ParquetUncompressedDataReadPerColumn - a summary counter that tracks
the amount of uncompressed data read per column for a scan node
The PerColumn counters are calculated by aggregating the number of bytes
read for each column across all scan ranges processed by a scan node.
Each sample in the counter is the size of a single column.
Here is an example of what the updated HDFS scan profile looks like:
- ParquetCompressedDataReadPerColumn: (Avg: 227.56 KB (233018) ;
Min: 225.14 KB (230540) ; Max: 229.98 KB (235496) ; Number of samples: 2)
- ParquetUncompressedDataReadPerColumn: (Avg: 227.96 KB (233426) ;
Min: 224.91 KB (230306) ; Max: 231.00 KB (236547) ; Number of samples: 2)
- ParquetCompressedPageSize: (Avg: 4.46 KB (4568) ; Min: 3.86 KB (3955) ;
Max: 5.19 KB (5315) ; Number of samples: 102)
- ParquetDecompressedPageSize: (Avg: 4.47 KB (4576) ; Min: 3.86 KB (3950)
; Max: 5.22 KB (5349) ; Number of samples: 102)
Testing:
* Added new tests to test_scanners.py that do some basic validation of
the new counters above
Change-Id: I322f9b324b6828df28e5caf79529085c43d7c817
Reviewed-on: http://gerrit.cloudera.org:8080/11575
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/hdfs-scan-node-base.cc | 45 ++++++++++++
be/src/exec/hdfs-scan-node-base.h | 33 +++++++++
be/src/exec/parquet/hdfs-parquet-scanner.cc | 15 ++++
be/src/exec/parquet/hdfs-parquet-scanner.h | 18 +++++
be/src/exec/parquet/parquet-column-readers.cc | 11 +++
be/src/util/runtime-profile.cc | 2 +-
tests/infra/test_utils.py | 17 +++++
tests/query_test/test_scanners.py | 101 +++++++++++++++++++++++---
tests/util/parse_util.py | 51 +++++++++++++
9 files changed, 283 insertions(+), 10 deletions(-)
diff --git a/be/src/exec/hdfs-scan-node-base.cc b/be/src/exec/hdfs-scan-node-base.cc
index 1e627b4..3470124 100644
--- a/be/src/exec/hdfs-scan-node-base.cc
+++ b/be/src/exec/hdfs-scan-node-base.cc
@@ -373,6 +373,11 @@ Status HdfsScanNodeBase::Open(RuntimeState* state) {
initial_range_actual_reservation_stats_ = ADD_SUMMARY_STATS_COUNTER(runtime_profile(),
"InitialRangeActualReservation", TUnit::BYTES);
+ uncompressed_bytes_read_per_column_counter_ = ADD_SUMMARY_STATS_COUNTER(
+ runtime_profile(), "ParquetUncompressedBytesReadPerColumn", TUnit::BYTES);
+ compressed_bytes_read_per_column_counter_ = ADD_SUMMARY_STATS_COUNTER(
+ runtime_profile(), "ParquetCompressedBytesReadPerColumn", TUnit::BYTES);
+
bytes_read_local_ = ADD_COUNTER(runtime_profile(), "BytesReadLocal",
TUnit::BYTES);
bytes_read_short_circuit_ = ADD_COUNTER(runtime_profile(), "BytesReadShortCircuit",
@@ -892,6 +897,25 @@ void HdfsScanNodeBase::StopAndFinalizeCounters() {
runtime_profile()->AppendExecOption(
Substitute("Codegen enabled: $0 out of $1", num_enabled, total));
+ // Locking here should not be necessary since bytes_read_per_col_ is only updated inside
+ // column readers, and all column readers should have completed at this point; however,
+ // we acquire a read lock in case the update semantics of bytes_read_per_col_ change
+ {
+ shared_lock<shared_mutex> bytes_read_per_col_guard_read_lock(
+ bytes_read_per_col_lock_);
+ for (const auto& bytes_read : bytes_read_per_col_) {
+ int64_t uncompressed_bytes_read = bytes_read.second.uncompressed_bytes_read.Load();
+ if (uncompressed_bytes_read > 0) {
+ uncompressed_bytes_read_per_column_counter_->UpdateCounter(
+ uncompressed_bytes_read);
+ }
+ int64_t compressed_bytes_read = bytes_read.second.compressed_bytes_read.Load();
+ if (compressed_bytes_read > 0) {
+ compressed_bytes_read_per_column_counter_->UpdateCounter(compressed_bytes_read);
+ }
+ }
+ }
+
if (reader_context_ != nullptr) {
bytes_read_local_->Set(reader_context_->bytes_read_local());
bytes_read_short_circuit_->Set(reader_context_->bytes_read_short_circuit());
@@ -926,3 +950,24 @@ void HdfsScanNodeBase::StopAndFinalizeCounters() {
Status HdfsScanNodeBase::ScanNodeDebugAction(TExecNodePhase::type phase) {
return ExecDebugAction(phase, runtime_state_);
}
+
+void HdfsScanNodeBase::UpdateBytesRead(
+ SlotId slot_id, int64_t uncompressed_bytes_read, int64_t compressed_bytes_read) {
+ // Acquire a read lock first and check if the slot_id is in bytes_read_per_col_, if it
+ // is then update the value and release the read lock; if not then release the read
+ // lock, acquire the write lock, and then initialize the slot_id with the give value for
+ // bytes_read
+ shared_lock<shared_mutex> bytes_read_per_col_guard_read_lock(
+ bytes_read_per_col_lock_);
+ auto bytes_read_itr = bytes_read_per_col_.find(slot_id);
+ if (bytes_read_itr != bytes_read_per_col_.end()) {
+ bytes_read_itr->second.uncompressed_bytes_read.Add(uncompressed_bytes_read);
+ bytes_read_itr->second.compressed_bytes_read.Add(compressed_bytes_read);
+ } else {
+ bytes_read_per_col_guard_read_lock.unlock();
+ lock_guard<shared_mutex> bytes_read_per_col_guard_write_lock(
+ bytes_read_per_col_lock_);
+ bytes_read_per_col_[slot_id].uncompressed_bytes_read.Add(uncompressed_bytes_read);
+ bytes_read_per_col_[slot_id].compressed_bytes_read.Add(compressed_bytes_read);
+ }
+}
diff --git a/be/src/exec/hdfs-scan-node-base.h b/be/src/exec/hdfs-scan-node-base.h
index 7da66cb..168bd18 100644
--- a/be/src/exec/hdfs-scan-node-base.h
+++ b/be/src/exec/hdfs-scan-node-base.h
@@ -363,6 +363,12 @@ class HdfsScanNodeBase : public ScanNode {
int64_t IncreaseReservationIncrementally(int64_t curr_reservation,
int64_t ideal_reservation);
+ /// Update the number of [un]compressed bytes read for the given SlotId. This is used
+ /// to track the number of bytes read per column and is meant to be called by
+ /// individual scanner classes.
+ void UpdateBytesRead(
+ SlotId slot_id, int64_t uncompressed_bytes_read, int64_t compressed_bytes_read);
+
protected:
friend class ScannerContext;
friend class HdfsScanner;
@@ -545,6 +551,15 @@ class HdfsScanNodeBase : public ScanNode {
RuntimeProfile::SummaryStatsCounter* initial_range_ideal_reservation_stats_ = nullptr;
RuntimeProfile::SummaryStatsCounter* initial_range_actual_reservation_stats_ = nullptr;
+ /// Track stats about the number of bytes read per column. Each sample in the counter is
+ /// the size of a single column that is scanned by the scan node. The scan node tracks
+ /// the number of bytes read for each column it processes, and when the scan node is
+ /// closed, it updates these counters with the size of each column.
+ RuntimeProfile::SummaryStatsCounter* compressed_bytes_read_per_column_counter_ =
+ nullptr;
+ RuntimeProfile::SummaryStatsCounter* uncompressed_bytes_read_per_column_counter_ =
+ nullptr;
+
/// Pool for allocating some amounts of memory that is shared between scanners.
/// e.g. partition key tuple and their string buffers
boost::scoped_ptr<MemPool> scan_node_pool_;
@@ -554,6 +569,24 @@ class HdfsScanNodeBase : public ScanNode {
/// scanner threads.
Status status_;
+ /// Struct that tracks the uncompressed and compressed bytes read. Used by the map
+ /// bytes_read_per_col_ to track the [un]compressed bytes read per column. Types are
+ /// atomic as the struct may be updated concurrently.
+ struct BytesRead {
+ AtomicInt64 uncompressed_bytes_read;
+ AtomicInt64 compressed_bytes_read;
+ };
+
+ /// Map from SlotId (column identifer) to a pair where the first entry is the number of
+ /// uncompressed bytes read for the column and the second entry is the number of
+ /// compressed bytes read for the column. This map is used to update the
+ /// [un]compressed_bytes_read_per_column counter.
+ std::unordered_map<SlotId, BytesRead> bytes_read_per_col_;
+
+ /// Lock that controls access to bytes_read_per_col_ so that multiple scanners
+ /// can update the map concurrently
+ boost::shared_mutex bytes_read_per_col_lock_;
+
/// Performs dynamic partition pruning, i.e., applies runtime filters to files, and
/// issues initial ranges for all file types. Waits for runtime filters if necessary.
/// Only valid to call if !initial_ranges_issued_. Sets initial_ranges_issued_ to true.
diff --git a/be/src/exec/parquet/hdfs-parquet-scanner.cc b/be/src/exec/parquet/hdfs-parquet-scanner.cc
index 9cfcdb8..4fe9914 100644
--- a/be/src/exec/parquet/hdfs-parquet-scanner.cc
+++ b/be/src/exec/parquet/hdfs-parquet-scanner.cc
@@ -100,6 +100,8 @@ HdfsParquetScanner::HdfsParquetScanner(HdfsScanNodeBase* scan_node, RuntimeState
num_row_groups_counter_(nullptr),
num_scanners_with_no_reads_counter_(nullptr),
num_dict_filtered_row_groups_counter_(nullptr),
+ parquet_compressed_page_size_counter_(nullptr),
+ parquet_uncompressed_page_size_counter_(nullptr),
coll_items_read_counter_(0),
codegend_process_scratch_batch_fn_(nullptr) {
assemble_rows_timer_.Stop();
@@ -121,6 +123,10 @@ Status HdfsParquetScanner::Open(ScannerContext* context) {
ADD_COUNTER(scan_node_->runtime_profile(), "NumDictFilteredRowGroups", TUnit::UNIT);
process_footer_timer_stats_ =
ADD_SUMMARY_STATS_TIMER(scan_node_->runtime_profile(), "FooterProcessingTime");
+ parquet_compressed_page_size_counter_ = ADD_SUMMARY_STATS_COUNTER(
+ scan_node_->runtime_profile(), "ParquetCompressedPageSize", TUnit::BYTES);
+ parquet_uncompressed_page_size_counter_ = ADD_SUMMARY_STATS_COUNTER(
+ scan_node_->runtime_profile(), "ParquetUncompressedPageSize", TUnit::BYTES);
codegend_process_scratch_batch_fn_ = reinterpret_cast<ProcessScratchBatchFn>(
scan_node_->GetCodegenFn(THdfsFileFormat::PARQUET));
@@ -1681,4 +1687,13 @@ ParquetTimestampDecoder HdfsParquetScanner::CreateTimestampDecoder(
return ParquetTimestampDecoder(element, &state_->local_time_zone(),
timestamp_conversion_needed_for_int96_timestamps);
}
+
+void HdfsParquetScanner::UpdateCompressedPageSizeCounter(int64_t compressed_page_size) {
+ parquet_compressed_page_size_counter_->UpdateCounter(compressed_page_size);
+}
+
+void HdfsParquetScanner::UpdateUncompressedPageSizeCounter(
+ int64_t uncompressed_page_size) {
+ parquet_uncompressed_page_size_counter_->UpdateCounter(uncompressed_page_size);
+}
}
diff --git a/be/src/exec/parquet/hdfs-parquet-scanner.h b/be/src/exec/parquet/hdfs-parquet-scanner.h
index 94d4714..09da2bb 100644
--- a/be/src/exec/parquet/hdfs-parquet-scanner.h
+++ b/be/src/exec/parquet/hdfs-parquet-scanner.h
@@ -452,6 +452,16 @@ class HdfsParquetScanner : public HdfsScanner {
/// Number of row groups skipped due to dictionary filter
RuntimeProfile::Counter* num_dict_filtered_row_groups_counter_;
+ /// Tracks the size of any compressed pages read. If no compressed pages are read, this
+ /// counter is empty
+ RuntimeProfile::SummaryStatsCounter* parquet_compressed_page_size_counter_;
+
+ /// Tracks the size of any de-compressed pages reads. This counter is updated in two
+ /// places: (1) when a compressed page is de-compressed, the de-compressed size is added
+ /// to this counter, (2) when a page that is not compressed is read, its size is added
+ /// to this counter
+ RuntimeProfile::SummaryStatsCounter* parquet_uncompressed_page_size_counter_;
+
/// Number of collection items read in current row batch. It is a scanner-local counter
/// used to reduce the frequency of updating HdfsScanNode counter. It is updated by the
/// callees of AssembleRows() and is merged into the HdfsScanNode counter at the end of
@@ -640,6 +650,14 @@ class HdfsParquetScanner : public HdfsScanner {
/// no values that pass the relevant conjuncts, then the row group can be skipped.
Status EvalDictionaryFilters(const parquet::RowGroup& row_group,
bool* skip_row_group) WARN_UNUSED_RESULT;
+
+ /// Updates the counter parquet_compressed_page_size_counter_ with the given compressed
+ /// page size. Called by ParquetColumnReader for each page read.
+ void UpdateCompressedPageSizeCounter(int64_t compressed_page_size);
+
+ /// Updates the counter parquet_uncompressed_page_size_counter_ with the given
+ /// uncompressed page size. Called by ParquetColumnReader for each page read.
+ void UpdateUncompressedPageSizeCounter(int64_t uncompressed_page_size);
};
} // namespace impala
diff --git a/be/src/exec/parquet/parquet-column-readers.cc b/be/src/exec/parquet/parquet-column-readers.cc
index fce14b1..f6ad734 100644
--- a/be/src/exec/parquet/parquet-column-readers.cc
+++ b/be/src/exec/parquet/parquet-column-readers.cc
@@ -1355,6 +1355,13 @@ Status BaseScalarColumnReader::ReadDataPage() {
data_ = decompressed_buffer;
data_size = current_page_header_.uncompressed_page_size;
data_end_ = data_ + data_size;
+ if (slot_desc() != nullptr) {
+ parent_->scan_node_->UpdateBytesRead(slot_desc()->id(), uncompressed_size,
+ current_page_header_.compressed_page_size);
+ parent_->UpdateUncompressedPageSizeCounter(uncompressed_size);
+ parent_->UpdateCompressedPageSizeCounter(
+ current_page_header_.compressed_page_size);
+ }
} else {
DCHECK_EQ(metadata_->codec, parquet::CompressionCodec::UNCOMPRESSED);
if (current_page_header_.compressed_page_size != uncompressed_size) {
@@ -1373,6 +1380,10 @@ Status BaseScalarColumnReader::ReadDataPage() {
data_ = copy_buffer;
data_end_ = data_ + uncompressed_size;
}
+ if (slot_desc() != nullptr) {
+ parent_->scan_node_->UpdateBytesRead(slot_desc()->id(), uncompressed_size, 0);
+ parent_->UpdateUncompressedPageSizeCounter(uncompressed_size);
+ }
}
// Initialize the repetition level data
diff --git a/be/src/util/runtime-profile.cc b/be/src/util/runtime-profile.cc
index cdb9dd9..9892c28 100644
--- a/be/src/util/runtime-profile.cc
+++ b/be/src/util/runtime-profile.cc
@@ -795,7 +795,7 @@ void RuntimeProfile::PrettyPrint(ostream* s, const string& prefix) const {
for (const SummaryStatsCounterMap::value_type& v: summary_stats_map_) {
if (v.second->TotalNumValues() == 0) {
// No point printing all the stats if number of samples is zero.
- stream << prefix << " - " << v.first << ": "
+ stream << prefix << " - " << v.first << ": "
<< PrettyPrinter::Print(v.second->value(), v.second->unit(), true)
<< " (Number of samples: " << v.second->TotalNumValues() << ")" << endl;
} else {
diff --git a/tests/infra/test_utils.py b/tests/infra/test_utils.py
index c998703..1c2e282 100644
--- a/tests/infra/test_utils.py
+++ b/tests/infra/test_utils.py
@@ -18,6 +18,8 @@
# This module contains tests for some of the tests/util code.
from tests.util.filesystem_utils import prepend_with_fs
+from tests.util.parse_util import get_bytes_summary_stats_counter
+
def test_filesystem_utils():
# Verify that empty FS prefix gives back the same path.
@@ -29,3 +31,18 @@ def test_filesystem_utils():
path = "/fake-warehouse"
assert prepend_with_fs(fs, path) == fs + path
assert prepend_with_fs(fs, prepend_with_fs(fs, path)) == fs + path
+
+
+def test_get_bytes_summary_stats_counter():
+ """Test get_bytes_summary_stats_counter(counter_name, runtime_profile) using a dummy
+ runtime profile.
+ """
+ runtime_profile = "- ExampleCounter: (Avg: 8.00 KB (8192) ; " \
+ "Min: 6.00 KB (6144) ; " \
+ "Max: 10.00 KB (10240) ; " \
+ "Number of samples: 4)"
+ summary_stats = get_bytes_summary_stats_counter("ExampleCounter",
+ runtime_profile)
+ assert len(summary_stats) == 1
+ assert summary_stats[0].sum == 32768 and summary_stats[0].min_value == 6144 and \
+ summary_stats[0].max_value == 10240 and summary_stats[0].total_num_values == 4
diff --git a/tests/query_test/test_scanners.py b/tests/query_test/test_scanners.py
index 35fc4d8..6d03269 100644
--- a/tests/query_test/test_scanners.py
+++ b/tests/query_test/test_scanners.py
@@ -55,6 +55,7 @@ from tests.common.test_vector import ImpalaTestDimension
from tests.util.filesystem_utils import WAREHOUSE, get_fs_path
from tests.util.hdfs_util import NAMENODE
from tests.util.get_parquet_metadata import get_parquet_metadata
+from tests.util.parse_util import get_bytes_summary_stats_counter
from tests.util.test_file_parser import QueryTestSectionReader
# Test scanners with denial of reservations at varying frequency. This will affect the
@@ -535,9 +536,8 @@ class TestParquet(ImpalaTestSuite):
assert (not result.log and not log_prefix) or \
(log_prefix and result.log.startswith(log_prefix))
- runtime_profile = str(result.runtime_profile)
num_scanners_with_no_reads_list = re.findall(
- 'NumScannersWithNoReads: ([0-9]*)', runtime_profile)
+ 'NumScannersWithNoReads: ([0-9]*)', result.runtime_profile)
# This will fail if the number of impalads != 3
# The fourth fragment is the "Averaged Fragment"
@@ -594,11 +594,11 @@ class TestParquet(ImpalaTestSuite):
assert len(result.data) == 1
assert result.data[0] == str(rows_in_table)
- runtime_profile = str(result.runtime_profile)
- num_row_groups_list = re.findall('NumRowGroups: ([0-9]*)', runtime_profile)
+ num_row_groups_list = re.findall('NumRowGroups: ([0-9]*)', result.runtime_profile)
scan_ranges_complete_list = re.findall(
- 'ScanRangesComplete: ([0-9]*)', runtime_profile)
- num_rows_read_list = re.findall('RowsRead: [0-9.K]* \(([0-9]*)\)', runtime_profile)
+ 'ScanRangesComplete: ([0-9]*)', result.runtime_profile)
+ num_rows_read_list = re.findall('RowsRead: [0-9.K]* \(([0-9]*)\)',
+ result.runtime_profile)
REGEX_UNIT_SECOND = "[0-9]*[s]*[0-9]*[.]*[0-9]*[nm]*[s]*"
REGEX_MIN_MAX_FOOTER_PROCESSING_TIME = \
@@ -606,7 +606,7 @@ class TestParquet(ImpalaTestSuite):
"Number of samples: %s\)" % (REGEX_UNIT_SECOND, REGEX_UNIT_SECOND,
REGEX_UNIT_SECOND, "[0-9]*"))
footer_processing_time_list = re.findall(
- REGEX_MIN_MAX_FOOTER_PROCESSING_TIME, runtime_profile)
+ REGEX_MIN_MAX_FOOTER_PROCESSING_TIME, result.runtime_profile)
# This will fail if the number of impalads != 3
# The fourth fragment is the "Averaged Fragment"
@@ -807,6 +807,90 @@ class TestParquet(ImpalaTestSuite):
self.run_test_case(
'QueryTest/parquet-int64-timestamps', vector, unique_database)
+ def _is_summary_stats_counter_empty(self, counter):
+ """Returns true if the given TSummaryStatCounter is empty, false otherwise"""
+ return counter.max_value == counter.min_value == counter.sum ==\
+ counter.total_num_values == 0
+
+ def test_page_size_counters(self, vector):
+ """IMPALA-6964: Test that the counter Parquet[Un]compressedPageSize is updated
+ when reading [un]compressed Parquet files, and that the counter
+ Parquet[Un]compressedPageSize is not updated."""
+ # lineitem_sixblocks is not compressed so ParquetCompressedPageSize should be empty,
+ # but ParquetUncompressedPageSize should have been updated
+ result = self.client.execute("select * from functional_parquet.lineitem_sixblocks"
+ " limit 10")
+
+ compressed_page_size_summaries = get_bytes_summary_stats_counter(
+ "ParquetCompressedPageSize", result.runtime_profile)
+
+ assert len(compressed_page_size_summaries) > 0
+ for summary in compressed_page_size_summaries:
+ assert self._is_summary_stats_counter_empty(summary)
+
+ uncompressed_page_size_summaries = get_bytes_summary_stats_counter(
+ "ParquetUncompressedPageSize", result.runtime_profile)
+
+ # validate that some uncompressed data has been read; we don't validate the exact
+ # amount as the value can change depending on Parquet format optimizations, Impala
+ # scanner optimizations, etc.
+ assert len(uncompressed_page_size_summaries) > 0
+ for summary in uncompressed_page_size_summaries:
+ assert not self._is_summary_stats_counter_empty(summary)
+
+ # alltypestiny is compressed so both ParquetCompressedPageSize and
+ # ParquetUncompressedPageSize should have been updated
+ result = self.client.execute("select * from functional_parquet.alltypestiny"
+ " limit 10")
+
+ for summary_name in ("ParquetCompressedPageSize", "ParquetUncompressedPageSize"):
+ page_size_summaries = get_bytes_summary_stats_counter(
+ summary_name, result.runtime_profile)
+ assert len(page_size_summaries) > 0
+ for summary in page_size_summaries:
+ assert not self._is_summary_stats_counter_empty(summary)
+
+ def test_bytes_read_per_column(self, vector):
+ """IMPALA-6964: Test that the counter Parquet[Un]compressedBytesReadPerColumn is
+ updated when reading [un]compressed Parquet files, and that the counter
+ Parquet[Un]CompressedBytesReadPerColumn is not updated."""
+ # lineitem_sixblocks is not compressed so ParquetCompressedBytesReadPerColumn should
+ # be empty, but ParquetUncompressedBytesReadPerColumn should have been updated
+ result = self.client.execute("select * from functional_parquet.lineitem_sixblocks"
+ " limit 10")
+
+ compressed_bytes_read_per_col_summaries = get_bytes_summary_stats_counter(
+ "ParquetCompressedBytesReadPerColumn", result.runtime_profile)
+
+ assert len(compressed_bytes_read_per_col_summaries) > 0
+ for summary in compressed_bytes_read_per_col_summaries:
+ assert self._is_summary_stats_counter_empty(summary)
+
+ uncompressed_bytes_read_per_col_summaries = get_bytes_summary_stats_counter(
+ "ParquetUncompressedBytesReadPerColumn", result.runtime_profile)
+
+ assert len(uncompressed_bytes_read_per_col_summaries) > 0
+ for summary in uncompressed_bytes_read_per_col_summaries:
+ assert not self._is_summary_stats_counter_empty(summary)
+ # There are 16 columns in lineitem_sixblocks so there should be 16 samples
+ assert summary.total_num_values == 16
+
+ # alltypestiny is compressed so both ParquetCompressedBytesReadPerColumn and
+ # ParquetUncompressedBytesReadPerColumn should have been updated
+ result = self.client.execute("select * from functional_parquet.alltypestiny"
+ " limit 10")
+
+ for summary_name in ("ParquetCompressedBytesReadPerColumn",
+ "ParquetUncompressedBytesReadPerColumn"):
+ bytes_read_per_col_summaries = get_bytes_summary_stats_counter(summary_name,
+ result.runtime_profile)
+ assert len(bytes_read_per_col_summaries) > 0
+ for summary in bytes_read_per_col_summaries:
+ assert not self._is_summary_stats_counter_empty(summary)
+ # There are 11 columns in alltypestiny so there should be 11 samples
+ assert summary.total_num_values == 11
+
+
# We use various scan range lengths to exercise corner cases in the HDFS scanner more
# thoroughly. In particular, it will exercise:
# 1. default scan range
@@ -1121,9 +1205,8 @@ class TestOrc(ImpalaTestSuite):
result = self.client.execute(query)
assert len(result.data) == rows_in_table
- runtime_profile = str(result.runtime_profile)
num_scanners_with_no_reads_list = re.findall(
- 'NumScannersWithNoReads: ([0-9]*)', runtime_profile)
+ 'NumScannersWithNoReads: ([0-9]*)', result.runtime_profile)
# This will fail if the number of impalads != 3
# The fourth fragment is the "Averaged Fragment"
diff --git a/tests/util/parse_util.py b/tests/util/parse_util.py
index 3d3c950..5dbaff0 100644
--- a/tests/util/parse_util.py
+++ b/tests/util/parse_util.py
@@ -17,6 +17,7 @@
import re
from datetime import datetime
+from RuntimeProfile.ttypes import TSummaryStatsCounter
# IMPALA-6715: Every so often the stress test or the TPC workload directories get
# changed, and the stress test loses the ability to run the full set of queries. Set
@@ -125,3 +126,53 @@ def match_memory_estimate(explain_lines):
if None in (mem_limit, units):
raise Exception('could not parse explain string:\n' + '\n'.join(explain_lines))
return mem_limit, units
+
+
+def get_bytes_summary_stats_counter(counter_name, runtime_profile):
+ """Extracts a list of TSummaryStatsCounters from a given runtime profile where the units
+ are in bytes. Each entry in the returned list corresponds to a single occurrence of
+ the counter in the profile. If the counter is present, but it has not been updated,
+ an empty TSummaryStatsCounter is returned for that entry. If the counter is not in
+ the given profile, an empty list is returned. Here is an example of how this method
+ should be used:
+
+ # A single line in a runtime profile used for example purposes.
+ runtime_profile = "- ExampleCounter: (Avg: 8.00 KB (8192) ; " \
+ "Min: 8.00 KB (8192) ; " \
+ "Max: 8.00 KB (8192) ; " \
+ "Number of samples: 4)"
+ summary_stats = get_bytes_summary_stats_counter("ExampleCounter",
+ runtime_profile)
+ assert len(summary_stats) == 1
+ assert summary_stats[0].sum == summary_stats[0].min_value == \
+ summary_stats[0].max_value == 8192 and \
+ summary_stats[0].total_num_values == 1
+ """
+
+ regex_summary_stat = re.compile(r"""\(
+ Avg:[^\(]*\((?P<avg>[0-9]+)\)\s;\s # Matches Avg: [?].[?] [?]B (?)
+ Min:[^\(]*\((?P<min>[0-9]+)\)\s;\s # Matches Min: [?].[?] [?]B (?)
+ Max:[^\(]*\((?P<max>[0-9]+)\)\s;\s # Matches Max: [?].[?] [?]B (?)
+ Number\sof\ssamples:\s(?P<samples>[0-9]+)\) # Matches Number of samples: ?)""",
+ re.VERBOSE)
+
+ # First, find all lines that contain the counter name, and then extract the summary
+ # stats from each line. If the summary stats cannot be extracted, return a dictionary
+ # with values of 0 for all keys.
+ summary_stats = []
+ for counter in re.findall(counter_name + ".*", runtime_profile):
+ summary_stat = re.search(regex_summary_stat, counter)
+ # We need to special-case when the counter has not been updated at all because empty
+ # summary counters have a different format than updated ones.
+ if not summary_stat:
+ assert "0 (Number of samples: 0)" in counter
+ summary_stats.append(TSummaryStatsCounter(sum=0, total_num_values=0, min_value=0,
+ max_value=0))
+ else:
+ summary_stat = summary_stat.groupdict()
+ num_samples = int(summary_stat['samples'])
+ summary_stats.append(TSummaryStatsCounter(sum=num_samples *
+ int(summary_stat['avg']), total_num_values=num_samples,
+ min_value=int(summary_stat['min']), max_value=int(summary_stat['max'])))
+
+ return summary_stats