IMPALA-4815, IMPALA-4817, IMPALA-4819: Write and Read Parquet Statistics for remaining types
This change adds functionality to write and read parquet::Statistics for Decimal, String, and Timestamp values. As an exception, we don't read statistics for CHAR columns, since CHAR support is broken in Impala (IMPALA-1652). This change also switches from using the deprecated fields 'min' and 'max' to populate the new fields 'min_value' and 'max_value' in parquet::Statistics, that were added in parquet-format pull request #46. The HdfsParquetScanner will preferably read the new fields if they are populated and if the column order 'TypeDefinedOrder' has been used to compute the statistics. For columns without a column order set or with only the deprecated fields populated, the scanner will read them only if they are of simple numeric type, i.e. boolean, integer, or floating point. This change removes the validation of the Parquet Statistics we write to Hive from the tests, since Hive does not write the new fields. Instead it adds a parquet file written by Hive that uses the deprecated fields for its statistics. It uses that file to exercise the fallback logic for supported types in a test. This change also cleans up the interface of ParquetPlainEncoder in parquet-common.h. Change-Id: I3ef4a5d25a57c82577fd498d6d1c4297ecf39312 Reviewed-on: http://gerrit.cloudera.org:8080/6563 Reviewed-by: Lars Volker <[email protected]> Tested-by: Lars Volker <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/92703468 Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/92703468 Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/92703468 Branch: refs/heads/master Commit: 9270346825b0bbc7708d458be076d7c26038edfc Parents: 249632b Author: Lars Volker <[email protected]> Authored: Mon Apr 3 12:55:56 2017 +0200 Committer: Lars Volker <[email protected]> Committed: Tue May 9 15:47:21 2017 +0000 ---------------------------------------------------------------------- be/src/exec/hdfs-parquet-scanner.cc | 24 +- be/src/exec/hdfs-parquet-scanner.h | 8 +- be/src/exec/hdfs-parquet-table-writer.cc | 45 ++- be/src/exec/parquet-column-stats.cc | 133 ++++++--- be/src/exec/parquet-column-stats.h | 115 +++++--- be/src/exec/parquet-column-stats.inline.h | 136 ++++++--- be/src/exec/parquet-common.h | 91 +++--- be/src/exec/parquet-metadata-utils.cc | 7 - be/src/exec/parquet-metadata-utils.h | 3 - be/src/exec/parquet-plain-test.cc | 10 +- be/src/util/dict-encoding.h | 2 +- common/thrift/parquet.thrift | 97 +++++- testdata/data/README | 4 + testdata/data/deprecated_statistics.parquet | Bin 0 -> 4456 bytes .../QueryTest/parquet-deprecated-stats.test | 144 +++++++++ .../queries/QueryTest/parquet-filtering.test | 3 +- .../queries/QueryTest/parquet_stats.test | 295 ++++++++++++++----- tests/query_test/test_insert_parquet.py | 178 +++++++---- tests/query_test/test_parquet_stats.py | 27 ++ 19 files changed, 974 insertions(+), 348 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/hdfs-parquet-scanner.cc ---------------------------------------------------------------------- diff --git a/be/src/exec/hdfs-parquet-scanner.cc b/be/src/exec/hdfs-parquet-scanner.cc index a6831b7..1d5d807 100644 --- a/be/src/exec/hdfs-parquet-scanner.cc +++ b/be/src/exec/hdfs-parquet-scanner.cc @@ -491,7 +491,8 @@ Status HdfsParquetScanner::GetNextInternal(RowBatch* row_batch) { return Status::OK(); } -Status HdfsParquetScanner::EvaluateStatsConjuncts(const parquet::RowGroup& row_group, +Status HdfsParquetScanner::EvaluateStatsConjuncts( + const parquet::FileMetaData& file_metadata, const parquet::RowGroup& row_group, bool* skip_row_group) { *skip_row_group = false; @@ -538,25 +539,27 @@ Status HdfsParquetScanner::EvaluateStatsConjuncts(const parquet::RowGroup& row_g int col_idx = node->col_idx; DCHECK(col_idx < row_group.columns.size()); - if (!ParquetMetadataUtils::HasRowGroupStats(row_group, col_idx)) continue; - const parquet::Statistics& stats = row_group.columns[col_idx].meta_data.statistics; + const vector<parquet::ColumnOrder>& col_orders = file_metadata.column_orders; + const parquet::ColumnOrder* col_order = nullptr; + if (col_idx < col_orders.size()) col_order = &col_orders[col_idx]; + const parquet::ColumnChunk& col_chunk = row_group.columns[col_idx]; + const ColumnType& col_type = slot_desc->type(); bool stats_read = false; void* slot = min_max_tuple->GetSlot(slot_desc->tuple_offset()); - const ColumnType& col_type = slot_desc->type(); - const string& fn_name = conjunct->root()->function_name(); if (fn_name == "lt" || fn_name == "le") { // We need to get min stats. - stats_read = ColumnStatsBase::ReadFromThrift(stats, col_type, - ColumnStatsBase::StatsField::MIN, slot); + stats_read = ColumnStatsBase::ReadFromThrift( + col_chunk, col_type, col_order, ColumnStatsBase::StatsField::MIN, slot); } else if (fn_name == "gt" || fn_name == "ge") { // We need to get max stats. - stats_read = ColumnStatsBase::ReadFromThrift(stats, col_type, - ColumnStatsBase::StatsField::MAX, slot); + stats_read = ColumnStatsBase::ReadFromThrift( + col_chunk, col_type, col_order, ColumnStatsBase::StatsField::MAX, slot); } else { DCHECK(false) << "Unsupported function name for statistics evaluation: " << fn_name; } + if (stats_read) min_max_conjuncts_ctxs_to_eval_.push_back(conjunct); } @@ -628,7 +631,8 @@ Status HdfsParquetScanner::NextRowGroup() { // Evaluate row group statistics. bool skip_row_group_on_stats; - RETURN_IF_ERROR(EvaluateStatsConjuncts(row_group, &skip_row_group_on_stats)); + RETURN_IF_ERROR( + EvaluateStatsConjuncts(file_metadata_, row_group, &skip_row_group_on_stats)); if (skip_row_group_on_stats) { COUNTER_ADD(num_stats_filtered_row_groups_counter_, 1); continue; http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/hdfs-parquet-scanner.h ---------------------------------------------------------------------- diff --git a/be/src/exec/hdfs-parquet-scanner.h b/be/src/exec/hdfs-parquet-scanner.h index 18c14fe..59a06bc 100644 --- a/be/src/exec/hdfs-parquet-scanner.h +++ b/be/src/exec/hdfs-parquet-scanner.h @@ -486,9 +486,11 @@ class HdfsParquetScanner : public HdfsScanner { virtual Status GetNextInternal(RowBatch* row_batch); /// Evaluates the min/max predicates of the 'scan_node_' using the parquet::Statistics - /// of 'row_group'. Sets 'skip_row_group' to true if the row group can be skipped, - /// 'false' otherwise. - Status EvaluateStatsConjuncts(const parquet::RowGroup& row_group, bool* skip_row_group); + /// of 'row_group'. 'file_metadata' is used to determine the ordering that was used to + /// compute the statistics. Sets 'skip_row_group' to true if the row group can be + /// skipped, 'false' otherwise. + Status EvaluateStatsConjuncts(const parquet::FileMetaData& file_metadata, + const parquet::RowGroup& row_group, bool* skip_row_group); /// Check runtime filters' effectiveness every BATCHES_PER_FILTER_SELECTIVITY_CHECK /// row batches. Will update 'filter_stats_'. http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/hdfs-parquet-table-writer.cc ---------------------------------------------------------------------- diff --git a/be/src/exec/hdfs-parquet-table-writer.cc b/be/src/exec/hdfs-parquet-table-writer.cc index ea0059d..295b10b 100644 --- a/be/src/exec/hdfs-parquet-table-writer.cc +++ b/be/src/exec/hdfs-parquet-table-writer.cc @@ -130,6 +130,13 @@ class HdfsParquetTableWriter::BaseColumnWriter { Status Flush(int64_t* file_pos, int64_t* first_data_page, int64_t* first_dictionary_page); + // Materializes the column statistics to the per-file MemPool so they are available + // after their row batch buffer has been freed. + void MaterializeStatsValues() { + row_group_stats_base_->MaterializeStringValuesToInternalBuffers(); + page_stats_base_->MaterializeStringValuesToInternalBuffers(); + } + // Encodes the row group statistics into a parquet::Statistics object and attaches it to // 'meta_data'. void EncodeRowGroupStats(ColumnMetaData* meta_data) { @@ -142,7 +149,7 @@ class HdfsParquetTableWriter::BaseColumnWriter { } // Resets all the data accumulated for this column. Memory can now be reused for - // the next row group + // the next row group. // Any data for previous row groups must be reset (e.g. dictionaries). // Subclasses must call this if they override this function. virtual void Reset() { @@ -284,12 +291,8 @@ class HdfsParquetTableWriter::ColumnWriter : : BaseColumnWriter(parent, ctx, codec), num_values_since_dict_size_check_(0), plain_encoded_value_size_( - ParquetPlainEncoder::EncodedByteSize(ctx->root()->type())), - page_stats_(plain_encoded_value_size_), - row_group_stats_(plain_encoded_value_size_) { + ParquetPlainEncoder::EncodedByteSize(ctx->root()->type())) { DCHECK_NE(ctx->root()->type().type, TYPE_BOOLEAN); - page_stats_base_ = &page_stats_; - row_group_stats_base_ = &row_group_stats_; } virtual void Reset() { @@ -300,8 +303,12 @@ class HdfsParquetTableWriter::ColumnWriter : dict_encoder_.reset( new DictEncoder<T>(parent_->per_file_mem_pool_.get(), plain_encoded_value_size_)); dict_encoder_base_ = dict_encoder_.get(); - page_stats_.Reset(); - row_group_stats_.Reset(); + page_stats_.reset( + new ColumnStats<T>(parent_->per_file_mem_pool_.get(), plain_encoded_value_size_)); + page_stats_base_ = page_stats_.get(); + row_group_stats_.reset( + new ColumnStats<T>(parent_->per_file_mem_pool_.get(), plain_encoded_value_size_)); + row_group_stats_base_ = row_group_stats_.get(); } protected: @@ -332,14 +339,14 @@ class HdfsParquetTableWriter::ColumnWriter : } uint8_t* dst_ptr = values_buffer_ + current_page_->header.uncompressed_page_size; int64_t written_len = - ParquetPlainEncoder::Encode(dst_ptr, plain_encoded_value_size_, *v); + ParquetPlainEncoder::Encode(*v, plain_encoded_value_size_, dst_ptr); DCHECK_EQ(*bytes_needed, written_len); current_page_->header.uncompressed_page_size += written_len; } else { // TODO: support other encodings here DCHECK(false); } - page_stats_.Update(*CastValue(value)); + page_stats_->Update(*CastValue(value)); return true; } @@ -366,10 +373,10 @@ class HdfsParquetTableWriter::ColumnWriter : StringValue temp_; // Tracks statistics per page. - ColumnStats<T> page_stats_; + scoped_ptr<ColumnStats<T>> page_stats_; // Tracks statistics per row group. This gets reset when starting a new row group. - ColumnStats<T> row_group_stats_; + scoped_ptr<ColumnStats<T>> row_group_stats_; // Converts a slot pointer to a raw value suitable for encoding inline T* CastValue(void* value) { @@ -394,7 +401,9 @@ class HdfsParquetTableWriter::BoolColumnWriter : public: BoolColumnWriter(HdfsParquetTableWriter* parent, ExprContext* ctx, const THdfsCompression::type& codec) - : BaseColumnWriter(parent, ctx, codec), page_stats_(-1), row_group_stats_(-1) { + : BaseColumnWriter(parent, ctx, codec), + page_stats_(parent_->reusable_col_mem_pool_.get(), -1), + row_group_stats_(parent_->reusable_col_mem_pool_.get(), -1) { DCHECK_EQ(ctx->root()->type().type, TYPE_BOOLEAN); bool_values_ = parent_->state_->obj_pool()->Add( new BitWriter(values_buffer_, values_buffer_len_)); @@ -985,6 +994,9 @@ Status HdfsParquetTableWriter::AppendRows( } } + // We exhausted the batch, so we materialize the statistics before releasing the memory. + for (unique_ptr<BaseColumnWriter>& column : columns_) column->MaterializeStatsValues(); + // Reset the row_idx_ when we exhaust the batch. We can exit before exhausting // the batch if we run out of file space and will continue from the last index. row_idx_ = 0; @@ -997,6 +1009,13 @@ Status HdfsParquetTableWriter::Finalize() { // At this point we write out the rest of the file. We first update the file // metadata, now that all the values have been seen. file_metadata_.num_rows = row_count_; + + // Set the ordering used to write parquet statistics for columns in the file. + ColumnOrder col_order = ColumnOrder(); + col_order.__set_TYPE_ORDER(TypeDefinedOrder()); + file_metadata_.column_orders.assign(columns_.size(), col_order); + file_metadata_.__isset.column_orders = true; + RETURN_IF_ERROR(FlushCurrentRowGroup()); RETURN_IF_ERROR(WriteFileFooter()); stats_.__set_parquet_stats(parquet_insert_stats_); http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/parquet-column-stats.cc ---------------------------------------------------------------------- diff --git a/be/src/exec/parquet-column-stats.cc b/be/src/exec/parquet-column-stats.cc index 5985017..bcd6fa4 100644 --- a/be/src/exec/parquet-column-stats.cc +++ b/be/src/exec/parquet-column-stats.cc @@ -17,70 +17,131 @@ #include "parquet-column-stats.inline.h" +#include <algorithm> #include <limits> +#include "common/names.h" + namespace impala { -bool ColumnStatsBase::ReadFromThrift(const parquet::Statistics& thrift_stats, - const ColumnType& col_type, const StatsField& stats_field, void* slot) { +bool ColumnStatsBase::ReadFromThrift(const parquet::ColumnChunk& col_chunk, + const ColumnType& col_type, const parquet::ColumnOrder* col_order, + StatsField stats_field, void* slot) { + if (!(col_chunk.__isset.meta_data && col_chunk.meta_data.__isset.statistics)) { + return false; + } + const parquet::Statistics& stats = col_chunk.meta_data.statistics; + + // Try to read the requested stats field. If it is not set, we may fall back to reading + // the old stats, based on the column type. + const string* stat_value = nullptr; + switch (stats_field) { + case StatsField::MIN: + if (stats.__isset.min_value && CanUseStats(col_type, col_order)) { + stat_value = &stats.min_value; + break; + } + if (stats.__isset.min && CanUseDeprecatedStats(col_type, col_order)) { + stat_value = &stats.min; + } + break; + case StatsField::MAX: + if (stats.__isset.max_value && CanUseStats(col_type, col_order)) { + stat_value = &stats.max_value; + break; + } + if (stats.__isset.max && CanUseDeprecatedStats(col_type, col_order)) { + stat_value = &stats.max; + } + break; + default: + DCHECK(false) << "Unsupported statistics field requested"; + } + if (stat_value == nullptr) return false; + switch (col_type.type) { case TYPE_BOOLEAN: - return ColumnStats<bool>::ReadFromThrift(thrift_stats, stats_field, slot); + return ColumnStats<bool>::DecodePlainValue(*stat_value, slot); case TYPE_TINYINT: { - // parquet::Statistics encodes INT_8 values using 4 bytes. - int32_t col_stats; - bool ret = ColumnStats<int32_t>::ReadFromThrift(thrift_stats, stats_field, - &col_stats); - if (!ret || col_stats < std::numeric_limits<int8_t>::min() || - col_stats > std::numeric_limits<int8_t>::max()) { - return false; - } - *static_cast<int8_t*>(slot) = col_stats; - return true; + // parquet::Statistics encodes INT_8 values using 4 bytes. + int32_t col_stats; + bool ret = ColumnStats<int32_t>::DecodePlainValue(*stat_value, &col_stats); + if (!ret || col_stats < std::numeric_limits<int8_t>::min() || + col_stats > std::numeric_limits<int8_t>::max()) { + return false; } + *static_cast<int8_t*>(slot) = col_stats; + return true; + } case TYPE_SMALLINT: { - // parquet::Statistics encodes INT_16 values using 4 bytes. - int32_t col_stats; - bool ret = ColumnStats<int32_t>::ReadFromThrift(thrift_stats, stats_field, - &col_stats); - if (!ret || col_stats < std::numeric_limits<int16_t>::min() || - col_stats > std::numeric_limits<int16_t>::max()) { - return false; - } - *static_cast<int16_t*>(slot) = col_stats; - return true; + // parquet::Statistics encodes INT_16 values using 4 bytes. + int32_t col_stats; + bool ret = ColumnStats<int32_t>::DecodePlainValue(*stat_value, &col_stats); + if (!ret || col_stats < std::numeric_limits<int16_t>::min() || + col_stats > std::numeric_limits<int16_t>::max()) { + return false; } + *static_cast<int16_t*>(slot) = col_stats; + return true; + } case TYPE_INT: - return ColumnStats<int32_t>::ReadFromThrift(thrift_stats, stats_field, slot); + return ColumnStats<int32_t>::DecodePlainValue(*stat_value, slot); case TYPE_BIGINT: - return ColumnStats<int64_t>::ReadFromThrift(thrift_stats, stats_field, slot); + return ColumnStats<int64_t>::DecodePlainValue(*stat_value, slot); case TYPE_FLOAT: - return ColumnStats<float>::ReadFromThrift(thrift_stats, stats_field, slot); + return ColumnStats<float>::DecodePlainValue(*stat_value, slot); case TYPE_DOUBLE: - return ColumnStats<double>::ReadFromThrift(thrift_stats, stats_field, slot); + return ColumnStats<double>::DecodePlainValue(*stat_value, slot); case TYPE_TIMESTAMP: - /// TODO add support for TimestampValue (IMPALA-4819) - break; + return ColumnStats<TimestampValue>::DecodePlainValue(*stat_value, slot); case TYPE_STRING: case TYPE_VARCHAR: + return ColumnStats<StringValue>::DecodePlainValue(*stat_value, slot); case TYPE_CHAR: - /// TODO add support for StringValue (IMPALA-4817) - break; + /// We don't read statistics for CHAR columns, since CHAR support is broken in + /// Impala (IMPALA-1652). + return false; case TYPE_DECIMAL: - /// TODO add support for DecimalValue (IMPALA-4815) switch (col_type.GetByteSize()) { case 4: - break; + return ColumnStats<Decimal4Value>::DecodePlainValue(*stat_value, slot); case 8: - break; + return ColumnStats<Decimal8Value>::DecodePlainValue(*stat_value, slot); case 16: - break; + return ColumnStats<Decimal16Value>::DecodePlainValue(*stat_value, slot); } - break; + DCHECK(false) << "Unknown decimal byte size: " << col_type.GetByteSize(); default: DCHECK(false) << col_type.DebugString(); } return false; } +void ColumnStatsBase::CopyToBuffer(StringBuffer* buffer, StringValue* value) { + if (value->ptr == buffer->buffer()) return; + buffer->Clear(); + buffer->Append(value->ptr, value->len); + value->ptr = buffer->buffer(); +} + +bool ColumnStatsBase::CanUseStats( + const ColumnType& col_type, const parquet::ColumnOrder* col_order) { + // If column order is not set, only statistics for numeric types can be trusted. + if (col_order == nullptr) { + return col_type.IsBooleanType() || col_type.IsIntegerType() + || col_type.IsFloatingPointType(); + } + // Stats can be used if the column order is TypeDefinedOrder (see parquet.thrift). + return col_order->__isset.TYPE_ORDER; +} + +bool ColumnStatsBase::CanUseDeprecatedStats( + const ColumnType& col_type, const parquet::ColumnOrder* col_order) { + // If column order is set to something other than TypeDefinedOrder, we shall not use the + // stats (see parquet.thrift). + if (col_order != nullptr && !col_order->__isset.TYPE_ORDER) return false; + return col_type.IsBooleanType() || col_type.IsIntegerType() + || col_type.IsFloatingPointType(); +} + } // end ns impala http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/parquet-column-stats.h ---------------------------------------------------------------------- diff --git a/be/src/exec/parquet-column-stats.h b/be/src/exec/parquet-column-stats.h index 995f20a..53fe548 100644 --- a/be/src/exec/parquet-column-stats.h +++ b/be/src/exec/parquet-column-stats.h @@ -21,30 +21,32 @@ #include <string> #include <type_traits> +#include "runtime/decimal-value.h" +#include "runtime/string-buffer.h" #include "runtime/timestamp-value.h" #include "runtime/types.h" -#include "exec/parquet-common.h" namespace impala { -/// This class, together with its derivatives, is used to track column statistics when +/// This class, together with its derivatives, is used to update column statistics when /// writing parquet files. It provides an interface to populate a parquet::Statistics /// object and attach it to an object supplied by the caller. It can also be used to /// decode parquet::Statistics into slots. /// -/// We currently support tracking 'min' and 'max' values for statistics. The other two -/// statistical values in parquet.thrift, 'null_count' and 'distinct_count' are not -/// tracked or populated. +/// We currently support writing the 'min_value' and 'max_value' fields in +/// parquet::Statistics. The other two statistical values - 'null_count' and +/// 'distinct_count' - are not tracked or populated. We do not populate the deprecated +/// 'min' and 'max' fields. /// -/// Regarding the ordering of values, we follow the parquet-mr reference implementation. +/// Regarding the ordering of values, we follow the parquet-format specification for +/// logical types (LogicalTypes.md in parquet-format): /// -/// Numeric values (BOOLEAN, INT, FLOAT, DOUBLE) are ordered by their numeric -/// value (as opposed to their binary representation). +/// - Numeric values (BOOLEAN, INT, FLOAT, DOUBLE, DECIMAL) are ordered by their numeric +/// value (as opposed to their binary representation). /// -/// We currently don't write statistics for DECIMAL values and character array values -/// (CHAR, VARCHAR, STRING) due to several issues with parquet-mr and subsequently, Hive -/// (PARQUET-251, PARQUET-686). For those types, the Update() method is empty, so that the -/// stats are not tracked. +/// - Strings are ordered using bytewise, unsigned comparison. +/// +/// - Timestamps are compared by numerically comparing the points in time they represent. /// /// NULL values are not considered for min/max statistics, and if a column consists only /// of NULL values, then no min/max statistics are written. @@ -55,23 +57,32 @@ namespace impala { /// TODO: Populate null_count and distinct_count. class ColumnStatsBase { public: - /// Enum to select statistics value when reading from parquet::Statistics structs. - enum class StatsField { MAX, MIN, NULL_COUNT, DISTINCT_COUNT }; + /// Enum to select whether to read minimum or maximum statistics. Values do not + /// correspond to fields in parquet::Statistics, but instead select between retrieving + /// the minimum or maximum value. + enum class StatsField { MIN, MAX }; ColumnStatsBase() : has_values_(false) {} virtual ~ColumnStatsBase() {} - /// Decodes the parquet::Statistics from 'row_group' and writes the value selected by + /// Decodes the parquet::Statistics from 'col_chunk' and writes the value selected by /// 'stats_field' into the buffer pointed to by 'slot', based on 'col_type'. Returns - /// 'true' if reading statistics for columns of type 'col_type' is supported and - /// decoding was successful, 'false' otherwise. - static bool ReadFromThrift(const parquet::Statistics& thrift_stats, - const ColumnType& col_type, const StatsField& stats_field, void* slot); + /// true if reading statistics for columns of type 'col_type' is supported and decoding + /// was successful, false otherwise. + static bool ReadFromThrift(const parquet::ColumnChunk& col_chunk, + const ColumnType& col_type, const parquet::ColumnOrder* col_order, + StatsField stats_field, void* slot); /// Merges this statistics object with values from 'other'. If other has not been /// initialized, then this object will not be changed. virtual void Merge(const ColumnStatsBase& other) = 0; + /// Copies the contents of this object's statistics values to internal buffers. Some + /// data types (e.g. StringValue) need to be copied at the end of processing a row + /// batch, since the batch memory will be released. Overwrite this method in derived + /// classes to provide the functionality. + virtual void MaterializeStringValuesToInternalBuffers() {} + /// Returns the number of bytes needed to encode the current statistics into a /// parquet::Statistics object. virtual int64_t BytesNeeded() const = 0; @@ -85,13 +96,32 @@ class ColumnStatsBase { bool has_values() const { return has_values_; } protected: + // Copies the memory of 'value' into 'buffer' and make 'value' point to 'buffer'. + // 'buffer' is reset before making the copy. + static void CopyToBuffer(StringBuffer* buffer, StringValue* value); + /// Stores whether the current object has been initialized with a set of values. bool has_values_; + + private: + /// Returns true if we support reading statistics stored in the fields 'min_value' and + /// 'max_value' in parquet::Statistics for the type 'col_type' and the column order + /// 'col_order'. Otherwise, returns false. If 'col_order' is nullptr, only primitive + /// numeric types are supported. + static bool CanUseStats( + const ColumnType& col_type, const parquet::ColumnOrder* col_order); + + /// Returns true if we consider statistics stored in the deprecated fields 'min' and + /// 'max' in parquet::Statistics to be correct for the type 'col_type' and the column + /// order 'col_order'. Otherwise, returns false. + static bool CanUseDeprecatedStats( + const ColumnType& col_type, const parquet::ColumnOrder* col_order); }; -/// This class contains the type-specific behavior to track statistics per column. +/// This class contains behavior specific to our in-memory formats for different types. template <typename T> class ColumnStats : public ColumnStatsBase { + friend class ColumnStatsBase; // We explicitly require types to be listed here in order to support column statistics. // When adding a type here, users of this class need to ensure that the statistics // follow the ordering semantics of parquet's min/max statistics for the new type. @@ -108,34 +138,40 @@ class ColumnStats : public ColumnStatsBase { T>::type; public: - ColumnStats(int plain_encoded_value_size) - : ColumnStatsBase(), plain_encoded_value_size_(plain_encoded_value_size) {} - - /// Decodes the parquet::Statistics from 'row_group' and writes the value selected by - /// 'stats_field' into the buffer pointed to by 'slot'. Returns 'true' if reading - /// statistics for columns of type 'col_type' is supported and decoding was successful, - /// 'false' otherwise. - static bool ReadFromThrift(const parquet::Statistics& thrift_stats, - const StatsField& stats_field, void* slot); + /// 'mem_pool' is used to materialize string values so that the user of this class can + /// free the memory of the original values. + /// 'plain_encoded_value_size' specifies the size of each encoded value in plain + /// encoding, -1 if the type is variable-length. + ColumnStats(MemPool* mem_pool, int plain_encoded_value_size) + : ColumnStatsBase(), + plain_encoded_value_size_(plain_encoded_value_size), + mem_pool_(mem_pool), + min_buffer_(mem_pool), + max_buffer_(mem_pool) {} /// Updates the statistics based on the value 'v'. If necessary, initializes the - /// statistics. + /// statistics. It may keep a reference to 'v' until + /// MaterializeStringValuesToInternalBuffers() gets called. void Update(const T& v); virtual void Merge(const ColumnStatsBase& other) override; + virtual void MaterializeStringValuesToInternalBuffers() override {} + virtual int64_t BytesNeeded() const override; virtual void EncodeToThrift(parquet::Statistics* out) const override; protected: - /// Encodes a single value using parquet's PLAIN encoding and stores it into the - /// binary string 'out'. - void EncodeValueToString(const T& v, std::string* out) const; + /// Encodes a single value using parquet's plain encoding and stores it into the + /// binary string 'out'. String values are stored without additional encoding. + static void EncodePlainValue(const T& v, int64_t bytes_needed, std::string* out); - /// Decodes a statistics values from 'buffer' into 'result'. - static bool DecodeValueFromThrift(const std::string& buffer, T* result); + /// Decodes the plain encoded stats value from 'buffer' and writes the result into the + /// buffer pointed to by 'slot'. Returns true if decoding was successful, false + /// otherwise. For timestamps, an additional validation will be performed. + static bool DecodePlainValue(const std::string& buffer, void* slot); /// Returns the number of bytes needed to encode value 'v'. - int64_t BytesNeededInternal(const T& v) const; + int64_t BytesNeeded(const T& v) const; // Size of each encoded value in plain encoding, -1 if the type is variable-length. int plain_encoded_value_size_; @@ -145,6 +181,13 @@ class ColumnStats : public ColumnStatsBase { // Maximum value since the last call to Reset(). T max_value_; + + // Memory pool to allocate from when making copies of the statistics data. + MemPool* mem_pool_; + + // Local buffers to copy statistics data into. + StringBuffer min_buffer_; + StringBuffer max_buffer_; }; } // end ns impala http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/parquet-column-stats.inline.h ---------------------------------------------------------------------- diff --git a/be/src/exec/parquet-column-stats.inline.h b/be/src/exec/parquet-column-stats.inline.h index 3738980..48c1b2e 100644 --- a/be/src/exec/parquet-column-stats.inline.h +++ b/be/src/exec/parquet-column-stats.inline.h @@ -18,26 +18,13 @@ #ifndef IMPALA_EXEC_PARQUET_COLUMN_STATS_INLINE_H #define IMPALA_EXEC_PARQUET_COLUMN_STATS_INLINE_H +#include "exec/parquet-common.h" #include "parquet-column-stats.h" +#include "runtime/string-value.inline.h" namespace impala { template <typename T> -inline bool ColumnStats<T>::ReadFromThrift(const parquet::Statistics& thrift_stats, - const StatsField& stats_field, void* slot) { - T* out = reinterpret_cast<T*>(slot); - switch (stats_field) { - case StatsField::MIN: - return DecodeValueFromThrift(thrift_stats.min, out); - case StatsField::MAX: - return DecodeValueFromThrift(thrift_stats.max, out); - default: - DCHECK(false) << "Unsupported statistics field requested"; - return false; - } -} - -template <typename T> inline void ColumnStats<T>::Update(const T& v) { if (!has_values_) { has_values_ = true; @@ -66,40 +53,40 @@ inline void ColumnStats<T>::Merge(const ColumnStatsBase& other) { template <typename T> inline int64_t ColumnStats<T>::BytesNeeded() const { - return BytesNeededInternal(min_value_) + BytesNeededInternal(max_value_); + return BytesNeeded(min_value_) + BytesNeeded(max_value_); } template <typename T> inline void ColumnStats<T>::EncodeToThrift(parquet::Statistics* out) const { DCHECK(has_values_); std::string min_str; - EncodeValueToString(min_value_, &min_str); - out->__set_min(move(min_str)); + EncodePlainValue(min_value_, BytesNeeded(min_value_), &min_str); + out->__set_min_value(move(min_str)); std::string max_str; - EncodeValueToString(max_value_, &max_str); - out->__set_max(move(max_str)); + EncodePlainValue(max_value_, BytesNeeded(max_value_), &max_str); + out->__set_max_value(move(max_str)); } template <typename T> -inline void ColumnStats<T>::EncodeValueToString(const T& v, std::string* out) const { - int64_t bytes_needed = BytesNeededInternal(v); +inline void ColumnStats<T>::EncodePlainValue( + const T& v, int64_t bytes_needed, std::string* out) { out->resize(bytes_needed); int64_t bytes_written = ParquetPlainEncoder::Encode( - reinterpret_cast<uint8_t*>(&(*out)[0]), bytes_needed, v); + v, bytes_needed, reinterpret_cast<uint8_t*>(&(*out)[0])); DCHECK_EQ(bytes_needed, bytes_written); } template <typename T> -inline bool ColumnStats<T>::DecodeValueFromThrift(const std::string& buffer, T* result) { +inline bool ColumnStats<T>::DecodePlainValue(const std::string& buffer, void* slot) { + T* result = reinterpret_cast<T*>(slot); int size = buffer.size(); - // The ParquetPlainEncoder interface expects mutable pointers. - uint8_t* data = const_cast<uint8_t*>(reinterpret_cast<const uint8_t*>(&buffer[0])); - if (ParquetPlainEncoder::Decode(data, data + size, -1, result) == -1) return false; + const uint8_t* data = reinterpret_cast<const uint8_t*>(&buffer[0]); + if (ParquetPlainEncoder::Decode(data, data + size, size, result) == -1) return false; return true; } template <typename T> -inline int64_t ColumnStats<T>::BytesNeededInternal(const T& v) const { +inline int64_t ColumnStats<T>::BytesNeeded(const T& v) const { return plain_encoded_value_size_ < 0 ? ParquetPlainEncoder::ByteSize<T>(v) : plain_encoded_value_size_; } @@ -107,46 +94,107 @@ inline int64_t ColumnStats<T>::BytesNeededInternal(const T& v) const { /// Plain encoding for Boolean values is not handled by the ParquetPlainEncoder and thus /// needs special handling here. template <> -inline void ColumnStats<bool>::EncodeValueToString(const bool& v, std::string* out) const -{ +inline void ColumnStats<bool>::EncodePlainValue( + const bool& v, int64_t bytes_needed, std::string* out) { char c = v; out->assign(1, c); } template <> -inline bool ColumnStats<bool>::DecodeValueFromThrift(const std::string& buffer, - bool* result) { +inline bool ColumnStats<bool>::DecodePlainValue(const std::string& buffer, void* slot) { + bool* result = reinterpret_cast<bool*>(slot); DCHECK(buffer.size() == 1); *result = (buffer[0] != 0); return true; } template <> -inline int64_t ColumnStats<bool>::BytesNeededInternal(const bool& v) const { +inline int64_t ColumnStats<bool>::BytesNeeded(const bool& v) const { return 1; } -/// parquet-mr and subsequently Hive currently do not handle the following types -/// correctly (PARQUET-251, PARQUET-686), so we disable support for them. -/// The relevant Impala Jiras are for -/// - StringValue IMPALA-4817 -/// - TimestampValue IMPALA-4819 -/// - DecimalValue IMPALA-4815 +/// Timestamp values need validation. template <> -inline void ColumnStats<StringValue>::Update(const StringValue& v) {} +inline bool ColumnStats<TimestampValue>::DecodePlainValue( + const std::string& buffer, void* slot) { + TimestampValue* result = reinterpret_cast<TimestampValue*>(slot); + int size = buffer.size(); + const uint8_t* data = reinterpret_cast<const uint8_t*>(&buffer[0]); + if (ParquetPlainEncoder::Decode(data, data + size, size, result) == -1) return false; + // We don't need to convert the value here, since we don't support reading timestamp + // statistics written by Hive / old versions of parquet-mr. Should Hive add support for + // writing new statistics for the deprecated timestamp type, we will have to add support + // for conversion here. + return result->IsValidDate(); +} +/// parquet::Statistics stores string values directly and does not use plain encoding. template <> -inline void ColumnStats<TimestampValue>::Update(const TimestampValue& v) {} +inline void ColumnStats<StringValue>::EncodePlainValue( + const StringValue& v, int64_t bytes_needed, string* out) { + out->assign(v.ptr, v.len); +} template <> -inline void ColumnStats<Decimal4Value>::Update(const Decimal4Value& v) {} +inline bool ColumnStats<StringValue>::DecodePlainValue( + const std::string& buffer, void* slot) { + StringValue* result = reinterpret_cast<StringValue*>(slot); + result->ptr = const_cast<char*>(&buffer[0]); + result->len = buffer.size(); + return true; +} template <> -inline void ColumnStats<Decimal8Value>::Update(const Decimal8Value& v) {} +inline void ColumnStats<StringValue>::Update(const StringValue& v) { + if (!has_values_) { + has_values_ = true; + min_value_ = v; + min_buffer_.Clear(); + max_value_ = v; + max_buffer_.Clear(); + } else { + if (v < min_value_) { + min_value_ = v; + min_buffer_.Clear(); + } + if (v > max_value_) { + max_value_ = v; + max_buffer_.Clear(); + } + } +} template <> -inline void ColumnStats<Decimal16Value>::Update(const Decimal16Value& v) {} +inline void ColumnStats<StringValue>::Merge(const ColumnStatsBase& other) { + DCHECK(dynamic_cast<const ColumnStats<StringValue>*>(&other)); + const ColumnStats<StringValue>* cs = + static_cast<const ColumnStats<StringValue>*>(&other); + if (!cs->has_values_) return; + if (!has_values_) { + has_values_ = true; + min_value_ = cs->min_value_; + min_buffer_.Clear(); + max_value_ = cs->max_value_; + max_buffer_.Clear(); + } else { + if (cs->min_value_ < min_value_) { + min_value_ = cs->min_value_; + min_buffer_.Clear(); + } + if (cs->max_value_ > max_value_) { + max_value_ = cs->max_value_; + max_buffer_.Clear(); + } + } +} +// StringValues need to be copied at the end of processing a row batch, since the batch +// memory will be released. +template <> +inline void ColumnStats<StringValue>::MaterializeStringValuesToInternalBuffers() { + if (min_buffer_.IsEmpty()) CopyToBuffer(&min_buffer_, &min_value_); + if (max_buffer_.IsEmpty()) CopyToBuffer(&max_buffer_, &max_value_); +} } // end ns impala #endif http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/parquet-common.h ---------------------------------------------------------------------- diff --git a/be/src/exec/parquet-common.h b/be/src/exec/parquet-common.h index ff82fed..53de54a 100644 --- a/be/src/exec/parquet-common.h +++ b/be/src/exec/parquet-common.h @@ -85,7 +85,7 @@ const parquet::CompressionCodec::type IMPALA_TO_PARQUET_CODEC[] = { class ParquetPlainEncoder { public: /// Returns the byte size of 'v'. - template<typename T> + template <typename T> static int ByteSize(const T& v) { return sizeof(T); } /// Returns the encoded size of values of type t. Returns -1 if it is variable @@ -167,8 +167,8 @@ class ParquetPlainEncoder { /// be preallocated and big enough. Buffer need not be aligned. /// 'fixed_len_size' is only applicable for data encoded using FIXED_LEN_BYTE_ARRAY and /// is the number of bytes the plain encoder should use. - template<typename T> - static int Encode(uint8_t* buffer, int fixed_len_size, const T& t) { + template <typename T> + static int Encode(const T& t, int fixed_len_size, uint8_t* buffer) { memcpy(buffer, &t, ByteSize(t)); return ByteSize(t); } @@ -177,8 +177,8 @@ class ParquetPlainEncoder { /// need not be aligned. For types that are stored as FIXED_LEN_BYTE_ARRAY, /// 'fixed_len_size' is the size of the object. Otherwise, it is unused. /// Returns the number of bytes read or -1 if the value was not decoded successfully. - template<typename T> - static int Decode(uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, + template <typename T> + static int Decode(const uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, T* v) { int byte_size = ByteSize(*v); if (UNLIKELY(buffer_end - buffer < byte_size)) return -1; @@ -189,56 +189,58 @@ class ParquetPlainEncoder { /// Calling this with arguments of type ColumnType is certainly a programmer error, so we /// disallow it. -template <> -int ParquetPlainEncoder::ByteSize(const ColumnType& t); +template <> int ParquetPlainEncoder::ByteSize(const ColumnType& t); /// Disable for bools. Plain encoding is not used for booleans. -template<> int ParquetPlainEncoder::ByteSize(const bool& b); -template<> int ParquetPlainEncoder::Encode(uint8_t*, int fixed_len_size, const bool&); -template<> int ParquetPlainEncoder::Decode(uint8_t*, const uint8_t*, int fixed_len_size, - bool* v); +template <> int ParquetPlainEncoder::ByteSize(const bool& b); +template <> int ParquetPlainEncoder::Encode(const bool&, int fixed_len_size, uint8_t*); +template <> int ParquetPlainEncoder::Decode(const uint8_t*, const uint8_t*, + int fixed_len_size, bool* v); /// Not used for decimals since the plain encoding encodes them using /// FIXED_LEN_BYTE_ARRAY. -template<> inline int ParquetPlainEncoder::ByteSize(const Decimal4Value&) { +template <> +inline int ParquetPlainEncoder::ByteSize(const Decimal4Value&) { DCHECK(false); return -1; } -template<> inline int ParquetPlainEncoder::ByteSize(const Decimal8Value&) { +template <> +inline int ParquetPlainEncoder::ByteSize(const Decimal8Value&) { DCHECK(false); return -1; } -template<> inline int ParquetPlainEncoder::ByteSize(const Decimal16Value&) { +template <> +inline int ParquetPlainEncoder::ByteSize(const Decimal16Value&) { DCHECK(false); return -1; } /// Parquet doesn't have 8-bit or 16-bit ints. They are converted to 32-bit. -template<> +template <> inline int ParquetPlainEncoder::ByteSize(const int8_t& v) { return sizeof(int32_t); } -template<> +template <> inline int ParquetPlainEncoder::ByteSize(const int16_t& v) { return sizeof(int32_t); } -template<> +template <> inline int ParquetPlainEncoder::ByteSize(const StringValue& v) { return sizeof(int32_t) + v.len; } -template<> +template <> inline int ParquetPlainEncoder::ByteSize(const TimestampValue& v) { return 12; } -template<> -inline int ParquetPlainEncoder::Decode(uint8_t* buffer, const uint8_t* buffer_end, +template <> +inline int ParquetPlainEncoder::Decode(const uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, int8_t* v) { int byte_size = ByteSize(*v); if (UNLIKELY(buffer_end - buffer < byte_size)) return -1; *v = *buffer; return byte_size; } -template<> -inline int ParquetPlainEncoder::Decode(uint8_t* buffer, const uint8_t* buffer_end, +template <> +inline int ParquetPlainEncoder::Decode(const uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, int16_t* v) { int byte_size = ByteSize(*v); if (UNLIKELY(buffer_end - buffer < byte_size)) return -1; @@ -246,38 +248,38 @@ inline int ParquetPlainEncoder::Decode(uint8_t* buffer, const uint8_t* buffer_en return byte_size; } -template<> +template <> inline int ParquetPlainEncoder::Encode( - uint8_t* buffer, int fixed_len_size, const int8_t& v) { + const int8_t& v, int fixed_len_size, uint8_t* buffer) { int32_t val = v; memcpy(buffer, &val, sizeof(int32_t)); return ByteSize(v); } -template<> +template <> inline int ParquetPlainEncoder::Encode( - uint8_t* buffer, int fixed_len_size, const int16_t& v) { + const int16_t& v, int fixed_len_size, uint8_t* buffer) { int32_t val = v; memcpy(buffer, &val, sizeof(int32_t)); return ByteSize(v); } -template<> +template <> inline int ParquetPlainEncoder::Encode( - uint8_t* buffer, int fixed_len_size, const StringValue& v) { + const StringValue& v, int fixed_len_size, uint8_t* buffer) { memcpy(buffer, &v.len, sizeof(int32_t)); memcpy(buffer + sizeof(int32_t), v.ptr, v.len); return ByteSize(v); } -template<> -inline int ParquetPlainEncoder::Decode( - uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, StringValue* v) { +template <> +inline int ParquetPlainEncoder::Decode(const uint8_t* buffer, const uint8_t* buffer_end, + int fixed_len_size, StringValue* v) { if (UNLIKELY(buffer_end - buffer < sizeof(int32_t))) return -1; memcpy(&v->len, buffer, sizeof(int32_t)); int byte_size = ByteSize(*v); if (UNLIKELY(v->len < 0 || buffer_end - buffer < byte_size)) return -1; - v->ptr = reinterpret_cast<char*>(buffer) + sizeof(int32_t); + v->ptr = reinterpret_cast<char*>(const_cast<uint8_t*>(buffer)) + sizeof(int32_t); if (fixed_len_size > 0) v->len = std::min(v->len, fixed_len_size); // we still read byte_size bytes, even if we truncate return byte_size; @@ -288,45 +290,45 @@ inline int ParquetPlainEncoder::Decode( /// that the value in the in-memory format has leading zeros or negative 1's. /// For example, precision 2 fits in 1 byte. All decimals stored as Decimal4Value /// will have 3 bytes of leading zeros, we will only store the interesting byte. -template<> +template <> inline int ParquetPlainEncoder::Encode( - uint8_t* buffer, int fixed_len_size, const Decimal4Value& v) { + const Decimal4Value& v, int fixed_len_size, uint8_t* buffer) { DecimalUtil::EncodeToFixedLenByteArray(buffer, fixed_len_size, v); return fixed_len_size; } -template<> +template <> inline int ParquetPlainEncoder::Encode( - uint8_t* buffer, int fixed_len_size, const Decimal8Value& v) { + const Decimal8Value& v, int fixed_len_size, uint8_t* buffer) { DecimalUtil::EncodeToFixedLenByteArray(buffer, fixed_len_size, v); return fixed_len_size; } -template<> +template <> inline int ParquetPlainEncoder::Encode( - uint8_t* buffer, int fixed_len_size, const Decimal16Value& v) { + const Decimal16Value& v, int fixed_len_size, uint8_t* buffer) { DecimalUtil::EncodeToFixedLenByteArray(buffer, fixed_len_size, v); return fixed_len_size; } -template<> -inline int ParquetPlainEncoder::Decode(uint8_t* buffer, const uint8_t* buffer_end, +template <> +inline int ParquetPlainEncoder::Decode(const uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, Decimal4Value* v) { if (UNLIKELY(buffer_end - buffer < fixed_len_size)) return -1; DecimalUtil::DecodeFromFixedLenByteArray(buffer, fixed_len_size, v); return fixed_len_size; } -template<> -inline int ParquetPlainEncoder::Decode(uint8_t* buffer, const uint8_t* buffer_end, +template <> +inline int ParquetPlainEncoder::Decode(const uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, Decimal8Value* v) { if (UNLIKELY(buffer_end - buffer < fixed_len_size)) return -1; DecimalUtil::DecodeFromFixedLenByteArray(buffer, fixed_len_size, v); return fixed_len_size; } -template<> -inline int ParquetPlainEncoder::Decode(uint8_t* buffer, const uint8_t* buffer_end, +template <> +inline int ParquetPlainEncoder::Decode(const uint8_t* buffer, const uint8_t* buffer_end, int fixed_len_size, Decimal16Value* v) { if (UNLIKELY(buffer_end - buffer < fixed_len_size)) return -1; DecimalUtil::DecodeFromFixedLenByteArray(buffer, fixed_len_size, v); @@ -334,5 +336,4 @@ inline int ParquetPlainEncoder::Decode(uint8_t* buffer, const uint8_t* buffer_en } } - #endif http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/parquet-metadata-utils.cc ---------------------------------------------------------------------- diff --git a/be/src/exec/parquet-metadata-utils.cc b/be/src/exec/parquet-metadata-utils.cc index 931161a..8ec0abd 100644 --- a/be/src/exec/parquet-metadata-utils.cc +++ b/be/src/exec/parquet-metadata-utils.cc @@ -227,13 +227,6 @@ Status ParquetMetadataUtils::ValidateColumn(const parquet::FileMetaData& file_me return Status::OK(); } -bool ParquetMetadataUtils::HasRowGroupStats(const parquet::RowGroup& row_group, - int col_idx) { - DCHECK(col_idx < row_group.columns.size()); - const parquet::ColumnChunk& col_chunk = row_group.columns[col_idx]; - return col_chunk.__isset.meta_data && col_chunk.meta_data.__isset.statistics; -} - ParquetFileVersion::ParquetFileVersion(const string& created_by) { string created_by_lower = created_by; std::transform(created_by_lower.begin(), created_by_lower.end(), http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/parquet-metadata-utils.h ---------------------------------------------------------------------- diff --git a/be/src/exec/parquet-metadata-utils.h b/be/src/exec/parquet-metadata-utils.h index 5e655b6..767a1a3 100644 --- a/be/src/exec/parquet-metadata-utils.h +++ b/be/src/exec/parquet-metadata-utils.h @@ -50,9 +50,6 @@ class ParquetMetadataUtils { const char* filename, int row_group_idx, int col_idx, const parquet::SchemaElement& schema_element, const SlotDescriptor* slot_desc, RuntimeState* state); - - /// Returns whether column 'col_idx' in 'row_group' has statistics attached to it. - static bool HasRowGroupStats(const parquet::RowGroup& row_group, int col_idx); }; struct ParquetFileVersion { http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/exec/parquet-plain-test.cc ---------------------------------------------------------------------- diff --git a/be/src/exec/parquet-plain-test.cc b/be/src/exec/parquet-plain-test.cc index 19c09ce..c86a33b 100644 --- a/be/src/exec/parquet-plain-test.cc +++ b/be/src/exec/parquet-plain-test.cc @@ -33,7 +33,7 @@ namespace impala { template <typename T> void TestTruncate(const T& v, int expected_byte_size) { uint8_t buffer[expected_byte_size]; - int encoded_size = ParquetPlainEncoder::Encode(buffer, expected_byte_size, v); + int encoded_size = ParquetPlainEncoder::Encode(v, expected_byte_size, buffer); EXPECT_EQ(encoded_size, expected_byte_size); // Check all possible truncations of the buffer. @@ -52,7 +52,7 @@ void TestTruncate(const T& v, int expected_byte_size) { template <typename T> void TestType(const T& v, int expected_byte_size) { uint8_t buffer[expected_byte_size]; - int encoded_size = ParquetPlainEncoder::Encode(buffer, expected_byte_size, v); + int encoded_size = ParquetPlainEncoder::Encode(v, expected_byte_size, buffer); EXPECT_EQ(encoded_size, expected_byte_size); T result; @@ -134,15 +134,15 @@ TEST(PlainEncoding, DecimalBigEndian) { memcpy(&d8, buffer, sizeof(d8)); memcpy(&d16, buffer, sizeof(d16)); - int size = ParquetPlainEncoder::Encode(result_buffer, sizeof(d4), d4); + int size = ParquetPlainEncoder::Encode(d4, sizeof(d4), result_buffer); ASSERT_EQ(size, sizeof(d4)); ASSERT_EQ(memcmp(result_buffer, buffer_swapped + 16 - sizeof(d4), sizeof(d4)), 0); - size = ParquetPlainEncoder::Encode(result_buffer, sizeof(d8), d8); + size = ParquetPlainEncoder::Encode(d8, sizeof(d8), result_buffer); ASSERT_EQ(size, sizeof(d8)); ASSERT_EQ(memcmp(result_buffer, buffer_swapped + 16 - sizeof(d8), sizeof(d8)), 0); - size = ParquetPlainEncoder::Encode(result_buffer, sizeof(d16), d16); + size = ParquetPlainEncoder::Encode(d16, sizeof(d16), result_buffer); ASSERT_EQ(size, sizeof(d16)); ASSERT_EQ(memcmp(result_buffer, buffer_swapped + 16 - sizeof(d16), sizeof(d16)), 0); } http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/be/src/util/dict-encoding.h ---------------------------------------------------------------------- diff --git a/be/src/util/dict-encoding.h b/be/src/util/dict-encoding.h index 0b74468..e842495 100644 --- a/be/src/util/dict-encoding.h +++ b/be/src/util/dict-encoding.h @@ -314,7 +314,7 @@ inline bool DictDecoder<Decimal16Value>::GetNextValue(Decimal16Value* value) { template<typename T> inline void DictEncoder<T>::WriteDict(uint8_t* buffer) { for (const Node& node: nodes_) { - buffer += ParquetPlainEncoder::Encode(buffer, encoded_value_size_, node.value); + buffer += ParquetPlainEncoder::Encode(node.value, encoded_value_size_, buffer); } } http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/common/thrift/parquet.thrift ---------------------------------------------------------------------- diff --git a/common/thrift/parquet.thrift b/common/thrift/parquet.thrift index 81073f4..c4afb77 100644 --- a/common/thrift/parquet.thrift +++ b/common/thrift/parquet.thrift @@ -9,24 +9,36 @@ * * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. */ /** * File format description for the parquet file format */ namespace cpp parquet -namespace java parquet.format +namespace java org.apache.parquet.format /** * Types supported by Parquet. These types are intended to be used in combination * with the encodings to control the on disk storage format. * For example INT16 is not included as a type since a good encoding of INT32 * would handle this. + * + * When a logical type is not present, the type-defined sort order of these + * physical types are: + * * BOOLEAN - false, true + * * INT32 - signed comparison + * * INT64 - signed comparison + * * INT96 - signed comparison + * * FLOAT - signed comparison + * * DOUBLE - signed comparison + * * BYTE_ARRAY - unsigned byte-wise comparison + * * FIXED_LEN_BYTE_ARRAY - unsigned byte-wise comparison */ enum Type { BOOLEAN = 0; @@ -92,7 +104,14 @@ enum ConvertedType { * as an INT32 physical type. */ TIME_MILLIS = 7; - // RESERVED = 8; + + /** + * A time. + * + * The total number of microseconds since midnight. The value is stored as + * an INT64 physical type. + */ + TIME_MICROS = 8; /** * A date/time combination @@ -101,7 +120,14 @@ enum ConvertedType { * a physical type of INT64. */ TIMESTAMP_MILLIS = 9; - // RESERVED = 10; + + /** + * A date/time combination + * + * Date and time recorded as microseconds since the Unix epoch. The value is + * stored as an INT64 physical type. + */ + TIMESTAMP_MICROS = 10; /** @@ -180,13 +206,33 @@ enum FieldRepetitionType { * All fields are optional. */ struct Statistics { - /** min and max value of the column, encoded in PLAIN encoding */ + /** + * DEPRECATED: min and max value of the column. Use min_value and max_value. + * + * Values are encoded using PLAIN encoding, except that variable-length byte + * arrays do not include a length prefix. + * + * These fields encode min and max values determined by SIGNED comparison + * only. New files should use the correct order for a column's logical type + * and store the values in the min_value and max_value fields. + * + * To support older readers, these may be set when the column order is + * SIGNED. + */ 1: optional binary max; 2: optional binary min; /** count of null value in the column */ 3: optional i64 null_count; /** count of distinct values occurring */ 4: optional i64 distinct_count; + /** + * Min and max values for the column, determined by its ColumnOrder. + * + * Values are encoded using PLAIN encoding, except that variable-length byte + * arrays do not include a length prefix. + */ + 5: optional binary max_value; + 6: optional binary min_value; } /** @@ -268,7 +314,7 @@ enum Encoding { */ PLAIN_DICTIONARY = 2; - /** Group packed run length encoding. Usable for definition/repetition levels + /** Group packed run length encoding. Usable for definition/reptition levels * encoding and Booleans (on one bit: 0 is false; 1 is true.) */ RLE = 3; @@ -306,6 +352,7 @@ enum CompressionCodec { SNAPPY = 1; GZIP = 2; LZO = 3; + BROTLI = 4; } enum PageType { @@ -440,6 +487,7 @@ struct PageEncodingStats { /** number of pages of this type with this encoding **/ 3: required i32 count; + } /** @@ -506,6 +554,9 @@ struct ColumnChunk { } struct RowGroup { + /** Metadata for each column chunk in this row group. + * This list must have the same order as the SchemaElement list in FileMetaData. + **/ 1: required list<ColumnChunk> columns /** Total byte size of all the uncompressed column data in this row group **/ @@ -520,6 +571,23 @@ struct RowGroup { 4: optional list<SortingColumn> sorting_columns } +/** Empty struct to signal the order defined by the physical or logical type */ +struct TypeDefinedOrder {} + +/** + * Union to specify the order used for min, max, and sorting values in a column. + * + * Possible values are: + * * TypeDefinedOrder - the column uses the order defined by its logical or + * physical type (if there is no logical type). + * + * If the reader does not support the value of this union, min and max stats + * for this column should be ignored. + */ +union ColumnOrder { + 1: TypeDefinedOrder TYPE_ORDER; +} + /** * Description for file metadata */ @@ -549,5 +617,14 @@ struct FileMetaData { * e.g. impala version 1.0 (build 6cf94d29b2b7115df4de2c06e2ab4326d721eb55) **/ 6: optional string created_by + + /** + * Sort order used for each column in this file. + * + * If this list is not present, then the order for each column is assumed to + * be Signed. In addition, min and max values for INTERVAL or DECIMAL stored + * as fixed or bytes should be ignored. + */ + 7: optional list<ColumnOrder> column_orders; } http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/testdata/data/README ---------------------------------------------------------------------- diff --git a/testdata/data/README b/testdata/data/README index 23b0586..4066a8d 100644 --- a/testdata/data/README +++ b/testdata/data/README @@ -102,3 +102,7 @@ Created with a text editor, contains two header lines before the data rows. table_with_header.gz, table_with_header_2.gz: Generated by gzip'ing table_with_header.csv and table_with_header_2.csv. +deprecated_statistics.parquet: +Generated with with hive shell, which uses parquet-mr version 1.5.0-cdh5.12.0-SNAPSHOT +Contains a copy of the data in functional.alltypessmall with statistics that use the old +'min'/'max' fields. http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/testdata/data/deprecated_statistics.parquet ---------------------------------------------------------------------- diff --git a/testdata/data/deprecated_statistics.parquet b/testdata/data/deprecated_statistics.parquet new file mode 100644 index 0000000..fa23deb Binary files /dev/null and b/testdata/data/deprecated_statistics.parquet differ http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/testdata/workloads/functional-query/queries/QueryTest/parquet-deprecated-stats.test ---------------------------------------------------------------------- diff --git a/testdata/workloads/functional-query/queries/QueryTest/parquet-deprecated-stats.test b/testdata/workloads/functional-query/queries/QueryTest/parquet-deprecated-stats.test new file mode 100644 index 0000000..be6e345 --- /dev/null +++ b/testdata/workloads/functional-query/queries/QueryTest/parquet-deprecated-stats.test @@ -0,0 +1,144 @@ +==== +---- QUERY +select id, bool_col from deprecated_stats where int_col < 0 +---- RESULTS +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where tinyint_col < 0 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where smallint_col < 0 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where int_col < 0 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where bigint_col < 0 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where float_col < 0 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where double_col < 0 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +# Test with inverted predicate +select id, bool_col from deprecated_stats where -1 > int_col +---- RESULTS +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where tinyint_col > 9 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where smallint_col > 9 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select id, bool_col from deprecated_stats where int_col > 9 +---- RESULTS +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where bigint_col > 90 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where float_col > 9.9 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where double_col > 99 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from deprecated_stats where string_col < "0" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +==== +---- QUERY +select count(*) from deprecated_stats where string_col > ":" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +==== +---- QUERY +select count(*) from deprecated_stats where timestamp_col = "2119-02-01 00:00:00" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +==== +---- QUERY +# Test that adding a column without stats will not disable stats-based pruning. +select count(*) from deprecated_stats where int_col < 0 and timestamp_col != "2009-02-01 00:00:00" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/testdata/workloads/functional-query/queries/QueryTest/parquet-filtering.test ---------------------------------------------------------------------- diff --git a/testdata/workloads/functional-query/queries/QueryTest/parquet-filtering.test b/testdata/workloads/functional-query/queries/QueryTest/parquet-filtering.test index 932ba60..5e863f1 100644 --- a/testdata/workloads/functional-query/queries/QueryTest/parquet-filtering.test +++ b/testdata/workloads/functional-query/queries/QueryTest/parquet-filtering.test @@ -181,7 +181,8 @@ select count(*) from functional_parquet.alltypes where date_string_col = '01/01/ 0 ---- RUNTIME_PROFILE aggregation(SUM, NumRowGroups): 24 -aggregation(SUM, NumDictFilteredRowGroups): 24 +aggregation(SUM, NumDictFilteredRowGroups): 2 +aggregation(SUM, NumStatsFilteredRowGroups): 22 ==== ---- QUERY # string_col: All values pass http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/92703468/testdata/workloads/functional-query/queries/QueryTest/parquet_stats.test ---------------------------------------------------------------------- diff --git a/testdata/workloads/functional-query/queries/QueryTest/parquet_stats.test b/testdata/workloads/functional-query/queries/QueryTest/parquet_stats.test index f88d1df..c149dc6 100644 --- a/testdata/workloads/functional-query/queries/QueryTest/parquet_stats.test +++ b/testdata/workloads/functional-query/queries/QueryTest/parquet_stats.test @@ -3,8 +3,8 @@ select id, bool_col from functional_parquet.alltypessmall where int_col < 0 ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY set explain_level=2; @@ -17,119 +17,119 @@ select count(*) from functional_parquet.alltypessmall where tinyint_col < 0 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where smallint_col < 0 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where int_col < 0 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where bigint_col < 0 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where float_col < 0 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where double_col < 0 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY # Test with inverted predicate select id, bool_col from functional_parquet.alltypessmall where -1 > int_col ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where tinyint_col > 9 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where smallint_col > 9 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select id, bool_col from functional_parquet.alltypessmall where int_col > 9 ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where bigint_col > 90 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where float_col > 9.9 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where double_col > 99 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where tinyint_col >= 10 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where tinyint_col <= 0 ---- RESULTS 12 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 0 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 0 ==== ==== ---- QUERY @@ -137,24 +137,24 @@ select count(*) from functional_parquet.alltypessmall where tinyint_col >= 9 ---- RESULTS 8 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 0 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 0 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where tinyint_col = -1 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select count(*) from functional_parquet.alltypessmall where tinyint_col = 10 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY set explain_level=2; @@ -167,8 +167,8 @@ select count(*) from functional_parquet.alltypessmall where id >= 30 and id <= 8 ---- RESULTS 51 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 1 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 1 ==== ---- QUERY # Mix with partitioning columns @@ -176,31 +176,22 @@ select count(*) from functional_parquet.alltypessmall where int_col < 0 and year ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* -==== ----- QUERY -# Test that adding a column without stats will not disable stats-based pruning. -select count(*) from functional_parquet.alltypessmall where int_col < 0 and string_col < "" ----- RESULTS -0 ----- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select id, bool_col from functional_parquet.alltypessmall where int_col < 3 - 3 ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select id, bool_col from functional_parquet.alltypessmall where int_col < 3 - 3 ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY # Test that without expr rewrite and thus without constant folding, constant exprs still @@ -209,15 +200,15 @@ set enable_expr_rewrites=0; select id, bool_col from functional_parquet.alltypessmall where int_col < 3 - 3 ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY select id, bool_col from functional_parquet.alltypessmall where 5 + 5 < int_col ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY # Test that without expr rewrite and thus without constant folding, constant exprs still @@ -226,8 +217,8 @@ set enable_expr_rewrites=0; select id, bool_col from functional_parquet.alltypessmall where 5 + 5 < int_col ---- RESULTS ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY # Test name based column resolution @@ -240,8 +231,8 @@ select count(*) from name_resolve where id > 10; ---- RESULTS 89 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 1 .* -row_regex: .*NumStatsFilteredRowGroups: 0 .* +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 0 ==== ---- QUERY # Query that has an implicit cast to a larger integer type @@ -249,8 +240,8 @@ select count(*) from functional_parquet.alltypessmall where tinyint_col > 100000 ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 4 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 ==== ---- QUERY # Predicates with explicit casts are not supported when evaluating parquet::Statistics. @@ -258,8 +249,8 @@ select count(*) from functional_parquet.alltypessmall where '0' > cast(tinyint_c ---- RESULTS 0 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 4 .* -row_regex: .*NumStatsFilteredRowGroups: 0 .* +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 0 ==== ---- QUERY # Explicit casts between numerical types can violate the transitivity of "min()", so they @@ -268,16 +259,16 @@ select count(*) from functional_parquet.alltypes where cast(id as tinyint) < 10; ---- RESULTS 3878 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 24 .* -row_regex: .*NumStatsFilteredRowGroups: 0 .* +aggregation(SUM, NumRowGroups): 24 +aggregation(SUM, NumStatsFilteredRowGroups): 0 ==== ---- QUERY select count(*) from functional_parquet.complextypestbl.int_array where pos < 5; ---- RESULTS 9 ---- RUNTIME_PROFILE -row_regex: .*NumRowGroups: 2 .* -row_regex: .*NumStatsFilteredRowGroups: 0 .* +aggregation(SUM, NumRowGroups): 2 +aggregation(SUM, NumStatsFilteredRowGroups): 0 ==== ---- QUERY # Test the conversion of constant IN lists to min/max predicats @@ -295,4 +286,164 @@ select count(*) from functional_parquet.alltypes where id IN (1,25,49); ---- RUNTIME_PROFILE aggregation(SUM, NumRowGroups): 24 aggregation(SUM, NumStatsFilteredRowGroups): 23 -==== \ No newline at end of file +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where string_col < "0" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where string_col <= "/" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where string_col < "1" +---- RESULTS +12 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where string_col >= "9" +---- RESULTS +8 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where string_col > ":" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where timestamp_col < "2009-01-01 00:00:00" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where timestamp_col <= "2009-01-01 00:00:00" +---- RESULTS +1 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 3 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where timestamp_col = "2009-01-01 00:00:00" +---- RESULTS +1 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 3 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where timestamp_col > "2009-04-03 00:24:00.96" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 4 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where timestamp_col >= "2009-04-03 00:24:00.96" +---- RESULTS +1 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 3 +==== +---- QUERY +select count(*) from functional_parquet.alltypessmall where timestamp_col = "2009-04-03 00:24:00.96" +---- RESULTS +1 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 4 +aggregation(SUM, NumStatsFilteredRowGroups): 3 +==== +---- QUERY +select count(*) from functional_parquet.decimal_tbl where d1 < 1234 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from functional_parquet.decimal_tbl where d3 < 1.23456789 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from functional_parquet.decimal_tbl where d3 = 1.23456788 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from functional_parquet.decimal_tbl where d3 = 1.23456789 +---- RESULTS +1 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +==== +---- QUERY +select count(*) from functional_parquet.decimal_tbl where d4 > 0.123456789 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +select count(*) from functional_parquet.decimal_tbl where d4 >= 0.12345678 +---- RESULTS +5 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +==== +---- QUERY +select count(*) from functional_parquet.decimal_tbl where d4 >= 0.12345679 +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 1 +==== +---- QUERY +# Test that stats are disabled for CHAR type columns. +create table chars (id int, c char(4)) stored as parquet; +insert into chars values (1, cast("abaa" as char(4))), (2, cast("abab" as char(4))); +select count(*) from chars; +---- RESULTS +2 +==== +---- QUERY +select count(*) from chars where c <= "aaaa" +---- RESULTS +0 +---- RUNTIME_PROFILE +aggregation(SUM, NumRowGroups): 1 +aggregation(SUM, NumStatsFilteredRowGroups): 0 +====
