You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2018/08/08 21:07:52 UTC
[2/2] impala git commit: IMPALA-7403: fix child batch mem mgmt in
analytic
IMPALA-7403: fix child batch mem mgmt in analytic
The core of the fix is in ProcessChildBatches(), where we copy
'prev_input_tuple_' to 'prev_input_tuple_pool_' and reset
the child batch *before* the call to child(0)->GetNext(). This
solves a couple of problems:
* prev_input_tuple_ may be referencing memory from the child
that had the needs_deep_copy() flag set and therefore will
be freed or recycled when calling child(0)->GetNext() again.
* 'prev_child_batch_' may have been holding onto resources that
the child had flushed, which means they need to be freed before
the next GetNext() call.
Also refactors the logic around child_cmp_row_ to make the variable
lifetime and data flow clearer.
Testing:
Add regression test. The test passes both with this patch alone and with
IMPALA-7333 reapplied.
Change-Id: I09eb6213d47287f2addb72f8c1304085d2d48c55
Reviewed-on: http://gerrit.cloudera.org:8080/11155
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/a6c35685
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/a6c35685
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/a6c35685
Branch: refs/heads/master
Commit: a6c356850bd4606f8506fa763d5f543229dce780
Parents: dd8a25d
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Tue Aug 7 14:17:10 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Wed Aug 8 20:51:45 2018 +0000
----------------------------------------------------------------------
be/src/exec/analytic-eval-node.cc | 81 ++++++++++----------
be/src/exec/analytic-eval-node.h | 61 ++++++++-------
be/src/exprs/scalar-expr-evaluator.cc | 22 +++---
be/src/exprs/scalar-expr-evaluator.h | 22 +++---
.../queries/QueryTest/analytic-fns-tpcds.test | 46 ++++++++++-
5 files changed, 137 insertions(+), 95 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/a6c35685/be/src/exec/analytic-eval-node.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/analytic-eval-node.cc b/be/src/exec/analytic-eval-node.cc
index c153b64..f51a5b5 100644
--- a/be/src/exec/analytic-eval-node.cc
+++ b/be/src/exec/analytic-eval-node.cc
@@ -53,7 +53,6 @@ AnalyticEvalNode::AnalyticEvalNode(
rows_end_offset_(0),
has_first_val_null_offset_(false),
first_val_null_offset_(0),
- child_tuple_cmp_row_(nullptr),
last_result_idx_(-1),
prev_pool_last_result_idx_(-1),
prev_pool_last_window_idx_(-1),
@@ -161,6 +160,7 @@ Status AnalyticEvalNode::Prepare(RuntimeState* state) {
curr_tuple_pool_.reset(new MemPool(mem_tracker()));
prev_tuple_pool_.reset(new MemPool(mem_tracker()));
mem_pool_.reset(new MemPool(mem_tracker()));
+ prev_input_tuple_pool_.reset(new MemPool(mem_tracker()));
evaluation_timer_ = ADD_TIMER(runtime_profile(), "EvaluationTime");
DCHECK_EQ(result_tuple_desc_->slots().size(), analytic_fns_.size());
@@ -222,13 +222,6 @@ Status AnalyticEvalNode::Open(RuntimeState* state) {
RETURN_IF_ERROR(order_by_eq_expr_eval_->Open(state));
}
- if (buffered_tuple_desc_ != nullptr) {
- // The backing mem_pool_ is freed in Reset(), so we need to allocate
- // a new row every time we Open().
- child_tuple_cmp_row_ = reinterpret_cast<TupleRow*>(
- mem_pool_->Allocate(sizeof(Tuple*) * 2));
- }
-
// An intermediate tuple is only allocated once and is reused.
curr_tuple_ = Tuple::Create(intermediate_tuple_desc_->byte_size(), mem_pool_.get());
AggFnEvaluator::Init(analytic_fn_evals_, curr_tuple_);
@@ -241,8 +234,6 @@ Status AnalyticEvalNode::Open(RuntimeState* state) {
// Initialize state for the first partition.
RETURN_IF_ERROR(InitNextPartition(state, 0));
- prev_child_batch_.reset(new RowBatch(child(0)->row_desc(), state->batch_size(),
- mem_tracker()));
curr_child_batch_.reset(new RowBatch(child(0)->row_desc(), state->batch_size(),
mem_tracker()));
return Status::OK();
@@ -403,22 +394,22 @@ Status AnalyticEvalNode::AddResultTuple(int64_t stream_idx) {
return Status::OK();
}
-inline Status AnalyticEvalNode::TryAddResultTupleForPrevRow(bool next_partition,
- int64_t stream_idx, TupleRow* row) {
+inline Status AnalyticEvalNode::TryAddResultTupleForPrevRow(
+ const TupleRow* child_tuple_cmp_row, bool next_partition, int64_t stream_idx) {
// The analytic fns are finalized after the previous row if we found a new partition
// or the window is a RANGE and the order by exprs changed. For ROWS windows we do not
// need to compare the current row to the previous row.
VLOG_ROW << id() << " TryAddResultTupleForPrevRow partition=" << next_partition
<< " idx=" << stream_idx;
- if (fn_scope_ != ROWS && (next_partition || (fn_scope_ == RANGE &&
- window_.__isset.window_end && !PrevRowCompare(order_by_eq_expr_eval_)))) {
+ if (fn_scope_ != ROWS && (next_partition
+ || (fn_scope_ == RANGE && window_.__isset.window_end
+ && !PrevRowCompare(order_by_eq_expr_eval_, child_tuple_cmp_row)))) {
RETURN_IF_ERROR(AddResultTuple(stream_idx - 1));
}
return Status::OK();
}
-inline Status AnalyticEvalNode::TryAddResultTupleForCurrRow(int64_t stream_idx,
- TupleRow* row) {
+inline Status AnalyticEvalNode::TryAddResultTupleForCurrRow(int64_t stream_idx) {
VLOG_ROW << id() << " TryAddResultTupleForCurrRow idx=" << stream_idx;
// We only add results at this point for ROWS windows (unless unbounded following)
// Nothing to add if the end offset is before the start of the partition.
@@ -586,9 +577,10 @@ inline Status AnalyticEvalNode::InitNextPartition(RuntimeState* state,
return Status::OK();
}
-inline bool AnalyticEvalNode::PrevRowCompare(ScalarExprEvaluator* pred_eval) {
+inline bool AnalyticEvalNode::PrevRowCompare(
+ ScalarExprEvaluator* pred_eval, const TupleRow* child_tuple_cmp_row) {
DCHECK(pred_eval != nullptr);
- BooleanVal result = pred_eval->GetBooleanVal(child_tuple_cmp_row_);
+ BooleanVal result = pred_eval->GetBooleanVal(child_tuple_cmp_row);
DCHECK(!result.is_null);
return result.val;
}
@@ -599,24 +591,31 @@ Status AnalyticEvalNode::ProcessChildBatches(RuntimeState* state) {
// allows us to simplify the logic dealing with last_result_idx_ and result_tuples_.
while (!input_eos_ && NumOutputRowsReady() < state->batch_size() + 1) {
RETURN_IF_CANCELLED(state);
+ if (has_partition_or_order_by_expr_eval() && prev_input_tuple_ != nullptr
+ && curr_child_batch_->num_rows() > 0) {
+ // 'prev_input_tuple_' is from the last row in 'curr_child_batch_' and is needed
+ // by subsequent calls to ProcessChildBatch(). Deep copy it so that we can safely
+ // free the memory backing it. 'prev_input_tuple_pool_' is not backing the previous
+ // input tuple any more - it is from the last row in 'curr_child_batch_' and
+ // therefore backed by 'curr_child_batch_'.
+ prev_input_tuple_pool_->Clear();
+ prev_input_tuple_ = prev_input_tuple_->DeepCopy(
+ *child(0)->row_desc()->tuple_descriptors()[0], prev_input_tuple_pool_.get());
+ }
+ curr_child_batch_->Reset();
RETURN_IF_ERROR(child(0)->GetNext(state, curr_child_batch_.get(), &input_eos_));
RETURN_IF_ERROR(QueryMaintenance(state));
RETURN_IF_ERROR(ProcessChildBatch(state));
// TODO: DCHECK that the size of result_tuples_ is bounded. It shouldn't be larger
// than 2x the batch size unless the end bound has an offset preceding, in which
// case it may be slightly larger (proportional to the offset but still bounded).
- prev_child_batch_->Reset();
- prev_child_batch_.swap(curr_child_batch_);
- }
- if (input_eos_) {
- curr_child_batch_.reset();
- prev_child_batch_.reset();
}
+ if (input_eos_) curr_child_batch_.reset();
return Status::OK();
}
Status AnalyticEvalNode::ProcessChildBatch(RuntimeState* state) {
- // TODO: DCHECK input is sorted (even just first row vs prev_input_row_)
+ // TODO: DCHECK input is sorted (even just first row vs prev_input_tuple_)
VLOG_FILE << id() << " ProcessChildBatch: " << DebugStateString()
<< " input batch size:" << curr_child_batch_->num_rows()
<< " tuple pool size:" << curr_tuple_pool_->total_allocated_bytes();
@@ -634,23 +633,24 @@ Status AnalyticEvalNode::ProcessChildBatch(RuntimeState* state) {
if (UNLIKELY(stream_idx == 0 && curr_child_batch_->num_rows() > 0)) {
TupleRow* row = curr_child_batch_->GetRow(0);
RETURN_IF_ERROR(AddRow(0, row));
- RETURN_IF_ERROR(TryAddResultTupleForCurrRow(0, row));
- prev_input_row_ = row;
+ RETURN_IF_ERROR(TryAddResultTupleForCurrRow(0));
+ prev_input_tuple_ = row->GetTuple(0);
++batch_idx;
++stream_idx;
}
-
+ Tuple* child_tuple_cmp_row_tuples[2] = {nullptr, nullptr};
+ TupleRow* child_tuple_cmp_row =
+ reinterpret_cast<TupleRow*>(child_tuple_cmp_row_tuples);
for (; batch_idx < curr_child_batch_->num_rows(); ++batch_idx, ++stream_idx) {
TupleRow* row = curr_child_batch_->GetRow(batch_idx);
- if (partition_by_eq_expr_eval_ != nullptr ||
- order_by_eq_expr_eval_ != nullptr) {
- // Only set the tuples in child_tuple_cmp_row_ if there are partition exprs or
+ if (has_partition_or_order_by_expr_eval()) {
+ // Only set the tuples in child_tuple_cmp_row if there are partition exprs or
// order by exprs that require comparing the current and previous rows. If there
// aren't partition or order by exprs (i.e. empty OVER() clause), there was no
// sort and there could be nullable tuples (whereas the sort node does not produce
// them), see IMPALA-1562.
- child_tuple_cmp_row_->SetTuple(0, prev_input_row_->GetTuple(0));
- child_tuple_cmp_row_->SetTuple(1, row->GetTuple(0));
+ child_tuple_cmp_row->SetTuple(0, prev_input_tuple_);
+ child_tuple_cmp_row->SetTuple(1, row->GetTuple(0));
}
TryRemoveRowsBeforeWindow(stream_idx);
@@ -667,16 +667,17 @@ Status AnalyticEvalNode::ProcessChildBatch(RuntimeState* state) {
bool next_partition = false;
if (partition_by_eq_expr_eval_ != nullptr) {
// partition_by_eq_expr_eval_ checks equality over the predicate exprs
- next_partition = !PrevRowCompare(partition_by_eq_expr_eval_);
+ next_partition = !PrevRowCompare(partition_by_eq_expr_eval_, child_tuple_cmp_row);
}
- RETURN_IF_ERROR(TryAddResultTupleForPrevRow(next_partition, stream_idx, row));
+ RETURN_IF_ERROR(TryAddResultTupleForPrevRow(
+ child_tuple_cmp_row, next_partition, stream_idx));
if (next_partition) RETURN_IF_ERROR(InitNextPartition(state, stream_idx));
// The analytic_fn_evals_ are updated with the current row.
RETURN_IF_ERROR(AddRow(stream_idx, row));
- RETURN_IF_ERROR(TryAddResultTupleForCurrRow(stream_idx, row));
- prev_input_row_ = row;
+ RETURN_IF_ERROR(TryAddResultTupleForCurrRow(stream_idx));
+ prev_input_tuple_ = row->GetTuple(0);
}
if (UNLIKELY(input_eos_ && stream_idx > curr_partition_idx_)) {
@@ -842,10 +843,8 @@ Status AnalyticEvalNode::Reset(RuntimeState* state) {
DCHECK(input_stream_ == nullptr || input_stream_->is_closed());
input_stream_.reset();
curr_tuple_ = nullptr;
- child_tuple_cmp_row_ = nullptr;
dummy_result_tuple_ = nullptr;
- prev_input_row_ = nullptr;
- prev_child_batch_.reset();
+ prev_input_tuple_ = nullptr;
curr_child_batch_.reset();
return ExecNode::Reset(state);
}
@@ -882,11 +881,11 @@ void AnalyticEvalNode::Close(RuntimeState* state) {
if (order_by_eq_expr_eval_ != nullptr) order_by_eq_expr_eval_->Close(state);
order_by_eq_expr_->Close();
}
- if (prev_child_batch_.get() != nullptr) prev_child_batch_.reset();
if (curr_child_batch_.get() != nullptr) curr_child_batch_.reset();
if (curr_tuple_pool_.get() != nullptr) curr_tuple_pool_->FreeAll();
if (prev_tuple_pool_.get() != nullptr) prev_tuple_pool_->FreeAll();
if (mem_pool_.get() != nullptr) mem_pool_->FreeAll();
+ if (prev_input_tuple_pool_.get() != nullptr) prev_input_tuple_pool_->FreeAll();
ExecNode::Close(state);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/a6c35685/be/src/exec/analytic-eval-node.h
----------------------------------------------------------------------
diff --git a/be/src/exec/analytic-eval-node.h b/be/src/exec/analytic-eval-node.h
index c7e7873..bf15ebf 100644
--- a/be/src/exec/analytic-eval-node.h
+++ b/be/src/exec/analytic-eval-node.h
@@ -18,6 +18,8 @@
#ifndef IMPALA_EXEC_ANALYTIC_EVAL_NODE_H
#define IMPALA_EXEC_ANALYTIC_EVAL_NODE_H
+#include <memory>
+
#include "exec/exec-node.h"
#include "runtime/buffered-tuple-stream.h"
#include "runtime/tuple.h"
@@ -128,19 +130,20 @@ class AnalyticEvalNode : public ExecNode {
/// Adds the row to the evaluators and the tuple stream.
Status AddRow(int64_t stream_idx, TupleRow* row);
- /// Determines if there is a window ending at the previous row, and if so, calls
- /// AddResultTuple() with the index of the previous row in 'input_stream_'.
- /// 'next_partition' indicates if the current row is the start of a new partition.
- /// 'stream_idx' is the index of the current input row from 'input_stream_'.
- /// Returns an error when memory limit is exceeded.
- Status TryAddResultTupleForPrevRow(bool next_partition, int64_t stream_idx,
- TupleRow* row);
+ /// Determines if there is a window ending at the previous row by evaluating
+ /// 'child_tuple_cmp_row', and if so, calls AddResultTuple() with the index
+ /// of the previous row in 'input_stream_'. 'next_partition' indicates if
+ /// the current row is the start of a new partition. 'stream_idx' is the
+ /// index of the current input row from 'input_stream_'. Returns an error
+ /// when memory limit is exceeded.
+ Status TryAddResultTupleForPrevRow(
+ const TupleRow* child_tuple_cmp_row, bool next_partition, int64_t stream_idx);
/// Determines if there is a window ending at the current row, and if so, calls
/// AddResultTuple() with the index of the current row in 'input_stream_'.
/// 'stream_idx' is the index of the current input row from 'input_stream_'.
/// Returns an error when memory limit is exceeded.
- Status TryAddResultTupleForCurrRow(int64_t stream_idx, TupleRow* row);
+ Status TryAddResultTupleForCurrRow(int64_t stream_idx);
/// Adds additional result tuples at the end of a partition, e.g. if the end bound is
/// FOLLOWING. partition_idx is the index into input_stream_ of the new partition,
@@ -172,10 +175,17 @@ class AnalyticEvalNode : public ExecNode {
/// This is necessary to produce the default value (set by Init()).
void ResetLeadFnSlots();
- /// Evaluates the predicate pred_eval over child_tuple_cmp_row_, which is
- /// a TupleRow* containing the previous row and the current row set during
- /// ProcessChildBatch().
- bool PrevRowCompare(ScalarExprEvaluator* pred_eval);
+ /// Evaluates the predicate pred_eval over child_tuple_cmp_row, which is
+ /// a TupleRow* containing the previous row and the current row.
+ bool PrevRowCompare(
+ ScalarExprEvaluator* pred_eval, const TupleRow* child_tuple_cmp_row);
+
+ /// Return true if there are partition or order by expression evaluators. These
+ /// evaluators are created in Prepare() if PARTITION BY or ORDER BY clauses exist
+ /// for the analytics.
+ bool has_partition_or_order_by_expr_eval() const {
+ return partition_by_eq_expr_eval_ != nullptr || order_by_eq_expr_eval_ != nullptr;
+ }
/// Debug string containing current state. If 'detailed', per-row state is included.
std::string DebugStateString(bool detailed) const;
@@ -252,11 +262,6 @@ class AnalyticEvalNode : public ExecNode {
/////////////////////////////////////////
/// BEGIN: Members that must be Reset()
- /// TupleRow* composed of the first child tuple and the buffered tuple, used by
- /// partition_by_eq_expr_eval_ and order_by_eq_expr_eval_. Set in Open() if
- /// buffered_tuple_desc_ is not NULL, allocated from mem_pool_.
- TupleRow* child_tuple_cmp_row_ = nullptr;
-
/// Queue of tuples which are ready to be set in output rows, with the index into
/// the input_stream_ stream of the last TupleRow that gets the Tuple, i.e. this is a
/// sparse structure. For example, if result_tuples_ contains tuples with indexes x1 and
@@ -308,17 +313,17 @@ class AnalyticEvalNode : public ExecNode {
/// Index of the row in input_stream_ at which the current partition started.
int64_t curr_partition_idx_;
- /// Previous input row used to compare partition boundaries and to determine when the
- /// order-by expressions change.
- TupleRow* prev_input_row_ = nullptr;
-
- /// Current and previous input row batches from the child. RowBatches are allocated
- /// once and reused. Previous input row batch owns prev_input_row_ between calls to
- /// ProcessChildBatch(). The prev batch is Reset() after calling ProcessChildBatch()
- /// and then swapped with the curr batch so the RowBatch owning prev_input_row_ is
- /// stored in prev_child_batch_ for the next call to ProcessChildBatch().
- boost::scoped_ptr<RowBatch> prev_child_batch_;
- boost::scoped_ptr<RowBatch> curr_child_batch_;
+ /// Previous input tuple used to compare partition boundaries and to determine when the
+ /// order-by expressions change. We only need to store the first tuple of the row
+ /// because all the partitioning and ordering columns are in the first tuple. Initially
+ /// this points to the first tuple of the last row processed from 'curr_child_batch_',
+ /// but it is later deep copied into 'prev_input_tuple_pool_' before 'curr_child_batch_'
+ /// is reset.
+ Tuple* prev_input_tuple_ = nullptr;
+ std::unique_ptr<MemPool> prev_input_tuple_pool_;
+
+ /// Current input row batch from the child. Allocated once and reused.
+ std::unique_ptr<RowBatch> curr_child_batch_;
/// Buffers input rows added in ProcessChildBatch() until enough rows are able to
/// be returned by GetNextOutputBatch(), in which case row batches are returned from
http://git-wip-us.apache.org/repos/asf/impala/blob/a6c35685/be/src/exprs/scalar-expr-evaluator.cc
----------------------------------------------------------------------
diff --git a/be/src/exprs/scalar-expr-evaluator.cc b/be/src/exprs/scalar-expr-evaluator.cc
index 0cadcc2..342b8e1 100644
--- a/be/src/exprs/scalar-expr-evaluator.cc
+++ b/be/src/exprs/scalar-expr-evaluator.cc
@@ -364,47 +364,47 @@ void ScalarExprEvaluator::PrintValue(const TupleRow* row, stringstream* stream)
RawValue::PrintValue(GetValue(row), root_.type(), output_scale_, stream);
}
-BooleanVal ScalarExprEvaluator::GetBooleanVal(TupleRow* row) {
+BooleanVal ScalarExprEvaluator::GetBooleanVal(const TupleRow* row) {
return root_.GetBooleanVal(this, row);
}
-TinyIntVal ScalarExprEvaluator::GetTinyIntVal(TupleRow* row) {
+TinyIntVal ScalarExprEvaluator::GetTinyIntVal(const TupleRow* row) {
return root_.GetTinyIntVal(this, row);
}
-SmallIntVal ScalarExprEvaluator::GetSmallIntVal(TupleRow* row) {
+SmallIntVal ScalarExprEvaluator::GetSmallIntVal(const TupleRow* row) {
return root_.GetSmallIntVal(this, row);
}
-IntVal ScalarExprEvaluator::GetIntVal(TupleRow* row) {
+IntVal ScalarExprEvaluator::GetIntVal(const TupleRow* row) {
return root_.GetIntVal(this, row);
}
-BigIntVal ScalarExprEvaluator::GetBigIntVal(TupleRow* row) {
+BigIntVal ScalarExprEvaluator::GetBigIntVal(const TupleRow* row) {
return root_.GetBigIntVal(this, row);
}
-FloatVal ScalarExprEvaluator::GetFloatVal(TupleRow* row) {
+FloatVal ScalarExprEvaluator::GetFloatVal(const TupleRow* row) {
return root_.GetFloatVal(this, row);
}
-DoubleVal ScalarExprEvaluator::GetDoubleVal(TupleRow* row) {
+DoubleVal ScalarExprEvaluator::GetDoubleVal(const TupleRow* row) {
return root_.GetDoubleVal(this, row);
}
-StringVal ScalarExprEvaluator::GetStringVal(TupleRow* row) {
+StringVal ScalarExprEvaluator::GetStringVal(const TupleRow* row) {
return root_.GetStringVal(this, row);
}
-CollectionVal ScalarExprEvaluator::GetCollectionVal(TupleRow* row) {
+CollectionVal ScalarExprEvaluator::GetCollectionVal(const TupleRow* row) {
return root_.GetCollectionVal(this, row);
}
-TimestampVal ScalarExprEvaluator::GetTimestampVal(TupleRow* row) {
+TimestampVal ScalarExprEvaluator::GetTimestampVal(const TupleRow* row) {
return root_.GetTimestampVal(this, row);
}
-DecimalVal ScalarExprEvaluator::GetDecimalVal(TupleRow* row) {
+DecimalVal ScalarExprEvaluator::GetDecimalVal(const TupleRow* row) {
return root_.GetDecimalVal(this, row);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/a6c35685/be/src/exprs/scalar-expr-evaluator.h
----------------------------------------------------------------------
diff --git a/be/src/exprs/scalar-expr-evaluator.h b/be/src/exprs/scalar-expr-evaluator.h
index 9a27951..ba8cf17 100644
--- a/be/src/exprs/scalar-expr-evaluator.h
+++ b/be/src/exprs/scalar-expr-evaluator.h
@@ -151,17 +151,17 @@ class ScalarExprEvaluator {
/// Evaluates the expression of this evaluator on tuple row 'row' and returns
/// the results. One function for each data type implemented.
- BooleanVal GetBooleanVal(TupleRow* row);
- TinyIntVal GetTinyIntVal(TupleRow* row);
- SmallIntVal GetSmallIntVal(TupleRow* row);
- IntVal GetIntVal(TupleRow* row);
- BigIntVal GetBigIntVal(TupleRow* row);
- FloatVal GetFloatVal(TupleRow* row);
- DoubleVal GetDoubleVal(TupleRow* row);
- StringVal GetStringVal(TupleRow* row);
- CollectionVal GetCollectionVal(TupleRow* row);
- TimestampVal GetTimestampVal(TupleRow* row);
- DecimalVal GetDecimalVal(TupleRow* row);
+ BooleanVal GetBooleanVal(const TupleRow* row);
+ TinyIntVal GetTinyIntVal(const TupleRow* row);
+ SmallIntVal GetSmallIntVal(const TupleRow* row);
+ IntVal GetIntVal(const TupleRow* row);
+ BigIntVal GetBigIntVal(const TupleRow* row);
+ FloatVal GetFloatVal(const TupleRow* row);
+ DoubleVal GetDoubleVal(const TupleRow* row);
+ StringVal GetStringVal(const TupleRow* row);
+ CollectionVal GetCollectionVal(const TupleRow* row);
+ TimestampVal GetTimestampVal(const TupleRow* row);
+ DecimalVal GetDecimalVal(const TupleRow* row);
/// Returns an error status if there was any error in evaluating the expression
/// or its sub-expressions. 'start_idx' and 'end_idx' correspond to the range
http://git-wip-us.apache.org/repos/asf/impala/blob/a6c35685/testdata/workloads/functional-query/queries/QueryTest/analytic-fns-tpcds.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/analytic-fns-tpcds.test b/testdata/workloads/functional-query/queries/QueryTest/analytic-fns-tpcds.test
index 9d98dec..5a440c6 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/analytic-fns-tpcds.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/analytic-fns-tpcds.test
@@ -1,6 +1,6 @@
====
---- QUERY
-# Analyic function with no partition
+# Analytic function with no partition
select i_item_sk, i_current_price,
SUM (i_current_price)
OVER (ORDER BY i_item_sk) running_total
@@ -72,7 +72,7 @@ limit 10
BIGINT, STRING, DECIMAL, DECIMAL
====
---- QUERY
-# Nested analyic functions
+# Nested analytic functions
select i_item_sk, i_brand, running_total,
SUM (running_total)
OVER (partition by i_manufact_id ORDER BY running_total) running_total2
@@ -104,7 +104,7 @@ limit 10
BIGINT, STRING, DECIMAL, DECIMAL
====
---- QUERY
-# 2 analyic functions on different partition and order by columns
+# 2 analytic functions on different partition and order by columns
select i_item_sk, i_brand, i_current_price, i_manufact_id,
SUM (i_current_price)
OVER (partition by i_brand ORDER BY i_current_price) running_total,
@@ -173,4 +173,42 @@ from (
2815.26
---- TYPES
DECIMAL
-====
\ No newline at end of file
+====
+---- QUERY
+# Regression test for IMPALA-7403 that reproduces a memory management bug.
+# The plan has two analytic operators with no intervening sort. A low NDV
+# sort key seems to help with reproducing the bug.
+SELECT c_birth_country, a, b, count(*)
+FROM (
+ SELECT c_birth_country,
+ LAG(c_birth_country) OVER (ORDER BY c_birth_country) a,
+ MIN(c_birth_country) OVER (ORDER BY c_birth_country) b
+ FROM customer
+) v
+GROUP BY c_birth_country, a, b
+ORDER BY count(*), c_birth_country, a, b
+LIMIT 20
+---- RESULTS
+'AFGHANISTAN','NULL','AFGHANISTAN',1
+'ALAND ISLANDS','AFGHANISTAN','AFGHANISTAN',1
+'ALBANIA','ALAND ISLANDS','AFGHANISTAN',1
+'ALGERIA','ALBANIA','AFGHANISTAN',1
+'AMERICAN SAMOA','ALGERIA','AFGHANISTAN',1
+'ANDORRA','AMERICAN SAMOA','AFGHANISTAN',1
+'ANGOLA','ANDORRA','AFGHANISTAN',1
+'ANGUILLA','ANGOLA','AFGHANISTAN',1
+'ANTARCTICA','ANGUILLA','AFGHANISTAN',1
+'ANTIGUA AND BARBUDA','ANTARCTICA','AFGHANISTAN',1
+'ARGENTINA','ANTIGUA AND BARBUDA','AFGHANISTAN',1
+'ARMENIA','ARGENTINA','AFGHANISTAN',1
+'ARUBA','ARMENIA','AFGHANISTAN',1
+'AUSTRALIA','ARUBA','AFGHANISTAN',1
+'AUSTRIA','AUSTRALIA','AFGHANISTAN',1
+'AZERBAIJAN','AUSTRIA','AFGHANISTAN',1
+'BAHAMAS','AZERBAIJAN','AFGHANISTAN',1
+'BAHRAIN','BAHAMAS','AFGHANISTAN',1
+'BANGLADESH','BAHRAIN','AFGHANISTAN',1
+'BARBADOS','BANGLADESH','AFGHANISTAN',1
+---- TYPES
+STRING,STRING,STRING,BIGINT
+====