You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2021/01/23 10:15:39 UTC
[impala] branch master updated (ac7f605 -> 6d4756d)
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from ac7f605 IMPALA-10421: [DOCS] Documented the JOIN_ROWS_PRODUCED_LIMIT query option
new 5852a00 IMPALA-10440: Import Theta functionality from DataSketches
new 18acca9 IMPALA-10435: Extend 'compute incremental stats' syntax to support a list of columns
new 1ada739 IMPALA-10296: Fix analytic limit pushdown when predicates are present
new 6d4756d IMPALA-10448: Build impala-profile-tool early for Docker-based tests
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/exec/topn-node-ir.cc | 94 ++-
be/src/exec/topn-node.cc | 143 ++--
be/src/exec/topn-node.h | 141 +++-
be/src/exprs/CMakeLists.txt | 2 +-
be/src/exprs/datasketches-test.cc | 58 ++
.../thirdparty/datasketches/HllUnion-internal.hpp | 28 -
be/src/thirdparty/datasketches/README.md | 9 +-
be/src/thirdparty/datasketches/hll.hpp | 15 -
.../datasketches/kll_quantile_calculator_impl.hpp | 2 +-
be/src/thirdparty/datasketches/kll_sketch.hpp | 12 +-
be/src/thirdparty/datasketches/kll_sketch_impl.hpp | 10 +-
be/src/thirdparty/datasketches/theta_a_not_b.hpp | 73 ++
.../thirdparty/datasketches/theta_a_not_b_impl.hpp | 83 ++
.../thirdparty/datasketches/theta_intersection.hpp | 88 ++
.../datasketches/theta_intersection_impl.hpp | 130 +++
be/src/thirdparty/datasketches/theta_sketch.hpp | 533 ++++++++++++
.../thirdparty/datasketches/theta_sketch_impl.hpp | 939 +++++++++++++++++++++
be/src/thirdparty/datasketches/theta_union.hpp | 122 +++
.../thirdparty/datasketches/theta_union_impl.hpp | 109 +++
be/src/util/tuple-row-compare.h | 6 +
common/thrift/PlanNodes.thrift | 6 +
docker/entrypoint.sh | 2 +-
fe/src/main/cup/sql-parser.cup | 10 +-
.../org/apache/impala/analysis/AnalyticExpr.java | 10 +-
.../apache/impala/analysis/ComputeStatsStmt.java | 5 +-
.../apache/impala/planner/AnalyticEvalNode.java | 180 ++--
.../apache/impala/planner/DistributedPlanner.java | 69 +-
.../apache/impala/planner/SingleNodePlanner.java | 48 +-
.../java/org/apache/impala/planner/SortNode.java | 101 ++-
.../org/apache/impala/analysis/AnalyzeDDLTest.java | 9 +
.../org/apache/impala/analysis/ParserTest.java | 12 +
.../PlannerTest/limit-pushdown-analytic.test | 493 +++++++++--
.../queries/PlannerTest/resource-requirements.test | 33 +-
.../queries/PlannerTest/tpcds/tpcds-q67.test | 43 +-
.../QueryTest/compute-stats-incremental.test | 309 +++++++
.../queries/limit-pushdown-analytic.test | 96 +++
.../tpch/queries/limit-pushdown-analytic.test | 149 ++++
tests/query_test/test_limit_pushdown_analytic.py | 24 +-
38 files changed, 3772 insertions(+), 424 deletions(-)
create mode 100644 be/src/thirdparty/datasketches/theta_a_not_b.hpp
create mode 100644 be/src/thirdparty/datasketches/theta_a_not_b_impl.hpp
create mode 100644 be/src/thirdparty/datasketches/theta_intersection.hpp
create mode 100644 be/src/thirdparty/datasketches/theta_intersection_impl.hpp
create mode 100644 be/src/thirdparty/datasketches/theta_sketch.hpp
create mode 100644 be/src/thirdparty/datasketches/theta_sketch_impl.hpp
create mode 100644 be/src/thirdparty/datasketches/theta_union.hpp
create mode 100644 be/src/thirdparty/datasketches/theta_union_impl.hpp
create mode 100644 testdata/workloads/functional-query/queries/limit-pushdown-analytic.test
[impala] 01/04: IMPALA-10440: Import Theta functionality from
DataSketches
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 5852a0028b2b89d64f2b3281c91c3e23f2788f14
Author: Fucun Chu <ch...@hotmail.com>
AuthorDate: Sun Jan 17 10:22:48 2021 +0800
IMPALA-10440: Import Theta functionality from DataSketches
This patch imports the functionality needed for Theta approximate
algorithm from Apache DataSketches.
First, I updated our existing snapshot of DataSketches to the
following commit:b2f749ed5ce6ba650f4259602b133c310c3a5ee4"
Merge pull request #182 from chufucun/include_type"
This affects files originated from hll/, kll/ and theta/ directories
of the DataSketches repo.
Then I copied all the files needed for Theta into our snapshot
directory.
Browse the source files here:
https://github.com/apache/datasketches-cpp/tree/b2f749ed5ce6ba650f4259602b133c310c3a5ee4
Change-Id: I8485d6829f50b130c84ec8bef0a4b5895255ba6c
Reviewed-on: http://gerrit.cloudera.org:8080/16959
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exprs/CMakeLists.txt | 2 +-
be/src/exprs/datasketches-test.cc | 58 ++
.../thirdparty/datasketches/HllUnion-internal.hpp | 28 -
be/src/thirdparty/datasketches/README.md | 9 +-
be/src/thirdparty/datasketches/hll.hpp | 15 -
.../datasketches/kll_quantile_calculator_impl.hpp | 2 +-
be/src/thirdparty/datasketches/kll_sketch.hpp | 12 +-
be/src/thirdparty/datasketches/kll_sketch_impl.hpp | 10 +-
be/src/thirdparty/datasketches/theta_a_not_b.hpp | 73 ++
.../thirdparty/datasketches/theta_a_not_b_impl.hpp | 83 ++
.../thirdparty/datasketches/theta_intersection.hpp | 88 ++
.../datasketches/theta_intersection_impl.hpp | 130 +++
be/src/thirdparty/datasketches/theta_sketch.hpp | 533 ++++++++++++
.../thirdparty/datasketches/theta_sketch_impl.hpp | 939 +++++++++++++++++++++
be/src/thirdparty/datasketches/theta_union.hpp | 122 +++
.../thirdparty/datasketches/theta_union_impl.hpp | 109 +++
16 files changed, 2153 insertions(+), 60 deletions(-)
diff --git a/be/src/exprs/CMakeLists.txt b/be/src/exprs/CMakeLists.txt
index 9652a5b..a28218e 100644
--- a/be/src/exprs/CMakeLists.txt
+++ b/be/src/exprs/CMakeLists.txt
@@ -89,7 +89,7 @@ add_library(ExprsTests STATIC
add_dependencies(ExprsTests gen-deps)
ADD_UNIFIED_BE_LSAN_TEST(datasketches-test
- "TestDataSketchesHll.*:TestDataSketchesKll.*:TestDataSketchesCpc.*")
+ "TestDataSketchesHll.*:TestDataSketchesKll.*:TestDataSketchesCpc.*:TestDataSketchesTheta.*")
ADD_UNIFIED_BE_LSAN_TEST(iceberg-functions-test "TestIcebergFunctions.*")
ADD_UNIFIED_BE_LSAN_TEST(expr-test "Instantiations/ExprTest.*")
# Exception to unified be tests: custom main initiailizes LLVM
diff --git a/be/src/exprs/datasketches-test.cc b/be/src/exprs/datasketches-test.cc
index 8bee1c3..687f070 100644
--- a/be/src/exprs/datasketches-test.cc
+++ b/be/src/exprs/datasketches-test.cc
@@ -19,6 +19,9 @@
#include "thirdparty/datasketches/cpc_sketch.hpp"
#include "thirdparty/datasketches/cpc_union.hpp"
#include "thirdparty/datasketches/kll_sketch.hpp"
+#include "thirdparty/datasketches/theta_sketch.hpp"
+#include "thirdparty/datasketches/theta_union.hpp"
+#include "thirdparty/datasketches/theta_intersection.hpp"
#include <sstream>
#include <stdlib.h>
@@ -153,4 +156,59 @@ TEST(TestDataSketchesKll, UseDataSketchesInterface) {
}
}
+// This test is meant to cover that the Theta algorithm from the DataSketches library
+// can be imported into Impala, builds without errors and the basic functionality is
+// available to use.
+// The below code is mostly translated from the example code found on the official
+// DataSketches web page:
+// https://datasketches.apache.org/docs/Theta/ThetaJavaExample.html
+// The purpose is to create 2 Theta sketches that have overlap in their data, serialize
+// them, deserialize them, and then obtain a cardinality estimate after combining the
+// 2 sketches and a cardinality estimate after intersecting the 2 sketches.
+TEST(TestDataSketchesTheta, UseDataSketchesInterface) {
+ std::stringstream sketch_stream1;
+ std::stringstream sketch_stream2;
+ // this section generates two sketches with some overlap and serializes them into files
+ {
+ // 100000 distinct keys
+ datasketches::update_theta_sketch sketch1 =
+ datasketches::update_theta_sketch::builder().build();
+ for (int key = 0; key < 100000; key++) sketch1.update(key);
+ sketch1.serialize(sketch_stream1);
+
+ // 100000 distinct keys
+ datasketches::update_theta_sketch sketch2 =
+ datasketches::update_theta_sketch::builder().build();
+ for (int key = 50000; key < 150000; key++) sketch2.update(key);
+ sketch2.serialize(sketch_stream2);
+ }
+
+ // this section deserializes the sketches, produces union and intersection
+ {
+ datasketches::update_theta_sketch sketch1 =
+ datasketches::update_theta_sketch::deserialize(sketch_stream1);
+
+ datasketches::update_theta_sketch sketch2 =
+ datasketches::update_theta_sketch::deserialize(sketch_stream2);
+
+ // union opertion
+ datasketches::theta_union u = datasketches::theta_union::builder().build();
+ u.update(sketch1);
+ u.update(sketch2);
+ datasketches::compact_theta_sketch unionResult = u.get_result();
+
+ // Like HLL, the order of the inputs fed to the sketches is fix here so we get the
+ // same estimate every time we run this test.
+ EXPECT_EQ(149586, (int)unionResult.get_estimate());
+
+ // intersection opertaion
+ datasketches::theta_intersection intersection;
+ intersection.update(sketch1);
+ intersection.update(sketch2);
+ datasketches::compact_theta_sketch intersectionResult = intersection.get_result();
+
+ EXPECT_EQ(48249, (int)intersectionResult.get_estimate());
+ }
+}
+
}
diff --git a/be/src/thirdparty/datasketches/HllUnion-internal.hpp b/be/src/thirdparty/datasketches/HllUnion-internal.hpp
index 5aa184f..0d12fd3 100644
--- a/be/src/thirdparty/datasketches/HllUnion-internal.hpp
+++ b/be/src/thirdparty/datasketches/HllUnion-internal.hpp
@@ -38,34 +38,6 @@ hll_union_alloc<A>::hll_union_alloc(const int lg_max_k):
{}
template<typename A>
-hll_union_alloc<A> hll_union_alloc<A>::deserialize(const void* bytes, size_t len) {
- hll_sketch_alloc<A> sk(hll_sketch_alloc<A>::deserialize(bytes, len));
- // we're using the sketch's lg_config_k to initialize the union so
- // we can initialize the Union with it as long as it's HLL_8.
- hll_union_alloc<A> hllUnion(sk.get_lg_config_k());
- if (sk.get_target_type() == HLL_8) {
- std::swap(hllUnion.gadget.sketch_impl, sk.sketch_impl);
- } else {
- hllUnion.update(sk);
- }
- return hllUnion;
-}
-
-template<typename A>
-hll_union_alloc<A> hll_union_alloc<A>::deserialize(std::istream& is) {
- hll_sketch_alloc<A> sk(hll_sketch_alloc<A>::deserialize(is));
- // we're using the sketch's lg_config_k to initialize the union so
- // we can initialize the Union with it as long as it's HLL_8.
- hll_union_alloc<A> hllUnion(sk.get_lg_config_k());
- if (sk.get_target_type() == HLL_8) {
- std::swap(hllUnion.gadget.sketch_impl, sk.sketch_impl);
- } else {
- hllUnion.update(sk);
- }
- return hllUnion;
-}
-
-template<typename A>
hll_sketch_alloc<A> hll_union_alloc<A>::get_result(target_hll_type target_type) const {
return hll_sketch_alloc<A>(gadget, target_type);
}
diff --git a/be/src/thirdparty/datasketches/README.md b/be/src/thirdparty/datasketches/README.md
index 9292ffc..1c0433d 100644
--- a/be/src/thirdparty/datasketches/README.md
+++ b/be/src/thirdparty/datasketches/README.md
@@ -1,16 +1,17 @@
-The content of this folder imports the functionality needed for HLL CPC and KLL
-approximate algorithms from Apache DataSketches by copying the necessary files
+The content of this folder imports the functionality needed for CPC, HLL, KLL and
+Theta approximate algorithms from Apache DataSketches by copying the necessary files
from that project into this folder. Note, that the original structure of files was
changed during this process as originally the following folders were affected:
hll/include/
cpc/include/
+ theta/include/
kll/include/
common/include/
I copied the content of these folders into the same directory so that Impala
can compile them without rewriting the include paths in the files themselves.
The git branch of the snapshot I used as a source for the files:
-The hash: 2b84e213067b681b696ec883d245ddf911790ff2
+The hash: b2f749ed5ce6ba650f4259602b133c310c3a5ee4
Browse the source files here:
-https://github.com/apache/incubator-datasketches-cpp/tree/2b84e213067b681b696ec883d245ddf911790ff2
+https://github.com/apache/datasketches-cpp/tree/b2f749ed5ce6ba650f4259602b133c310c3a5ee4
diff --git a/be/src/thirdparty/datasketches/hll.hpp b/be/src/thirdparty/datasketches/hll.hpp
index c76ba48..3898dda 100644
--- a/be/src/thirdparty/datasketches/hll.hpp
+++ b/be/src/thirdparty/datasketches/hll.hpp
@@ -434,21 +434,6 @@ class hll_union_alloc {
explicit hll_union_alloc(int lg_max_k);
/**
- * Construct an hll_union operator from the given std::istream, which
- * must be a valid serialized image of an hll_union.
- * @param is The input stream from which to read.
- */
- static hll_union_alloc deserialize(std::istream& is);
-
- /**
- * Construct an hll_union operator from the given byte array, which
- * must be a valid serialized image of an hll_union.
- * @param bytes The byte array to read.
- * @param len Byte array length in bytes.
- */
- static hll_union_alloc deserialize(const void* bytes, size_t len);
-
- /**
* Returns the current cardinality estimate
* @return the cardinality estimate
*/
diff --git a/be/src/thirdparty/datasketches/kll_quantile_calculator_impl.hpp b/be/src/thirdparty/datasketches/kll_quantile_calculator_impl.hpp
index cb92674..f580819 100644
--- a/be/src/thirdparty/datasketches/kll_quantile_calculator_impl.hpp
+++ b/be/src/thirdparty/datasketches/kll_quantile_calculator_impl.hpp
@@ -81,7 +81,7 @@ template <typename T, typename C, typename A>
void kll_quantile_calculator<T, C, A>::convert_to_preceding_cummulative() {
uint64_t subtotal = 0;
for (auto& entry: entries_) {
- const uint32_t new_subtotal = subtotal + entry.second;
+ const uint64_t new_subtotal = subtotal + entry.second;
entry.second = subtotal;
subtotal = new_subtotal;
}
diff --git a/be/src/thirdparty/datasketches/kll_sketch.hpp b/be/src/thirdparty/datasketches/kll_sketch.hpp
index 8b3409c..a4530c9 100644
--- a/be/src/thirdparty/datasketches/kll_sketch.hpp
+++ b/be/src/thirdparty/datasketches/kll_sketch.hpp
@@ -315,8 +315,8 @@ class kll_sketch {
*
* <p>If the sketch is empty this returns an empty vector.
*
- * @param split_points an array of <i>m</i> unique, monotonically increasing float values
- * that divide the real number line into <i>m+1</i> consecutive disjoint intervals.
+ * @param split_points an array of <i>m</i> unique, monotonically increasing values
+ * that divide the input domain into <i>m+1</i> consecutive disjoint intervals.
* The definition of an "interval" is inclusive of the left split point (or minimum value) and
* exclusive of the right split point, with the exception that the last interval will include
* the maximum value.
@@ -338,8 +338,8 @@ class kll_sketch {
*
* <p>If the sketch is empty this returns an empty vector.
*
- * @param split_points an array of <i>m</i> unique, monotonically increasing float values
- * that divide the real number line into <i>m+1</i> consecutive disjoint intervals.
+ * @param split_points an array of <i>m</i> unique, monotonically increasing values
+ * that divide the input domain into <i>m+1</i> consecutive disjoint intervals.
* The definition of an "interval" is inclusive of the left split point (or minimum value) and
* exclusive of the right split point, with the exception that the last interval will include
* the maximum value.
@@ -401,7 +401,7 @@ class kll_sketch {
* @param is input stream
* @return an instance of a sketch
*/
- static kll_sketch<T, C, S, A> deserialize(std::istream& is);
+ static kll_sketch deserialize(std::istream& is);
/**
* This method deserializes a sketch from a given array of bytes.
@@ -409,7 +409,7 @@ class kll_sketch {
* @param size the size of the array
* @return an instance of a sketch
*/
- static kll_sketch<T, C, S, A> deserialize(const void* bytes, size_t size);
+ static kll_sketch deserialize(const void* bytes, size_t size);
/*
* Gets the normalized rank error given k and pmf.
diff --git a/be/src/thirdparty/datasketches/kll_sketch_impl.hpp b/be/src/thirdparty/datasketches/kll_sketch_impl.hpp
index d051135..f0c5ff3 100644
--- a/be/src/thirdparty/datasketches/kll_sketch_impl.hpp
+++ b/be/src/thirdparty/datasketches/kll_sketch_impl.hpp
@@ -91,7 +91,7 @@ is_level_zero_sorted_(other.is_level_zero_sorted_)
template<typename T, typename C, typename S, typename A>
kll_sketch<T, C, S, A>& kll_sketch<T, C, S, A>::operator=(const kll_sketch& other) {
- kll_sketch<T, C, S, A> copy(other);
+ kll_sketch copy(other);
std::swap(k_, copy.k_);
std::swap(m_, copy.m_);
std::swap(min_k_, copy.min_k_);
@@ -271,6 +271,7 @@ T kll_sketch<T, C, S, A>::get_quantile(double fraction) const {
template<typename T, typename C, typename S, typename A>
std::vector<T, A> kll_sketch<T, C, S, A>::get_quantiles(const double* fractions, uint32_t size) const {
std::vector<T, A> quantiles;
+ quantiles.reserve(size);
if (is_empty()) return quantiles;
std::unique_ptr<kll_quantile_calculator<T, C, A>, std::function<void(kll_quantile_calculator<T, C, A>*)>> quantile_calculator;
quantiles.reserve(size);
@@ -469,9 +470,9 @@ kll_sketch<T, C, S, A> kll_sketch<T, C, S, A>::deserialize(std::istream& is) {
check_serial_version(serial_version);
check_family_id(family_id);
+ if (!is.good()) throw std::runtime_error("error reading from std::istream");
const bool is_empty(flags_byte & (1 << flags::IS_EMPTY));
- if (!is.good()) throw std::runtime_error("error reading from std::istream");
- if (is_empty) return kll_sketch<T, C, S, A>(k);
+ if (is_empty) return kll_sketch(k);
uint64_t n;
uint16_t min_k;
@@ -524,8 +525,7 @@ kll_sketch<T, C, S, A> kll_sketch<T, C, S, A>::deserialize(std::istream& is) {
// copy did not throw, repackage with destrtuctor
max_value = std::unique_ptr<T, item_deleter>(max_value_buffer.release(), item_deleter());
}
- if (!is.good())
- throw std::runtime_error("error reading from std::istream");
+ if (!is.good()) throw std::runtime_error("error reading from std::istream");
return kll_sketch(k, min_k, n, num_levels, std::move(levels), std::move(items), capacity,
std::move(min_value), std::move(max_value), is_level_zero_sorted);
}
diff --git a/be/src/thirdparty/datasketches/theta_a_not_b.hpp b/be/src/thirdparty/datasketches/theta_a_not_b.hpp
new file mode 100644
index 0000000..db66ac7
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_a_not_b.hpp
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_A_NOT_B_HPP_
+#define THETA_A_NOT_B_HPP_
+
+#include <memory>
+#include <functional>
+#include <climits>
+
+#include "theta_sketch.hpp"
+#include "common_defs.hpp"
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+template<typename A>
+class theta_a_not_b_alloc {
+public:
+ /**
+ * Creates an instance of the a-not-b operation (set difference) with a given has seed.
+ * @param seed hash seed
+ */
+ explicit theta_a_not_b_alloc(uint64_t seed = DEFAULT_SEED);
+
+ /**
+ * Computes the a-not-b set operation given two sketches.
+ * @return the result of a-not-b
+ */
+ compact_theta_sketch_alloc<A> compute(const theta_sketch_alloc<A>& a, const theta_sketch_alloc<A>& b, bool ordered = true) const;
+
+private:
+ typedef typename std::allocator_traits<A>::template rebind_alloc<uint64_t> AllocU64;
+ uint16_t seed_hash_;
+
+ class less_than {
+ public:
+ explicit less_than(uint64_t value): value(value) {}
+ bool operator()(uint64_t value) const { return value < this->value; }
+ private:
+ uint64_t value;
+ };
+};
+
+// alias with default allocator for convenience
+typedef theta_a_not_b_alloc<std::allocator<void>> theta_a_not_b;
+
+} /* namespace datasketches */
+
+#include "theta_a_not_b_impl.hpp"
+
+# endif
diff --git a/be/src/thirdparty/datasketches/theta_a_not_b_impl.hpp b/be/src/thirdparty/datasketches/theta_a_not_b_impl.hpp
new file mode 100644
index 0000000..4343ee3
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_a_not_b_impl.hpp
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_A_NOT_B_IMPL_HPP_
+#define THETA_A_NOT_B_IMPL_HPP_
+
+#include <algorithm>
+
+#include "conditional_back_inserter.hpp"
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+template<typename A>
+theta_a_not_b_alloc<A>::theta_a_not_b_alloc(uint64_t seed):
+seed_hash_(theta_sketch_alloc<A>::get_seed_hash(seed))
+{}
+
+template<typename A>
+compact_theta_sketch_alloc<A> theta_a_not_b_alloc<A>::compute(const theta_sketch_alloc<A>& a, const theta_sketch_alloc<A>& b, bool ordered) const {
+ if (a.is_empty() || a.get_num_retained() == 0 || b.is_empty()) return compact_theta_sketch_alloc<A>(a, ordered);
+ if (a.get_seed_hash() != seed_hash_) throw std::invalid_argument("A seed hash mismatch");
+ if (b.get_seed_hash() != seed_hash_) throw std::invalid_argument("B seed hash mismatch");
+
+ const uint64_t theta = std::min(a.get_theta64(), b.get_theta64());
+ vector_u64<A> keys;
+ bool is_empty = a.is_empty();
+
+ if (b.get_num_retained() == 0) {
+ std::copy_if(a.begin(), a.end(), std::back_inserter(keys), less_than(theta));
+ } else {
+ if (a.is_ordered() && b.is_ordered()) { // sort-based
+ std::set_difference(a.begin(), a.end(), b.begin(), b.end(), conditional_back_inserter(keys, less_than(theta)));
+ } else { // hash-based
+ const uint8_t lg_size = lg_size_from_count(b.get_num_retained(), update_theta_sketch_alloc<A>::REBUILD_THRESHOLD);
+ vector_u64<A> b_hash_table(1 << lg_size, 0);
+ for (auto key: b) {
+ if (key < theta) {
+ update_theta_sketch_alloc<A>::hash_search_or_insert(key, b_hash_table.data(), lg_size);
+ } else if (b.is_ordered()) {
+ break; // early stop
+ }
+ }
+
+ // scan A lookup B
+ for (auto key: a) {
+ if (key < theta) {
+ if (!update_theta_sketch_alloc<A>::hash_search(key, b_hash_table.data(), lg_size)) keys.push_back(key);
+ } else if (a.is_ordered()) {
+ break; // early stop
+ }
+ }
+ }
+ }
+ if (keys.empty() && theta == theta_sketch_alloc<A>::MAX_THETA) is_empty = true;
+ if (ordered && !a.is_ordered()) std::sort(keys.begin(), keys.end());
+ return compact_theta_sketch_alloc<A>(is_empty, theta, std::move(keys), seed_hash_, a.is_ordered() || ordered);
+}
+
+} /* namespace datasketches */
+
+# endif
diff --git a/be/src/thirdparty/datasketches/theta_intersection.hpp b/be/src/thirdparty/datasketches/theta_intersection.hpp
new file mode 100644
index 0000000..5945c52
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_intersection.hpp
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_INTERSECTION_HPP_
+#define THETA_INTERSECTION_HPP_
+
+#include <memory>
+#include <functional>
+#include <climits>
+
+#include "theta_sketch.hpp"
+#include "common_defs.hpp"
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+template<typename A>
+class theta_intersection_alloc {
+public:
+ /**
+ * Creates an instance of the intersection with a given hash seed.
+ * @param seed hash seed
+ */
+ explicit theta_intersection_alloc(uint64_t seed = DEFAULT_SEED);
+
+ /**
+ * Updates the intersection with a given sketch.
+ * The intersection can be viewed as starting from the "universe" set, and every update
+ * can reduce the current set to leave the overlapping subset only.
+ * @param sketch represents input set for the intersection
+ */
+ void update(const theta_sketch_alloc<A>& sketch);
+
+ /**
+ * Produces a copy of the current state of the intersection.
+ * If update() was not called, the state is the infinite "universe",
+ * which is considered an undefined state, and throws an exception.
+ * @param ordered optional flag to specify if ordered sketch should be produced
+ * @return the result of the intersection
+ */
+ compact_theta_sketch_alloc<A> get_result(bool ordered = true) const;
+
+ /**
+ * Returns true if the state of the intersection is defined (not infinite "universe").
+ * @return true if the state is valid
+ */
+ bool has_result() const;
+
+private:
+ typedef typename std::allocator_traits<A>::template rebind_alloc<uint64_t> AllocU64;
+ bool is_valid_;
+ bool is_empty_;
+ uint64_t theta_;
+ uint8_t lg_size_;
+ vector_u64<A> keys_;
+ uint32_t num_keys_;
+ uint16_t seed_hash_;
+};
+
+// alias with default allocator for convenience
+typedef theta_intersection_alloc<std::allocator<void>> theta_intersection;
+
+} /* namespace datasketches */
+
+#include "theta_intersection_impl.hpp"
+
+# endif
diff --git a/be/src/thirdparty/datasketches/theta_intersection_impl.hpp b/be/src/thirdparty/datasketches/theta_intersection_impl.hpp
new file mode 100644
index 0000000..d090b3a
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_intersection_impl.hpp
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_INTERSECTION_IMPL_HPP_
+#define THETA_INTERSECTION_IMPL_HPP_
+
+#include <algorithm>
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+template<typename A>
+theta_intersection_alloc<A>::theta_intersection_alloc(uint64_t seed):
+is_valid_(false),
+is_empty_(false),
+theta_(theta_sketch_alloc<A>::MAX_THETA),
+lg_size_(0),
+keys_(),
+num_keys_(0),
+seed_hash_(theta_sketch_alloc<A>::get_seed_hash(seed))
+{}
+
+template<typename A>
+void theta_intersection_alloc<A>::update(const theta_sketch_alloc<A>& sketch) {
+ if (is_empty_) return;
+ if (!sketch.is_empty() && sketch.get_seed_hash() != seed_hash_) throw std::invalid_argument("seed hash mismatch");
+ is_empty_ |= sketch.is_empty();
+ theta_ = std::min(theta_, sketch.get_theta64());
+ if (is_valid_ && num_keys_ == 0) return;
+ if (sketch.get_num_retained() == 0) {
+ is_valid_ = true;
+ if (keys_.size() > 0) {
+ keys_.resize(0);
+ lg_size_ = 0;
+ num_keys_ = 0;
+ }
+ return;
+ }
+ if (!is_valid_) { // first update, clone incoming sketch
+ is_valid_ = true;
+ lg_size_ = lg_size_from_count(sketch.get_num_retained(), update_theta_sketch_alloc<A>::REBUILD_THRESHOLD);
+ keys_.resize(1 << lg_size_, 0);
+ for (auto key: sketch) {
+ if (!update_theta_sketch_alloc<A>::hash_search_or_insert(key, keys_.data(), lg_size_)) {
+ throw std::invalid_argument("duplicate key, possibly corrupted input sketch");
+ }
+ ++num_keys_;
+ }
+ if (num_keys_ != sketch.get_num_retained()) throw std::invalid_argument("num keys mismatch, possibly corrupted input sketch");
+ } else { // intersection
+ const uint32_t max_matches = std::min(num_keys_, sketch.get_num_retained());
+ vector_u64<A> matched_keys(max_matches);
+ uint32_t match_count = 0;
+ uint32_t count = 0;
+ for (auto key: sketch) {
+ if (key < theta_) {
+ if (update_theta_sketch_alloc<A>::hash_search(key, keys_.data(), lg_size_)) {
+ if (match_count == max_matches) throw std::invalid_argument("max matches exceeded, possibly corrupted input sketch");
+ matched_keys[match_count++] = key;
+ }
+ } else if (sketch.is_ordered()) {
+ break; // early stop
+ }
+ ++count;
+ }
+ if (count > sketch.get_num_retained()) {
+ throw std::invalid_argument(" more keys then expected, possibly corrupted input sketch");
+ } else if (!sketch.is_ordered() && count < sketch.get_num_retained()) {
+ throw std::invalid_argument(" fewer keys then expected, possibly corrupted input sketch");
+ }
+ if (match_count == 0) {
+ keys_.resize(0);
+ lg_size_ = 0;
+ num_keys_ = 0;
+ if (theta_ == theta_sketch_alloc<A>::MAX_THETA) is_empty_ = true;
+ } else {
+ const uint8_t lg_size = lg_size_from_count(match_count, update_theta_sketch_alloc<A>::REBUILD_THRESHOLD);
+ if (lg_size != lg_size_) {
+ lg_size_ = lg_size;
+ keys_.resize(1 << lg_size_);
+ }
+ std::fill(keys_.begin(), keys_.end(), 0);
+ for (uint32_t i = 0; i < match_count; i++) {
+ update_theta_sketch_alloc<A>::hash_search_or_insert(matched_keys[i], keys_.data(), lg_size_);
+ }
+ num_keys_ = match_count;
+ }
+ }
+}
+
+template<typename A>
+compact_theta_sketch_alloc<A> theta_intersection_alloc<A>::get_result(bool ordered) const {
+ if (!is_valid_) throw std::invalid_argument("calling get_result() before calling update() is undefined");
+ vector_u64<A> keys(num_keys_);
+ if (num_keys_ > 0) {
+ std::copy_if(keys_.begin(), keys_.end(), keys.begin(), [](uint64_t key) { return key != 0; });
+ if (ordered) std::sort(keys.begin(), keys.end());
+ }
+ return compact_theta_sketch_alloc<A>(is_empty_, theta_, std::move(keys), seed_hash_, ordered);
+}
+
+template<typename A>
+bool theta_intersection_alloc<A>::has_result() const {
+ return is_valid_;
+}
+
+} /* namespace datasketches */
+
+# endif
diff --git a/be/src/thirdparty/datasketches/theta_sketch.hpp b/be/src/thirdparty/datasketches/theta_sketch.hpp
new file mode 100644
index 0000000..b809f71
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_sketch.hpp
@@ -0,0 +1,533 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_SKETCH_HPP_
+#define THETA_SKETCH_HPP_
+
+#include <memory>
+#include <functional>
+#include <climits>
+#include <vector>
+
+#include "common_defs.hpp"
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+// forward-declarations
+template<typename A> class theta_sketch_alloc;
+template<typename A> class update_theta_sketch_alloc;
+template<typename A> class compact_theta_sketch_alloc;
+template<typename A> class theta_union_alloc;
+template<typename A> class theta_intersection_alloc;
+template<typename A> class theta_a_not_b_alloc;
+
+// for serialization as raw bytes
+template<typename A> using AllocU8 = typename std::allocator_traits<A>::template rebind_alloc<uint8_t>;
+template<typename A> using vector_u8 = std::vector<uint8_t, AllocU8<A>>;
+
+template<typename A>
+class theta_sketch_alloc {
+public:
+ static const uint64_t MAX_THETA = LLONG_MAX; // signed max for compatibility with Java
+ static const uint8_t SERIAL_VERSION = 3;
+
+ virtual ~theta_sketch_alloc() = default;
+
+ /**
+ * @return true if this sketch represents an empty set (not the same as no retained entries!)
+ */
+ bool is_empty() const;
+
+ /**
+ * @return estimate of the distinct count of the input stream
+ */
+ double get_estimate() const;
+
+ /**
+ * Returns the approximate lower error bound given a number of standard deviations.
+ * This parameter is similar to the number of standard deviations of the normal distribution
+ * and corresponds to approximately 67%, 95% and 99% confidence intervals.
+ * @param num_std_devs number of Standard Deviations (1, 2 or 3)
+ * @return the lower bound
+ */
+ double get_lower_bound(uint8_t num_std_devs) const;
+
+ /**
+ * Returns the approximate upper error bound given a number of standard deviations.
+ * This parameter is similar to the number of standard deviations of the normal distribution
+ * and corresponds to approximately 67%, 95% and 99% confidence intervals.
+ * @param num_std_devs number of Standard Deviations (1, 2 or 3)
+ * @return the upper bound
+ */
+ double get_upper_bound(uint8_t num_std_devs) const;
+
+ /**
+ * @return true if the sketch is in estimation mode (as opposed to exact mode)
+ */
+ bool is_estimation_mode() const;
+
+ /**
+ * @return theta as a fraction from 0 to 1 (effective sampling rate)
+ */
+ double get_theta() const;
+
+ /**
+ * @return theta as a positive integer between 0 and LLONG_MAX
+ */
+ uint64_t get_theta64() const;
+
+ /**
+ * @return the number of retained entries in the sketch
+ */
+ virtual uint32_t get_num_retained() const = 0;
+
+ virtual uint16_t get_seed_hash() const = 0;
+
+ /**
+ * @return true if retained entries are ordered
+ */
+ virtual bool is_ordered() const = 0;
+
+ /**
+ * Writes a human-readable summary of this sketch to a given stream
+ * @param print_items if true include the list of items retained by the sketch
+ */
+ virtual string<A> to_string(bool print_items = false) const = 0;
+
+ /**
+ * This method serializes the sketch into a given stream in a binary form
+ * @param os output stream
+ */
+ virtual void serialize(std::ostream& os) const = 0;
+
+ // This is a convenience alias for users
+ // The type returned by the following serialize method
+ typedef vector_u8<A> vector_bytes;
+
+ /**
+ * This method serializes the sketch as a vector of bytes.
+ * An optional header can be reserved in front of the sketch.
+ * It is an uninitialized space of a given size.
+ * This header is used in Datasketches PostgreSQL extension.
+ * @param header_size_bytes space to reserve in front of the sketch
+ */
+ virtual vector_bytes serialize(unsigned header_size_bytes = 0) const = 0;
+
+ // This is a convenience alias for users
+ // The type returned by the following deserialize methods
+ // It is not possible to return instances of an abstract type, so this has to be a pointer
+ typedef std::unique_ptr<theta_sketch_alloc<A>, std::function<void(theta_sketch_alloc<A>*)>> unique_ptr;
+
+ /**
+ * This method deserializes a sketch from a given stream.
+ * @param is input stream
+ * @param seed the seed for the hash function that was used to create the sketch
+ * @return an instance of a sketch as a unique_ptr
+ */
+ static unique_ptr deserialize(std::istream& is, uint64_t seed = DEFAULT_SEED);
+
+ /**
+ * This method deserializes a sketch from a given array of bytes.
+ * @param bytes pointer to the array of bytes
+ * @param size the size of the array
+ * @param seed the seed for the hash function that was used to create the sketch
+ * @return an instance of the sketch
+ */
+ static unique_ptr deserialize(const void* bytes, size_t size, uint64_t seed = DEFAULT_SEED);
+
+ class const_iterator;
+
+ /**
+ * Iterator over hash values in this sketch.
+ * @return begin iterator
+ */
+ virtual const_iterator begin() const = 0;
+
+ /**
+ * Iterator pointing past the valid range.
+ * Not to be incremented or dereferenced.
+ * @return end iterator
+ */
+ virtual const_iterator end() const = 0;
+
+protected:
+ enum flags { IS_BIG_ENDIAN, IS_READ_ONLY, IS_EMPTY, IS_COMPACT, IS_ORDERED };
+
+ bool is_empty_;
+ uint64_t theta_;
+
+ theta_sketch_alloc(bool is_empty, uint64_t theta);
+
+ static uint16_t get_seed_hash(uint64_t seed);
+
+ static void check_sketch_type(uint8_t actual, uint8_t expected);
+ static void check_serial_version(uint8_t actual, uint8_t expected);
+ static void check_seed_hash(uint16_t actual, uint16_t expected);
+
+ friend theta_intersection_alloc<A>;
+ friend theta_a_not_b_alloc<A>;
+};
+
+// update sketch
+
+template<typename A> using AllocU64 = typename std::allocator_traits<A>::template rebind_alloc<uint64_t>;
+template<typename A> using vector_u64 = std::vector<uint64_t, AllocU64<A>>;
+
+template<typename A>
+class update_theta_sketch_alloc: public theta_sketch_alloc<A> {
+public:
+ class builder;
+ enum resize_factor { X1, X2, X4, X8 };
+ static const uint8_t SKETCH_TYPE = 2;
+
+ // No constructor here. Use builder instead.
+
+ virtual ~update_theta_sketch_alloc() = default;
+
+ virtual uint32_t get_num_retained() const;
+ virtual uint16_t get_seed_hash() const;
+ virtual bool is_ordered() const;
+ virtual string<A> to_string(bool print_items = false) const;
+ virtual void serialize(std::ostream& os) const;
+ typedef vector_u8<A> vector_bytes; // alias for users
+ // header space is reserved, but not initialized
+ virtual vector_bytes serialize(unsigned header_size_bytes = 0) const;
+
+ /**
+ * Update this sketch with a given string.
+ * @param value string to update the sketch with
+ */
+ void update(const std::string& value);
+
+ /**
+ * Update this sketch with a given unsigned 64-bit integer.
+ * @param value uint64_t to update the sketch with
+ */
+ void update(uint64_t value);
+
+ /**
+ * Update this sketch with a given signed 64-bit integer.
+ * @param value int64_t to update the sketch with
+ */
+ void update(int64_t value);
+
+ /**
+ * Update this sketch with a given unsigned 32-bit integer.
+ * For compatibility with Java implementation.
+ * @param value uint32_t to update the sketch with
+ */
+ void update(uint32_t value);
+
+ /**
+ * Update this sketch with a given signed 32-bit integer.
+ * For compatibility with Java implementation.
+ * @param value int32_t to update the sketch with
+ */
+ void update(int32_t value);
+
+ /**
+ * Update this sketch with a given unsigned 16-bit integer.
+ * For compatibility with Java implementation.
+ * @param value uint16_t to update the sketch with
+ */
+ void update(uint16_t value);
+
+ /**
+ * Update this sketch with a given signed 16-bit integer.
+ * For compatibility with Java implementation.
+ * @param value int16_t to update the sketch with
+ */
+ void update(int16_t value);
+
+ /**
+ * Update this sketch with a given unsigned 8-bit integer.
+ * For compatibility with Java implementation.
+ * @param value uint8_t to update the sketch with
+ */
+ void update(uint8_t value);
+
+ /**
+ * Update this sketch with a given signed 8-bit integer.
+ * For compatibility with Java implementation.
+ * @param value int8_t to update the sketch with
+ */
+ void update(int8_t value);
+
+ /**
+ * Update this sketch with a given double-precision floating point value.
+ * For compatibility with Java implementation.
+ * @param value double to update the sketch with
+ */
+ void update(double value);
+
+ /**
+ * Update this sketch with a given floating point value.
+ * For compatibility with Java implementation.
+ * @param value float to update the sketch with
+ */
+ void update(float value);
+
+ /**
+ * Update this sketch with given data of any type.
+ * This is a "universal" update that covers all cases above,
+ * but may produce different hashes.
+ * Be very careful to hash input values consistently using the same approach
+ * both over time and on different platforms
+ * and while passing sketches between C++ environment and Java environment.
+ * Otherwise two sketches that should represent overlapping sets will be disjoint
+ * For instance, for signed 32-bit values call update(int32_t) method above,
+ * which does widening conversion to int64_t, if compatibility with Java is expected
+ * @param data pointer to the data
+ * @param length of the data in bytes
+ */
+ void update(const void* data, unsigned length);
+
+ /**
+ * Remove retained entries in excess of the nominal size k (if any)
+ */
+ void trim();
+
+ /**
+ * Converts this sketch to a compact sketch (ordered or unordered).
+ * @param ordered optional flag to specify if ordered sketch should be produced
+ * @return compact sketch
+ */
+ compact_theta_sketch_alloc<A> compact(bool ordered = true) const;
+
+ virtual typename theta_sketch_alloc<A>::const_iterator begin() const;
+ virtual typename theta_sketch_alloc<A>::const_iterator end() const;
+
+ /**
+ * This method deserializes a sketch from a given stream.
+ * @param is input stream
+ * @param seed the seed for the hash function that was used to create the sketch
+ * @return an instance of a sketch
+ */
+ static update_theta_sketch_alloc<A> deserialize(std::istream& is, uint64_t seed = DEFAULT_SEED);
+
+ /**
+ * This method deserializes a sketch from a given array of bytes.
+ * @param bytes pointer to the array of bytes
+ * @param size the size of the array
+ * @param seed the seed for the hash function that was used to create the sketch
+ * @return an instance of the sketch
+ */
+ static update_theta_sketch_alloc<A> deserialize(const void* bytes, size_t size, uint64_t seed = DEFAULT_SEED);
+
+private:
+ // resize threshold = 0.5 tuned for speed
+ static constexpr double RESIZE_THRESHOLD = 0.5;
+ // hash table rebuild threshold = 15/16
+ static constexpr double REBUILD_THRESHOLD = 15.0 / 16.0;
+
+ static constexpr uint8_t STRIDE_HASH_BITS = 7;
+ static constexpr uint32_t STRIDE_MASK = (1 << STRIDE_HASH_BITS) - 1;
+
+ uint8_t lg_cur_size_;
+ uint8_t lg_nom_size_;
+ vector_u64<A> keys_;
+ uint32_t num_keys_;
+ resize_factor rf_;
+ float p_;
+ uint64_t seed_;
+ uint32_t capacity_;
+
+ // for builder
+ update_theta_sketch_alloc(uint8_t lg_cur_size, uint8_t lg_nom_size, resize_factor rf, float p, uint64_t seed);
+
+ // for deserialize
+ update_theta_sketch_alloc(bool is_empty, uint64_t theta, uint8_t lg_cur_size, uint8_t lg_nom_size, vector_u64<A>&& keys, uint32_t num_keys, resize_factor rf, float p, uint64_t seed);
+
+ void resize();
+ void rebuild();
+
+ friend theta_union_alloc<A>;
+ void internal_update(uint64_t hash);
+
+ friend theta_intersection_alloc<A>;
+ friend theta_a_not_b_alloc<A>;
+ static inline uint32_t get_capacity(uint8_t lg_cur_size, uint8_t lg_nom_size);
+ static inline uint32_t get_stride(uint64_t hash, uint8_t lg_size);
+ static bool hash_search_or_insert(uint64_t hash, uint64_t* table, uint8_t lg_size);
+ static bool hash_search(uint64_t hash, const uint64_t* table, uint8_t lg_size);
+
+ friend theta_sketch_alloc<A>;
+ static update_theta_sketch_alloc<A> internal_deserialize(std::istream& is, resize_factor rf, uint8_t lg_cur_size, uint8_t lg_nom_size, uint8_t flags_byte, uint64_t seed);
+ static update_theta_sketch_alloc<A> internal_deserialize(const void* bytes, size_t size, resize_factor rf, uint8_t lg_cur_size, uint8_t lg_nom_size, uint8_t flags_byte, uint64_t seed);
+};
+
+// compact sketch
+
+template<typename A>
+class compact_theta_sketch_alloc: public theta_sketch_alloc<A> {
+public:
+ static const uint8_t SKETCH_TYPE = 3;
+
+ // No constructor here.
+ // Instances of this type can be obtained:
+ // - by compacting an update_theta_sketch
+ // - as a result of a set operation
+ // - by deserializing a previously serialized compact sketch
+
+ compact_theta_sketch_alloc(const theta_sketch_alloc<A>& other, bool ordered);
+ virtual ~compact_theta_sketch_alloc() = default;
+
+ virtual uint32_t get_num_retained() const;
+ virtual uint16_t get_seed_hash() const;
+ virtual bool is_ordered() const;
+ virtual string<A> to_string(bool print_items = false) const;
+ virtual void serialize(std::ostream& os) const;
+ typedef vector_u8<A> vector_bytes; // alias for users
+ // header space is reserved, but not initialized
+ virtual vector_bytes serialize(unsigned header_size_bytes = 0) const;
+
+ virtual typename theta_sketch_alloc<A>::const_iterator begin() const;
+ virtual typename theta_sketch_alloc<A>::const_iterator end() const;
+
+ /**
+ * This method deserializes a sketch from a given stream.
+ * @param is input stream
+ * @param seed the seed for the hash function that was used to create the sketch
+ * @return an instance of a sketch
+ */
+ static compact_theta_sketch_alloc<A> deserialize(std::istream& is, uint64_t seed = DEFAULT_SEED);
+
+ /**
+ * This method deserializes a sketch from a given array of bytes.
+ * @param bytes pointer to the array of bytes
+ * @param size the size of the array
+ * @param seed the seed for the hash function that was used to create the sketch
+ * @return an instance of the sketch
+ */
+ static compact_theta_sketch_alloc<A> deserialize(const void* bytes, size_t size, uint64_t seed = DEFAULT_SEED);
+
+private:
+ typedef typename std::allocator_traits<A>::template rebind_alloc<uint64_t> AllocU64;
+
+ vector_u64<A> keys_;
+ uint16_t seed_hash_;
+ bool is_ordered_;
+
+ friend theta_sketch_alloc<A>;
+ friend update_theta_sketch_alloc<A>;
+ friend theta_union_alloc<A>;
+ friend theta_intersection_alloc<A>;
+ friend theta_a_not_b_alloc<A>;
+ compact_theta_sketch_alloc(bool is_empty, uint64_t theta, vector_u64<A>&& keys, uint16_t seed_hash, bool is_ordered);
+ static compact_theta_sketch_alloc<A> internal_deserialize(std::istream& is, uint8_t preamble_longs, uint8_t flags_byte, uint16_t seed_hash);
+ static compact_theta_sketch_alloc<A> internal_deserialize(const void* bytes, size_t size, uint8_t preamble_longs, uint8_t flags_byte, uint16_t seed_hash);
+};
+
+// builder
+
+template<typename A>
+class update_theta_sketch_alloc<A>::builder {
+public:
+ static const uint8_t MIN_LG_K = 5;
+ static const uint8_t DEFAULT_LG_K = 12;
+ static const resize_factor DEFAULT_RESIZE_FACTOR = X8;
+
+ /**
+ * Creates and instance of the builder with default parameters.
+ */
+ builder();
+
+ /**
+ * Set log2(k), where k is a nominal number of entries in the sketch
+ * @param lg_k base 2 logarithm of nominal number of entries
+ * @return this builder
+ */
+ builder& set_lg_k(uint8_t lg_k);
+
+ /**
+ * Set resize factor for the internal hash table (defaults to 8)
+ * @param rf resize factor
+ * @return this builder
+ */
+ builder& set_resize_factor(resize_factor rf);
+
+ /**
+ * Set sampling probability (initial theta). The default is 1, so the sketch retains
+ * all entries until it reaches the limit, at which point it goes into the estimation mode
+ * and reduces the effective sampling probability (theta) as necessary.
+ * @param p sampling probability
+ * @return this builder
+ */
+ builder& set_p(float p);
+
+ /**
+ * Set the seed for the hash function. Should be used carefully if needed.
+ * Sketches produced with different seed are not compatible
+ * and cannot be mixed in set operations.
+ * @param seed hash seed
+ * @return this builder
+ */
+ builder& set_seed(uint64_t seed);
+
+ /**
+ * This is to create an instance of the sketch with predefined parameters.
+ * @return and instance of the sketch
+ */
+ update_theta_sketch_alloc<A> build() const;
+
+private:
+ uint8_t lg_k_;
+ resize_factor rf_;
+ float p_;
+ uint64_t seed_;
+
+ static uint8_t starting_sub_multiple(uint8_t lg_tgt, uint8_t lg_min, uint8_t lg_rf);
+};
+
+// iterator
+template<typename A>
+class theta_sketch_alloc<A>::const_iterator: public std::iterator<std::input_iterator_tag, uint64_t> {
+public:
+ const_iterator& operator++();
+ const_iterator operator++(int);
+ bool operator==(const const_iterator& other) const;
+ bool operator!=(const const_iterator& other) const;
+ uint64_t operator*() const;
+
+private:
+ const uint64_t* keys_;
+ uint32_t size_;
+ uint32_t index_;
+ const_iterator(const uint64_t* keys, uint32_t size, uint32_t index);
+ friend class update_theta_sketch_alloc<A>;
+ friend class compact_theta_sketch_alloc<A>;
+};
+
+
+// aliases with default allocator for convenience
+typedef theta_sketch_alloc<std::allocator<void>> theta_sketch;
+typedef update_theta_sketch_alloc<std::allocator<void>> update_theta_sketch;
+typedef compact_theta_sketch_alloc<std::allocator<void>> compact_theta_sketch;
+
+} /* namespace datasketches */
+
+#include "theta_sketch_impl.hpp"
+
+#endif
diff --git a/be/src/thirdparty/datasketches/theta_sketch_impl.hpp b/be/src/thirdparty/datasketches/theta_sketch_impl.hpp
new file mode 100644
index 0000000..579a675
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_sketch_impl.hpp
@@ -0,0 +1,939 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_SKETCH_IMPL_HPP_
+#define THETA_SKETCH_IMPL_HPP_
+
+#include <algorithm>
+#include <cmath>
+#include <memory>
+#include <functional>
+#include <istream>
+#include <ostream>
+#include <sstream>
+
+#include "MurmurHash3.h"
+#include "serde.hpp"
+#include "binomial_bounds.hpp"
+#include "memory_operations.hpp"
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+template<typename A>
+theta_sketch_alloc<A>::theta_sketch_alloc(bool is_empty, uint64_t theta):
+is_empty_(is_empty), theta_(theta)
+{}
+
+template<typename A>
+bool theta_sketch_alloc<A>::is_empty() const {
+ return is_empty_;
+}
+
+template<typename A>
+double theta_sketch_alloc<A>::get_estimate() const {
+ return get_num_retained() / get_theta();
+}
+
+template<typename A>
+double theta_sketch_alloc<A>::get_lower_bound(uint8_t num_std_devs) const {
+ if (!is_estimation_mode()) return get_num_retained();
+ return binomial_bounds::get_lower_bound(get_num_retained(), get_theta(), num_std_devs);
+}
+
+template<typename A>
+double theta_sketch_alloc<A>::get_upper_bound(uint8_t num_std_devs) const {
+ if (!is_estimation_mode()) return get_num_retained();
+ return binomial_bounds::get_upper_bound(get_num_retained(), get_theta(), num_std_devs);
+}
+
+template<typename A>
+bool theta_sketch_alloc<A>::is_estimation_mode() const {
+ return theta_ < MAX_THETA && !is_empty_;
+}
+
+template<typename A>
+double theta_sketch_alloc<A>::get_theta() const {
+ return (double) theta_ / MAX_THETA;
+}
+
+template<typename A>
+uint64_t theta_sketch_alloc<A>::get_theta64() const {
+ return theta_;
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::unique_ptr theta_sketch_alloc<A>::deserialize(std::istream& is, uint64_t seed) {
+ uint8_t preamble_longs;
+ is.read((char*)&preamble_longs, sizeof(preamble_longs));
+ uint8_t serial_version;
+ is.read((char*)&serial_version, sizeof(serial_version));
+ uint8_t type;
+ is.read((char*)&type, sizeof(type));
+ uint8_t lg_nom_size;
+ is.read((char*)&lg_nom_size, sizeof(lg_nom_size));
+ uint8_t lg_cur_size;
+ is.read((char*)&lg_cur_size, sizeof(lg_cur_size));
+ uint8_t flags_byte;
+ is.read((char*)&flags_byte, sizeof(flags_byte));
+ uint16_t seed_hash;
+ is.read((char*)&seed_hash, sizeof(seed_hash));
+
+ check_serial_version(serial_version, SERIAL_VERSION);
+
+ if (type == update_theta_sketch_alloc<A>::SKETCH_TYPE) {
+ check_seed_hash(seed_hash, get_seed_hash(seed));
+ typename update_theta_sketch_alloc<A>::resize_factor rf = static_cast<typename update_theta_sketch_alloc<A>::resize_factor>(preamble_longs >> 6);
+ typedef typename std::allocator_traits<A>::template rebind_alloc<update_theta_sketch_alloc<A>> AU;
+ return unique_ptr(
+ static_cast<theta_sketch_alloc<A>*>(new (AU().allocate(1)) update_theta_sketch_alloc<A>(update_theta_sketch_alloc<A>::internal_deserialize(is, rf, lg_cur_size, lg_nom_size, flags_byte, seed))),
+ [](theta_sketch_alloc<A>* ptr) {
+ ptr->~theta_sketch_alloc();
+ AU().deallocate(static_cast<update_theta_sketch_alloc<A>*>(ptr), 1);
+ }
+ );
+ } else if (type == compact_theta_sketch_alloc<A>::SKETCH_TYPE) {
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ if (!is_empty) check_seed_hash(seed_hash, get_seed_hash(seed));
+ typedef typename std::allocator_traits<A>::template rebind_alloc<compact_theta_sketch_alloc<A>> AC;
+ return unique_ptr(
+ static_cast<theta_sketch_alloc<A>*>(new (AC().allocate(1)) compact_theta_sketch_alloc<A>(compact_theta_sketch_alloc<A>::internal_deserialize(is, preamble_longs, flags_byte, seed_hash))),
+ [](theta_sketch_alloc<A>* ptr) {
+ ptr->~theta_sketch_alloc();
+ AC().deallocate(static_cast<compact_theta_sketch_alloc<A>*>(ptr), 1);
+ }
+ );
+ }
+ throw std::invalid_argument("unsupported sketch type " + std::to_string((int) type));
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::unique_ptr theta_sketch_alloc<A>::deserialize(const void* bytes, size_t size, uint64_t seed) {
+ ensure_minimum_memory(size, static_cast<size_t>(8));
+ const char* ptr = static_cast<const char*>(bytes);
+ uint8_t preamble_longs;
+ ptr += copy_from_mem(ptr, &preamble_longs, sizeof(preamble_longs));
+ uint8_t serial_version;
+ ptr += copy_from_mem(ptr, &serial_version, sizeof(serial_version));
+ uint8_t type;
+ ptr += copy_from_mem(ptr, &type, sizeof(type));
+ uint8_t lg_nom_size;
+ ptr += copy_from_mem(ptr, &lg_nom_size, sizeof(lg_nom_size));
+ uint8_t lg_cur_size;
+ ptr += copy_from_mem(ptr, &lg_cur_size, sizeof(lg_cur_size));
+ uint8_t flags_byte;
+ ptr += copy_from_mem(ptr, &flags_byte, sizeof(flags_byte));
+ uint16_t seed_hash;
+ ptr += copy_from_mem(ptr, &seed_hash, sizeof(seed_hash));
+
+ check_serial_version(serial_version, SERIAL_VERSION);
+
+ if (type == update_theta_sketch_alloc<A>::SKETCH_TYPE) {
+ check_seed_hash(seed_hash, get_seed_hash(seed));
+ typename update_theta_sketch_alloc<A>::resize_factor rf = static_cast<typename update_theta_sketch_alloc<A>::resize_factor>(preamble_longs >> 6);
+ typedef typename std::allocator_traits<A>::template rebind_alloc<update_theta_sketch_alloc<A>> AU;
+ return unique_ptr(
+ static_cast<theta_sketch_alloc<A>*>(new (AU().allocate(1)) update_theta_sketch_alloc<A>(
+ update_theta_sketch_alloc<A>::internal_deserialize(ptr, size - (ptr - static_cast<const char*>(bytes)), rf, lg_cur_size, lg_nom_size, flags_byte, seed))
+ ),
+ [](theta_sketch_alloc<A>* ptr) {
+ ptr->~theta_sketch_alloc();
+ AU().deallocate(static_cast<update_theta_sketch_alloc<A>*>(ptr), 1);
+ }
+ );
+ } else if (type == compact_theta_sketch_alloc<A>::SKETCH_TYPE) {
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ if (!is_empty) check_seed_hash(seed_hash, get_seed_hash(seed));
+ typedef typename std::allocator_traits<A>::template rebind_alloc<compact_theta_sketch_alloc<A>> AC;
+ return unique_ptr(
+ static_cast<theta_sketch_alloc<A>*>(new (AC().allocate(1)) compact_theta_sketch_alloc<A>(
+ compact_theta_sketch_alloc<A>::internal_deserialize(ptr, size - (ptr - static_cast<const char*>(bytes)), preamble_longs, flags_byte, seed_hash))
+ ),
+ [](theta_sketch_alloc<A>* ptr) {
+ ptr->~theta_sketch_alloc();
+ AC().deallocate(static_cast<compact_theta_sketch_alloc<A>*>(ptr), 1);
+ }
+ );
+ }
+ throw std::invalid_argument("unsupported sketch type " + std::to_string((int) type));
+}
+
+template<typename A>
+uint16_t theta_sketch_alloc<A>::get_seed_hash(uint64_t seed) {
+ HashState hashes;
+ MurmurHash3_x64_128(&seed, sizeof(seed), 0, hashes);
+ return hashes.h1;
+}
+
+template<typename A>
+void theta_sketch_alloc<A>::check_sketch_type(uint8_t actual, uint8_t expected) {
+ if (actual != expected) {
+ throw std::invalid_argument("Sketch type mismatch: expected " + std::to_string((int)expected) + ", actual " + std::to_string((int)actual));
+ }
+}
+
+template<typename A>
+void theta_sketch_alloc<A>::check_serial_version(uint8_t actual, uint8_t expected) {
+ if (actual != expected) {
+ throw std::invalid_argument("Sketch serial version mismatch: expected " + std::to_string((int)expected) + ", actual " + std::to_string((int)actual));
+ }
+}
+
+template<typename A>
+void theta_sketch_alloc<A>::check_seed_hash(uint16_t actual, uint16_t expected) {
+ if (actual != expected) {
+ throw std::invalid_argument("Sketch seed hash mismatch: expected " + std::to_string(expected) + ", actual " + std::to_string(actual));
+ }
+}
+
+// update sketch
+
+template<typename A>
+update_theta_sketch_alloc<A>::update_theta_sketch_alloc(uint8_t lg_cur_size, uint8_t lg_nom_size, resize_factor rf, float p, uint64_t seed):
+theta_sketch_alloc<A>(true, theta_sketch_alloc<A>::MAX_THETA),
+lg_cur_size_(lg_cur_size),
+lg_nom_size_(lg_nom_size),
+keys_(1 << lg_cur_size_, 0),
+num_keys_(0),
+rf_(rf),
+p_(p),
+seed_(seed),
+capacity_(get_capacity(lg_cur_size, lg_nom_size))
+{
+ if (p < 1) this->theta_ *= p;
+}
+
+template<typename A>
+update_theta_sketch_alloc<A>::update_theta_sketch_alloc(bool is_empty, uint64_t theta, uint8_t lg_cur_size, uint8_t lg_nom_size, vector_u64<A>&& keys, uint32_t num_keys, resize_factor rf, float p, uint64_t seed):
+theta_sketch_alloc<A>(is_empty, theta),
+lg_cur_size_(lg_cur_size),
+lg_nom_size_(lg_nom_size),
+keys_(std::move(keys)),
+num_keys_(num_keys),
+rf_(rf),
+p_(p),
+seed_(seed),
+capacity_(get_capacity(lg_cur_size, lg_nom_size))
+{}
+
+template<typename A>
+uint32_t update_theta_sketch_alloc<A>::get_num_retained() const {
+ return num_keys_;
+}
+
+template<typename A>
+uint16_t update_theta_sketch_alloc<A>::get_seed_hash() const {
+ return theta_sketch_alloc<A>::get_seed_hash(seed_);
+}
+
+template<typename A>
+bool update_theta_sketch_alloc<A>::is_ordered() const {
+ return false;
+}
+
+template<typename A>
+string<A> update_theta_sketch_alloc<A>::to_string(bool print_items) const {
+ std::basic_ostringstream<char, std::char_traits<char>, AllocChar<A>> os;
+ os << "### Update Theta sketch summary:" << std::endl;
+ os << " lg nominal size : " << (int) lg_nom_size_ << std::endl;
+ os << " lg current size : " << (int) lg_cur_size_ << std::endl;
+ os << " num retained keys : " << num_keys_ << std::endl;
+ os << " resize factor : " << (1 << rf_) << std::endl;
+ os << " sampling probability : " << p_ << std::endl;
+ os << " seed hash : " << this->get_seed_hash() << std::endl;
+ os << " empty? : " << (this->is_empty() ? "true" : "false") << std::endl;
+ os << " ordered? : " << (this->is_ordered() ? "true" : "false") << std::endl;
+ os << " estimation mode? : " << (this->is_estimation_mode() ? "true" : "false") << std::endl;
+ os << " theta (fraction) : " << this->get_theta() << std::endl;
+ os << " theta (raw 64-bit) : " << this->theta_ << std::endl;
+ os << " estimate : " << this->get_estimate() << std::endl;
+ os << " lower bound 95% conf : " << this->get_lower_bound(2) << std::endl;
+ os << " upper bound 95% conf : " << this->get_upper_bound(2) << std::endl;
+ os << "### End sketch summary" << std::endl;
+ if (print_items) {
+ os << "### Retained keys" << std::endl;
+ for (auto key: *this) os << " " << key << std::endl;
+ os << "### End retained keys" << std::endl;
+ }
+ return os.str();
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::serialize(std::ostream& os) const {
+ const uint8_t preamble_longs_and_rf = 3 | (rf_ << 6);
+ os.write((char*)&preamble_longs_and_rf, sizeof(preamble_longs_and_rf));
+ const uint8_t serial_version = theta_sketch_alloc<A>::SERIAL_VERSION;
+ os.write((char*)&serial_version, sizeof(serial_version));
+ const uint8_t type = SKETCH_TYPE;
+ os.write((char*)&type, sizeof(type));
+ os.write((char*)&lg_nom_size_, sizeof(lg_nom_size_));
+ os.write((char*)&lg_cur_size_, sizeof(lg_cur_size_));
+ const uint8_t flags_byte(
+ (this->is_empty() ? 1 << theta_sketch_alloc<A>::flags::IS_EMPTY : 0)
+ );
+ os.write((char*)&flags_byte, sizeof(flags_byte));
+ const uint16_t seed_hash = get_seed_hash();
+ os.write((char*)&seed_hash, sizeof(seed_hash));
+ os.write((char*)&num_keys_, sizeof(num_keys_));
+ os.write((char*)&p_, sizeof(p_));
+ os.write((char*)&(this->theta_), sizeof(uint64_t));
+ os.write((char*)keys_.data(), sizeof(uint64_t) * keys_.size());
+}
+
+template<typename A>
+vector_u8<A> update_theta_sketch_alloc<A>::serialize(unsigned header_size_bytes) const {
+ const uint8_t preamble_longs = 3;
+ const size_t size = header_size_bytes + sizeof(uint64_t) * preamble_longs + sizeof(uint64_t) * keys_.size();
+ vector_u8<A> bytes(size);
+ uint8_t* ptr = bytes.data() + header_size_bytes;
+
+ const uint8_t preamble_longs_and_rf = preamble_longs | (rf_ << 6);
+ ptr += copy_to_mem(&preamble_longs_and_rf, ptr, sizeof(preamble_longs_and_rf));
+ const uint8_t serial_version = theta_sketch_alloc<A>::SERIAL_VERSION;
+ ptr += copy_to_mem(&serial_version, ptr, sizeof(serial_version));
+ const uint8_t type = SKETCH_TYPE;
+ ptr += copy_to_mem(&type, ptr, sizeof(type));
+ ptr += copy_to_mem(&lg_nom_size_, ptr, sizeof(lg_nom_size_));
+ ptr += copy_to_mem(&lg_cur_size_, ptr, sizeof(lg_cur_size_));
+ const uint8_t flags_byte(
+ (this->is_empty() ? 1 << theta_sketch_alloc<A>::flags::IS_EMPTY : 0)
+ );
+ ptr += copy_to_mem(&flags_byte, ptr, sizeof(flags_byte));
+ const uint16_t seed_hash = get_seed_hash();
+ ptr += copy_to_mem(&seed_hash, ptr, sizeof(seed_hash));
+ ptr += copy_to_mem(&num_keys_, ptr, sizeof(num_keys_));
+ ptr += copy_to_mem(&p_, ptr, sizeof(p_));
+ ptr += copy_to_mem(&(this->theta_), ptr, sizeof(uint64_t));
+ ptr += copy_to_mem(keys_.data(), ptr, sizeof(uint64_t) * keys_.size());
+
+ return bytes;
+}
+
+template<typename A>
+update_theta_sketch_alloc<A> update_theta_sketch_alloc<A>::deserialize(std::istream& is, uint64_t seed) {
+ uint8_t preamble_longs;
+ is.read((char*)&preamble_longs, sizeof(preamble_longs));
+ resize_factor rf = static_cast<resize_factor>(preamble_longs >> 6);
+ preamble_longs &= 0x3f; // remove resize factor
+ uint8_t serial_version;
+ is.read((char*)&serial_version, sizeof(serial_version));
+ uint8_t type;
+ is.read((char*)&type, sizeof(type));
+ uint8_t lg_nom_size;
+ is.read((char*)&lg_nom_size, sizeof(lg_nom_size));
+ uint8_t lg_cur_size;
+ is.read((char*)&lg_cur_size, sizeof(lg_cur_size));
+ uint8_t flags_byte;
+ is.read((char*)&flags_byte, sizeof(flags_byte));
+ uint16_t seed_hash;
+ is.read((char*)&seed_hash, sizeof(seed_hash));
+ theta_sketch_alloc<A>::check_sketch_type(type, SKETCH_TYPE);
+ theta_sketch_alloc<A>::check_serial_version(serial_version, theta_sketch_alloc<A>::SERIAL_VERSION);
+ theta_sketch_alloc<A>::check_seed_hash(seed_hash, theta_sketch_alloc<A>::get_seed_hash(seed));
+ return internal_deserialize(is, rf, lg_cur_size, lg_nom_size, flags_byte, seed);
+}
+
+template<typename A>
+update_theta_sketch_alloc<A> update_theta_sketch_alloc<A>::internal_deserialize(std::istream& is, resize_factor rf, uint8_t lg_cur_size, uint8_t lg_nom_size, uint8_t flags_byte, uint64_t seed) {
+ uint32_t num_keys;
+ is.read((char*)&num_keys, sizeof(num_keys));
+ float p;
+ is.read((char*)&p, sizeof(p));
+ uint64_t theta;
+ is.read((char*)&theta, sizeof(theta));
+ vector_u64<A> keys(1 << lg_cur_size);
+ is.read((char*)keys.data(), sizeof(uint64_t) * keys.size());
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ if (!is.good()) throw std::runtime_error("error reading from std::istream");
+ return update_theta_sketch_alloc<A>(is_empty, theta, lg_cur_size, lg_nom_size, std::move(keys), num_keys, rf, p, seed);
+}
+
+template<typename A>
+update_theta_sketch_alloc<A> update_theta_sketch_alloc<A>::deserialize(const void* bytes, size_t size, uint64_t seed) {
+ ensure_minimum_memory(size, 8);
+ const char* ptr = static_cast<const char*>(bytes);
+ uint8_t preamble_longs;
+ ptr += copy_from_mem(ptr, &preamble_longs, sizeof(preamble_longs));
+ resize_factor rf = static_cast<resize_factor>(preamble_longs >> 6);
+ preamble_longs &= 0x3f; // remove resize factor
+ uint8_t serial_version;
+ ptr += copy_from_mem(ptr, &serial_version, sizeof(serial_version));
+ uint8_t type;
+ ptr += copy_from_mem(ptr, &type, sizeof(type));
+ uint8_t lg_nom_size;
+ ptr += copy_from_mem(ptr, &lg_nom_size, sizeof(lg_nom_size));
+ uint8_t lg_cur_size;
+ ptr += copy_from_mem(ptr, &lg_cur_size, sizeof(lg_cur_size));
+ uint8_t flags_byte;
+ ptr += copy_from_mem(ptr, &flags_byte, sizeof(flags_byte));
+ uint16_t seed_hash;
+ ptr += copy_from_mem(ptr, &seed_hash, sizeof(seed_hash));
+ theta_sketch_alloc<A>::check_sketch_type(type, SKETCH_TYPE);
+ theta_sketch_alloc<A>::check_serial_version(serial_version, theta_sketch_alloc<A>::SERIAL_VERSION);
+ theta_sketch_alloc<A>::check_seed_hash(seed_hash, theta_sketch_alloc<A>::get_seed_hash(seed));
+ return internal_deserialize(ptr, size - (ptr - static_cast<const char*>(bytes)), rf, lg_cur_size, lg_nom_size, flags_byte, seed);
+}
+
+template<typename A>
+update_theta_sketch_alloc<A> update_theta_sketch_alloc<A>::internal_deserialize(const void* bytes, size_t size, resize_factor rf, uint8_t lg_cur_size, uint8_t lg_nom_size, uint8_t flags_byte, uint64_t seed) {
+ const uint32_t table_size = 1 << lg_cur_size;
+ ensure_minimum_memory(size, 16 + sizeof(uint64_t) * table_size);
+ const char* ptr = static_cast<const char*>(bytes);
+ uint32_t num_keys;
+ ptr += copy_from_mem(ptr, &num_keys, sizeof(num_keys));
+ float p;
+ ptr += copy_from_mem(ptr, &p, sizeof(p));
+ uint64_t theta;
+ ptr += copy_from_mem(ptr, &theta, sizeof(theta));
+ vector_u64<A> keys(table_size);
+ ptr += copy_from_mem(ptr, keys.data(), sizeof(uint64_t) * table_size);
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ return update_theta_sketch_alloc<A>(is_empty, theta, lg_cur_size, lg_nom_size, std::move(keys), num_keys, rf, p, seed);
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(const std::string& value) {
+ if (value.empty()) return;
+ update(value.c_str(), value.length());
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(uint64_t value) {
+ update(&value, sizeof(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(int64_t value) {
+ update(&value, sizeof(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(uint32_t value) {
+ update(static_cast<int32_t>(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(int32_t value) {
+ update(static_cast<int64_t>(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(uint16_t value) {
+ update(static_cast<int16_t>(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(int16_t value) {
+ update(static_cast<int64_t>(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(uint8_t value) {
+ update(static_cast<int8_t>(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(int8_t value) {
+ update(static_cast<int64_t>(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(double value) {
+ union {
+ int64_t long_value;
+ double double_value;
+ } long_double_union;
+
+ if (value == 0.0) {
+ long_double_union.double_value = 0.0; // canonicalize -0.0 to 0.0
+ } else if (std::isnan(value)) {
+ long_double_union.long_value = 0x7ff8000000000000L; // canonicalize NaN using value from Java's Double.doubleToLongBits()
+ } else {
+ long_double_union.double_value = value;
+ }
+ update(&long_double_union, sizeof(long_double_union));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(float value) {
+ update(static_cast<double>(value));
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::update(const void* data, unsigned length) {
+ HashState hashes;
+ MurmurHash3_x64_128(data, length, seed_, hashes);
+ const uint64_t hash = hashes.h1 >> 1; // Java implementation does logical shift >>> to make values positive
+ internal_update(hash);
+}
+
+template<typename A>
+compact_theta_sketch_alloc<A> update_theta_sketch_alloc<A>::compact(bool ordered) const {
+ return compact_theta_sketch_alloc<A>(*this, ordered);
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::internal_update(uint64_t hash) {
+ this->is_empty_ = false;
+ if (hash >= this->theta_ || hash == 0) return; // hash == 0 is reserved to mark empty slots in the table
+ if (hash_search_or_insert(hash, keys_.data(), lg_cur_size_)) {
+ num_keys_++;
+ if (num_keys_ > capacity_) {
+ if (lg_cur_size_ <= lg_nom_size_) {
+ resize();
+ } else {
+ rebuild();
+ }
+ }
+ }
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::trim() {
+ if (num_keys_ > static_cast<uint32_t>(1 << lg_nom_size_)) rebuild();
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::resize() {
+ const uint8_t lg_tgt_size = lg_nom_size_ + 1;
+ const uint8_t factor = std::max(1, std::min(static_cast<int>(rf_), lg_tgt_size - lg_cur_size_));
+ const uint8_t lg_new_size = lg_cur_size_ + factor;
+ const uint32_t new_size = 1 << lg_new_size;
+ vector_u64<A> new_keys(new_size, 0);
+ for (uint32_t i = 0; i < keys_.size(); i++) {
+ if (keys_[i] != 0) {
+ hash_search_or_insert(keys_[i], new_keys.data(), lg_new_size); // TODO hash_insert
+ }
+ }
+ keys_ = std::move(new_keys);
+ lg_cur_size_ += factor;
+ capacity_ = get_capacity(lg_cur_size_, lg_nom_size_);
+}
+
+template<typename A>
+void update_theta_sketch_alloc<A>::rebuild() {
+ const uint32_t pivot = (1 << lg_nom_size_) + keys_.size() - num_keys_;
+ std::nth_element(keys_.begin(), keys_.begin() + pivot, keys_.end());
+ this->theta_ = keys_[pivot];
+ vector_u64<A> new_keys(keys_.size(), 0);
+ num_keys_ = 0;
+ for (uint32_t i = 0; i < keys_.size(); i++) {
+ if (keys_[i] != 0 && keys_[i] < this->theta_) {
+ hash_search_or_insert(keys_[i], new_keys.data(), lg_cur_size_); // TODO hash_insert
+ num_keys_++;
+ }
+ }
+ keys_ = std::move(new_keys);
+}
+
+template<typename A>
+uint32_t update_theta_sketch_alloc<A>::get_capacity(uint8_t lg_cur_size, uint8_t lg_nom_size) {
+ const double fraction = (lg_cur_size <= lg_nom_size) ? RESIZE_THRESHOLD : REBUILD_THRESHOLD;
+ return std::floor(fraction * (1 << lg_cur_size));
+}
+
+template<typename A>
+uint32_t update_theta_sketch_alloc<A>::get_stride(uint64_t hash, uint8_t lg_size) {
+ // odd and independent of index assuming lg_size lowest bits of the hash were used for the index
+ return (2 * static_cast<uint32_t>((hash >> lg_size) & STRIDE_MASK)) + 1;
+}
+
+template<typename A>
+bool update_theta_sketch_alloc<A>::hash_search_or_insert(uint64_t hash, uint64_t* table, uint8_t lg_size) {
+ const uint32_t mask = (1 << lg_size) - 1;
+ const uint32_t stride = get_stride(hash, lg_size);
+ uint32_t cur_probe = static_cast<uint32_t>(hash) & mask;
+
+ // search for duplicate or zero
+ const uint32_t loop_index = cur_probe;
+ do {
+ const uint64_t value = table[cur_probe];
+ if (value == 0) {
+ table[cur_probe] = hash; // insert value
+ return true;
+ } else if (value == hash) {
+ return false; // found a duplicate
+ }
+ cur_probe = (cur_probe + stride) & mask;
+ } while (cur_probe != loop_index);
+ throw std::logic_error("key not found and no empty slots!");
+}
+
+template<typename A>
+bool update_theta_sketch_alloc<A>::hash_search(uint64_t hash, const uint64_t* table, uint8_t lg_size) {
+ const uint32_t mask = (1 << lg_size) - 1;
+ const uint32_t stride = update_theta_sketch_alloc<A>::get_stride(hash, lg_size);
+ uint32_t cur_probe = static_cast<uint32_t>(hash) & mask;
+ const uint32_t loop_index = cur_probe;
+ do {
+ const uint64_t value = table[cur_probe];
+ if (value == 0) {
+ return false;
+ } else if (value == hash) {
+ return true;
+ }
+ cur_probe = (cur_probe + stride) & mask;
+ } while (cur_probe != loop_index);
+ throw std::logic_error("key not found and search wrapped");
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::const_iterator update_theta_sketch_alloc<A>::begin() const {
+ return typename theta_sketch_alloc<A>::const_iterator(keys_.data(), keys_.size(), 0);
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::const_iterator update_theta_sketch_alloc<A>::end() const {
+ return typename theta_sketch_alloc<A>::const_iterator(keys_.data(), keys_.size(), keys_.size());
+}
+
+// compact sketch
+
+template<typename A>
+compact_theta_sketch_alloc<A>::compact_theta_sketch_alloc(bool is_empty, uint64_t theta, vector_u64<A>&& keys, uint16_t seed_hash, bool is_ordered):
+theta_sketch_alloc<A>(is_empty, theta),
+keys_(std::move(keys)),
+seed_hash_(seed_hash),
+is_ordered_(is_ordered)
+{}
+
+template<typename A>
+compact_theta_sketch_alloc<A>::compact_theta_sketch_alloc(const theta_sketch_alloc<A>& other, bool ordered):
+theta_sketch_alloc<A>(other),
+keys_(other.get_num_retained()),
+seed_hash_(other.get_seed_hash()),
+is_ordered_(other.is_ordered() || ordered)
+{
+ std::copy(other.begin(), other.end(), keys_.begin());
+ if (ordered && !other.is_ordered()) std::sort(keys_.begin(), keys_.end());
+}
+
+template<typename A>
+uint32_t compact_theta_sketch_alloc<A>::get_num_retained() const {
+ return keys_.size();
+}
+
+template<typename A>
+uint16_t compact_theta_sketch_alloc<A>::get_seed_hash() const {
+ return seed_hash_;
+}
+
+template<typename A>
+bool compact_theta_sketch_alloc<A>::is_ordered() const {
+ return is_ordered_;
+}
+
+template<typename A>
+string<A> compact_theta_sketch_alloc<A>::to_string(bool print_items) const {
+ std::basic_ostringstream<char, std::char_traits<char>, AllocChar<A>> os;
+ os << "### Compact Theta sketch summary:" << std::endl;
+ os << " num retained keys : " << keys_.size() << std::endl;
+ os << " seed hash : " << this->get_seed_hash() << std::endl;
+ os << " empty? : " << (this->is_empty() ? "true" : "false") << std::endl;
+ os << " ordered? : " << (this->is_ordered() ? "true" : "false") << std::endl;
+ os << " estimation mode? : " << (this->is_estimation_mode() ? "true" : "false") << std::endl;
+ os << " theta (fraction) : " << this->get_theta() << std::endl;
+ os << " theta (raw 64-bit) : " << this->theta_ << std::endl;
+ os << " estimate : " << this->get_estimate() << std::endl;
+ os << " lower bound 95% conf : " << this->get_lower_bound(2) << std::endl;
+ os << " upper bound 95% conf : " << this->get_upper_bound(2) << std::endl;
+ os << "### End sketch summary" << std::endl;
+ if (print_items) {
+ os << "### Retained keys" << std::endl;
+ for (auto key: *this) os << " " << key << std::endl;
+ os << "### End retained keys" << std::endl;
+ }
+ return os.str();
+}
+
+template<typename A>
+void compact_theta_sketch_alloc<A>::serialize(std::ostream& os) const {
+ const bool is_single_item = keys_.size() == 1 && !this->is_estimation_mode();
+ const uint8_t preamble_longs = this->is_empty() || is_single_item ? 1 : this->is_estimation_mode() ? 3 : 2;
+ os.write(reinterpret_cast<const char*>(&preamble_longs), sizeof(preamble_longs));
+ const uint8_t serial_version = theta_sketch_alloc<A>::SERIAL_VERSION;
+ os.write(reinterpret_cast<const char*>(&serial_version), sizeof(serial_version));
+ const uint8_t type = SKETCH_TYPE;
+ os.write(reinterpret_cast<const char*>(&type), sizeof(type));
+ const uint16_t unused16 = 0;
+ os.write(reinterpret_cast<const char*>(&unused16), sizeof(unused16));
+ const uint8_t flags_byte(
+ (1 << theta_sketch_alloc<A>::flags::IS_COMPACT) |
+ (1 << theta_sketch_alloc<A>::flags::IS_READ_ONLY) |
+ (this->is_empty() ? 1 << theta_sketch_alloc<A>::flags::IS_EMPTY : 0) |
+ (this->is_ordered() ? 1 << theta_sketch_alloc<A>::flags::IS_ORDERED : 0)
+ );
+ os.write(reinterpret_cast<const char*>(&flags_byte), sizeof(flags_byte));
+ const uint16_t seed_hash = get_seed_hash();
+ os.write((char*)&seed_hash, sizeof(seed_hash));
+ if (!this->is_empty()) {
+ if (!is_single_item) {
+ const uint32_t num_keys = keys_.size();
+ os.write((char*)&num_keys, sizeof(num_keys));
+ const uint32_t unused32 = 0;
+ os.write((char*)&unused32, sizeof(unused32));
+ if (this->is_estimation_mode()) {
+ os.write((char*)&(this->theta_), sizeof(uint64_t));
+ }
+ }
+ os.write((char*)keys_.data(), sizeof(uint64_t) * keys_.size());
+ }
+}
+
+template<typename A>
+vector_u8<A> compact_theta_sketch_alloc<A>::serialize(unsigned header_size_bytes) const {
+ const bool is_single_item = keys_.size() == 1 && !this->is_estimation_mode();
+ const uint8_t preamble_longs = this->is_empty() || is_single_item ? 1 : this->is_estimation_mode() ? 3 : 2;
+ const size_t size = header_size_bytes + sizeof(uint64_t) * preamble_longs + sizeof(uint64_t) * keys_.size();
+ vector_u8<A> bytes(size);
+ uint8_t* ptr = bytes.data() + header_size_bytes;
+
+ ptr += copy_to_mem(&preamble_longs, ptr, sizeof(preamble_longs));
+ const uint8_t serial_version = theta_sketch_alloc<A>::SERIAL_VERSION;
+ ptr += copy_to_mem(&serial_version, ptr, sizeof(serial_version));
+ const uint8_t type = SKETCH_TYPE;
+ ptr += copy_to_mem(&type, ptr, sizeof(type));
+ const uint16_t unused16 = 0;
+ ptr += copy_to_mem(&unused16, ptr, sizeof(unused16));
+ const uint8_t flags_byte(
+ (1 << theta_sketch_alloc<A>::flags::IS_COMPACT) |
+ (1 << theta_sketch_alloc<A>::flags::IS_READ_ONLY) |
+ (this->is_empty() ? 1 << theta_sketch_alloc<A>::flags::IS_EMPTY : 0) |
+ (this->is_ordered() ? 1 << theta_sketch_alloc<A>::flags::IS_ORDERED : 0)
+ );
+ ptr += copy_to_mem(&flags_byte, ptr, sizeof(flags_byte));
+ const uint16_t seed_hash = get_seed_hash();
+ ptr += copy_to_mem(&seed_hash, ptr, sizeof(seed_hash));
+ if (!this->is_empty()) {
+ if (!is_single_item) {
+ const uint32_t num_keys = keys_.size();
+ ptr += copy_to_mem(&num_keys, ptr, sizeof(num_keys));
+ const uint32_t unused32 = 0;
+ ptr += copy_to_mem(&unused32, ptr, sizeof(unused32));
+ if (this->is_estimation_mode()) {
+ ptr += copy_to_mem(&(this->theta_), ptr, sizeof(uint64_t));
+ }
+ }
+ ptr += copy_to_mem(keys_.data(), ptr, sizeof(uint64_t) * keys_.size());
+ }
+
+ return bytes;
+}
+
+template<typename A>
+compact_theta_sketch_alloc<A> compact_theta_sketch_alloc<A>::deserialize(std::istream& is, uint64_t seed) {
+ uint8_t preamble_longs;
+ is.read((char*)&preamble_longs, sizeof(preamble_longs));
+ uint8_t serial_version;
+ is.read((char*)&serial_version, sizeof(serial_version));
+ uint8_t type;
+ is.read((char*)&type, sizeof(type));
+ uint16_t unused16;
+ is.read((char*)&unused16, sizeof(unused16));
+ uint8_t flags_byte;
+ is.read((char*)&flags_byte, sizeof(flags_byte));
+ uint16_t seed_hash;
+ is.read((char*)&seed_hash, sizeof(seed_hash));
+ theta_sketch_alloc<A>::check_sketch_type(type, SKETCH_TYPE);
+ theta_sketch_alloc<A>::check_serial_version(serial_version, theta_sketch_alloc<A>::SERIAL_VERSION);
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ if (!is_empty) theta_sketch_alloc<A>::check_seed_hash(seed_hash, theta_sketch_alloc<A>::get_seed_hash(seed));
+ return internal_deserialize(is, preamble_longs, flags_byte, seed_hash);
+}
+
+template<typename A>
+compact_theta_sketch_alloc<A> compact_theta_sketch_alloc<A>::internal_deserialize(std::istream& is, uint8_t preamble_longs, uint8_t flags_byte, uint16_t seed_hash) {
+ uint64_t theta = theta_sketch_alloc<A>::MAX_THETA;
+ uint32_t num_keys = 0;
+
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ if (!is_empty) {
+ if (preamble_longs == 1) {
+ num_keys = 1;
+ } else {
+ is.read((char*)&num_keys, sizeof(num_keys));
+ uint32_t unused32;
+ is.read((char*)&unused32, sizeof(unused32));
+ if (preamble_longs > 2) {
+ is.read((char*)&theta, sizeof(theta));
+ }
+ }
+ }
+ vector_u64<A> keys(num_keys);
+ if (!is_empty) is.read((char*)keys.data(), sizeof(uint64_t) * keys.size());
+
+ const bool is_ordered = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_ORDERED);
+ if (!is.good()) throw std::runtime_error("error reading from std::istream");
+ return compact_theta_sketch_alloc<A>(is_empty, theta, std::move(keys), seed_hash, is_ordered);
+}
+
+template<typename A>
+compact_theta_sketch_alloc<A> compact_theta_sketch_alloc<A>::deserialize(const void* bytes, size_t size, uint64_t seed) {
+ ensure_minimum_memory(size, 8);
+ const char* ptr = static_cast<const char*>(bytes);
+ uint8_t preamble_longs;
+ ptr += copy_from_mem(ptr, &preamble_longs, sizeof(preamble_longs));
+ uint8_t serial_version;
+ ptr += copy_from_mem(ptr, &serial_version, sizeof(serial_version));
+ uint8_t type;
+ ptr += copy_from_mem(ptr, &type, sizeof(type));
+ uint16_t unused16;
+ ptr += copy_from_mem(ptr, &unused16, sizeof(unused16));
+ uint8_t flags_byte;
+ ptr += copy_from_mem(ptr, &flags_byte, sizeof(flags_byte));
+ uint16_t seed_hash;
+ ptr += copy_from_mem(ptr, &seed_hash, sizeof(seed_hash));
+ theta_sketch_alloc<A>::check_sketch_type(type, SKETCH_TYPE);
+ theta_sketch_alloc<A>::check_serial_version(serial_version, theta_sketch_alloc<A>::SERIAL_VERSION);
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ if (!is_empty) theta_sketch_alloc<A>::check_seed_hash(seed_hash, theta_sketch_alloc<A>::get_seed_hash(seed));
+ return internal_deserialize(ptr, size - (ptr - static_cast<const char*>(bytes)), preamble_longs, flags_byte, seed_hash);
+}
+
+template<typename A>
+compact_theta_sketch_alloc<A> compact_theta_sketch_alloc<A>::internal_deserialize(const void* bytes, size_t size, uint8_t preamble_longs, uint8_t flags_byte, uint16_t seed_hash) {
+ const char* ptr = static_cast<const char*>(bytes);
+ const char* base = ptr;
+
+ uint64_t theta = theta_sketch_alloc<A>::MAX_THETA;
+ uint32_t num_keys = 0;
+
+ const bool is_empty = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_EMPTY);
+ if (!is_empty) {
+ if (preamble_longs == 1) {
+ num_keys = 1;
+ } else {
+ ensure_minimum_memory(size, 8); // read the first prelong before this method
+ ptr += copy_from_mem(ptr, &num_keys, sizeof(num_keys));
+ uint32_t unused32;
+ ptr += copy_from_mem(ptr, &unused32, sizeof(unused32));
+ if (preamble_longs > 2) {
+ ensure_minimum_memory(size, (preamble_longs - 1) << 3);
+ ptr += copy_from_mem(ptr, &theta, sizeof(theta));
+ }
+ }
+ }
+ const size_t keys_size_bytes = sizeof(uint64_t) * num_keys;
+ check_memory_size(ptr - base + keys_size_bytes, size);
+ vector_u64<A> keys(num_keys);
+ if (!is_empty) ptr += copy_from_mem(ptr, keys.data(), keys_size_bytes);
+
+ const bool is_ordered = flags_byte & (1 << theta_sketch_alloc<A>::flags::IS_ORDERED);
+ return compact_theta_sketch_alloc<A>(is_empty, theta, std::move(keys), seed_hash, is_ordered);
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::const_iterator compact_theta_sketch_alloc<A>::begin() const {
+ return typename theta_sketch_alloc<A>::const_iterator(keys_.data(), keys_.size(), 0);
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::const_iterator compact_theta_sketch_alloc<A>::end() const {
+ return typename theta_sketch_alloc<A>::const_iterator(keys_.data(), keys_.size(), keys_.size());
+}
+
+// builder
+
+template<typename A>
+update_theta_sketch_alloc<A>::builder::builder():
+lg_k_(DEFAULT_LG_K), rf_(DEFAULT_RESIZE_FACTOR), p_(1), seed_(DEFAULT_SEED) {}
+
+template<typename A>
+typename update_theta_sketch_alloc<A>::builder& update_theta_sketch_alloc<A>::builder::set_lg_k(uint8_t lg_k) {
+ if (lg_k < MIN_LG_K) {
+ throw std::invalid_argument("lg_k must not be less than " + std::to_string(MIN_LG_K) + ": " + std::to_string(lg_k));
+ }
+ lg_k_ = lg_k;
+ return *this;
+}
+
+template<typename A>
+typename update_theta_sketch_alloc<A>::builder& update_theta_sketch_alloc<A>::builder::set_resize_factor(resize_factor rf) {
+ rf_ = rf;
+ return *this;
+}
+
+template<typename A>
+typename update_theta_sketch_alloc<A>::builder& update_theta_sketch_alloc<A>::builder::set_p(float p) {
+ p_ = p;
+ return *this;
+}
+
+template<typename A>
+typename update_theta_sketch_alloc<A>::builder& update_theta_sketch_alloc<A>::builder::set_seed(uint64_t seed) {
+ seed_ = seed;
+ return *this;
+}
+
+template<typename A>
+uint8_t update_theta_sketch_alloc<A>::builder::starting_sub_multiple(uint8_t lg_tgt, uint8_t lg_min, uint8_t lg_rf) {
+ return (lg_tgt <= lg_min) ? lg_min : (lg_rf == 0) ? lg_tgt : ((lg_tgt - lg_min) % lg_rf) + lg_min;
+}
+
+template<typename A>
+update_theta_sketch_alloc<A> update_theta_sketch_alloc<A>::builder::build() const {
+ return update_theta_sketch_alloc<A>(starting_sub_multiple(lg_k_ + 1, MIN_LG_K, static_cast<uint8_t>(rf_)), lg_k_, rf_, p_, seed_);
+}
+
+// iterator
+
+template<typename A>
+theta_sketch_alloc<A>::const_iterator::const_iterator(const uint64_t* keys, uint32_t size, uint32_t index):
+keys_(keys), size_(size), index_(index) {
+ while (index_ < size_ && keys_[index_] == 0) ++index_;
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::const_iterator& theta_sketch_alloc<A>::const_iterator::operator++() {
+ do {
+ ++index_;
+ } while (index_ < size_ && keys_[index_] == 0);
+ return *this;
+}
+
+template<typename A>
+typename theta_sketch_alloc<A>::const_iterator theta_sketch_alloc<A>::const_iterator::operator++(int) {
+ const_iterator tmp(*this);
+ operator++();
+ return tmp;
+}
+
+template<typename A>
+bool theta_sketch_alloc<A>::const_iterator::operator==(const const_iterator& other) const {
+ return index_ == other.index_;
+}
+
+template<typename A>
+bool theta_sketch_alloc<A>::const_iterator::operator!=(const const_iterator& other) const {
+ return index_ != other.index_;
+}
+
+template<typename A>
+uint64_t theta_sketch_alloc<A>::const_iterator::operator*() const {
+ return keys_[index_];
+}
+
+} /* namespace datasketches */
+
+#endif
diff --git a/be/src/thirdparty/datasketches/theta_union.hpp b/be/src/thirdparty/datasketches/theta_union.hpp
new file mode 100644
index 0000000..6cf8ccc
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_union.hpp
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_UNION_HPP_
+#define THETA_UNION_HPP_
+
+#include <memory>
+#include <functional>
+#include <climits>
+
+#include "theta_sketch.hpp"
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+template<typename A>
+class theta_union_alloc {
+public:
+ class builder;
+
+ // No constructor here. Use builder instead.
+
+ /**
+ * This method is to update the union with a given sketch
+ * @param sketch to update the union with
+ */
+ void update(const theta_sketch_alloc<A>& sketch);
+
+ /**
+ * This method produces a copy of the current state of the union as a compact sketch.
+ * @param ordered optional flag to specify if ordered sketch should be produced
+ * @return the result of the union
+ */
+ compact_theta_sketch_alloc<A> get_result(bool ordered = true) const;
+
+private:
+ bool is_empty_;
+ uint64_t theta_;
+ update_theta_sketch_alloc<A> state_;
+
+ // for builder
+ theta_union_alloc(uint64_t theta, update_theta_sketch_alloc<A>&& state);
+};
+
+// builder
+
+template<typename A>
+class theta_union_alloc<A>::builder {
+public:
+ typedef typename update_theta_sketch_alloc<A>::resize_factor resize_factor;
+
+ /**
+ * Set log2(k), where k is a nominal number of entries in the sketch
+ * @param lg_k base 2 logarithm of nominal number of entries
+ * @return this builder
+ */
+ builder& set_lg_k(uint8_t lg_k);
+
+ /**
+ * Set resize factor for the internal hash table (defaults to 8)
+ * @param rf resize factor
+ * @return this builder
+ */
+ builder& set_resize_factor(resize_factor rf);
+
+ /**
+ * Set sampling probability (initial theta). The default is 1, so the sketch retains
+ * all entries until it reaches the limit, at which point it goes into the estimation mode
+ * and reduces the effective sampling probability (theta) as necessary.
+ * @param p sampling probability
+ * @return this builder
+ */
+ builder& set_p(float p);
+
+ /**
+ * Set the seed for the hash function. Should be used carefully if needed.
+ * Sketches produced with different seed are not compatible
+ * and cannot be mixed in set operations.
+ * @param seed hash seed
+ * @return this builder
+ */
+ builder& set_seed(uint64_t seed);
+
+ /**
+ * This is to create an instance of the union with predefined parameters.
+ * @return and instance of the union
+ */
+ theta_union_alloc<A> build() const;
+
+private:
+ typename update_theta_sketch_alloc<A>::builder sketch_builder;
+};
+
+// alias with default allocator for convenience
+typedef theta_union_alloc<std::allocator<void>> theta_union;
+
+} /* namespace datasketches */
+
+#include "theta_union_impl.hpp"
+
+# endif
diff --git a/be/src/thirdparty/datasketches/theta_union_impl.hpp b/be/src/thirdparty/datasketches/theta_union_impl.hpp
new file mode 100644
index 0000000..4d8ebaa
--- /dev/null
+++ b/be/src/thirdparty/datasketches/theta_union_impl.hpp
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * 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.
+ */
+
+#ifndef THETA_UNION_IMPL_HPP_
+#define THETA_UNION_IMPL_HPP_
+
+namespace datasketches {
+
+/*
+ * author Alexander Saydakov
+ * author Lee Rhodes
+ * author Kevin Lang
+ */
+
+template<typename A>
+theta_union_alloc<A>::theta_union_alloc(uint64_t theta, update_theta_sketch_alloc<A>&& state):
+is_empty_(true), theta_(theta), state_(std::move(state)) {}
+
+template<typename A>
+void theta_union_alloc<A>::update(const theta_sketch_alloc<A>& sketch) {
+ if (sketch.is_empty()) return;
+ if (sketch.get_seed_hash() != state_.get_seed_hash()) throw std::invalid_argument("seed hash mismatch");
+ is_empty_ = false;
+ if (sketch.get_theta64() < theta_) theta_ = sketch.get_theta64();
+ if (sketch.is_ordered()) {
+ for (auto hash: sketch) {
+ if (hash >= theta_) break; // early stop
+ state_.internal_update(hash);
+ }
+ } else {
+ for (auto hash: sketch) if (hash < theta_) state_.internal_update(hash);
+ }
+ if (state_.get_theta64() < theta_) theta_ = state_.get_theta64();
+}
+
+template<typename A>
+compact_theta_sketch_alloc<A> theta_union_alloc<A>::get_result(bool ordered) const {
+ if (is_empty_) return state_.compact(ordered);
+ const uint32_t nom_num_keys = 1 << state_.lg_nom_size_;
+ if (theta_ >= state_.theta_ && state_.get_num_retained() <= nom_num_keys) return state_.compact(ordered);
+ uint64_t theta = std::min(theta_, state_.get_theta64());
+ vector_u64<A> keys(state_.get_num_retained());
+ uint32_t num_keys = 0;
+ for (auto key: state_) {
+ if (key < theta) keys[num_keys++] = key;
+ }
+ if (num_keys > nom_num_keys) {
+ std::nth_element(keys.begin(), keys.begin() + nom_num_keys, keys.begin() + num_keys);
+ theta = keys[nom_num_keys];
+ num_keys = nom_num_keys;
+ }
+ if (num_keys != state_.get_num_retained()) {
+ keys.resize(num_keys);
+ }
+ if (ordered) std::sort(keys.begin(), keys.end());
+ return compact_theta_sketch_alloc<A>(false, theta, std::move(keys), state_.get_seed_hash(), ordered);
+}
+
+// builder
+
+template<typename A>
+typename theta_union_alloc<A>::builder& theta_union_alloc<A>::builder::set_lg_k(uint8_t lg_k) {
+ sketch_builder.set_lg_k(lg_k);
+ return *this;
+}
+
+template<typename A>
+typename theta_union_alloc<A>::builder& theta_union_alloc<A>::builder::set_resize_factor(resize_factor rf) {
+ sketch_builder.set_resize_factor(rf);
+ return *this;
+}
+
+template<typename A>
+typename theta_union_alloc<A>::builder& theta_union_alloc<A>::builder::set_p(float p) {
+ sketch_builder.set_p(p);
+ return *this;
+}
+
+template<typename A>
+typename theta_union_alloc<A>::builder& theta_union_alloc<A>::builder::set_seed(uint64_t seed) {
+ sketch_builder.set_seed(seed);
+ return *this;
+}
+
+template<typename A>
+theta_union_alloc<A> theta_union_alloc<A>::builder::build() const {
+ update_theta_sketch_alloc<A> sketch = sketch_builder.build();
+ return theta_union_alloc(sketch.get_theta64(), std::move(sketch));
+}
+
+} /* namespace datasketches */
+
+# endif
[impala] 02/04: IMPALA-10435: Extend 'compute incremental stats'
syntax to support a list of columns
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 18acca92ee802c138db4eaaea8a6bc53f8ec0faf
Author: liuyao <54...@163.com>
AuthorDate: Wed Jan 13 10:37:38 2021 +0800
IMPALA-10435: Extend 'compute incremental stats' syntax
to support a list of columns
Modified parser to support compute incremental stats
columns.No need to modify the code of other modules
because it already supports
Change-Id: I4dcc2d4458679c39581446f6d87bb7903803f09b
Reviewed-on: http://gerrit.cloudera.org:8080/16947
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
---
fe/src/main/cup/sql-parser.cup | 10 +-
.../apache/impala/analysis/ComputeStatsStmt.java | 5 +-
.../org/apache/impala/analysis/AnalyzeDDLTest.java | 9 +
.../org/apache/impala/analysis/ParserTest.java | 12 +
.../QueryTest/compute-stats-incremental.test | 309 +++++++++++++++++++++
5 files changed, 340 insertions(+), 5 deletions(-)
diff --git a/fe/src/main/cup/sql-parser.cup b/fe/src/main/cup/sql-parser.cup
index 7967ac4..6fabdd7 100644
--- a/fe/src/main/cup/sql-parser.cup
+++ b/fe/src/main/cup/sql-parser.cup
@@ -2261,9 +2261,15 @@ compute_stats_stmt ::=
opt_tablesample:tblsmpl
{: RESULT = ComputeStatsStmt.createStatsStmt(table, tblsmpl, cols); :}
| KW_COMPUTE KW_INCREMENTAL KW_STATS table_name:table
- {: RESULT = ComputeStatsStmt.createIncrementalStatsStmt(table, null); :}
+ {: RESULT = ComputeStatsStmt.createIncrementalStatsStmt(table, null, null); :}
+ | KW_COMPUTE KW_INCREMENTAL KW_STATS table_name:table
+ LPAREN opt_ident_list:cols RPAREN
+ {: RESULT = ComputeStatsStmt.createIncrementalStatsStmt(table, null, cols); :}
+ | KW_COMPUTE KW_INCREMENTAL KW_STATS table_name:table partition_set:parts
+ {: RESULT = ComputeStatsStmt.createIncrementalStatsStmt(table, parts, null); :}
| KW_COMPUTE KW_INCREMENTAL KW_STATS table_name:table partition_set:parts
- {: RESULT = ComputeStatsStmt.createIncrementalStatsStmt(table, parts); :}
+ LPAREN opt_ident_list:cols RPAREN
+ {: RESULT = ComputeStatsStmt.createIncrementalStatsStmt(table, parts, cols); :}
;
drop_stats_stmt ::=
diff --git a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
index 906a972..db3dd2d 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ComputeStatsStmt.java
@@ -202,7 +202,6 @@ public class ComputeStatsStmt extends StatementBase {
Preconditions.checkState(tableName != null && !tableName.isEmpty());
Preconditions.checkState(isIncremental || partitionSet == null);
Preconditions.checkState(!isIncremental || sampleParams == null);
- Preconditions.checkState(!isIncremental || columns == null);
tableName_ = tableName;
sampleParams_ = sampleParams;
table_ = null;
@@ -234,8 +233,8 @@ public class ComputeStatsStmt extends StatementBase {
* set of partitions whose stats should be computed.
*/
public static ComputeStatsStmt createIncrementalStatsStmt(TableName tableName,
- PartitionSet partitionSet) {
- return new ComputeStatsStmt(tableName, null, true, partitionSet, null);
+ PartitionSet partitionSet, List<String> columns) {
+ return new ComputeStatsStmt(tableName, null, true, partitionSet, columns);
}
private List<String> getBaseColumnStatsQuerySelectList(Analyzer analyzer) {
diff --git a/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java b/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
index c0b33cc..1bb2fc8 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
@@ -1757,6 +1757,15 @@ public class AnalyzeDDLTest extends FrontendTestBase {
AnalysisError("compute incremental stats functional.view_view",
"COMPUTE STATS not supported for view: functional.view_view");
+
+ checkComputeStatsStmt("compute incremental stats functional.alltypes " +
+ "(tinyint_col, smallint_col)");
+ checkComputeStatsStmt(
+ "compute incremental stats functional.alltypes partition(year=2010, month=10)" +
+ "(tinyint_col, smallint_col)");
+ checkComputeStatsStmt(
+ "compute incremental stats functional.alltypes partition(year<=2010)" +
+ "(tinyint_col, smallint_col)");
}
diff --git a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
index b244337..88b34be 100644
--- a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
@@ -4065,10 +4065,22 @@ public class ParserTest extends FrontendTestBase {
"COMPUTE INCREMENTAL STATS functional.alltypes PARTITION(month=10, year=2010)");
ParsesOk(
"DROP INCREMENTAL STATS functional.alltypes PARTITION(month=10, year=2010)");
+ ParsesOk("COMPUTE INCREMENTAL STATS functional.alltypes(tinyint_col, smallint_col)");
+ ParsesOk(
+ "COMPUTE INCREMENTAL STATS functional.alltypes PARTITION(month=10, year=2010)"
+ + "(tinyint_col, smallint_col)");
+
ParserError("COMPUTE INCREMENTAL STATS");
ParserError("COMPUTE INCREMENTAL functional.alltypes");
ParserError("DROP INCREMENTAL STATS functional.alltypes");
ParserError("COMPUTE INCREMENTAL STATS functional.alltypes TABLESAMPLE SYSTEM(10)");
+ //Missing closing parenthesis
+ ParserError(
+ "COMPUTE INCREMENTAL STATS functional.alltypes(tinyint_col, smallint_col");
+ //Missing opening parenthesis
+ ParserError(
+ "COMPUTE INCREMENTAL STATS functional.alltypes PARTITION(month=10, year=2010)"
+ + "tinyint_col, smallint_col)");
}
@Test
diff --git a/testdata/workloads/functional-query/queries/QueryTest/compute-stats-incremental.test b/testdata/workloads/functional-query/queries/QueryTest/compute-stats-incremental.test
index 1f687d5..3edfefe 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/compute-stats-incremental.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/compute-stats-incremental.test
@@ -622,3 +622,312 @@ compute incremental stats complextypestbl_part;
---- TYPES
STRING
====
+---- QUERY
+# test computing incremental stats on a partitioned text table with all types
+create table alltypes_incremental_cols like functional.alltypes;
+insert into alltypes_incremental_cols partition(year, month)
+select * from functional.alltypes;
+====
+---- QUERY
+compute incremental stats alltypes_incremental_cols(int_col, float_col, string_col)
+---- RESULTS
+'Updated 24 partition(s) and 3 column(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+show table stats alltypes_incremental_cols
+---- RESULTS
+'2009','1',310,1,'24.56KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','2',280,1,'22.27KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','3',310,1,'24.67KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','4',300,1,'24.06KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','1',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','2',280,1,'22.54KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','3',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','4',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'Total','',7300,24,'586.84KB','0B','','','',''
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
+====
+---- QUERY
+show column stats alltypes_incremental_cols
+---- LABELS
+COLUMN, TYPE, #DISTINCT VALUES, #NULLS, MAX SIZE, AVG SIZE, #TRUES, #FALSES
+---- RESULTS
+'id','INT',-1,-1,4,4,-1,-1
+'bool_col','BOOLEAN',-1,-1,1,1,-1,-1
+'tinyint_col','TINYINT',-1,-1,1,1,-1,-1
+'smallint_col','SMALLINT',-1,-1,2,2,-1,-1
+'int_col','INT',10,0,4,4,-1,-1
+'bigint_col','BIGINT',-1,-1,8,8,-1,-1
+'float_col','FLOAT',10,0,4,4,-1,-1
+'double_col','DOUBLE',-1,-1,8,8,-1,-1
+'date_string_col','STRING',-1,-1,-1,-1,-1,-1
+'string_col','STRING',10,0,1,1,-1,-1
+'timestamp_col','TIMESTAMP',-1,-1,16,16,-1,-1
+'year','INT',2,0,4,4,-1,-1
+'month','INT',12,0,4,4,-1,-1
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, BIGINT, DOUBLE, BIGINT, BIGINT
+====
+---- QUERY
+drop incremental stats alltypes_incremental_cols partition(year=2010, month=12)
+====
+---- QUERY
+show table stats alltypes_incremental_cols;
+---- RESULTS
+'2009','1',310,1,'24.56KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','2',280,1,'22.27KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','3',310,1,'24.67KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','4',300,1,'24.06KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','1',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','2',280,1,'22.54KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','3',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','4',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','false',regex:.*
+'Total','',7300,24,'586.84KB','0B','','','',''
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
+====
+---- QUERY
+compute incremental stats alltypes_incremental_cols partition(year=2010, month=12)
+(int_col, float_col, string_col);
+---- RESULTS
+'Updated 1 partition(s) and 3 column(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+show table stats alltypes_incremental_cols;
+---- RESULTS
+'2009','1',310,1,'24.56KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','2',280,1,'22.27KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','3',310,1,'24.67KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','4',300,1,'24.06KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','1',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','2',280,1,'22.54KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','3',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','4',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'Total','',7300,24,'586.84KB','0B','','','',''
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
+====
+---- QUERY
+show column stats alltypes_incremental_cols
+---- LABELS
+COLUMN, TYPE, #DISTINCT VALUES, #NULLS, MAX SIZE, AVG SIZE, #TRUES, #FALSES
+---- RESULTS
+'id','INT',-1,-1,4,4,-1,-1
+'bool_col','BOOLEAN',-1,-1,1,1,-1,-1
+'tinyint_col','TINYINT',-1,-1,1,1,-1,-1
+'smallint_col','SMALLINT',-1,-1,2,2,-1,-1
+'int_col','INT',10,0,4,4,-1,-1
+'bigint_col','BIGINT',-1,-1,8,8,-1,-1
+'float_col','FLOAT',10,0,4,4,-1,-1
+'double_col','DOUBLE',-1,-1,8,8,-1,-1
+'date_string_col','STRING',-1,-1,-1,-1,-1,-1
+'string_col','STRING',10,0,1,1,-1,-1
+'timestamp_col','TIMESTAMP',-1,-1,16,16,-1,-1
+'year','INT',2,0,4,4,-1,-1
+'month','INT',12,0,4,4,-1,-1
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, BIGINT, DOUBLE, BIGINT, BIGINT
+====
+---- QUERY
+drop stats alltypes_incremental_cols
+====
+---- QUERY
+compute incremental stats alltypes_incremental_cols partition(year=2010, month=11)
+(float_col, id);
+compute incremental stats alltypes_incremental_cols partition(year=2010, month=12)
+(float_col, string_col);
+show column stats alltypes_incremental_cols;
+---- RESULTS
+'id','INT',300,0,4,4,-1,-1
+'bool_col','BOOLEAN',-1,-1,1,1,-1,-1
+'tinyint_col','TINYINT',-1,-1,1,1,-1,-1
+'smallint_col','SMALLINT',-1,-1,2,2,-1,-1
+'int_col','INT',-1,-1,4,4,-1,-1
+'bigint_col','BIGINT',-1,-1,8,8,-1,-1
+'float_col','FLOAT',10,0,4,4,-1,-1
+'double_col','DOUBLE',-1,-1,8,8,-1,-1
+'date_string_col','STRING',-1,-1,-1,-1,-1,-1
+'string_col','STRING',10,0,1,1,-1,-1
+'timestamp_col','TIMESTAMP',-1,-1,16,16,-1,-1
+'year','INT',2,0,4,4,-1,-1
+'month','INT',12,0,4,4,-1,-1
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, BIGINT, DOUBLE, BIGINT, BIGINT
+====
+---- QUERY
+compute incremental stats alltypes_incremental_cols partition(year=2010, month=12)
+(id);
+show column stats alltypes_incremental_cols;
+---- RESULTS
+'id','INT',610,0,4,4,-1,-1
+'bool_col','BOOLEAN',-1,-1,1,1,-1,-1
+'tinyint_col','TINYINT',-1,-1,1,1,-1,-1
+'smallint_col','SMALLINT',-1,-1,2,2,-1,-1
+'int_col','INT',-1,-1,4,4,-1,-1
+'bigint_col','BIGINT',-1,-1,8,8,-1,-1
+'float_col','FLOAT',10,0,4,4,-1,-1
+'double_col','DOUBLE',-1,-1,8,8,-1,-1
+'date_string_col','STRING',-1,-1,-1,-1,-1,-1
+'string_col','STRING',10,0,1,1,-1,-1
+'timestamp_col','TIMESTAMP',-1,-1,16,16,-1,-1
+'year','INT',2,0,4,4,-1,-1
+'month','INT',12,0,4,4,-1,-1
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, BIGINT, DOUBLE, BIGINT, BIGINT
+====
+---- QUERY
+drop stats alltypes_incremental_cols;
+compute incremental stats alltypes_incremental_cols (float_col, id);
+show column stats alltypes_incremental_cols;
+---- RESULTS
+'id','INT',7300,0,4,4,-1,-1
+'bool_col','BOOLEAN',-1,-1,1,1,-1,-1
+'tinyint_col','TINYINT',-1,-1,1,1,-1,-1
+'smallint_col','SMALLINT',-1,-1,2,2,-1,-1
+'int_col','INT',-1,-1,4,4,-1,-1
+'bigint_col','BIGINT',-1,-1,8,8,-1,-1
+'float_col','FLOAT',10,0,4,4,-1,-1
+'double_col','DOUBLE',-1,-1,8,8,-1,-1
+'date_string_col','STRING',-1,-1,-1,-1,-1,-1
+'string_col','STRING',-1,-1,-1,-1,-1,-1
+'timestamp_col','TIMESTAMP',-1,-1,16,16,-1,-1
+'year','INT',2,0,4,4,-1,-1
+'month','INT',12,0,4,4,-1,-1
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, BIGINT, DOUBLE, BIGINT, BIGINT
+====
+---- QUERY
+show table stats alltypes_incremental_cols;
+---- RESULTS
+'2009','1',310,1,'24.56KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','2',280,1,'22.27KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','3',310,1,'24.67KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','4',300,1,'24.06KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','1',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','2',280,1,'22.54KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','3',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','4',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'Total','',7300,24,'586.84KB','0B','','','',''
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
+====
+---- QUERY
+compute incremental stats alltypes_incremental_cols (float_col, string_col);
+show column stats alltypes_incremental_cols;
+---- RESULTS
+'id','INT',7300,0,4,4,-1,-1
+'bool_col','BOOLEAN',-1,-1,1,1,-1,-1
+'tinyint_col','TINYINT',-1,-1,1,1,-1,-1
+'smallint_col','SMALLINT',-1,-1,2,2,-1,-1
+'int_col','INT',-1,-1,4,4,-1,-1
+'bigint_col','BIGINT',-1,-1,8,8,-1,-1
+'float_col','FLOAT',10,0,4,4,-1,-1
+'double_col','DOUBLE',-1,-1,8,8,-1,-1
+'date_string_col','STRING',-1,-1,-1,-1,-1,-1
+'string_col','STRING',10,0,1,1,-1,-1
+'timestamp_col','TIMESTAMP',-1,-1,16,16,-1,-1
+'year','INT',2,0,4,4,-1,-1
+'month','INT',12,0,4,4,-1,-1
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, BIGINT, DOUBLE, BIGINT, BIGINT
+====
+---- QUERY
+show table stats alltypes_incremental_cols;
+---- RESULTS
+'2009','1',310,1,'24.56KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','2',280,1,'22.27KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','3',310,1,'24.67KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','4',300,1,'24.06KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2009','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','1',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','2',280,1,'22.54KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','3',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','4',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','5',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','6',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','7',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','8',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','9',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','10',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','11',300,1,'24.16KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'2010','12',310,1,'24.97KB','NOT CACHED','NOT CACHED','TEXT','true',regex:.*
+'Total','',7300,24,'586.84KB','0B','','','',''
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
+====
[impala] 04/04: IMPALA-10448: Build impala-profile-tool early for
Docker-based tests
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 6d4756da0123b8b4e0b49e1d7e37651dc02db7fd
Author: Laszlo Gaal <la...@cloudera.com>
AuthorDate: Wed Jan 20 14:43:45 2021 +0100
IMPALA-10448: Build impala-profile-tool early for Docker-based tests
impala-profile-tool is a new dependency for end-to-end tests.
The tool is built together with all the other backend tests
(so the buildall.sh flag '-notests' can turn off building it), it is
actually used in the parallel phase of end-to-end tests.
This means a problem for Docker-based builds for the following reasons:
- Docker-based tests run BE, FE and various phases of the EE test in
separate Docker containers for parallel executions
- Test binaries are only built inside the container running BE tests to
cut down on the build time and the size of the Docker image that all test
containers are based on.
- This means that the EE_TEST_PARALLEL container will miss the tool
required for running test designed to test it.
The solution is to build the tool early, at the end of the build phase
running in the build container. There is already another such tool built
there (parquet-reader) for similar reason, so just add
impala-profile-tool to the same 'make' command there.
Tested by running BE_TEST and EE_TEST_PARALLEL phases in a Docker-based
build.
Change-Id: I60e78ea883f3057c59a345feca38ef08a7f6a0b8
Reviewed-on: http://gerrit.cloudera.org:8080/16965
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
docker/entrypoint.sh | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh
index bb33e4f..81750e6 100755
--- a/docker/entrypoint.sh
+++ b/docker/entrypoint.sh
@@ -269,7 +269,7 @@ function build_impdev() {
# shards, depends on this binary. We build it here once,
# instead of building it during the startup of each container running
# a subset of E2E tests. Building it here is also a lot faster.
- make -j$(nproc) --load-average=$(nproc) parquet-reader
+ make -j$(nproc) --load-average=$(nproc) parquet-reader impala-profile-tool
# Dump current memory usage to logs, before shutting things down.
memory_usage
[impala] 03/04: IMPALA-10296: Fix analytic limit pushdown when
predicates are present
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 1ada739e81416a6f49f2e8bb287a007252371444
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Tue Jan 12 11:21:12 2021 -0800
IMPALA-10296: Fix analytic limit pushdown when predicates are present
This fixes the analytic push down optimization for the case where
the ORDER BY expressions are compatible with the partitioning of the
analytic *and* there is a rank() or row_number() predicate.
In this case the rows returned are going to come from the first partitions,
i.e. if the limit is 100, if we go through the partitions in order until
the row count adds up to 100, then we know that the rows must come from
those partitions.
The problem is that predicates can discard rows from the partitions,
meaning that a limit naively pushed down to the top-n will filter
out rows that could be returned from the query.
We can avoid the problem in the case where the partition limit >=
order by limit, however.
In this case the relevant set of partitions is the set of partitions
that include the first <limit> rows, since the top-level limit
generally kicks in before the per-partition limit. The only twist
is that the orderings may be different within a partition, so we
need to make sure to include all of the rows in the final partition.
The solution implemented in this patch is to increase the pushed
down limit so that it is always guaranteed to include all of the
rows in the final partition to be returned. E.g. if you had a
row_number() <= 100 predicate and limit 100, if you pushed down
limit 200, then you'd be guaranteed to capture all of the rows
in the final partition. One case we need to handle is that,
in the case of a rank() predicate, we can have more than that
number of rows in the partition because of ties.
This patch implements tie handling in the backend (I took most
of that implementation from my in-progress partitioned top-n patch,
with the intention of rebasing that onto this patch).
This also adds a check against TOPN_BYTES_LIMIT so that
the limit can't be increased to an arbitarily large value.
Testing:
* Add new planner test with negative case where it's rejected
because the transformation is incorrect.
* Update other planner tests to reflect new limit calculation
+ tie handling required for correctness.
* Add planner test for very high rank predicate that overflows int32
* Add planner test that checks TOPN_BYTES_LIMIT handling
* Add planner test that checks that dense_rank() can't be pushed.
* Existing planner tests already have adequate coverage for predicates
: <=, <, = and row_number().
* Add some end-to-end tests that repro bugs that fall under the jira
* Add an end-to-end test on TPC-H with more data to exercise the
tie-handling logic in the execnode more.
Perf:
Ran TPC-DS q67 with mt_dop=1 on a single node, confirmed there was
no measurable change in performance as a result of this patchset.
Ran TPC-H scale 30 on a single node, no significant perf change.
Ran a targeted query to check for regressions in the top-n node.
The elapsed time for this targeted query did not change:
use tpch30_parquet;
set mt_dop=1;
select l_extendedprice from lineitem
order by 1 limit 100
Change-Id: I801d7799b0d649c73d2dd1703729a9b58a662509
Reviewed-on: http://gerrit.cloudera.org:8080/16942
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/topn-node-ir.cc | 94 +++-
be/src/exec/topn-node.cc | 143 +++---
be/src/exec/topn-node.h | 141 ++++--
be/src/util/tuple-row-compare.h | 6 +
common/thrift/PlanNodes.thrift | 6 +
.../org/apache/impala/analysis/AnalyticExpr.java | 10 +-
.../apache/impala/planner/AnalyticEvalNode.java | 180 +++++---
.../apache/impala/planner/DistributedPlanner.java | 69 +--
.../apache/impala/planner/SingleNodePlanner.java | 48 +-
.../java/org/apache/impala/planner/SortNode.java | 101 ++++-
.../PlannerTest/limit-pushdown-analytic.test | 493 +++++++++++++++++----
.../queries/PlannerTest/resource-requirements.test | 33 +-
.../queries/PlannerTest/tpcds/tpcds-q67.test | 43 +-
.../queries/limit-pushdown-analytic.test | 96 ++++
.../tpch/queries/limit-pushdown-analytic.test | 149 +++++++
tests/query_test/test_limit_pushdown_analytic.py | 24 +-
16 files changed, 1278 insertions(+), 358 deletions(-)
diff --git a/be/src/exec/topn-node-ir.cc b/be/src/exec/topn-node-ir.cc
index 74ec297..d64b2f0 100644
--- a/be/src/exec/topn-node-ir.cc
+++ b/be/src/exec/topn-node-ir.cc
@@ -25,39 +25,97 @@ void TopNNode::InsertBatch(RowBatch* batch) {
// are hoisted out of the loop as possible (either via code changes or __restrict__)
// annotations.
FOREACH_ROW(batch, 0, iter) {
- bool replaced_existing_row = heap_->InsertTupleRow(this, iter.Get());
- if (replaced_existing_row) ++rows_to_reclaim_;
+ int num_to_reclaim = heap_->InsertTupleRow(this, iter.Get());
+ // Using a branch here instead of adding the value directly improves
+ // top-n performance by a couple of %.
+ if (num_to_reclaim != 0) rows_to_reclaim_ += num_to_reclaim;
}
}
-bool TopNNode::Heap::InsertTupleRow(TopNNode* node, TupleRow* input_row) {
+int TopNNode::Heap::InsertTupleRow(TopNNode* node, TupleRow* input_row) {
const TupleDescriptor& tuple_desc = *node->output_tuple_desc_;
- bool replaced_existing_row = false;
- Tuple* insert_tuple = nullptr;
if (priority_queue_.Size() < heap_capacity()) {
- // Add all tuples until we hit capacity.
- insert_tuple = reinterpret_cast<Tuple*>(
+ // Add all tuples (including any ties) until we hit capacity.
+ Tuple* insert_tuple = reinterpret_cast<Tuple*>(
node->tuple_pool_->Allocate(node->tuple_byte_size()));
insert_tuple->MaterializeExprs<false, false>(input_row, tuple_desc,
node->output_tuple_expr_evals_, node->tuple_pool_.get());
priority_queue_.Push(insert_tuple);
+ return 0;
+ }
+
+ // We're at capacity - compare to the first row in the priority queue to see if
+ // we need to insert this row into the queue.
+ DCHECK(!priority_queue_.Empty());
+ Tuple* top_tuple = priority_queue_.Top();
+ node->tmp_tuple_->MaterializeExprs<false, true>(input_row, tuple_desc,
+ node->output_tuple_expr_evals_, nullptr);
+ if (include_ties()) {
+ return InsertTupleWithTieHandling(node, node->tmp_tuple_);
} else {
- // We're at capacity - compare to the first row in the priority queue to see if
- // we need to insert this row into the queue.
- DCHECK(!priority_queue_.Empty());
- Tuple* top_tuple = priority_queue_.Top();
- node->tmp_tuple_->MaterializeExprs<false, true>(input_row, tuple_desc,
- node->output_tuple_expr_evals_, nullptr);
- // If the new tuple is less than the top, need to swap it with the top and re-heapify.
if (node->tuple_row_less_than_->Less(node->tmp_tuple_, top_tuple)) {
- // Deep copy the new tuple into 'top_tuple' to reuse the fixed-length memory
- // of 'top_tuple'. The top element in the heap points at 'top_tuple'.
+ // Pop off the old head, and replace with the new tuple. Deep copy into 'top_tuple'
+ // to reuse the fixed-length memory of 'top_tuple'.
node->tmp_tuple_->DeepCopy(top_tuple, tuple_desc, node->tuple_pool_.get());
// Re-heapify from the top element and down.
priority_queue_.HeapifyFromTop();
- replaced_existing_row = true;
+ return 1;
+ }
+ return 0;
+ }
+}
+
+int TopNNode::Heap::InsertTupleWithTieHandling(
+ TopNNode* node, Tuple* materialized_tuple) {
+ DCHECK(include_ties());
+ DCHECK_EQ(capacity_, priority_queue_.Size())
+ << "Ties only need special handling when heap is at capacity";
+ const TupleDescriptor& tuple_desc = *node->output_tuple_desc_;
+ Tuple* top_tuple = priority_queue_.Top();
+ // If we need to retain ties with the current head, the logic is more complex - we
+ // have a logical heap in indices [0, heap_capacity()) of 'priority_queue_' plus
+ // some number of tuples in 'overflowed_ties_' that are equal to priority_queue_.Top()
+ // according to 'intra_partition_tuple_row_less_than_'.
+ int cmp_result = node->tuple_row_less_than_->Compare(materialized_tuple, top_tuple);
+ if (cmp_result == 0) {
+ // This is a duplicate of the current min, we need to include it as a tie with min.
+ Tuple* insert_tuple =
+ reinterpret_cast<Tuple*>(node->tuple_pool_->Allocate(node->tuple_byte_size()));
+ materialized_tuple->DeepCopy(insert_tuple, tuple_desc, node->tuple_pool_.get());
+ overflowed_ties_.push_back(insert_tuple);
+ return 0;
+ } else if (cmp_result > 0) {
+ // Tuple does not belong in the heap.
+ return 1;
+ } else {
+ // 'materialized_tuple' needs to be added. Figure out which other tuples, if any,
+ // need to be removed from the heap.
+ DCHECK_LT(cmp_result, 0);
+ // Pop off the head.
+ priority_queue_.Pop();
+
+ // Check if 'top_tuple' (the tuple we just popped off) is tied with the new head.
+ if (heap_capacity() > 1 &&
+ node->tuple_row_less_than_->Compare(top_tuple, priority_queue_.Top()) == 0) {
+ // The new top is still tied with the tuples in 'overflowed_ties_' so we must keep
+ // it. The previous top becomes another overflowed tuple.
+ overflowed_ties_.push_back(top_tuple);
+ Tuple* insert_tuple =
+ reinterpret_cast<Tuple*>(node->tuple_pool_->Allocate(node->tuple_byte_size()));
+ materialized_tuple->DeepCopy(insert_tuple, tuple_desc, node->tuple_pool_.get());
+ // Push the new tuple onto the heap, but retain the tied tuple.
+ priority_queue_.Push(insert_tuple);
+ return 0;
+ } else {
+ // No tuples tied with 'top_tuple' are left.
+ // Reuse the fixed-length memory of 'top_tuple' to reduce allocations.
+ int64_t num_rows_replaced = overflowed_ties_.size() + 1;
+ overflowed_ties_.clear();
+ materialized_tuple->DeepCopy(top_tuple, tuple_desc, node->tuple_pool_.get());
+ priority_queue_.Push(top_tuple);
+ return num_rows_replaced;
}
}
- return replaced_existing_row;
}
+
diff --git a/be/src/exec/topn-node.cc b/be/src/exec/topn-node.cc
index fb69621..1ea16e8 100644
--- a/be/src/exec/topn-node.cc
+++ b/be/src/exec/topn-node.cc
@@ -111,11 +111,11 @@ void TopNPlanNode::Codegen(FragmentState* state) {
codegen->GetFunction(IRFunction::TOPN_NODE_INSERT_BATCH, true);
DCHECK(insert_batch_fn != NULL);
- // Generate two MaterializeExprs() functions, one using tuple_pool_ and
- // one with no pool.
+ // Generate two MaterializeExprs() functions, one with no pool that
+ // does a shallow copy and one with 'tuple_pool_' that does a deep copy of the data.
DCHECK(output_tuple_desc_ != NULL);
- llvm::Function* materialize_exprs_tuple_pool_fn;
- llvm::Function* materialize_exprs_no_pool_fn;
+ llvm::Function* materialize_exprs_tuple_pool_fn = nullptr;
+ llvm::Function* materialize_exprs_no_pool_fn = nullptr;
codegen_status = Tuple::CodegenMaterializeExprs(codegen, false,
*output_tuple_desc_, output_tuple_exprs_,
@@ -125,37 +125,45 @@ void TopNPlanNode::Codegen(FragmentState* state) {
codegen_status = Tuple::CodegenMaterializeExprs(codegen, false,
*output_tuple_desc_, output_tuple_exprs_,
false, &materialize_exprs_no_pool_fn);
+ }
- if (codegen_status.ok()) {
- int replaced = codegen->ReplaceCallSites(insert_batch_fn,
- materialize_exprs_tuple_pool_fn, Tuple::MATERIALIZE_EXPRS_SYMBOL);
- DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
-
- // The total number of calls to TupleRowComparator::Less(Tuple*, Tuple*)
- // is 3 in PriorityQueue and 1 in TopNNode::Heap::InsertTupleRow().
- // Each Less(Tuple*, Tuple*) indirectly calls TupleRowComparator::Compare()
- // once.
- replaced = codegen->ReplaceCallSites(insert_batch_fn,
- compare_fn, TupleRowComparator::COMPARE_SYMBOL);
- DCHECK_REPLACE_COUNT(replaced, 4) << LlvmCodeGen::Print(insert_batch_fn);
-
- replaced = codegen->ReplaceCallSites(insert_batch_fn,
- materialize_exprs_no_pool_fn, Tuple::MATERIALIZE_EXPRS_NULL_POOL_SYMBOL);
- DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
-
- replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
- codegen->GetI64Constant(tnode_->limit + offset()), "heap_capacity");
- DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
-
- int tuple_byte_size = output_tuple_desc_->byte_size();
- replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
- codegen->GetI32Constant(tuple_byte_size), "tuple_byte_size");
- DCHECK_REPLACE_COUNT(replaced, 1);
-
- insert_batch_fn = codegen->FinalizeFunction(insert_batch_fn);
- DCHECK(insert_batch_fn != NULL);
- codegen->AddFunctionToJit(insert_batch_fn, &codegend_insert_batch_fn_);
- }
+ if (codegen_status.ok()) {
+ int replaced;
+ replaced = codegen->ReplaceCallSites(insert_batch_fn,
+ materialize_exprs_tuple_pool_fn, Tuple::MATERIALIZE_EXPRS_SYMBOL);
+ DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
+
+ replaced = codegen->ReplaceCallSites(insert_batch_fn,
+ materialize_exprs_no_pool_fn, Tuple::MATERIALIZE_EXPRS_NULL_POOL_SYMBOL);
+ DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
+
+ // The total number of calls to tuple_row_less_than_->Compare() is 3 in
+ // PriorityQueue (called from 2 places), 1 in TopNNode::Heap::InsertTupleRow()
+ // and 3 in TopNNode::Heap::InsertTupleWithTieHandling
+ // Each tuple_row_less_than_->Less(Tuple*, Tuple*) indirectly calls Compare() once.
+ replaced = codegen->ReplaceCallSites(insert_batch_fn,
+ compare_fn, TupleRowComparator::COMPARE_SYMBOL);
+ DCHECK_REPLACE_COUNT(replaced, 10) << LlvmCodeGen::Print(insert_batch_fn);
+
+ replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
+ codegen->GetI64Constant(heap_capacity()), "heap_capacity");
+ DCHECK_REPLACE_COUNT(replaced, 2)
+ << LlvmCodeGen::Print(insert_batch_fn);
+
+ replaced = codegen->ReplaceCallSitesWithBoolConst(
+ insert_batch_fn, include_ties(), "include_ties");
+ DCHECK_REPLACE_COUNT(replaced, 1)
+ << LlvmCodeGen::Print(insert_batch_fn);
+
+ int tuple_byte_size = output_tuple_desc_->byte_size();
+ replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
+ codegen->GetI32Constant(tuple_byte_size), "tuple_byte_size");
+ DCHECK_REPLACE_COUNT(replaced, 3)
+ << LlvmCodeGen::Print(insert_batch_fn);
+
+ insert_batch_fn = codegen->FinalizeFunction(insert_batch_fn);
+ DCHECK(insert_batch_fn != NULL);
+ codegen->AddFunctionToJit(insert_batch_fn, &codegend_insert_batch_fn_);
}
}
AddCodegenStatus(codegen_status);
@@ -171,7 +179,9 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_CANCELLED(state);
RETURN_IF_ERROR(QueryMaintenance(state));
- heap_.reset(new Heap(*tuple_row_less_than_, limit_ + offset_));
+ const TopNPlanNode& pnode = static_cast<const TopNPlanNode&>(plan_node_);
+ heap_.reset(
+ new Heap(*tuple_row_less_than_, pnode.heap_capacity(), pnode.include_ties()));
// Allocate memory for a temporary tuple.
tmp_tuple_ = reinterpret_cast<Tuple*>(
@@ -180,7 +190,7 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_ERROR(child(0)->Open(state));
// Limit of 0, no need to fetch anything from children.
- if (limit_ != 0) {
+ if (pnode.heap_capacity() != 0) {
RowBatch batch(child(0)->row_desc(), state->batch_size(), mem_tracker());
bool eos;
do {
@@ -188,14 +198,14 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_ERROR(child(0)->GetNext(state, &batch, &eos));
{
SCOPED_TIMER(insert_batch_timer_);
- const TopNPlanNode::InsertBatchFn insert_batch_fn
- = codegend_insert_batch_fn_.load();
+ TopNPlanNode::InsertBatchFn insert_batch_fn = codegend_insert_batch_fn_.load();
if (insert_batch_fn != nullptr) {
insert_batch_fn(this, &batch);
} else {
InsertBatch(&batch);
}
- if (rows_to_reclaim_ > 2 * (limit_ + offset_)) {
+ DCHECK(heap_->DCheckConsistency());
+ if (rows_to_reclaim_ > 2 * pnode.heap_capacity()) {
RETURN_IF_ERROR(ReclaimTuplePool(state));
COUNTER_ADD(tuple_pool_reclaim_counter_, 1);
}
@@ -204,8 +214,6 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_ERROR(QueryMaintenance(state));
} while (!eos);
}
- // TODO: this wouldn't be valid for partitioned.
- DCHECK_LE(heap_->num_tuples(), limit_ + offset_);
PrepareForOutput();
// Unless we are inside a subplan expecting to call Open()/GetNext() on the child
@@ -248,6 +256,7 @@ Status TopNNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
Status TopNNode::Reset(RuntimeState* state, RowBatch* row_batch) {
heap_->Reset();
+ tmp_tuple_ = nullptr;
num_rows_skipped_ = 0;
// Transfer ownership of tuple data to output batch.
row_batch->tuple_data_pool()->AcquireData(tuple_pool_.get(), false);
@@ -266,7 +275,7 @@ void TopNNode::Close(RuntimeState* state) {
}
void TopNNode::PrepareForOutput() {
- // TODO: this will need to iterate through partitions.
+ DCHECK(heap_->DCheckConsistency());
heap_->PrepareForOutput(*this, &sorted_top_n_);
get_next_iter_ = sorted_top_n_.begin();
}
@@ -274,17 +283,35 @@ void TopNNode::PrepareForOutput() {
void TopNNode::Heap::PrepareForOutput(
const TopNNode& RESTRICT node, vector<Tuple*>* sorted_top_n) RESTRICT {
// Reverse the order of the tuples in the priority queue
- sorted_top_n->resize(priority_queue_.Size());
+ sorted_top_n->resize(num_tuples());
int64_t index = sorted_top_n->size() - 1;
- while (priority_queue_.Size() > 0) {
+ /// Any ties with the min will be the last elements in 'sorted_top_n'.
+ while (!overflowed_ties_.empty()) {
+ (*sorted_top_n)[index] = overflowed_ties_.back();
+ overflowed_ties_.pop_back();
+ --index;
+ }
+ while (!priority_queue_.Empty()) {
(*sorted_top_n)[index] = priority_queue_.Pop();
--index;
}
}
+bool TopNNode::Heap::DCheckConsistency() {
+ DCHECK_LE(num_tuples(), capacity_ + overflowed_ties_.size())
+ << num_tuples() << " > " << capacity_ << " + " << overflowed_ties_.size();
+ if (!overflowed_ties_.empty()) {
+ DCHECK(include_ties_);
+ DCHECK_EQ(capacity_, priority_queue_.Size())
+ << "Ties should only be present if heap is at capacity";
+ }
+ return true;
+}
+
Status TopNNode::ReclaimTuplePool(RuntimeState* state) {
unique_ptr<MemPool> temp_pool(new MemPool(mem_tracker()));
RETURN_IF_ERROR(heap_->RematerializeTuples(this, state, temp_pool.get()));
+ DCHECK(heap_->DCheckConsistency());
rows_to_reclaim_ = 0;
tmp_tuple_ = reinterpret_cast<Tuple*>(temp_pool->TryAllocate(
@@ -314,34 +341,44 @@ void TopNNode::DebugString(int indentation_level, stringstream* out) const {
*out << ")";
}
-TopNNode::Heap::Heap(const TupleRowComparator& c, int64_t capacity)
- : capacity_(capacity), priority_queue_(c) {
- priority_queue_.Reserve(capacity);
-}
+TopNNode::Heap::Heap(const TupleRowComparator& c, int64_t capacity, bool include_ties) :
+ capacity_(capacity), include_ties_(include_ties), priority_queue_(c) {}
void TopNNode::Heap::Reset() {
priority_queue_.Clear();
+ overflowed_ties_.clear();
}
void TopNNode::Heap::Close() {
priority_queue_.Clear();
+ overflowed_ties_.clear();
}
-Status TopNNode::Heap::RematerializeTuples(TopNNode* node,
- RuntimeState* state, MemPool* new_pool) {
+template <class T>
+Status TopNNode::Heap::RematerializeTuplesHelper(TopNNode* node,
+ RuntimeState* state, MemPool* new_pool, T begin_it, T end_it) {
const TupleDescriptor& tuple_desc = *node->output_tuple_desc_;
int tuple_size = tuple_desc.byte_size();
- for (int i = 0; i < priority_queue_.Size(); i++) {
+ for (T it = begin_it; it != end_it; ++it) {
Tuple* insert_tuple = reinterpret_cast<Tuple*>(new_pool->TryAllocate(tuple_size));
if (UNLIKELY(insert_tuple == nullptr)) {
return new_pool->mem_tracker()->MemLimitExceeded(state,
"Failed to allocate memory in TopNNode::ReclaimTuplePool.", tuple_size);
}
- priority_queue_[i]->DeepCopy(insert_tuple, tuple_desc, new_pool);
- priority_queue_[i] = insert_tuple;
+ (*it)->DeepCopy(insert_tuple, tuple_desc, new_pool);
+ *it = insert_tuple;
}
return Status::OK();
}
+Status TopNNode::Heap::RematerializeTuples(TopNNode* node,
+ RuntimeState* state, MemPool* new_pool) {
+ RETURN_IF_ERROR(RematerializeTuplesHelper(
+ node, state, new_pool, priority_queue_.Begin(), priority_queue_.End()));
+ RETURN_IF_ERROR(RematerializeTuplesHelper(
+ node, state, new_pool, overflowed_ties_.begin(), overflowed_ties_.end()));
+ return Status::OK();
+}
+
template class impala::PriorityQueue<Tuple*, TupleRowComparator>;
template class impala::PriorityQueueIterator<Tuple*, TupleRowComparator>;
diff --git a/be/src/exec/topn-node.h b/be/src/exec/topn-node.h
index 4a6de25..9f033c2 100644
--- a/be/src/exec/topn-node.h
+++ b/be/src/exec/topn-node.h
@@ -46,6 +46,16 @@ class TopNPlanNode : public PlanNode {
}
~TopNPlanNode(){}
+ /// Returns the per-heap capacity used for the Heap objects in this node.
+ int64_t heap_capacity() const {
+ int64_t limit = include_ties() ? tnode_->sort_node.limit_with_ties : tnode_->limit;
+ return limit + offset();
+ }
+
+ bool include_ties() const {
+ return tnode_->sort_node.include_ties;
+ }
+
/// Ordering expressions used for tuple comparison.
std::vector<ScalarExpr*> ordering_exprs_;
@@ -55,7 +65,7 @@ class TopNPlanNode : public PlanNode {
/// Materialization exprs for the output tuple and their evaluators.
std::vector<ScalarExpr*> output_tuple_exprs_;
- /// Config used to create a TupleRowComparator instance.
+ /// Config used to create a TupleRowComparator instance for 'ordering_exprs_'.
TupleRowComparatorConfig* row_comparator_config_ = nullptr;
/// Codegened version of TopNNode::InsertBatch().
@@ -63,12 +73,37 @@ class TopNPlanNode : public PlanNode {
CodegenFnPtr<InsertBatchFn> codegend_insert_batch_fn_;
};
-/// Node for in-memory TopN (ORDER BY ... LIMIT)
-/// This handles the case where the result fits in memory.
-/// This node will materialize its input rows into a new tuple using the expressions
-/// in sort_tuple_slot_exprs_ in its sort_exec_exprs_ member.
-/// TopN is implemented by storing rows in a priority queue.
-
+/// Node for in-memory TopN operator that sorts input tuples and applies a limit such
+/// that only the Top N tuples according to the sort order are returned by the operator.
+/// This node materializes its input rows into a new row format comprised of a single
+/// tuple using the output_tuple_exprs_.
+///
+/// In-memory priority queues, represented as binary heaps, are used to compute the Top N
+/// efficiently. Maintaining in-memory heaps with the current top rows allows discarding
+/// rows that are not in the top N as soon as possible, minimizing the memory requirements
+/// and processing time of the operator.
+///
+/// TODO: currently we only support a single top-n heap per operator. IMPALA-9979 will
+/// add a partitioned mode.
+///
+/// Unpartitioned TopN Implementation Details
+/// =========================================
+/// Unpartitioned mode uses a single in-memory priority queue and does not spill results
+/// to disk. Memory consumption is bounded by the limit. After the input is consumed,
+/// rows can be directly outputted from the priority queue by calling
+/// Heap::PrepareForOutput().
+///
+/// Memory Management
+/// =================
+/// In-memory heaps are backed by 'tuple_pool_' - all tuples in the heaps must reference
+/// only memory allocated from this pool. To reclaim memory from tuples that have been
+/// evicted from the heaps, the in-memory heaps must be re-materialized with a new
+/// MemPool - see ReclaimTuplePool(). In some cases the fixed-length portion of a
+/// tuple can be reused to avoid the need to reclaim all the time.
+///
+/// In unpartitioned mode, reclamation is triggered by 'rows_to_reclaim_' hitting a
+/// threshold, which indicates that enough unused memory may have accumulated to
+/// be worth reclaiming.
class TopNNode : public ExecNode {
public:
TopNNode(ObjectPool* pool, const TopNPlanNode& pnode, const DescriptorTbl& descs);
@@ -87,18 +122,24 @@ class TopNNode : public ExecNode {
friend class TupleLessThan;
- /// Inserts all the rows in 'batch' into the queue.
+ bool include_ties() const {
+ const TopNPlanNode& pnode = static_cast<const TopNPlanNode&>(plan_node_);
+ return pnode.include_ties();
+ }
+
+ /// Inserts all the input rows in 'batch' into 'heap_'.
void InsertBatch(RowBatch* batch);
- /// Prepare to output all of the rows in this operator in sorted order. Initializes
- /// 'sorted_top_n_' and 'get_next_iterator_'.
+ /// Prepare to start outputting rows. Called after consuming all rows from the child.
+ /// Collects all output rows in 'sorted_top_n_' and initializes 'get_next_iter_' to
+ /// point to the first row.
void PrepareForOutput();
- // Re-materialize all tuples that reference 'tuple_pool_' and release 'tuple_pool_',
- // replacing it with a new pool.
+ /// Re-materialize all tuples that reference 'tuple_pool_' and release 'tuple_pool_',
+ /// replacing it with a new pool.
Status ReclaimTuplePool(RuntimeState* state);
- IR_NO_INLINE int tuple_byte_size() const {
+ IR_NO_INLINE int tuple_byte_size() const noexcept {
return output_tuple_desc_->byte_size();
}
@@ -112,17 +153,13 @@ class TopNNode : public ExecNode {
/// Cached descriptor for the materialized tuple.
TupleDescriptor* const output_tuple_desc_;
- /// Comparator for priority_queue_.
+ /// Comparator for ordering tuples globally.
std::unique_ptr<TupleRowComparator> tuple_row_less_than_;
- /// After computing the TopN in the priority_queue, pop them and put them in this vector
+ /// Temporary staging vector for sorted tuples extracted from a Heap via
+ /// Heap::PrepareForOutput().
std::vector<Tuple*> sorted_top_n_;
- /// Tuple allocated once from tuple_pool_ and reused in InsertTupleRow to
- /// materialize input tuples if necessary. After materialization, tmp_tuple_ may be
- /// copied into the tuple pool and inserted into the priority queue.
- Tuple* tmp_tuple_ = nullptr;
-
/// Stores everything referenced in priority_queue_.
std::unique_ptr<MemPool> tuple_pool_;
@@ -133,10 +170,10 @@ class TopNNode : public ExecNode {
/// was used to create this instance.
const CodegenFnPtr<TopNPlanNode::InsertBatchFn>& codegend_insert_batch_fn_;
- /// Timer for time spent in InsertBatch() function (or codegen'd version)
+ /// Timer for time spent in InsertBatch() function (or codegen'd version).
RuntimeProfile::Counter* insert_batch_timer_;
- /// Number of rows to be reclaimed since tuple_pool_ was last created/reclaimed
+ /// Number of rows to be reclaimed since tuple_pool_ was last created/reclaimed.
int64_t rows_to_reclaim_;
/// Number of times tuple pool memory was reclaimed
@@ -145,10 +182,15 @@ class TopNNode : public ExecNode {
/////////////////////////////////////////
/// BEGIN: Members that must be Reset()
- /// A heap containing up to 'limit_' + 'offset_' rows.
+ /// Tuple allocated once from tuple_pool_ and reused in InsertTupleRow to
+ /// materialize input tuples if necessary. After materialization, tmp_tuple_ may be
+ /// copied into the tuple pool and inserted into the priority queue.
+ Tuple* tmp_tuple_ = nullptr;
+
+ // Single heap used as the main heap in unpartitioned Top-N.
std::unique_ptr<Heap> heap_;
- /// Number of rows skipped. Used for adhering to offset_.
+ /// Number of rows skipped. Used for adhering to offset_ in unpartitioned Top-N.
int64_t num_rows_skipped_;
/// END: Members that must be Reset()
@@ -159,7 +201,7 @@ class TopNNode : public ExecNode {
/// up to 'capacity' tuples.
class TopNNode::Heap {
public:
- Heap(const TupleRowComparator& c, int64_t capacity);
+ Heap(const TupleRowComparator& c, int64_t capacity, bool include_ties);
void Reset();
void Close();
@@ -168,8 +210,8 @@ class TopNNode::Heap {
/// copy of 'tuple_row', which it stores in 'tuple_pool'. Always inlined in IR into
/// TopNNode::InsertBatch() because codegen relies on this for substituting exprs
/// in the body of TopNNode.
- /// Returns true if a previous row was replaced.
- bool IR_ALWAYS_INLINE InsertTupleRow(TopNNode* node, TupleRow* input_row);
+ /// Returns the number of materialized tuples discarded that may need to be reclaimed.
+ int IR_ALWAYS_INLINE InsertTupleRow(TopNNode* node, TupleRow* input_row);
/// Copy the elements in the priority queue into a new tuple pool, and release
/// the previous pool.
@@ -180,26 +222,57 @@ class TopNNode::Heap {
void PrepareForOutput(
const TopNNode& RESTRICT node, std::vector<Tuple*>* sorted_top_n) RESTRICT;
+ /// Can be called to invoke DCHECKs if the heap is in an inconsistent state.
+ /// Returns a bool so it can be wrapped in a DCHECK() macro.
+ bool DCheckConsistency();
+
/// Returns number of tuples currently in heap.
- int64_t num_tuples() const { return priority_queue_.Size(); }
+ int64_t num_tuples() const { return priority_queue_.Size() + overflowed_ties_.size(); }
- IR_NO_INLINE int64_t heap_capacity() const { return capacity_; }
+ IR_NO_INLINE int64_t heap_capacity() const noexcept { return capacity_; }
+
+ IR_NO_INLINE bool include_ties() const noexcept { return include_ties_; }
private:
+ /// Helper for RematerializeTuples() that materializes the tuples in a container in the
+ /// range (begin_it, end_it].
+ template <class T>
+ Status RematerializeTuplesHelper(TopNNode* node, RuntimeState* state,
+ MemPool* new_pool, T begin_it, T end_it);
+
+ /// Helper to insert tuple row into a full priority queue with tie handling. This
+ /// should not be called until the heap is at capacity and tie handling is needed.
+ /// 'materialized_tuple' must be materialized into the output row format, i.e.
+ /// output_tuple_desc_. Returns the number of materialized tuples discarded as a result
+ /// of this function.
+ /// Always inlined in IR because codegen relies on this for substituting exprs in the
+ /// body of the function.
+ int IR_ALWAYS_INLINE InsertTupleWithTieHandling(
+ TopNNode* node, Tuple* materialized_tuple);
+
+
/// Limit on capacity of 'priority_queue_'. If inserting a tuple into the queue
/// would exceed this, a tuple is popped off the queue.
const int64_t capacity_;
+ /// If true, the heap may include more than 'capacity_' if multiple tuples are
+ /// tied to be the head of the heap.
+ const bool include_ties_;
+
/////////////////////////////////////////
/// BEGIN: Members that must be Reset()
- /// The priority queue (represented by a vector and modified using
- /// push_heap()/pop_heap() to maintain ordered heap invariants) will never have more
- /// elements in it than the LIMIT + OFFSET. The order of the queue is the opposite of
- /// what the ORDER BY clause specifies, such that the top of the queue is the last
- /// sorted element.
+ /// The priority queue is represented by a vector and modified using push_heap()/
+ /// pop_heap() to maintain ordered heap invariants. It has up to 'capacity_' elements
+ /// in the heap. The order of the queue is the opposite of what the ORDER BY clause
+ /// specifies, such that the head of the queue (i.e. index 0) is the last sorted
+ /// element.
PriorityQueue<Tuple*, TupleRowComparator> priority_queue_;
+ /// Tuples tied with the head of the priority queue in excess of the heap capacity.
+ /// Only used when 'include_ties_' is true.
+ std::vector<Tuple*> overflowed_ties_;
+
/// END: Members that must be Reset()
/////////////////////////////////////////
};
diff --git a/be/src/util/tuple-row-compare.h b/be/src/util/tuple-row-compare.h
index 25b9f4d..bf982c4 100644
--- a/be/src/util/tuple-row-compare.h
+++ b/be/src/util/tuple-row-compare.h
@@ -113,6 +113,12 @@ class TupleRowComparator {
return Compare(nullptr, nullptr, lhs, rhs);
}
+ int ALWAYS_INLINE Compare(const Tuple* lhs, const Tuple* rhs) const {
+ TupleRow* lhs_row = reinterpret_cast<TupleRow*>(&lhs);
+ TupleRow* rhs_row = reinterpret_cast<TupleRow*>(&rhs);
+ return Compare(lhs_row, rhs_row);
+ }
+
/// Returns true if lhs is strictly less than rhs.
/// All exprs (ordering_exprs_lhs_ and ordering_exprs_rhs_) must have been prepared
/// and opened before calling this.
diff --git a/common/thrift/PlanNodes.thrift b/common/thrift/PlanNodes.thrift
index 6987e8e..e6a9bb5 100644
--- a/common/thrift/PlanNodes.thrift
+++ b/common/thrift/PlanNodes.thrift
@@ -485,6 +485,12 @@ struct TSortNode {
// Estimated bytes of input that will go into this sort node across all backends.
// -1 if such estimate is unavailable.
4: optional i64 estimated_full_input_size
+
+ // Whether to include ties for the last place in the Top-N
+ 5: optional bool include_ties
+
+ // If include_ties is true, the limit including ties.
+ 6: optional i64 limit_with_ties
}
enum TAnalyticWindowType {
diff --git a/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java b/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
index 1fac321..c61bcbe 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
@@ -17,7 +17,6 @@
package org.apache.impala.analysis;
-import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
@@ -74,15 +73,16 @@ public class AnalyticExpr extends Expr {
// SQL string of this AnalyticExpr before standardization. Returned in toSqlImpl().
private String sqlString_;
+ public static String DENSERANK = "dense_rank";
+ public static String RANK = "rank";
+ public static String ROWNUMBER = "row_number";
+
private static String LEAD = "lead";
private static String LAG = "lag";
private static String FIRST_VALUE = "first_value";
private static String LAST_VALUE = "last_value";
private static String FIRST_VALUE_IGNORE_NULLS = "first_value_ignore_nulls";
private static String LAST_VALUE_IGNORE_NULLS = "last_value_ignore_nulls";
- private static String RANK = "rank";
- private static String DENSERANK = "dense_rank";
- private static String ROWNUMBER = "row_number";
private static String MIN = "min";
private static String MAX = "max";
private static String PERCENT_RANK = "percent_rank";
@@ -190,7 +190,7 @@ public class AnalyticExpr extends Expr {
&& ((AggregateFunction) fn).isAnalyticFn();
}
- private static boolean isAnalyticFn(Function fn, String fnName) {
+ public static boolean isAnalyticFn(Function fn, String fnName) {
return isAnalyticFn(fn) && fn.functionName().equals(fnName);
}
diff --git a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
index 458f228..70d5cba 100644
--- a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
@@ -38,6 +38,7 @@ import org.apache.impala.analysis.SlotRef;
import org.apache.impala.analysis.SortInfo;
import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.analysis.TupleId;
+import org.apache.impala.catalog.Function;
import org.apache.impala.common.Pair;
import org.apache.impala.thrift.TAnalyticNode;
import org.apache.impala.thrift.TExplainLevel;
@@ -221,7 +222,7 @@ public class AnalyticEvalNode extends PlanNode {
// compare elements[i]
Expr lhs = elements.get(i);
- Preconditions.checkState(lhs.isBound(inputTid));
+ Preconditions.checkState(lhs.isBound(inputTid), lhs.debugString());
Expr rhs = lhs.substitute(bufferedSmap, analyzer, false);
Expr bothNull = new CompoundPredicate(
@@ -362,6 +363,19 @@ public class AnalyticEvalNode extends PlanNode {
.setSpillableBufferBytes(bufferSize).setMaxRowBufferBytes(bufferSize).build();
}
+ public static class LimitPushdownInfo {
+ public LimitPushdownInfo(boolean includeTies, int additionalLimit) {
+ this.includeTies = includeTies;
+ this.additionalLimit = additionalLimit;
+ }
+
+ // Whether ties need to be returned from the top-n operator.
+ public final boolean includeTies;
+
+ // Amount that needs to be added to limit pushed down.
+ public final int additionalLimit;
+ }
+
/**
* Check if it is safe to push down limit to the Sort node of this AnalyticEval.
* Qualifying checks:
@@ -371,33 +385,33 @@ public class AnalyticEvalNode extends PlanNode {
* - If there is a predicate on the analytic function (provided through the
* selectNode), the predicate's eligibility is checked (see further below)
* @param sortInfo The sort info from the outer sort node
+ * @param sortInputSmap the input smap that can be applied to the sort exprs from
+ * sortInfo to get exprs referencing its input tuple.
* @param selectNode The selection node with predicates on analytic function.
* This can be null if no such predicate is present.
* @param limit Limit value from the outer sort node
* @param analyticNodeSort The analytic sort associated with this analytic node
- * @param sortExprsForPartitioning A placeholder list supplied by caller that is
- * populated with the sort exprs of the analytic sort that will be later used
- * for hash partitioning of the distributed TopN.
* @param analyzer analyzer instance
- * @return True if limit pushdown into analytic sort is safe, False if not
+ * @return parameters for limit pushdown into analytic sort if allowed, null if not
*/
- public boolean isLimitPushdownSafe(SortInfo sortInfo, SelectNode selectNode,
- long limit, SortNode analyticNodeSort, List<Expr> sortExprsForPartitioning,
- Analyzer analyzer) {
- if (analyticFnCalls_.size() != 1) return false;
+ public LimitPushdownInfo checkForLimitPushdown(SortInfo sortInfo,
+ ExprSubstitutionMap sortInputSmap, SelectNode selectNode, long limit,
+ SortNode analyticNodeSort, Analyzer analyzer) {
+ if (analyticFnCalls_.size() != 1) return null;
Expr expr = analyticFnCalls_.get(0);
if (!(expr instanceof FunctionCallExpr) ||
(!AnalyticExpr.isRankingFn(((FunctionCallExpr) expr).getFn()))) {
- return false;
+ return null;
}
+
List<Expr> analyticSortSortExprs = analyticNodeSort.getSortInfo().getSortExprs();
// In the mapping below, we use the original sort exprs that the sortInfo was
// created with, not the sort exprs that got mapped in SortInfo.createSortTupleInfo().
- // This allows us to substitute it using this node's output smap.
+ // This allows us to substitute it so that it references the analytic sort tuple.
List<Expr> origSortExprs = sortInfo != null ? sortInfo.getOrigSortExprs() :
new ArrayList<>();
- List<Expr> sortExprs = Expr.substituteList(origSortExprs, getOutputSmap(),
+ List<Expr> sortExprs = Expr.substituteList(origSortExprs, sortInputSmap,
analyzer, false);
// Also use substituted partition exprs such that they can be compared with the
// sort exprs
@@ -408,58 +422,51 @@ public class AnalyticEvalNode extends PlanNode {
// the limit to the sort below if there is no selection node or if
// the predicate in the selection node is eligible
if (selectNode == null) {
- return true;
+ return new LimitPushdownInfo(false, 0);
}
- Pair<Boolean, Double> status =
- isPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
- if (status.first) {
- sortExprsForPartitioning.addAll(analyticSortSortExprs);
+ Pair<LimitPushdownInfo, Double> status =
+ checkPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
+ if (status.first != null) {
selectNode.setSelectivity(status.second);
- return true;
+ return status.first;
}
- return false;
+ return null;
}
Preconditions.checkArgument(analyticSortSortExprs.size() >= pbExprs.size());
// Check if pby exprs are a prefix of the top level sort exprs
- // TODO: also check if subsequent expressions match. Need to check ASC and NULLS FIRST
- // compatibility more explicitly in the case.
- if (sortExprs.size() == 0) {
- sortExprsForPartitioning.addAll(pbExprs);
- } else {
+ if (sortExprs.size() > 0) {
if (!analyticSortExprsArePrefix(
sortInfo, sortExprs, analyticNodeSort.getSortInfo(), pbExprs)) {
- return false;
+ return null;
}
-
- // get the corresponding sort exprs from the analytic sort
- // since that's what will eventually be used for hash partitioning
- sortExprsForPartitioning.addAll(analyticSortSortExprs.subList(0, pbExprs.size()));
}
-
// check that the window frame is UNBOUNDED PRECEDING to CURRENT ROW
if (!(analyticWindow_.getLeftBoundary().getType() ==
AnalyticWindow.BoundaryType.UNBOUNDED_PRECEDING
&& analyticWindow_.getRightBoundary().getType() ==
AnalyticWindow.BoundaryType.CURRENT_ROW)) {
- return false;
+ return null;
}
if (selectNode == null) {
// Limit pushdown is valid if the pre-analytic sort puts rows into the same order
// as sortExprs. We check the prefix match, since extra analytic sort exprs do not
// affect the compatibility of the ordering with sortExprs.
- return analyticSortExprsArePrefix(sortInfo, sortExprs,
- analyticNodeSort.getSortInfo(), analyticSortSortExprs);
+ if (!analyticSortExprsArePrefix(sortInfo, sortExprs,
+ analyticNodeSort.getSortInfo(), analyticSortSortExprs)) {
+ return null;
+ }
+ return new LimitPushdownInfo(false, 0);
} else {
- Pair<Boolean, Double> status =
- isPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
- if (status.first) {
+ Pair<LimitPushdownInfo, Double> status =
+ checkPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
+ if (status.first != null) {
selectNode.setSelectivity(status.second);
- return true;
+ return status.first;
}
}
- return false;
+ return null;
}
/**
@@ -492,16 +499,47 @@ public class AnalyticEvalNode extends PlanNode {
/**
* Check eligibility of a predicate (provided as list of conjuncts) for limit
* pushdown optimization.
- * TODO: IMPALA-10296: this check is not strict enough to allow limit pushdown to
- * be safe in all circumstances.
+ *
+ * Pushing down the limit past the predicate is not, in general, possible because
+ * the predicate can filter out an arbitrary number of rows flowing returned from
+ * the analytic sort. However, if the predicate is a rank() or row_number() analytic
+ * predicate, in some cases we can determine an upper bound on the number of rows
+ * returned from the analytic sort that will end up in the output of the final top-n.
+ *
+ * The following conditions must be met:
+ * 1. The final top-n sort order must match the analytic partition sort order.
+ * The order within the partition doesn't matter for the purposes of the pushdown
+ * decision.
+ * 2. A single rank() or row_number() predicate evaluated by the analytic operator
+ * between the final top-n and analytic sort.
+ * 3. The max rows returned from each analytic partition - P - must be >= the limit
+ * being pushed down - N. (Note: ties in rank() means that P is not a strict limit
+ * on rows returned from each partition, but extra rows don't invalidate the
+ * argument below).
+ *
+ * In this case the rows returned from the final sort are going to come from the
+ * first analytic partitions because of #1. I.e. if we go through the partitions
+ * in order, until the row count adds up to N, then that is the set of partitions
+ * that can possibly end up in the output of the final top-n.
+ * If P >= N, we will never hit a partition limit before we hit N.
+ *
+ * If P < N, we could hit a partition limit and filter out an arbitrary number of
+ * rows before we find the next row to be returned, therefore the optimization
+ * can't be applied.
+ *
+ * The ordering within a partition in the analytic sort may be different from the
+ * final order, so we need to make sure to include all of the relevant rows in the
+ * final partition, which we can do by pushing down a limit of P + N, and including
+ * ties if the predicate is rank().
+ *
* @param conjuncts list of conjuncts from the predicate
- * @param limit limit from outer sort
- * @return a Pair whose first value is True if the conjuncts+limit allows pushdown,
- * False otherwise. Second value is the predicate's estimated selectivity
+ * @param limit limit from final sort
+ * @return a Pair whose first value is non-null if the conjuncts+limit allows pushdown,
+ * null otherwise. Second value is the predicate's estimated selectivity
*/
- private Pair<Boolean, Double> isPredEligibleForLimitPushdown(List<Expr> conjuncts,
- long limit) {
- Pair<Boolean, Double> falseStatus = new Pair<>(false, -1.0);
+ private Pair<LimitPushdownInfo, Double> checkPredEligibleForLimitPushdown(
+ List<Expr> conjuncts, long limit) {
+ Pair<LimitPushdownInfo, Double> falseStatus = new Pair<>(null, -1.0);
// Currently, single conjuncts are supported. In the future, multiple conjuncts
// involving a range e.g 'col >= 10 AND col <= 20' could potentially be supported
if (conjuncts.size() > 1) return falseStatus;
@@ -512,7 +550,8 @@ public class AnalyticEvalNode extends PlanNode {
Expr rhs = pred.getChild(1);
// Lhs of the binary predicate must be a ranking function.
// Also, it must be bound to the output tuple of this analytic eval node
- if (!(lhs instanceof SlotRef)) {
+ if (!(lhs instanceof SlotRef) ||
+ !lhs.isBound(outputTupleDesc_.getId())) {
return falseStatus;
}
List<Expr> lhsSourceExprs = ((SlotRef) lhs).getDesc().getSourceExprs();
@@ -520,11 +559,25 @@ public class AnalyticEvalNode extends PlanNode {
!(lhsSourceExprs.get(0) instanceof AnalyticExpr)) {
return falseStatus;
}
- if (!(AnalyticExpr.isRankingFn(((AnalyticExpr) lhsSourceExprs.
- get(0)).getFnCall().getFn()))
- || !lhs.isBound(outputTupleDesc_.getId())) {
+
+ boolean includeTies;
+ Function fn = ((AnalyticExpr) lhsSourceExprs.get(0)).getFnCall().getFn();
+ if (AnalyticExpr.isAnalyticFn(fn, AnalyticExpr.RANK)) {
+ // RANK() assigns equal values to rows that compare equal. Thus if there are
+ // ties for the max RANK() value to be returned, they all pass the filter.
+ // E.g. if the predicate is RANK() <= 3, and the values we are ordering by are
+ // 1, 42, 99, 99, 100, then the RANK() values are 1, 2, 3, 3, 5 and we need
+ // to return the top 4 rows:
+ // 1, 42, 99, 99. We do not need special handling for ties except at the limit
+ // value. E.g. if the ordering values were 1, 1, 42, 99, 100, then the RANK()
+ // values would be 1, 1, 3, 4, 5 and we only return the top 3.
+ includeTies = true;
+ } else if (AnalyticExpr.isAnalyticFn(fn, AnalyticExpr.ROWNUMBER)) {
+ includeTies = false;
+ } else {
return falseStatus;
}
+
// Restrict the pushdown for =, <, <= predicates because these ensure the
// qualifying rows are fully 'contained' within the LIMIT value. Other
// types of predicates would select rows that fall outside the LIMIT range.
@@ -533,12 +586,24 @@ public class AnalyticEvalNode extends PlanNode {
pred.getOp() == BinaryPredicate.Operator.LE)) {
return falseStatus;
}
- // Rhs of the predicate must be a numeric literal and its value
- // must be less than or equal to the limit.
- if (!(rhs instanceof NumericLiteral) ||
- ((NumericLiteral)rhs).getLongValue() > limit) {
- return falseStatus;
- }
+ // Rhs of the predicate must be a numeric literal we need to add the value
+ // to the limit to get correct results when the ordering expressions don't
+ // exactly match between the analytic sort and the parent sort.
+ if (!(rhs instanceof NumericLiteral)) return falseStatus;
+ long analyticLimit = ((NumericLiteral)rhs).getLongValue();
+ if (pred.getOp() == BinaryPredicate.Operator.LT) analyticLimit--;
+
+ // Equality predicates can filter out an arbitrary number of rows from
+ // each analytic partition, so it is not safe to push the limit down.
+ if (pred.getOp() == BinaryPredicate.Operator.EQ && limit > 1) return falseStatus;
+
+ // See method comment for explanation of why the analytic partition limit
+ // must be >= the pushed down limit.
+ if (analyticLimit < limit) return falseStatus;
+
+ // Avoid overflow of limit.
+ if (analyticLimit + (long)limit > Integer.MAX_VALUE) return falseStatus;
+
double selectivity = Expr.DEFAULT_SELECTIVITY;
// Since the predicate is qualified for limit pushdown, estimate its selectivity.
// For EQ conditions, leave it as the default. For LT and LE, assume all of the
@@ -547,7 +612,8 @@ public class AnalyticEvalNode extends PlanNode {
pred.getOp() == BinaryPredicate.Operator.LE) {
selectivity = 1.0;
}
- return new Pair<Boolean, Double>(true, selectivity);
+ return new Pair<LimitPushdownInfo, Double>(
+ new LimitPushdownInfo(includeTies, (int)analyticLimit), selectivity);
}
}
diff --git a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
index f284e32..9f7456d 100644
--- a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
@@ -1089,19 +1089,19 @@ public class DistributedPlanner {
// Make sure the childFragment's output is partitioned as required by the sortNode.
DataPartition sortPartition = sortNode.getInputPartition();
if (!childFragment.getDataPartition().equals(sortPartition)) {
- if (sortNode.hasLimit() && sortNode.isTypeTopN()) {
+ if (sortNode.isTypeTopN()) {
lowerTopN = sortNode;
childFragment.addPlanRoot(lowerTopN);
+ // Update partitioning exprs to reference sort tuple.
+ sortPartition.substitute(
+ sortNode.getSortInfo().getOutputSmap(), ctx_.getRootAnalyzer());
addedLowerTopN = true;
- DataPartition hashPartition =
- DataPartition.hashPartitioned(lowerTopN.getPartitioningExprs());
// When creating the analytic fragment, pass in a flag to unset the limit
// on the partition exchange. This ensures that the exchange does not
// prematurely stop sending rows in case there's a downstream operator
// that has a LIMIT - for instance a Sort with LIMIT after the
// Analytic operator.
- analyticFragment = createParentFragment(childFragment, hashPartition,
- true);
+ analyticFragment = createParentFragment(childFragment, sortPartition, true);
} else {
analyticFragment = createParentFragment(childFragment, sortPartition);
}
@@ -1109,11 +1109,12 @@ public class DistributedPlanner {
}
if (addedLowerTopN) {
// Create the upper TopN node
- SortNode upperTopN = SortNode.createTopNSortNode(ctx_.getNextNodeId(),
- childFragment.getPlanRoot(), lowerTopN.getSortInfo(), sortNode.getOffset());
+ SortNode upperTopN = SortNode.createTopNSortNode(ctx_.getQueryOptions(),
+ ctx_.getNextNodeId(), childFragment.getPlanRoot(),
+ lowerTopN.getSortInfo(), sortNode.getOffset(), lowerTopN.getSortLimit(),
+ lowerTopN.isIncludeTies());
upperTopN.setIsAnalyticSort(true);
upperTopN.init(ctx_.getRootAnalyzer());
- upperTopN.setLimit(lowerTopN.getLimit());
// connect this to the analytic eval node
analyticNode.setChild(0, upperTopN);
analyticFragment.addPlanRoot(upperTopN);
@@ -1137,33 +1138,47 @@ public class DistributedPlanner {
childFragment.addPlanRoot(node);
if (!childFragment.isPartitioned()) return childFragment;
- // Remember original offset and limit.
- boolean hasLimit = node.hasLimit();
- long limit = node.getLimit();
- long offset = node.getOffset();
-
// Create a new fragment for a sort-merging exchange.
PlanFragment mergeFragment =
createParentFragment(childFragment, DataPartition.UNPARTITIONED);
ExchangeNode exchNode = (ExchangeNode) mergeFragment.getPlanRoot();
-
- // Set limit, offset and merge parameters in the exchange node.
- exchNode.unsetLimit();
- if (hasLimit) exchNode.setLimit(limit);
- exchNode.setMergeInfo(node.getSortInfo(), offset);
-
- // Child nodes should not process the offset. If there is a limit,
- // the child nodes need only return (offset + limit) rows.
SortNode childSortNode = (SortNode) childFragment.getPlanRoot();
- Preconditions.checkState(node == childSortNode);
- if (hasLimit) {
- childSortNode.unsetLimit();
- childSortNode.setLimit(PlanNode.checkedAdd(limit, offset));
+ if (node.isIncludeTies()) {
+ Preconditions.checkState(node.getOffset() == 0,
+ "Tie handling with offset not supported");
+ // TopN that returns ties needs special handling because ties are not handled
+ // correctly by the ExchangeNode limit. We need to generate a top-n on top
+ // of the exchange to correctly merge the input.
+ SortNode parentSortNode = SortNode.createTopNSortNode(
+ ctx_.getQueryOptions(), ctx_.getNextNodeId(), exchNode,
+ childSortNode.getSortInfo(), 0, node.getSortLimit(),
+ childSortNode.isIncludeTies());
+ parentSortNode.init(ctx_.getRootAnalyzer());
+ mergeFragment.addPlanRoot(parentSortNode);
+ } else {
+ // Remember original offset and limit.
+ boolean hasLimit = node.hasLimit();
+ long limit = node.getLimit();
+ long offset = node.getOffset();
+
+ // Set limit, offset and merge parameters in the exchange node.
+ exchNode.unsetLimit();
+ if (hasLimit) exchNode.setLimit(limit);
+ exchNode.setMergeInfo(node.getSortInfo(), offset);
+
+ // Child nodes should not process the offset. If there is a limit,
+ // the child nodes need only return (offset + limit) rows.
+ Preconditions.checkState(node == childSortNode);
+ if (hasLimit) {
+ childSortNode.unsetLimit();
+ childSortNode.setLimit(PlanNode.checkedAdd(limit, offset));
+ }
+ childSortNode.setOffset(0);
}
- childSortNode.setOffset(0);
+
+
childSortNode.computeStats(ctx_.getRootAnalyzer());
exchNode.computeStats(ctx_.getRootAnalyzer());
-
return mergeFragment;
}
}
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 2a85d9e..a556d80 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -64,8 +64,8 @@ import org.apache.impala.catalog.FeDataSourceTable;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeHBaseTable;
-import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.ScalarType;
@@ -75,6 +75,7 @@ import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.NotImplementedException;
import org.apache.impala.common.Pair;
+import org.apache.impala.planner.AnalyticEvalNode.LimitPushdownInfo;
import org.apache.impala.planner.JoinNode.DistributionMode;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.util.AcidUtils;
@@ -336,35 +337,20 @@ public class SingleNodePlanner {
long limit, long offset, boolean hasLimit, boolean disableTopN)
throws ImpalaException {
SortNode sortNode;
- long topNBytesLimit = ctx_.getQueryOptions().topn_bytes_limit;
if (hasLimit && offset == 0) {
checkAndApplyLimitPushdown(root, sortInfo, limit, analyzer);
}
if (hasLimit && !disableTopN) {
- if (topNBytesLimit <= 0) {
- sortNode =
- SortNode.createTopNSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
- } else {
- long topNCardinality = PlanNode.capCardinalityAtLimit(root.cardinality_, limit);
- long estimatedTopNMaterializedSize =
- sortInfo.estimateTopNMaterializedSize(topNCardinality, offset);
-
- if (estimatedTopNMaterializedSize < topNBytesLimit) {
- sortNode =
- SortNode.createTopNSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
- } else {
- sortNode =
- SortNode.createTotalSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
- }
- }
+ sortNode = SortNode.createTopNSortNode(ctx_.getQueryOptions(),
+ ctx_.getNextNodeId(), root, sortInfo, offset, limit, false);
} else {
sortNode =
SortNode.createTotalSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
+ sortNode.setLimit(limit);
}
Preconditions.checkState(sortNode.hasValidStats());
- sortNode.setLimit(limit);
sortNode.init(analyzer);
return sortNode;
@@ -376,10 +362,9 @@ public class SingleNodePlanner {
*/
private void checkAndApplyLimitPushdown(PlanNode root, SortInfo sortInfo, long limit,
Analyzer analyzer) {
- boolean pushdownLimit = false;
+ LimitPushdownInfo pushdownLimit = null;
AnalyticEvalNode analyticNode = null;
List<PlanNode> intermediateNodes = new ArrayList<>();
- List<Expr> partitioningExprs = new ArrayList<>();
SortNode analyticNodeSort = null;
PlanNode descendant = findDescendantAnalyticNode(root, intermediateNodes);
if (descendant != null && intermediateNodes.size() <= 1) {
@@ -391,24 +376,30 @@ public class SingleNodePlanner {
return;
}
analyticNodeSort = (SortNode) analyticNode.getChild(0);
+ // Don't attempt conversion if there is already a limit or offset.
+ if (analyticNodeSort.hasLimit() || analyticNodeSort.hasOffset()) {
+ return;
+ }
int numNodes = intermediateNodes.size();
if (numNodes > 1 ||
(numNodes == 1 && !(intermediateNodes.get(0) instanceof SelectNode))) {
- pushdownLimit = false;
+ pushdownLimit = null;
} else if (numNodes == 0) {
- pushdownLimit = analyticNode.isLimitPushdownSafe(sortInfo, null,
- limit, analyticNodeSort, partitioningExprs, ctx_.getRootAnalyzer());
+ pushdownLimit = analyticNode.checkForLimitPushdown(sortInfo,
+ root.getOutputSmap(), null, limit, analyticNodeSort, ctx_.getRootAnalyzer());
} else {
SelectNode selectNode = (SelectNode) intermediateNodes.get(0);
- pushdownLimit = analyticNode.isLimitPushdownSafe(sortInfo, selectNode,
- limit, analyticNodeSort, partitioningExprs, ctx_.getRootAnalyzer());
+ pushdownLimit = analyticNode.checkForLimitPushdown(sortInfo,
+ root.getOutputSmap(), selectNode, limit, analyticNodeSort,
+ ctx_.getRootAnalyzer());
}
}
- if (pushdownLimit) {
+ if (pushdownLimit != null) {
Preconditions.checkArgument(analyticNode != null);
Preconditions.checkArgument(analyticNode.getChild(0) instanceof SortNode);
- analyticNodeSort.convertToTopN(limit, partitioningExprs, analyzer);
+ analyticNodeSort.tryConvertToTopN(limit + pushdownLimit.additionalLimit,
+ analyzer, pushdownLimit.includeTies);
// after the limit is pushed down, update stats for the analytic eval node
// and intermediate nodes
analyticNode.computeStats(analyzer);
@@ -1997,7 +1988,6 @@ public class SingleNodePlanner {
Expr.getIds(allJoinConjuncts, null, allSlotIds);
List<TupleId> joinInputTupleIds = joinInput.getTupleIds();
List<Expr> distinctExprs = new ArrayList<>();
- double estDistinctTupleSize = 0;
for (SlotDescriptor slot : analyzer.getSlotDescs(allSlotIds)) {
if (joinInputTupleIds.contains(slot.getParent().getId())) {
distinctExprs.add(new SlotRef(slot));
diff --git a/fe/src/main/java/org/apache/impala/planner/SortNode.java b/fe/src/main/java/org/apache/impala/planner/SortNode.java
index ff78ab1..bfcebd5 100644
--- a/fe/src/main/java/org/apache/impala/planner/SortNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/SortNode.java
@@ -68,9 +68,6 @@ public class SortNode extends PlanNode {
// analytic eval node otherwise null
private AnalyticEvalNode analyticEvalNode_;
- // set only for the analytic sort node
- private List<Expr> partitioningExprs_;
-
// info_.sortTupleSlotExprs_ substituted with the outputSmap_ for materialized slots
// in init().
private List<Expr> resolvedTupleExprs_;
@@ -78,6 +75,14 @@ public class SortNode extends PlanNode {
// The offset of the first row to return.
protected long offset_;
+ // Whether to include ties for the last place in the Top-N values.
+ // Only supported if type_ is TOPN.
+ protected boolean includeTies_;
+
+ // If includeTies_ is true, this is the limit used. We cannot use the default 'limit_'
+ // because the plan node may return more than this number of rows.
+ protected long limitWithTies_;
+
// The type of sort. Determines the exec node used in the BE.
private TSortType type_;
@@ -90,15 +95,30 @@ public class SortNode extends PlanNode {
*/
public static SortNode createPartialSortNode(
PlanNodeId id, PlanNode input, SortInfo info) {
- return new SortNode(id, input, info, 0, TSortType.PARTIAL);
+ return new SortNode(id, input, info, 0, -1, false, TSortType.PARTIAL);
}
/**
- * Creates a new SortNode with a limit that is executed with TopNNode in the BE.
+ * Creates a new SortNode with a limit that is executed with either TopNNode in the BE
+ * or SortNode in the backend dependent on TOPN_BYTES_LIMIT.
*/
- public static SortNode createTopNSortNode(
- PlanNodeId id, PlanNode input, SortInfo info, long offset) {
- return new SortNode(id, input, info, offset, TSortType.TOPN);
+ public static SortNode createTopNSortNode(TQueryOptions queryOptions,
+ PlanNodeId id, PlanNode input, SortInfo info, long offset, long limit,
+ boolean includeTies) {
+ long topNBytesLimit = queryOptions.topn_bytes_limit;
+ long topNCardinality = capCardinalityAtLimit(input.cardinality_, limit);
+ long estimatedTopNMaterializedSize =
+ info.estimateTopNMaterializedSize(topNCardinality, offset);
+
+ SortNode result;
+ if (topNBytesLimit <= 0 || estimatedTopNMaterializedSize < topNBytesLimit
+ || includeTies) {
+ result = new SortNode(id, input, info, offset, limit, includeTies, TSortType.TOPN);
+ } else {
+ result = SortNode.createTotalSortNode(id, input, info, offset);
+ result.setLimit(limit);
+ }
+ return result;
}
/**
@@ -106,22 +126,34 @@ public class SortNode extends PlanNode {
*/
public static SortNode createTotalSortNode(
PlanNodeId id, PlanNode input, SortInfo info, long offset) {
- return new SortNode(id, input, info, offset, TSortType.TOTAL);
+ return new SortNode(id, input, info, offset, -1, false, TSortType.TOTAL);
}
private SortNode(
- PlanNodeId id, PlanNode input, SortInfo info, long offset, TSortType type) {
+ PlanNodeId id, PlanNode input, SortInfo info, long offset, long limit,
+ boolean includeTies, TSortType type) {
super(id, info.getSortTupleDescriptor().getId().asList(), getDisplayName(type));
+ Preconditions.checkState(offset >= 0);
+ Preconditions.checkArgument(type != TSortType.TOPN || limit >= 0);
info_ = info;
children_.add(input);
offset_ = offset;
+ includeTies_ = includeTies;
+ limitWithTies_ = includeTies ? limit : -1;
type_ = type;
+ if (!includeTies) setLimit(limit);
}
public long getOffset() { return offset_; }
public void setOffset(long offset) { offset_ = offset; }
public boolean hasOffset() { return offset_ > 0; }
public boolean isTypeTopN() { return type_ == TSortType.TOPN; }
+ public boolean isIncludeTies() { return includeTies_; }
+ // Get the limit that applies to the sort, including ties or not.
+ public long getSortLimit() {
+ return includeTies_ ? limitWithTies_ : limit_;
+ }
+
public SortInfo getSortInfo() { return info_; }
public void setInputPartition(DataPartition inputPartition) {
inputPartition_ = inputPartition;
@@ -134,20 +166,33 @@ public class SortNode extends PlanNode {
/**
* Under special cases, the planner may decide to convert a total sort into a
- * TopN sort with limit
+ * TopN sort with limit. This does the conversion to top-n if the converted
+ * node would pass the TOPN_BYTES_LIMIT check. Otherwise does not modify this node.
*/
- public void convertToTopN(long limit, List<Expr> partitioningExprs,
- Analyzer analyzer) {
+ public void tryConvertToTopN(long limit, Analyzer analyzer, boolean includeTies) {
Preconditions.checkArgument(type_ == TSortType.TOTAL);
+ Preconditions.checkState(!hasLimit());
+ Preconditions.checkState(!hasOffset());
+ long topNBytesLimit = analyzer.getQueryOptions().topn_bytes_limit;
+ long topNCardinality = capCardinalityAtLimit(getChild(0).cardinality_, limit);
+ long estimatedTopNMaterializedSize =
+ info_.estimateTopNMaterializedSize(topNCardinality, offset_);
+
+ if (topNBytesLimit > 0 && estimatedTopNMaterializedSize >= topNBytesLimit) {
+ return;
+ }
type_ = TSortType.TOPN;
displayName_ = getDisplayName(type_);
- setLimit(limit);
- partitioningExprs_ = partitioningExprs;
+ includeTies_ = includeTies;
+ if (includeTies) {
+ unsetLimit();
+ limitWithTies_ = limit;
+ } else {
+ setLimit(limit);
+ }
computeStats(analyzer);
}
- public List<Expr> getPartitioningExprs() { return partitioningExprs_ ; }
-
@Override
public boolean allowPartitioned() {
if (isAnalyticSort_ && hasLimit()) return true;
@@ -203,7 +248,12 @@ public class SortNode extends PlanNode {
@Override
protected void computeStats(Analyzer analyzer) {
super.computeStats(analyzer);
- cardinality_ = capCardinalityAtLimit(getChild(0).cardinality_);
+ if (includeTies_) {
+ cardinality_ = capCardinalityAtLimit(getChild(0).cardinality_, limitWithTies_);
+ } else {
+ cardinality_ = capCardinalityAtLimit(getChild(0).cardinality_);
+ }
+
if (LOG.isTraceEnabled()) {
LOG.trace("stats Sort: cardinality=" + Long.toString(cardinality_));
}
@@ -221,13 +271,19 @@ public class SortNode extends PlanNode {
.add("is_asc", "[" + Joiner.on(" ").join(strings) + "]")
.add("nulls_first", "[" + Joiner.on(" ").join(info_.getNullsFirst()) + "]")
.add("offset_", offset_)
+ .add("includeTies_", includeTies_)
+ .add("limitWithTies_", limitWithTies_)
.addValue(super.debugString())
.toString();
}
@Override
protected void toThrift(TPlanNode msg) {
- Preconditions.checkState(!isTypeTopN() || hasLimit(), "Top-N must have limit");
+ Preconditions.checkState(!isTypeTopN() || hasLimit() ||
+ (includeTies_ && limitWithTies_ >= 0), "Top-N must have limit",
+ debugString());
+ Preconditions.checkState(!includeTies_ || (!hasLimit() && limitWithTies_ >= 0),
+ "Top-N with tie handling must set limitWithTies_ only");
Preconditions.checkState(offset_ >= 0);
msg.node_type = TPlanNodeType.SORT_NODE;
TSortInfo sort_info = new TSortInfo(Expr.treesToThrift(info_.getSortExprs()),
@@ -239,6 +295,10 @@ public class SortNode extends PlanNode {
TSortNode sort_node = new TSortNode(sort_info, type_);
sort_node.setOffset(offset_);
sort_node.setEstimated_full_input_size(estimatedFullInputSize_);
+ sort_node.setInclude_ties(includeTies_);
+ if (includeTies_) {
+ sort_node.setLimit_with_ties(limitWithTies_);
+ }
msg.sort_node = sort_node;
}
@@ -253,6 +313,9 @@ public class SortNode extends PlanNode {
output.append(getSortingOrderExplainString(info_.getSortExprs(),
info_.getIsAscOrder(), info_.getNullsFirstParams(), info_.getSortingOrder(),
info_.getNumLexicalKeysInZOrder()));
+ if (includeTies_) {
+ output.append(detailPrefix + "limit with ties: " + limitWithTies_ + "\n");
+ }
}
if (detailLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) {
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test b/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test
index 137243b..a6cdaa4 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test
@@ -16,18 +16,19 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -45,59 +46,61 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-06:TOP-N [LIMIT=10]
+06:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
05:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
row-size=14B cardinality=11.00K
====
-# row_number() predicate on equality instead of range.
+# row_number() predicate
# Limit pushdown into analytic sort should be applied.
select * from (
select int_col, bigint_col, smallint_col,
row_number() over (partition by int_col order by smallint_col desc) rk
from functional.alltypesagg) dt
-where rk = 5
+where rk <= 10
order by int_col, bigint_col, smallint_col, rk
-limit 10
+limit 5
---- PLAN
PLAN-ROOT SINK
|
-04:TOP-N [LIMIT=10]
+04:TOP-N [LIMIT=5]
| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
-| row-size=22B cardinality=1
+| row-size=22B cardinality=5
|
03:SELECT
-| predicates: row_number() = 5
-| row-size=22B cardinality=1
+| predicates: row_number() <= 10
+| row-size=22B cardinality=15
|
02:ANALYTIC
| functions: row_number()
| partition by: int_col
| order by: smallint_col DESC
| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=15
|
-01:TOP-N [LIMIT=10]
+01:TOP-N [LIMIT=15]
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| row-size=14B cardinality=15
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -107,32 +110,32 @@ PLAN-ROOT SINK
|
07:MERGING-EXCHANGE [UNPARTITIONED]
| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
-| limit: 10
+| limit: 5
|
-04:TOP-N [LIMIT=10]
+04:TOP-N [LIMIT=5]
| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
-| row-size=22B cardinality=1
+| row-size=22B cardinality=5
|
03:SELECT
-| predicates: row_number() = 5
-| row-size=22B cardinality=1
+| predicates: row_number() <= 10
+| row-size=22B cardinality=15
|
02:ANALYTIC
| functions: row_number()
| partition by: int_col
| order by: smallint_col DESC
| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=15
|
-06:TOP-N [LIMIT=10]
+06:TOP-N [LIMIT=15]
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| row-size=14B cardinality=15
|
05:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N [LIMIT=15]
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| row-size=14B cardinality=15
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -157,18 +160,19 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col, bigint_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, bigint_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -186,24 +190,26 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col, bigint_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-06:TOP-N [LIMIT=10]
+06:TOP-N
| order by: int_col ASC NULLS LAST, bigint_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
05:EXCHANGE [HASH(int_col,bigint_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, bigint_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -289,11 +295,12 @@ PLAN-ROOT SINK
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -314,17 +321,19 @@ PLAN-ROOT SINK
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-05:TOP-N [LIMIT=10]
+05:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
-04:EXCHANGE [HASH(int_col,smallint_col)]
+04:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -336,7 +345,7 @@ select * from (
select int_col, bigint_col, smallint_col,
rank() over (partition by int_col order by smallint_col desc) rk
from functional.alltypesagg) dt
-where 10 > rk
+where 11 > rk
order by int_col, bigint_col, smallint_col, rk
limit 10
---- PLAN
@@ -347,19 +356,20 @@ PLAN-ROOT SINK
| row-size=22B cardinality=10
|
03:SELECT
-| predicates: rank() < 10
-| row-size=22B cardinality=10
+| predicates: rank() < 11
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -376,25 +386,93 @@ PLAN-ROOT SINK
| row-size=22B cardinality=10
|
03:SELECT
-| predicates: rank() < 10
-| row-size=22B cardinality=10
+| predicates: rank() < 11
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-06:TOP-N [LIMIT=10]
+06:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
05:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=14B cardinality=11.00K
+====
+# IMPALA-10296: Limit pushdown should not be applied
+# when the limit on rank() is smaller than the limit on the order by.
+select * from (
+ select int_col, bigint_col, smallint_col,
+ rank() over (partition by int_col order by smallint_col desc) rk
+ from functional.alltypesagg) dt
+where rk < 10
+order by int_col, bigint_col, smallint_col, rk
+limit 10
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=10]
+| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
+| row-size=22B cardinality=10
+|
+03:SELECT
+| predicates: rank() < 10
+| row-size=22B cardinality=1.10K
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: int_col
+| order by: smallint_col DESC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=22B cardinality=11.00K
+|
+01:SORT
+| order by: int_col ASC NULLS LAST, smallint_col DESC
+| row-size=14B cardinality=11.00K
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=14B cardinality=11.00K
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+06:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
+| limit: 10
+|
+04:TOP-N [LIMIT=10]
+| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
+| row-size=22B cardinality=10
+|
+03:SELECT
+| predicates: rank() < 10
+| row-size=22B cardinality=1.10K
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: int_col
+| order by: smallint_col DESC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=22B cardinality=11.00K
+|
+01:SORT
+| order by: int_col ASC NULLS LAST, smallint_col DESC
+| row-size=14B cardinality=11.00K
+|
+05:EXCHANGE [HASH(int_col)]
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -420,18 +498,19 @@ PLAN-ROOT SINK
|
04:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
03:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-02:TOP-N [LIMIT=10]
+02:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
01:AGGREGATE [FINALIZE]
| group by: int_col, bigint_col, smallint_col
@@ -493,18 +572,19 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 20
-| row-size=22B cardinality=1.10K
+| row-size=22B cardinality=30
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=11.00K
+| row-size=22B cardinality=30
|
-01:SORT
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=11.00K
+| limit with ties: 30
+| row-size=14B cardinality=30
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -658,26 +738,74 @@ PLAN-ROOT SINK
row-size=14B cardinality=11.00K
====
# 2 ranking functions with different partition-by exprs. Predicate is on
-# the dense_rank() function. If the dense_rank() appears as the top
-# level analytic function, the limit pushdown can be applied. Otherwise,
-# it should not.
+# the row_number() function. If the row_number() appears as the top
+# level analytic function, like in this example, the limit pushdown can be
+# applied. Otherwise, it should not.
select * from (
select int_col, bigint_col, smallint_col,
- dense_rank() over (partition by bigint_col order by smallint_col desc) rk2,
+ rank() over (partition by int_col order by smallint_col desc) rk1,
+ row_number() over (partition by bigint_col order by smallint_col desc) rk2
+ from functional.alltypesagg) dt
+where rk2 <= 10
+order by bigint_col, int_col, smallint_col, rk1, rk2
+limit 10
+---- PLAN
+PLAN-ROOT SINK
+|
+06:TOP-N [LIMIT=10]
+| order by: bigint_col ASC, int_col ASC, smallint_col ASC, rk1 ASC, rk2 ASC
+| row-size=30B cardinality=10
+|
+05:SELECT
+| predicates: row_number() <= 10
+| row-size=30B cardinality=20
+|
+04:ANALYTIC
+| functions: row_number()
+| partition by: bigint_col
+| order by: smallint_col DESC
+| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=30B cardinality=20
+|
+03:TOP-N [LIMIT=20]
+| order by: bigint_col ASC NULLS LAST, smallint_col DESC
+| row-size=22B cardinality=20
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: int_col
+| order by: smallint_col DESC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=22B cardinality=11.00K
+|
+01:SORT
+| order by: int_col ASC NULLS LAST, smallint_col DESC
+| row-size=14B cardinality=11.00K
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=14B cardinality=11.00K
+====
+# 2 ranking functions with different partition-by exprs. Predicate is on
+# the row_number() function. row_number() is not the top analytic function
+# here, so limit pushdown cannot be applied.
+select * from (
+ select int_col, bigint_col, smallint_col,
+ row_number() over (partition by bigint_col order by smallint_col desc) rk2,
rank() over (partition by int_col order by smallint_col desc) rk1
from functional.alltypesagg) dt
where rk2 <= 10
-order by int_col, bigint_col, smallint_col, rk1, rk2
+order by bigint_col, int_col, smallint_col, rk1, rk2
limit 10
---- PLAN
PLAN-ROOT SINK
|
06:TOP-N [LIMIT=10]
-| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk1 ASC, rk2 ASC
+| order by: bigint_col ASC, int_col ASC, smallint_col ASC, rk1 ASC, rk2 ASC
| row-size=30B cardinality=10
|
05:SELECT
-| predicates: dense_rank() <= 10
+| predicates: row_number() <= 10
| row-size=30B cardinality=1.10K
|
04:ANALYTIC
@@ -692,10 +820,10 @@ PLAN-ROOT SINK
| row-size=22B cardinality=11.00K
|
02:ANALYTIC
-| functions: dense_rank()
+| functions: row_number()
| partition by: bigint_col
| order by: smallint_col DESC
-| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| row-size=22B cardinality=11.00K
|
01:SORT
@@ -939,3 +1067,210 @@ PLAN-ROOT SINK
HDFS partitions=4/24 files=4 size=34.10KB
row-size=24B cardinality=2.17K
====
+# Very high rank() predicate is not pushed down because the limit would not fit in
+# a 32-bit integer.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypestiny) v
+where rnk <= 9999999999
+order by tinyint_col, string_col desc, id desc
+limit 6
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=6]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=1
+|
+03:SELECT
+| predicates: rank() <= 9999999999
+| row-size=26B cardinality=1
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# Very high rank() predicate is pushed down because the limit fits in
+# a 32-bit integer and the top-n bytes limit does not kick in.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypestiny) v
+where rnk <= 999999999
+order by tinyint_col, string_col desc, id desc
+limit 6
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=6]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=6
+|
+03:SELECT
+| predicates: rank() <= 999999999
+| row-size=26B cardinality=8
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:TOP-N
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| limit with ties: 1000000005
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# rank() predicate is not pushed down because TOPN_BYTES_LIMIT prevents conversion
+# to top-n node.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypesagg) v
+where rnk <= 9999
+order by tinyint_col, string_col desc, id desc
+limit 6
+---- QUERYOPTIONS
+topn_bytes_limit=1000
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=6]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=28B cardinality=6
+|
+03:SELECT
+| predicates: rank() <= 9999
+| row-size=28B cardinality=11.00K
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=28B cardinality=11.00K
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| row-size=20B cardinality=11.00K
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=20B cardinality=11.00K
+====
+# Limit cannot be pushed down past a dense_rank() predicate.
+select tinyint_col, string_col, id, rnk from (
+ select *, dense_rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypestiny) v
+where rnk <= 10
+order by tinyint_col, string_col desc, id desc
+limit 10
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=10]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=1
+|
+03:SELECT
+| predicates: dense_rank() <= 10
+| row-size=26B cardinality=1
+|
+02:ANALYTIC
+| functions: dense_rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# Limit cannot be pushed down past = predicates, except for comparison to 1 where
+# it is equivalent to <= 1
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from functional.alltypestiny) v
+where rnk = 2
+order by tinyint_col, string_col desc, id desc
+limit 2
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=2]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=2
+|
+03:SELECT
+| predicates: rank() = 2
+| row-size=26B cardinality=8
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, id ASC
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# Limit cannot be pushed down past = predicates, except for comparison to 1 where
+# it is equivalent to <= 1
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from functional.alltypestiny) v
+where rnk = 1
+order by tinyint_col, string_col desc, id desc
+limit 1
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=1]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=1
+|
+03:SELECT
+| predicates: rank() = 1
+| row-size=26B cardinality=1
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=2
+|
+01:TOP-N
+| order by: tinyint_col ASC NULLS LAST, id ASC
+| limit with ties: 2
+| row-size=18B cardinality=2
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test b/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test
index 9d97d5d..30159a5 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test
@@ -6122,7 +6122,7 @@ PLAN-ROOT SINK
03:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 01(GETNEXT)
|
02:ANALYTIC
@@ -6131,13 +6131,14 @@ PLAN-ROOT SINK
| order by: l_orderkey DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 01(GETNEXT)
|
-01:TOP-N [LIMIT=100]
+01:TOP-N
| order by: l_partkey ASC NULLS LAST, l_orderkey DESC
-| mem-estimate=2.34KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| limit with ties: 200
+| mem-estimate=4.69KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [tpch_parquet.lineitem]
@@ -6181,7 +6182,7 @@ Per-Host Resources: mem-estimate=4.02MB mem-reservation=4.00MB thread-reservatio
03:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 06(GETNEXT)
|
02:ANALYTIC
@@ -6190,26 +6191,28 @@ Per-Host Resources: mem-estimate=4.02MB mem-reservation=4.00MB thread-reservatio
| order by: l_orderkey DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 06(GETNEXT)
|
-06:TOP-N [LIMIT=100]
+06:TOP-N
| order by: l_partkey ASC NULLS LAST, l_orderkey DESC
-| mem-estimate=2.34KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| limit with ties: 200
+| mem-estimate=4.69KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 06(GETNEXT), 01(OPEN)
|
05:EXCHANGE [HASH(l_partkey)]
-| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 01(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=80.00MB mem-reservation=16.00MB thread-reservation=2
-01:TOP-N [LIMIT=100]
+01:TOP-N
| order by: l_partkey ASC NULLS LAST, l_orderkey DESC
-| mem-estimate=2.34KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| limit with ties: 200
+| mem-estimate=4.69KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test
index ae96817..bc8328f 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test
@@ -58,7 +58,7 @@ PLAN-ROOT SINK
11:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 09(GETNEXT)
|
10:ANALYTIC
@@ -67,13 +67,14 @@ PLAN-ROOT SINK
| order by: aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 6 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 7 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 8 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 9 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 10 THEN su [...]
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 09(GETNEXT)
|
-09:TOP-N [LIMIT=100]
+09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
-| mem-estimate=8.98KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| limit with ties: 200
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 09(GETNEXT), 08(OPEN)
|
08:AGGREGATE [FINALIZE]
@@ -173,10 +174,10 @@ PLAN-ROOT SINK
| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds_parquet.store_sales]
- HDFS partitions=1824/1824 files=1824 size=200.95MB
+ HDFS partitions=1824/1824 files=1824 size=196.95MB
runtime filters: RF000[bloom] -> ss_item_sk, RF002[bloom] -> ss_store_sk, RF004[bloom] -> ss_sold_date_sk
stored statistics:
- table: rows=2.88M size=200.95MB
+ table: rows=2.88M size=196.95MB
partitions: 1824/1824 rows=2.88M
columns: all
extrapolated-rows=disabled max-scan-range-rows=130.09K
@@ -200,7 +201,7 @@ PLAN-ROOT SINK
| in pipelines: 12(GETNEXT)
|
F05:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END)] hosts=3 instances=3
-Per-Host Resources: mem-estimate=4.04MB mem-reservation=4.00MB thread-reservation=1
+Per-Host Resources: mem-estimate=4.08MB mem-reservation=4.00MB thread-reservation=1
12:TOP-N [LIMIT=100]
| order by: i_category ASC, i_class ASC, i_brand ASC, i_product_name ASC, d_year ASC, d_qoy ASC, d_moy ASC, s_store_id ASC, sumsales ASC, rk ASC
| mem-estimate=9.38KB mem-reservation=0B thread-reservation=0
@@ -210,7 +211,7 @@ Per-Host Resources: mem-estimate=4.04MB mem-reservation=4.00MB thread-reservatio
11:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 19(GETNEXT)
|
10:ANALYTIC
@@ -219,26 +220,28 @@ Per-Host Resources: mem-estimate=4.04MB mem-reservation=4.00MB thread-reservatio
| order by: aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 6 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 7 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 8 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 9 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 10 THEN su [...]
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 19(GETNEXT)
|
-19:TOP-N [LIMIT=100]
+19:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
-| mem-estimate=8.98KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| limit with ties: 200
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 19(GETNEXT), 09(OPEN)
|
18:EXCHANGE [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END)]
-| mem-estimate=31.12KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| mem-estimate=62.24KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 09(GETNEXT)
|
F04:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_category) WHEN 5 THEN murmur_hash(i_category) WHEN 6 THEN murmur_hash(i_category) WHEN 7 THEN murmur_hash(i_category) WHEN 8 THEN murmur_hash(i_category) WHEN 9 THEN murmur_hash(i_category) WHEN 10 THEN murmur_hash(i_category) WHEN 11 THEN murmur_hash(i_category) WHEN 12 THEN murmur_hash(NULL) END,CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_class) WHEN 5 THEN murmur_hash(i_class) WHE [...]
Per-Host Resources: mem-estimate=43.90GB mem-reservation=244.75MB thread-reservation=1
-09:TOP-N [LIMIT=100]
+09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
-| mem-estimate=8.98KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| limit with ties: 200
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 09(GETNEXT), 08(OPEN)
|
08:AGGREGATE [FINALIZE]
@@ -398,10 +401,10 @@ Per-Host Resources: mem-estimate=8.96GB mem-reservation=221.75MB thread-reservat
| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds_parquet.store_sales, RANDOM]
- HDFS partitions=1824/1824 files=1824 size=200.95MB
+ HDFS partitions=1824/1824 files=1824 size=196.95MB
runtime filters: RF000[bloom] -> ss_item_sk, RF002[bloom] -> ss_store_sk, RF004[bloom] -> ss_sold_date_sk
stored statistics:
- table: rows=2.88M size=200.95MB
+ table: rows=2.88M size=196.95MB
partitions: 1824/1824 rows=2.88M
columns: all
extrapolated-rows=disabled max-scan-range-rows=130.09K
diff --git a/testdata/workloads/functional-query/queries/limit-pushdown-analytic.test b/testdata/workloads/functional-query/queries/limit-pushdown-analytic.test
new file mode 100644
index 0000000..0554932
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/limit-pushdown-analytic.test
@@ -0,0 +1,96 @@
+====
+---- QUERY
+# IMPALA-10296: this query returns correct results even before
+# IMPALA-10296 because the limit is not hit. Reducing the limit
+# should simply truncate the results because the ordering is total.
+select tinyint_col, id from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from alltypestiny) v
+where rnk < 4
+order by tinyint_col, id desc
+limit 10
+---- TYPES
+TINYINT,INT
+---- RESULTS
+0,4
+0,2
+0,0
+1,5
+1,3
+1,1
+====
+---- QUERY
+# IMPALA-10296: it is not safe to push the limit down past the analytic.
+# This query reproduced incorrect results prior to the IMPALA-10296 fix.
+select tinyint_col, id from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from alltypestiny) v
+where rnk < 4
+order by tinyint_col, id desc
+limit 5
+---- TYPES
+TINYINT,INT
+---- RESULTS
+0,4
+0,2
+0,0
+1,5
+1,3
+====
+---- QUERY
+# IMPALA-10296: this query returns correct results even before
+# IMPALA-10296 because the limit is not hit. Reducing the limit
+# should simply truncate the results because the ordering is total.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from alltypestiny) v
+where rnk <= 5
+order by tinyint_col, string_col desc, id desc
+limit 10
+---- TYPES
+TINYINT,STRING,INT,BIGINT
+---- RESULTS
+0,'0',6,1
+0,'0',4,1
+0,'0',2,1
+0,'0',0,1
+1,'1',7,1
+1,'1',5,1
+1,'1',3,1
+1,'1',1,1
+====
+---- QUERY
+# IMPALA-10296: the limit can be pushed past the analytic operator,
+# but we need to increase the limit and include ties to guarantee
+# correct results.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from alltypestiny) v
+where rnk <= 5
+order by tinyint_col, string_col desc, id desc
+limit 5
+---- TYPES
+TINYINT,STRING,INT,BIGINT
+---- RESULTS
+0,'0',6,1
+0,'0',4,1
+0,'0',2,1
+0,'0',0,1
+1,'1',7,1
+====
+---- QUERY
+# IMPALA-10296: cannot push limit through an equality comparison to
+# rank the predicate can filter out an arbitrary number of rows returned
+# from the top-n sort.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from alltypestiny) v
+where rnk = 2
+order by tinyint_col, string_col desc, id desc
+limit 2
+---- TYPES
+TINYINT,STRING,INT,BIGINT
+---- RESULTS
+0,'0',2,2
+1,'1',3,2
+====
diff --git a/testdata/workloads/tpch/queries/limit-pushdown-analytic.test b/testdata/workloads/tpch/queries/limit-pushdown-analytic.test
index 4bd8b8b..5dc1635 100644
--- a/testdata/workloads/tpch/queries/limit-pushdown-analytic.test
+++ b/testdata/workloads/tpch/queries/limit-pushdown-analytic.test
@@ -89,3 +89,152 @@ limit 3
---- TYPES
INT, INT, BIGINT
====
+---- QUERY
+# IMPALA-10296: the limit pushdown through rank() requires tie handling in
+# the top-n. This query is designed to exercise that by using an analytic
+# order by with low cardinality and therefore many ties. The query returns
+# values from two analytic partitions.
+select l_orderkey, l_partkey, l_suppkey, l_linenumber, l_shipmode, rnk from (
+ select *, rank() over (partition by l_partkey order by l_shipmode) rnk
+ from lineitem) v
+where rnk <= 50
+order by l_partkey, l_orderkey, l_suppkey, l_linenumber, l_shipmode
+limit 50
+---- RESULTS
+504065,1,2502,6,'RAIL',15
+545153,1,2,4,'SHIP',24
+599361,1,5002,7,'AIR',1
+1139363,1,2,1,'FOB',8
+1225698,1,2,5,'FOB',8
+1695463,1,5002,5,'FOB',8
+1829732,1,2,3,'RAIL',15
+1876199,1,2502,3,'AIR',1
+2265089,1,2502,4,'RAIL',15
+2301222,1,2,5,'SHIP',24
+2485411,1,2502,6,'REG AIR',21
+2535332,1,5002,4,'MAIL',12
+2603011,1,5002,2,'MAIL',12
+2698307,1,2,3,'RAIL',15
+3438019,1,2,6,'RAIL',15
+3700610,1,5002,2,'AIR',1
+3743746,1,5002,6,'SHIP',24
+4076803,1,2,4,'SHIP',24
+4168578,1,5002,3,'TRUCK',30
+4343170,1,7502,2,'SHIP',24
+4422049,1,7502,5,'AIR',1
+4450401,1,7502,1,'MAIL',12
+4464900,1,2502,1,'AIR',1
+4810753,1,5002,5,'REG AIR',21
+4912929,1,5002,5,'TRUCK',30
+5120486,1,2,1,'SHIP',24
+5352710,1,7502,2,'AIR',1
+5362629,1,5002,3,'AIR',1
+5618244,1,7502,5,'REG AIR',21
+5862498,1,2,6,'RAIL',15
+5926723,1,2,2,'FOB',8
+16135,2,5003,1,'RAIL',11
+269286,2,3,2,'RAIL',11
+437670,2,3,5,'MAIL',6
+698981,2,3,1,'TRUCK',25
+864452,2,7503,6,'TRUCK',25
+957444,2,7503,2,'TRUCK',25
+976196,2,7503,2,'RAIL',11
+1140614,2,5003,2,'SHIP',22
+1204321,2,2503,3,'MAIL',6
+1339139,2,5003,5,'REG AIR',19
+1394727,2,2503,3,'AIR',1
+2189347,2,7503,3,'TRUCK',25
+2664550,2,7503,4,'AIR',1
+2774018,2,2503,1,'RAIL',11
+2838083,2,5003,5,'SHIP',22
+2859844,2,5003,2,'AIR',1
+2974054,2,2503,5,'REG AIR',19
+3204707,2,2503,1,'MAIL',6
+3815428,2,7503,3,'TRUCK',25
+---- TYPES
+BIGINT, BIGINT, BIGINT, INT, STRING, BIGINT
+====
+---- QUERY
+# IMPALA-10296: the same query as above with a higher limit. The limit pushdown
+# optimization should not be applied and the results should be the same as above,
+# but with additional rows.
+select l_orderkey, l_partkey, l_suppkey, l_linenumber, l_shipmode, rnk from (
+ select *, rank() over (partition by l_partkey order by l_shipmode) rnk
+ from lineitem) v
+where rnk <= 50
+order by l_partkey, l_orderkey, l_suppkey, l_linenumber, l_shipmode
+limit 70
+---- RESULTS
+504065,1,2502,6,'RAIL',15
+545153,1,2,4,'SHIP',24
+599361,1,5002,7,'AIR',1
+1139363,1,2,1,'FOB',8
+1225698,1,2,5,'FOB',8
+1695463,1,5002,5,'FOB',8
+1829732,1,2,3,'RAIL',15
+1876199,1,2502,3,'AIR',1
+2265089,1,2502,4,'RAIL',15
+2301222,1,2,5,'SHIP',24
+2485411,1,2502,6,'REG AIR',21
+2535332,1,5002,4,'MAIL',12
+2603011,1,5002,2,'MAIL',12
+2698307,1,2,3,'RAIL',15
+3438019,1,2,6,'RAIL',15
+3700610,1,5002,2,'AIR',1
+3743746,1,5002,6,'SHIP',24
+4076803,1,2,4,'SHIP',24
+4168578,1,5002,3,'TRUCK',30
+4343170,1,7502,2,'SHIP',24
+4422049,1,7502,5,'AIR',1
+4450401,1,7502,1,'MAIL',12
+4464900,1,2502,1,'AIR',1
+4810753,1,5002,5,'REG AIR',21
+4912929,1,5002,5,'TRUCK',30
+5120486,1,2,1,'SHIP',24
+5352710,1,7502,2,'AIR',1
+5362629,1,5002,3,'AIR',1
+5618244,1,7502,5,'REG AIR',21
+5862498,1,2,6,'RAIL',15
+5926723,1,2,2,'FOB',8
+16135,2,5003,1,'RAIL',11
+269286,2,3,2,'RAIL',11
+437670,2,3,5,'MAIL',6
+698981,2,3,1,'TRUCK',25
+864452,2,7503,6,'TRUCK',25
+957444,2,7503,2,'TRUCK',25
+976196,2,7503,2,'RAIL',11
+1140614,2,5003,2,'SHIP',22
+1204321,2,2503,3,'MAIL',6
+1339139,2,5003,5,'REG AIR',19
+1394727,2,2503,3,'AIR',1
+2189347,2,7503,3,'TRUCK',25
+2664550,2,7503,4,'AIR',1
+2774018,2,2503,1,'RAIL',11
+2838083,2,5003,5,'SHIP',22
+2859844,2,5003,2,'AIR',1
+2974054,2,2503,5,'REG AIR',19
+3204707,2,2503,1,'MAIL',6
+3815428,2,7503,3,'TRUCK',25
+3978178,2,2503,4,'REG AIR',19
+4052995,2,3,4,'MAIL',6
+4212039,2,3,2,'FOB',4
+4291943,2,5003,5,'MAIL',6
+4325446,2,2503,6,'SHIP',22
+4563589,2,2503,2,'TRUCK',25
+4701030,2,7503,1,'RAIL',11
+4849414,2,2503,4,'TRUCK',25
+5000327,2,5003,3,'RAIL',11
+5240129,2,7503,4,'RAIL',11
+5399651,2,3,1,'TRUCK',25
+5435686,2,7503,4,'FOB',4
+5879491,2,3,4,'RAIL',11
+367685,3,7504,1,'RAIL',20
+480000,3,5004,5,'MAIL',15
+601123,3,2504,3,'AIR',1
+789220,3,7504,5,'TRUCK',34
+810723,3,4,1,'AIR',1
+951425,3,2504,6,'RAIL',20
+993152,3,4,1,'SHIP',31
+---- TYPES
+BIGINT, BIGINT, BIGINT, INT, STRING, BIGINT
+====
diff --git a/tests/query_test/test_limit_pushdown_analytic.py b/tests/query_test/test_limit_pushdown_analytic.py
index dca9c6b..caac68b 100644
--- a/tests/query_test/test_limit_pushdown_analytic.py
+++ b/tests/query_test/test_limit_pushdown_analytic.py
@@ -19,9 +19,11 @@
# of ranking functions
from tests.common.impala_test_suite import ImpalaTestSuite
+from tests.common.test_dimensions import (create_single_exec_option_dimension,
+ extend_exec_option_dimension)
-class TestLimitPushdownAnalytic(ImpalaTestSuite):
+class TestLimitPushdownAnalyticTpch(ImpalaTestSuite):
@classmethod
def get_workload(cls):
@@ -29,7 +31,25 @@ class TestLimitPushdownAnalytic(ImpalaTestSuite):
@classmethod
def add_test_dimensions(cls):
- super(TestLimitPushdownAnalytic, cls).add_test_dimensions()
+ super(TestLimitPushdownAnalyticTpch, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_constraint(lambda v:
+ v.get_value('table_format').file_format in ['parquet'])
+
+ def test_limit_pushdown_analytic(self, vector):
+ self.run_test_case('limit-pushdown-analytic', vector)
+
+
+class TestLimitPushdownAnalyticFunctional(ImpalaTestSuite):
+ @classmethod
+ def get_workload(cls):
+ return 'functional-query'
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestLimitPushdownAnalyticFunctional, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
+ # Also run with num_nodes=1 because it's easier to reproduce IMPALA-10296.
+ extend_exec_option_dimension(cls, 'num_nodes', 1)
cls.ImpalaTestMatrix.add_constraint(lambda v:
v.get_value('table_format').file_format in ['parquet'])