You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2016/05/17 17:09:10 UTC
[1/4] incubator-impala git commit: IMPALA-3535: Ignore invalid
per-pool default query options
Repository: incubator-impala
Updated Branches:
refs/heads/master a59408b57 -> 265e39f89
IMPALA-3535: Ignore invalid per-pool default query options
In 2.5 we added the ability to set per-pool default query
options. A string of key-value pairs can be specified with a
pool configuration. However, if any options fail to parse,
then all the options are ignored. We want that behavior (and
returning an error) when parsing the process-wide default
query options on startup and when parsing the options sent
from a client (e.g. in beeswax server) because an error can
be returned immediately for the triggering action at that
time (i.e. starting the impalad or submitting a query with
the options set). This behavior is bad for the pool default
query options because (a) the configuration is set by the
administrator and there's nothing we can do until a query is
submitted and (b) one invalid option shouldn't mean that
other valid options aren't set.
Change-Id: If04733b775963091b0314c65286df126fd812358
Reviewed-on: http://gerrit.cloudera.org:8080/3056
Reviewed-by: Dan Hecht <dh...@cloudera.com>
Tested-by: Internal Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/f067929f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/f067929f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/f067929f
Branch: refs/heads/master
Commit: f067929f3a2b66f60c9bc020df268dd13d1dbc02
Parents: a59408b
Author: Matthew Jacobs <mj...@cloudera.com>
Authored: Thu May 12 17:34:42 2016 -0700
Committer: Tim Armstrong <ta...@cloudera.com>
Committed: Tue May 17 10:09:05 2016 -0700
----------------------------------------------------------------------
be/src/service/impala-server.cc | 9 ++-
be/src/service/query-options-test.cc | 27 +++++++++
be/src/service/query-options.cc | 13 ++--
be/src/service/query-options.h | 3 +-
.../impala/util/TestRequestPoolService.java | 4 +-
fe/src/test/resources/llama-site-test2.xml | 2 +-
.../functional-query/queries/QueryTest/set.test | 4 +-
tests/query_test/test_query_opts.py | 64 ++++++++++++++++++++
8 files changed, 112 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/be/src/service/impala-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 25ef571..7ab3a81 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -735,9 +735,8 @@ void ImpalaServer::AddPoolQueryOptions(TQueryCtx* ctx,
status = ParseQueryOptions(config.default_query_options, &pool_options,
&set_pool_options_mask);
if (!status.ok()) {
- VLOG_RPC << "Not adding pool query options for query=" << ctx->query_id
- << " ParseQueryOptions status: " << status.GetDetail();
- return;
+ VLOG_QUERY << "Ignoring errors while parsing default query options for pool="
+ << resolved_pool << ", message: " << status.GetDetail();
}
QueryOptionsMask overlay_mask = override_options_mask & set_pool_options_mask;
@@ -1143,8 +1142,8 @@ void ImpalaServer::TransmitData(
void ImpalaServer::InitializeConfigVariables() {
QueryOptionsMask set_query_options; // unused
- Status status = ParseQueryOptions(FLAGS_default_query_options, &default_query_options_,
- &set_query_options);
+ Status status = ParseQueryOptions(FLAGS_default_query_options,
+ &default_query_options_, &set_query_options);
if (!status.ok()) {
// Log error and exit if the default query options are invalid.
CLEAN_EXIT_WITH_ERROR(Substitute("Invalid default query options. Please check "
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/be/src/service/query-options-test.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options-test.cc b/be/src/service/query-options-test.cc
index 53767d9..f1d90c0 100644
--- a/be/src/service/query-options-test.cc
+++ b/be/src/service/query-options-test.cc
@@ -80,6 +80,33 @@ TEST(QueryOptions, SetFilterWait) {
EXPECT_EQ(numeric_limits<int32_t>::max(), options.runtime_filter_wait_time_ms);
}
+TEST(QueryOptions, ParseQueryOptions) {
+ QueryOptionsMask expectedMask;
+ expectedMask.set(TImpalaQueryOptions::NUM_NODES);
+ expectedMask.set(TImpalaQueryOptions::MEM_LIMIT);
+
+ {
+ TQueryOptions options;
+ QueryOptionsMask mask;
+ EXPECT_OK(ParseQueryOptions("num_nodes=1,mem_limit=42", &options, &mask));
+ EXPECT_EQ(options.num_nodes, 1);
+ EXPECT_EQ(options.mem_limit, 42);
+ EXPECT_EQ(mask, expectedMask);
+ }
+
+ {
+ TQueryOptions options;
+ QueryOptionsMask mask;
+ Status status = ParseQueryOptions("num_nodes=1,mem_limit:42,foo=bar,mem_limit=42",
+ &options, &mask);
+ EXPECT_EQ(options.num_nodes, 1);
+ EXPECT_EQ(options.mem_limit, 42);
+ EXPECT_EQ(mask, expectedMask);
+ EXPECT_FALSE(status.ok());
+ EXPECT_EQ(status.msg().details().size(), 2);
+ }
+}
+
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
MemInfo::Init();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/be/src/service/query-options.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index d12ce1f..83e3548 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -107,7 +107,7 @@ Status impala::SetQueryOption(const string& key, const string& value,
TQueryOptions* query_options, QueryOptionsMask* set_query_options_mask) {
int option = GetQueryOptionForKey(key);
if (option < 0) {
- return Status(Substitute("Ignoring invalid configuration option: $0", key));
+ return Status(Substitute("Invalid query option: $0", key));
} else {
switch (option) {
case TImpalaQueryOptions::ABORT_ON_ERROR:
@@ -421,17 +421,22 @@ Status impala::ParseQueryOptions(const string& options, TQueryOptions* query_opt
if (options.length() == 0) return Status::OK();
vector<string> kv_pairs;
split(kv_pairs, options, is_any_of(","), token_compress_on);
+ // Construct an error status which is used to aggregate errors encountered during
+ // parsing. It is only returned if the number of error details is greater than 0.
+ Status errorStatus = Status::Expected("Errors parsing query options");
for (string& kv_string: kv_pairs) {
trim(kv_string);
if (kv_string.length() == 0) continue;
vector<string> key_value;
split(key_value, kv_string, is_any_of("="), token_compress_on);
if (key_value.size() != 2) {
- return Status(Substitute("Ignoring invalid configuration option $0: bad format "
- "(expected 'key=value')", kv_string));
+ errorStatus.MergeStatus(
+ Status(Substitute("Invalid configuration option '$0'.", kv_string)));
+ continue;
}
- RETURN_IF_ERROR(SetQueryOption(key_value[0], key_value[1], query_options,
+ errorStatus.MergeStatus(SetQueryOption(key_value[0], key_value[1], query_options,
set_query_options_mask));
}
+ if (errorStatus.msg().details().size() > 0) return errorStatus;
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/be/src/service/query-options.h
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index 60c122d..c86bc6b 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -111,7 +111,8 @@ Status SetQueryOption(const std::string& key, const std::string& value,
/// Parse a "," separated key=value pair of query options and set it in 'query_options'.
/// If the same query option is specified more than once, the last one wins. The
/// set_query_options_mask bitmask is updated to reflect the query options which were
-/// set. Return an error if the input is invalid (bad format or invalid query option).
+/// set. Returns an error status containing an error detail for any invalid options (e.g.
+/// bad format or invalid query option), but all valid query options are still handled.
Status ParseQueryOptions(const std::string& options, TQueryOptions* query_options,
QueryOptionsMask* set_query_options_mask);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/fe/src/test/java/com/cloudera/impala/util/TestRequestPoolService.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/com/cloudera/impala/util/TestRequestPoolService.java b/fe/src/test/java/com/cloudera/impala/util/TestRequestPoolService.java
index c40ca0f..8e54b88 100644
--- a/fe/src/test/java/com/cloudera/impala/util/TestRequestPoolService.java
+++ b/fe/src/test/java/com/cloudera/impala/util/TestRequestPoolService.java
@@ -250,8 +250,10 @@ public class TestRequestPoolService {
// Test pool limit changes
checkPoolConfigResult("root", 15, 100, -1, 30000L, "");
+ // not_a_valid_option=foo.bar gets filtered out when parsing the query options on
+ // the backend, but it should be observed coming from the test file here.
checkPoolConfigResult("root.queueA", 1, 30, 100000 * ByteUnits.MEGABYTE,
- 50L, "mem_limit=128m,query_timeout_s=5");
+ 50L, "mem_limit=128m,query_timeout_s=5,not_a_valid_option=foo.bar");
checkPoolConfigResult("root.queueB", 5, 10, -1, 60000L, "");
checkPoolConfigResult("root.queueC", 10, 30, 128 * ByteUnits.MEGABYTE,
30000L, "mem_limit=2048m,query_timeout_s=60");
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/fe/src/test/resources/llama-site-test2.xml
----------------------------------------------------------------------
diff --git a/fe/src/test/resources/llama-site-test2.xml b/fe/src/test/resources/llama-site-test2.xml
index 8b1e3d1..2e01f2e 100644
--- a/fe/src/test/resources/llama-site-test2.xml
+++ b/fe/src/test/resources/llama-site-test2.xml
@@ -41,7 +41,7 @@
</property>
<property>
<name>impala.admission-control.pool-default-query-options.root.queueA</name>
- <value>mem_limit=128m,query_timeout_s=5</value>
+ <value>mem_limit=128m,query_timeout_s=5,not_a_valid_option=foo.bar</value>
</property>
<property>
<name>llama.am.throttling.maximum.placed.reservations.root.queueC</name>
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/testdata/workloads/functional-query/queries/QueryTest/set.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/set.test b/testdata/workloads/functional-query/queries/QueryTest/set.test
index 361fc9b..e405caf 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/set.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/set.test
@@ -135,7 +135,7 @@ The PARQUET_FILE_SIZE query option must be less than 2GB.
---- QUERY
set foo=bar
---- CATCH
-Ignoring invalid configuration option: foo
+Invalid query option: foo
====
---- QUERY
set parquet_compression_codec=bar
@@ -255,4 +255,4 @@ explain select count(distinct double_col) from functional.alltypesagg;
'03:EXCHANGE [HASH(double_col)]'
'01:AGGREGATE'
'00:SCAN HDFS [functional.alltypesagg]'
-====
\ No newline at end of file
+====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f067929f/tests/query_test/test_query_opts.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_query_opts.py b/tests/query_test/test_query_opts.py
new file mode 100644
index 0000000..6b6cd24
--- /dev/null
+++ b/tests/query_test/test_query_opts.py
@@ -0,0 +1,64 @@
+# Copyright (c) 2016 Cloudera, Inc. All rights reserved.
+#
+# Licensed 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.
+#
+# Tests for exercising query options that can be set in various ways.
+# TODO: Add custom cluster tests for process default_query_options, but we need
+# to make it easier to handle startup failures (right now it waits 60sec to
+# timeout).
+
+from ImpalaService import ImpalaHiveServer2Service
+from TCLIService import TCLIService
+
+from tests.beeswax.impala_beeswax import ImpalaBeeswaxException
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.impala_test_suite import ImpalaTestSuite
+from tests.common.test_dimensions import create_exec_option_dimension
+from tests.hs2.hs2_test_suite import HS2TestSuite, needs_session, operation_id_to_query_id
+
+class TestQueryOptions(ImpalaTestSuite):
+ @classmethod
+ def get_workload(cls):
+ return 'functional-query'
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestQueryOptions, cls).add_test_dimensions()
+ cls.TestMatrix.add_constraint(lambda v:\
+ v.get_value('table_format').file_format == 'text')
+ cls.TestMatrix.add_dimension(create_exec_option_dimension(
+ cluster_sizes=[0], disable_codegen_options=[False], batch_sizes=[0]))
+
+ def test_set_invalid_query_option(self, vector):
+ ex = self.execute_query_expect_failure(self.client, "select 1", {'foo':'bar'})
+ assert "invalid query option: foo" in str(ex).lower()
+
+class TestQueryOptionsHS2(HS2TestSuite):
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestQueryOptions, cls).add_test_dimensions()
+ cls.TestMatrix.add_constraint(lambda v:\
+ v.get_value('table_format').file_format == 'text')
+ cls.TestMatrix.add_dimension(create_exec_option_dimension(
+ cluster_sizes=[0], disable_codegen_options=[False], batch_sizes=[0]))
+
+ @needs_session()
+ def test_set_invalid_query_option(self):
+ """Tests that GetOperationStatus returns a valid result for a running query"""
+ execute_statement_req = TCLIService.TExecuteStatementReq()
+ execute_statement_req.sessionHandle = self.session_handle
+ execute_statement_req.confOverlay = {"foo":"bar"}
+ execute_statement_req.statement = "select 1"
+ execute_statement_resp = self.hs2_client.ExecuteStatement(execute_statement_req)
+ TestQueryOptionsHS2.check_response(execute_statement_resp,
+ TCLIService.TStatusCode.ERROR_STATUS, "Invalid query option: foo")
[2/4] incubator-impala git commit: IMPALA-1928: Fix Thrift client
transport wrapping order
Posted by ta...@apache.org.
IMPALA-1928: Fix Thrift client transport wrapping order
The thrift client incorrectly wraps the TSaslTransport
around the TBufferedTransport which leads to significant
performance issues. (Note that the server-side wraps the
transports in the correct order already.)
Currently: TSaslTransport(TBufferedTransport(socket))
Should be: TBufferedTransport(TSaslTransport(socket))
As a result, when we write a structure, we end up doing lots
of write calls which hit the TSaslTransport which does no
buffering. So it ends up producing output that looks like:
[0, 0, 0, 1], <one char>, [0, 0, 0, 1], <one char>, etc.
for each individual write call.
These end up buffered so we don't get lots of tiny packets
on the send side. However, on the receiver side we are doing
one recv call per Sasl frame.
This patch reorders the wrapping of transports in the thrift
client, so that it matches the order on the thrift server
which improves exhange performance making it within 10% of
non-kerberos.
Change-Id: I81d30b3d8d10fe6dcd8eb88cca49734af09f9d91
Reviewed-on: http://gerrit.cloudera.org:8080/3093
Reviewed-by: Dan Hecht <dh...@cloudera.com>
Tested-by: Internal Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/9172f4b8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/9172f4b8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/9172f4b8
Branch: refs/heads/master
Commit: 9172f4b82444596da182955cda65951b60460501
Parents: f067929
Author: Matthew Jacobs <mj...@cloudera.com>
Authored: Mon May 16 17:22:42 2016 -0700
Committer: Tim Armstrong <ta...@cloudera.com>
Committed: Tue May 17 10:09:06 2016 -0700
----------------------------------------------------------------------
be/src/rpc/thrift-client.h | 27 +++++++++++++++++----------
1 file changed, 17 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9172f4b8/be/src/rpc/thrift-client.h
----------------------------------------------------------------------
diff --git a/be/src/rpc/thrift-client.h b/be/src/rpc/thrift-client.h
index 347e673..d80caf3 100644
--- a/be/src/rpc/thrift-client.h
+++ b/be/src/rpc/thrift-client.h
@@ -127,26 +127,33 @@ class ThriftClient : public ThriftClientImpl {
template <class InterfaceType>
ThriftClient<InterfaceType>::ThriftClient(const std::string& ipaddress, int port,
- const std::string& service_name,
- AuthProvider* auth_provider, bool ssl)
+ const std::string& service_name, AuthProvider* auth_provider, bool ssl)
: ThriftClientImpl(ipaddress, port, ssl),
iface_(new InterfaceType(protocol_)),
auth_provider_(auth_provider) {
- // Below is one line of code in ThriftClientImpl::Close(),
- // if (transport_.get != NULL && transport_->isOpen()) transport_->close();
- // Here transport_->isOpen() will call socker_->isOpen(), when socket_ is NULL,
- // it will crash
- if (socket_ != NULL) {
- ThriftServer::BufferedTransportFactory factory;
- transport_ = factory.getTransport(socket_);
- }
if (auth_provider_ == NULL) {
auth_provider_ = AuthManager::GetInstance()->GetInternalAuthProvider();
+ DCHECK(auth_provider_ != NULL);
+ }
+
+ // If socket_ is NULL (because ThriftClientImpl::CreateSocket() failed in the base
+ // class constructor, nothing else should be constructed. Open()/Reopen() will return
+ // the error that the socket couldn't be created and the caller should be careful to
+ // not use the client after that.
+ // TODO: Move initialization code that can fail into a separate Init() method.
+ if (socket_ == NULL) {
+ DCHECK(!socket_create_status_.ok());
+ return;
}
+ // transport_ is created by wrapping the socket_ in the TTransport provided by the
+ // auth_provider_ and then a TBufferedTransport (IMPALA-1928).
+ transport_ = socket_;
auth_provider_->WrapClientTransport(address_.hostname, transport_, service_name,
&transport_);
+ ThriftServer::BufferedTransportFactory factory;
+ transport_ = factory.getTransport(transport_);
protocol_.reset(new apache::thrift::protocol::TBinaryProtocol(transport_));
iface_.reset(new InterfaceType(protocol_));
[4/4] incubator-impala git commit: IMPALA-3286: prefetching for
PartitionedAggregationNode
Posted by ta...@apache.org.
IMPALA-3286: prefetching for PartitionedAggregationNode
This patch builds on top of the prefetching infrastructure to add
prefetching to PartitionedAggregationNode. Input batches are evaluated
in prefetch groups and hash table buckets are prefetched if the
prefetch_mode query option is set to HT_BUCKET.
We avoid some pointer indirections on the critical path by caching hash
tables in a 'hash_tbls_' array.
There is also a bit of cleanup to directly instantiate the templated
ProcessBatch() method to remove the ProcessBatch_true() and
ProcessBatch_false() hack, and also to separate out
ProcessBatchNoGrouping() so that it doesn't have to have the same
argument list as ProcessBatch().
Co-author: Michael Ho <kw...@cloudera.com>
Change-Id: I7726454efb416d61080c4e11db0ee7ada18c149b
Reviewed-on: http://gerrit.cloudera.org:8080/3070
Reviewed-by: Michael Ho <kw...@cloudera.com>
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Internal Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/9f4276ee
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/9f4276ee
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/9f4276ee
Branch: refs/heads/master
Commit: 9f4276eea80a2e162a3045cefcc38f936df271d4
Parents: 9172f4b
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Mon May 16 13:58:13 2016 -0700
Committer: Tim Armstrong <ta...@cloudera.com>
Committed: Tue May 17 10:09:06 2016 -0700
----------------------------------------------------------------------
be/src/codegen/gen_ir_descriptions.py | 12 +-
be/src/exec/hash-table.cc | 9 +-
be/src/exec/partitioned-aggregation-node-ir.cc | 152 +++++++++++++-------
be/src/exec/partitioned-aggregation-node.cc | 99 ++++++++-----
be/src/exec/partitioned-aggregation-node.h | 92 +++++++-----
5 files changed, 230 insertions(+), 134 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f4276ee/be/src/codegen/gen_ir_descriptions.py
----------------------------------------------------------------------
diff --git a/be/src/codegen/gen_ir_descriptions.py b/be/src/codegen/gen_ir_descriptions.py
index 71fb1a7..5cc992c 100755
--- a/be/src/codegen/gen_ir_descriptions.py
+++ b/be/src/codegen/gen_ir_descriptions.py
@@ -40,10 +40,14 @@ options, args = parser.parse_args()
ir_functions = [
["AGG_NODE_PROCESS_ROW_BATCH_WITH_GROUPING", "ProcessRowBatchWithGrouping"],
["AGG_NODE_PROCESS_ROW_BATCH_NO_GROUPING", "ProcessRowBatchNoGrouping"],
- ["PART_AGG_NODE_PROCESS_BATCH_TRUE", "ProcessBatch_true"],
- ["PART_AGG_NODE_PROCESS_BATCH_FALSE", "ProcessBatch_false"],
- ["PART_AGG_NODE_PROCESS_BATCH_NO_GROUPING", "ProcessBatchNoGrouping"],
- ["PART_AGG_NODE_PROCESS_BATCH_STREAMING", "ProcessBatchStreaming"],
+ ["PART_AGG_NODE_PROCESS_BATCH_UNAGGREGATED",
+ "PartitionedAggregationNode12ProcessBatchILb0"],
+ ["PART_AGG_NODE_PROCESS_BATCH_AGGREGATED",
+ "PartitionedAggregationNode12ProcessBatchILb1"],
+ ["PART_AGG_NODE_PROCESS_BATCH_NO_GROUPING",
+ "PartitionedAggregationNode22ProcessBatchNoGrouping"],
+ ["PART_AGG_NODE_PROCESS_BATCH_STREAMING",
+ "PartitionedAggregationNode21ProcessBatchStreaming"],
["AVG_UPDATE_BIGINT", "9AvgUpdateIN10impala_udf9BigIntVal"],
["AVG_UPDATE_DOUBLE", "9AvgUpdateIN10impala_udf9DoubleVal"],
["AVG_UPDATE_TIMESTAMP", "TimestampAvgUpdate"],
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f4276ee/be/src/exec/hash-table.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hash-table.cc b/be/src/exec/hash-table.cc
index 953ddce..719b600 100644
--- a/be/src/exec/hash-table.cc
+++ b/be/src/exec/hash-table.cc
@@ -854,9 +854,12 @@ Status HashTableCtx::CodegenHashCurrentRow(RuntimeState* state, bool use_murmur,
Value* cur_expr_values = builder.CreateLoad(cur_expr_values_ptr);
// Load cur_expr_values_null_ into a LLVM pointer.
- Value* cur_expr_values_null_ptr =
- codegen->CastPtrToLlvmPtr(buffer_ptr_type, &expr_values_cache_.cur_expr_values_null_);
- Value* cur_expr_values_null = builder.CreateLoad(cur_expr_values_null_ptr);
+ Value* cur_expr_values_null = NULL;
+ if (stores_nulls_) {
+ Value* cur_expr_values_null_ptr = codegen->CastPtrToLlvmPtr(
+ buffer_ptr_type, &expr_values_cache_.cur_expr_values_null_);
+ cur_expr_values_null = builder.CreateLoad(cur_expr_values_null_ptr);
+ }
// Call GetHashSeed() to get seeds_[level_]
Function* get_hash_seed_fn =
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f4276ee/be/src/exec/partitioned-aggregation-node-ir.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/partitioned-aggregation-node-ir.cc b/be/src/exec/partitioned-aggregation-node-ir.cc
index d575c01..d479189 100644
--- a/be/src/exec/partitioned-aggregation-node-ir.cc
+++ b/be/src/exec/partitioned-aggregation-node-ir.cc
@@ -23,8 +23,7 @@
using namespace impala;
-Status PartitionedAggregationNode::ProcessBatchNoGrouping(RowBatch* batch,
- const HashTableCtx* ht_ctx) { // 'ht_ctx' is unused
+Status PartitionedAggregationNode::ProcessBatchNoGrouping(RowBatch* batch) {
Tuple* output_tuple = singleton_output_tuple_;
FOREACH_ROW(batch, 0, batch_iter) {
UpdateTuple(&agg_fn_ctxs_[0], output_tuple, batch_iter.Get());
@@ -34,7 +33,7 @@ Status PartitionedAggregationNode::ProcessBatchNoGrouping(RowBatch* batch,
template<bool AGGREGATED_ROWS>
Status PartitionedAggregationNode::ProcessBatch(RowBatch* batch,
- HashTableCtx* __restrict__ ht_ctx) {
+ TPrefetchMode::type prefetch_mode, HashTableCtx* __restrict__ ht_ctx) {
DCHECK(!hash_partitions_.empty());
DCHECK(!is_streaming_preagg_);
@@ -44,39 +43,78 @@ Status PartitionedAggregationNode::ProcessBatch(RowBatch* batch,
// TODO: Once we have a histogram with the number of rows per partition, we will have
// accurate resize calls.
RETURN_IF_ERROR(CheckAndResizeHashPartitions(batch->num_rows(), ht_ctx));
- FOREACH_ROW(batch, 0, batch_iter) {
- RETURN_IF_ERROR(ProcessRow<AGGREGATED_ROWS>(batch_iter.Get(), ht_ctx));
+
+ HashTableCtx::ExprValuesCache* expr_vals_cache = ht_ctx->expr_values_cache();
+ const int cache_size = expr_vals_cache->capacity();
+ const int num_rows = batch->num_rows();
+ for (int group_start = 0; group_start < num_rows; group_start += cache_size) {
+ EvalAndHashPrefetchGroup<AGGREGATED_ROWS>(batch, group_start, prefetch_mode, ht_ctx);
+
+ FOREACH_ROW_LIMIT(batch, group_start, cache_size, batch_iter) {
+ RETURN_IF_ERROR(ProcessRow<AGGREGATED_ROWS>(batch_iter.Get(), ht_ctx));
+ expr_vals_cache->NextRow();
+ }
+ DCHECK(expr_vals_cache->AtEnd());
}
return Status::OK();
}
template<bool AGGREGATED_ROWS>
-Status PartitionedAggregationNode::ProcessRow(TupleRow* __restrict__ row,
- HashTableCtx* __restrict__ ht_ctx) {
- if (AGGREGATED_ROWS) {
- if (!ht_ctx->EvalAndHashBuild(row)) return Status::OK();
- } else {
- if (!ht_ctx->EvalAndHashProbe(row)) return Status::OK();
+void IR_ALWAYS_INLINE PartitionedAggregationNode::EvalAndHashPrefetchGroup(
+ RowBatch* batch, int start_row_idx, TPrefetchMode::type prefetch_mode,
+ HashTableCtx* ht_ctx) {
+ HashTableCtx::ExprValuesCache* expr_vals_cache = ht_ctx->expr_values_cache();
+ const int cache_size = expr_vals_cache->capacity();
+
+ expr_vals_cache->Reset();
+ FOREACH_ROW_LIMIT(batch, start_row_idx, cache_size, batch_iter) {
+ TupleRow* row = batch_iter.Get();
+ bool is_null;
+ if (AGGREGATED_ROWS) {
+ is_null = !ht_ctx->EvalAndHashBuild(row);
+ } else {
+ is_null = !ht_ctx->EvalAndHashProbe(row);
+ }
+ // Hoist lookups out of non-null branch to speed up non-null case.
+ const uint32_t hash = expr_vals_cache->ExprValuesHash();
+ const uint32_t partition_idx = hash >> (32 - NUM_PARTITIONING_BITS);
+ HashTable* hash_tbl = GetHashTable(partition_idx);
+ if (is_null) {
+ expr_vals_cache->SetRowNull();
+ } else if (prefetch_mode != TPrefetchMode::NONE) {
+ if (LIKELY(hash_tbl != NULL)) hash_tbl->PrefetchBucket<false>(hash);
+ }
+ expr_vals_cache->NextRow();
}
- uint32_t hash = ht_ctx->expr_values_cache()->ExprValuesHash();
+ expr_vals_cache->ResetForRead();
+}
+
+template<bool AGGREGATED_ROWS>
+Status PartitionedAggregationNode::ProcessRow(TupleRow* __restrict__ row,
+ HashTableCtx* __restrict__ ht_ctx) {
+ HashTableCtx::ExprValuesCache* expr_vals_cache = ht_ctx->expr_values_cache();
+ // Hoist lookups out of non-null branch to speed up non-null case.
+ const uint32_t hash = expr_vals_cache->ExprValuesHash();
+ const uint32_t partition_idx = hash >> (32 - NUM_PARTITIONING_BITS);
+ if (expr_vals_cache->IsRowNull()) return Status::OK();
// To process this row, we first see if it can be aggregated or inserted into this
// partition's hash table. If we need to insert it and that fails, due to OOM, we
// spill the partition. The partition to spill is not necessarily dst_partition,
// so we can try again to insert the row.
- Partition* dst_partition = hash_partitions_[hash >> (32 - NUM_PARTITIONING_BITS)];
- if (dst_partition->is_spilled()) {
+ HashTable* hash_tbl = GetHashTable(partition_idx);
+ Partition* dst_partition = hash_partitions_[partition_idx];
+ DCHECK_EQ(dst_partition->is_spilled(), hash_tbl == NULL);
+ if (hash_tbl == NULL) {
// This partition is already spilled, just append the row.
return AppendSpilledRow<AGGREGATED_ROWS>(dst_partition, row);
}
- HashTable* ht = dst_partition->hash_tbl.get();
- DCHECK(ht != NULL);
DCHECK(dst_partition->aggregated_row_stream->is_pinned());
bool found;
// Find the appropriate bucket in the hash table. There will always be a free
// bucket because we checked the size above.
- HashTable::Iterator it = ht->FindBuildRowBucket(ht_ctx, &found);
+ HashTable::Iterator it = hash_tbl->FindBuildRowBucket(ht_ctx, &found);
DCHECK(!it.AtEnd()) << "Hash table had no free buckets";
if (AGGREGATED_ROWS) {
// If the row is already an aggregate row, it cannot match anything in the
@@ -130,50 +168,47 @@ Status PartitionedAggregationNode::AppendSpilledRow(Partition* __restrict__ part
return AppendSpilledRow(stream, row);
}
-Status PartitionedAggregationNode::ProcessBatch_false(
- RowBatch* batch, HashTableCtx* __restrict__ ht_ctx) {
- return ProcessBatch<false>(batch, ht_ctx);
-}
-
-Status PartitionedAggregationNode::ProcessBatch_true(
- RowBatch* batch, HashTableCtx* __restrict__ ht_ctx) {
- return ProcessBatch<true>(batch, ht_ctx);
-}
-
Status PartitionedAggregationNode::ProcessBatchStreaming(bool needs_serialize,
- RowBatch* in_batch, RowBatch* out_batch, HashTableCtx* __restrict__ ht_ctx,
- int remaining_capacity[PARTITION_FANOUT]) {
+ TPrefetchMode::type prefetch_mode, RowBatch* in_batch, RowBatch* out_batch,
+ HashTableCtx* __restrict__ ht_ctx, int remaining_capacity[PARTITION_FANOUT]) {
DCHECK(is_streaming_preagg_);
DCHECK_EQ(out_batch->num_rows(), 0);
DCHECK_LE(in_batch->num_rows(), out_batch->capacity());
RowBatch::Iterator out_batch_iterator(out_batch, out_batch->num_rows());
- FOREACH_ROW(in_batch, 0, in_batch_iter) {
- TupleRow* in_row = in_batch_iter.Get();
- if (!ht_ctx->EvalAndHashProbe(in_row)) continue;
- uint32_t hash = ht_ctx->expr_values_cache()->ExprValuesHash();
- const uint32_t partition_idx = hash >> (32 - NUM_PARTITIONING_BITS);
-
- if (TryAddToHashTable(ht_ctx, hash_partitions_[partition_idx], in_row, hash,
- &remaining_capacity[partition_idx], &process_batch_status_)) {
- continue;
- }
- RETURN_IF_ERROR(process_batch_status_);
+ HashTableCtx::ExprValuesCache* expr_vals_cache = ht_ctx->expr_values_cache();
+ const int num_rows = in_batch->num_rows();
+ const int cache_size = expr_vals_cache->capacity();
+ for (int group_start = 0; group_start < num_rows; group_start += cache_size) {
+ EvalAndHashPrefetchGroup<false>(in_batch, group_start, prefetch_mode, ht_ctx);
- // Tuple is not going into hash table, add it to the output batch.
- Tuple* intermediate_tuple = ConstructIntermediateTuple(agg_fn_ctxs_,
- out_batch->tuple_data_pool(), &process_batch_status_);
- if (UNLIKELY(intermediate_tuple == NULL)) {
- DCHECK(!process_batch_status_.ok());
- return process_batch_status_;
+ FOREACH_ROW_LIMIT(in_batch, group_start, cache_size, in_batch_iter) {
+ // Hoist lookups out of non-null branch to speed up non-null case.
+ TupleRow* in_row = in_batch_iter.Get();
+ const uint32_t hash = expr_vals_cache->ExprValuesHash();
+ const uint32_t partition_idx = hash >> (32 - NUM_PARTITIONING_BITS);
+ if (!expr_vals_cache->IsRowNull() &&
+ !TryAddToHashTable(ht_ctx, hash_partitions_[partition_idx],
+ GetHashTable(partition_idx), in_row, hash, &remaining_capacity[partition_idx],
+ &process_batch_status_)) {
+ RETURN_IF_ERROR(process_batch_status_);
+ // Tuple is not going into hash table, add it to the output batch.
+ Tuple* intermediate_tuple = ConstructIntermediateTuple(agg_fn_ctxs_,
+ out_batch->tuple_data_pool(), &process_batch_status_);
+ if (UNLIKELY(intermediate_tuple == NULL)) {
+ DCHECK(!process_batch_status_.ok());
+ return process_batch_status_;
+ }
+ UpdateTuple(&agg_fn_ctxs_[0], intermediate_tuple, in_row, false);
+ out_batch_iterator.Get()->SetTuple(0, intermediate_tuple);
+ out_batch_iterator.Next();
+ out_batch->CommitLastRow();
+ }
+ DCHECK(process_batch_status_.ok());
+ expr_vals_cache->NextRow();
}
- UpdateTuple(&agg_fn_ctxs_[0], intermediate_tuple, in_row, false);
-
- out_batch_iterator.Get()->SetTuple(0, intermediate_tuple);
- out_batch_iterator.Next();
- out_batch->CommitLastRow();
+ DCHECK(expr_vals_cache->AtEnd());
}
-
if (needs_serialize) {
FOREACH_ROW(out_batch, 0, out_batch_iter) {
AggFnEvaluator::Serialize(aggregate_evaluators_, agg_fn_ctxs_,
@@ -186,13 +221,14 @@ Status PartitionedAggregationNode::ProcessBatchStreaming(bool needs_serialize,
bool PartitionedAggregationNode::TryAddToHashTable(
HashTableCtx* __restrict__ ht_ctx, Partition* __restrict__ partition,
- TupleRow* __restrict__ in_row, uint32_t hash, int* __restrict__ remaining_capacity,
- Status* status) {
+ HashTable* __restrict__ hash_tbl, TupleRow* __restrict__ in_row,
+ uint32_t hash, int* __restrict__ remaining_capacity, Status* status) {
DCHECK(remaining_capacity != NULL);
+ DCHECK_EQ(hash_tbl, partition->hash_tbl.get());
DCHECK_GE(*remaining_capacity, 0);
bool found;
// This is called from ProcessBatchStreaming() so the rows are not aggregated.
- HashTable::Iterator it = partition->hash_tbl->FindBuildRowBucket(ht_ctx, &found);
+ HashTable::Iterator it = hash_tbl->FindBuildRowBucket(ht_ctx, &found);
Tuple* intermediate_tuple;
if (found) {
intermediate_tuple = it.GetTuple();
@@ -214,3 +250,9 @@ bool PartitionedAggregationNode::TryAddToHashTable(
UpdateTuple(&partition->agg_fn_ctxs[0], intermediate_tuple, in_row, false);
return true;
}
+
+// Instantiate required templates.
+template Status PartitionedAggregationNode::ProcessBatch<false>(RowBatch*,
+ TPrefetchMode::type, HashTableCtx*);
+template Status PartitionedAggregationNode::ProcessBatch<true>(RowBatch*,
+ TPrefetchMode::type, HashTableCtx*);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f4276ee/be/src/exec/partitioned-aggregation-node.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/partitioned-aggregation-node.cc b/be/src/exec/partitioned-aggregation-node.cc
index 2cde059..0a6a811 100644
--- a/be/src/exec/partitioned-aggregation-node.cc
+++ b/be/src/exec/partitioned-aggregation-node.cc
@@ -109,7 +109,8 @@ PartitionedAggregationNode::PartitionedAggregationNode(
needs_serialize_(false),
block_mgr_client_(NULL),
output_partition_(NULL),
- process_row_batch_fn_(NULL),
+ process_batch_no_grouping_fn_(NULL),
+ process_batch_fn_(NULL),
process_batch_streaming_fn_(NULL),
build_timer_(NULL),
ht_resize_timer_(NULL),
@@ -274,25 +275,9 @@ Status PartitionedAggregationNode::Prepare(RuntimeState* state) {
bool codegen_enabled = false;
Status codegen_status;
if (state->codegen_enabled()) {
- LlvmCodeGen* codegen;
- RETURN_IF_ERROR(state->GetCodegen(&codegen));
- if (is_streaming_preagg_) {
- Function* codegen_process_batch_streaming_fn;
- codegen_status = CodegenProcessBatchStreaming(&codegen_process_batch_streaming_fn);
- if (codegen_status.ok()) {
- codegen->AddFunctionToJit(codegen_process_batch_streaming_fn,
- reinterpret_cast<void**>(&process_batch_streaming_fn_));
- codegen_enabled = true;
- }
- } else {
- Function* codegen_process_row_batch_fn;
- codegen_status = CodegenProcessBatch(&codegen_process_row_batch_fn);
- if (codegen_status.ok()) {
- codegen->AddFunctionToJit(codegen_process_row_batch_fn,
- reinterpret_cast<void**>(&process_row_batch_fn_));
- codegen_enabled = true;
- }
- }
+ codegen_status = is_streaming_preagg_ ? CodegenProcessBatchStreaming()
+ : CodegenProcessBatch();
+ codegen_enabled = codegen_status.ok();
}
AddCodegenExecOption(codegen_enabled, codegen_status);
return Status::OK();
@@ -330,14 +315,21 @@ Status PartitionedAggregationNode::Open(RuntimeState* state) {
}
}
+ TPrefetchMode::type prefetch_mode = state->query_options().prefetch_mode;
SCOPED_TIMER(build_timer_);
- if (process_row_batch_fn_ != NULL) {
- RETURN_IF_ERROR(process_row_batch_fn_(this, &batch, ht_ctx_.get()));
- } else if (grouping_expr_ctxs_.empty()) {
- RETURN_IF_ERROR(ProcessBatchNoGrouping(&batch));
+ if (grouping_expr_ctxs_.empty()) {
+ if (process_batch_no_grouping_fn_ != NULL) {
+ RETURN_IF_ERROR(process_batch_no_grouping_fn_(this, &batch));
+ } else {
+ RETURN_IF_ERROR(ProcessBatchNoGrouping(&batch));
+ }
} else {
// There is grouping, so we will do partitioned aggregation.
- RETURN_IF_ERROR(ProcessBatch<false>(&batch, ht_ctx_.get()));
+ if (process_batch_fn_ != NULL) {
+ RETURN_IF_ERROR(process_batch_fn_(this, &batch, prefetch_mode, ht_ctx_.get()));
+ } else {
+ RETURN_IF_ERROR(ProcessBatch<false>(&batch, prefetch_mode, ht_ctx_.get()));
+ }
}
batch.Reset();
} while (!eos);
@@ -542,8 +534,9 @@ Status PartitionedAggregationNode::GetRowsStreaming(RuntimeState* state,
int remaining_capacity[PARTITION_FANOUT];
bool ht_needs_expansion = false;
for (int i = 0; i < PARTITION_FANOUT; ++i) {
- DCHECK(hash_partitions_[i]->hash_tbl != NULL);
- remaining_capacity[i] = hash_partitions_[i]->hash_tbl->NumInsertsBeforeResize();
+ HashTable* hash_tbl = GetHashTable(i);
+ DCHECK(hash_tbl != NULL);
+ remaining_capacity[i] = hash_tbl->NumInsertsBeforeResize();
ht_needs_expansion |= remaining_capacity[i] < child_batch_->num_rows();
}
@@ -554,7 +547,7 @@ Status PartitionedAggregationNode::GetRowsStreaming(RuntimeState* state,
// should always use the remaining space in the hash table to avoid wasting memory.
if (ht_needs_expansion && ShouldExpandPreaggHashTables()) {
for (int i = 0; i < PARTITION_FANOUT; ++i) {
- HashTable* ht = hash_partitions_[i]->hash_tbl.get();
+ HashTable* ht = GetHashTable(i);
if (remaining_capacity[i] < child_batch_->num_rows()) {
SCOPED_TIMER(ht_resize_timer_);
if (ht->CheckAndResize(child_batch_->num_rows(), ht_ctx_.get())) {
@@ -564,12 +557,13 @@ Status PartitionedAggregationNode::GetRowsStreaming(RuntimeState* state,
}
}
+ TPrefetchMode::type prefetch_mode = state->query_options().prefetch_mode;
if (process_batch_streaming_fn_ != NULL) {
- RETURN_IF_ERROR(process_batch_streaming_fn_(this, needs_serialize_,
+ RETURN_IF_ERROR(process_batch_streaming_fn_(this, needs_serialize_, prefetch_mode,
child_batch_.get(), out_batch, ht_ctx_.get(), remaining_capacity));
} else {
- RETURN_IF_ERROR(ProcessBatchStreaming(needs_serialize_, child_batch_.get(),
- out_batch, ht_ctx_.get(), remaining_capacity ));
+ RETURN_IF_ERROR(ProcessBatchStreaming(needs_serialize_, prefetch_mode,
+ child_batch_.get(), out_batch, ht_ctx_.get(), remaining_capacity ));
}
child_batch_->Reset(); // All rows from child_batch_ were processed.
@@ -1145,6 +1139,7 @@ Status PartitionedAggregationNode::CreateHashPartitions(int level) {
DCHECK(new_partition != NULL);
hash_partitions_.push_back(partition_pool_->Add(new_partition));
RETURN_IF_ERROR(new_partition->InitStreams());
+ hash_tbls_[i] = NULL;
}
if (!is_streaming_preagg_) {
DCHECK_GT(state_->block_mgr()->num_reserved_buffers_remaining(block_mgr_client_), 0);
@@ -1166,6 +1161,7 @@ Status PartitionedAggregationNode::CreateHashPartitions(int level) {
}
RETURN_IF_ERROR(hash_partitions_[i]->Spill());
}
+ hash_tbls_[i] = hash_partitions_[i]->hash_tbl.get();
}
COUNTER_ADD(partitions_created_, hash_partitions_.size());
@@ -1293,12 +1289,14 @@ Status PartitionedAggregationNode::ProcessStream(BufferedTupleStream* input_stre
RETURN_IF_ERROR(SpillPartition());
}
+ TPrefetchMode::type prefetch_mode = state_->query_options().prefetch_mode;
bool eos = false;
RowBatch batch(AGGREGATED_ROWS ? *intermediate_row_desc_ : children_[0]->row_desc(),
state_->batch_size(), mem_tracker());
do {
RETURN_IF_ERROR(input_stream->GetNext(&batch, &eos));
- RETURN_IF_ERROR(ProcessBatch<AGGREGATED_ROWS>(&batch, ht_ctx_.get()));
+ RETURN_IF_ERROR(
+ ProcessBatch<AGGREGATED_ROWS>(&batch, prefetch_mode, ht_ctx_.get()));
RETURN_IF_ERROR(state_->GetQueryStatus());
FreeLocalAllocations();
batch.Reset();
@@ -1331,6 +1329,7 @@ Status PartitionedAggregationNode::SpillPartition() {
return state_->block_mgr()->MemLimitTooLowError(block_mgr_client_, id());
}
+ hash_tbls_[partition_idx] = NULL;
return hash_partitions_[partition_idx]->Spill();
}
@@ -1398,6 +1397,7 @@ void PartitionedAggregationNode::ClosePartitions() {
aggregated_partitions_.clear();
spilled_partitions_.clear();
hash_partitions_.clear();
+ memset(hash_tbls_, 0, sizeof(hash_tbls_));
partition_pool_->Clear();
}
@@ -1792,7 +1792,7 @@ Status PartitionedAggregationNode::CodegenUpdateTuple(Function** fn) {
return Status::OK();
}
-Status PartitionedAggregationNode::CodegenProcessBatch(Function** fn) {
+Status PartitionedAggregationNode::CodegenProcessBatch() {
LlvmCodeGen* codegen;
RETURN_IF_ERROR(state_->GetCodegen(&codegen));
SCOPED_TIMER(codegen->codegen_timer());
@@ -1802,7 +1802,7 @@ Status PartitionedAggregationNode::CodegenProcessBatch(Function** fn) {
// Get the cross compiled update row batch function
IRFunction::Type ir_fn = (!grouping_expr_ctxs_.empty() ?
- IRFunction::PART_AGG_NODE_PROCESS_BATCH_FALSE :
+ IRFunction::PART_AGG_NODE_PROCESS_BATCH_UNAGGREGATED :
IRFunction::PART_AGG_NODE_PROCESS_BATCH_NO_GROUPING);
Function* process_batch_fn = codegen->GetFunction(ir_fn, true);
DCHECK(process_batch_fn != NULL);
@@ -1810,6 +1810,13 @@ Status PartitionedAggregationNode::CodegenProcessBatch(Function** fn) {
int replaced;
if (!grouping_expr_ctxs_.empty()) {
// Codegen for grouping using hash table
+
+ // Replace prefetch_mode with constant so branches can be optimised out.
+ TPrefetchMode::type prefetch_mode = state_->query_options().prefetch_mode;
+ Value* prefetch_mode_arg = codegen->GetArgument(process_batch_fn, 3);
+ prefetch_mode_arg->replaceAllUsesWith(
+ ConstantInt::get(Type::getInt32Ty(codegen->context()), prefetch_mode));
+
// The codegen'd ProcessBatch function is only used in Open() with level_ = 0,
// so don't use murmur hash
Function* hash_fn;
@@ -1838,15 +1845,20 @@ Status PartitionedAggregationNode::CodegenProcessBatch(Function** fn) {
replaced = codegen->ReplaceCallSites(process_batch_fn, update_tuple_fn, "UpdateTuple");
DCHECK_GE(replaced, 1);
- *fn = codegen->FinalizeFunction(process_batch_fn);
- if (*fn == NULL) {
+ process_batch_fn = codegen->FinalizeFunction(process_batch_fn);
+ if (process_batch_fn == NULL) {
return Status("PartitionedAggregationNode::CodegenProcessBatch(): codegen'd "
"ProcessBatch() function failed verification, see log");
}
+
+ void **codegened_fn_ptr = grouping_expr_ctxs_.empty() ?
+ reinterpret_cast<void**>(&process_batch_no_grouping_fn_) :
+ reinterpret_cast<void**>(&process_batch_fn_);
+ codegen->AddFunctionToJit(process_batch_fn, codegened_fn_ptr);
return Status::OK();
}
-Status PartitionedAggregationNode::CodegenProcessBatchStreaming(Function** fn) {
+Status PartitionedAggregationNode::CodegenProcessBatchStreaming() {
DCHECK(is_streaming_preagg_);
LlvmCodeGen* codegen;
RETURN_IF_ERROR(state_->GetCodegen(&codegen));
@@ -1861,6 +1873,12 @@ Status PartitionedAggregationNode::CodegenProcessBatchStreaming(Function** fn) {
needs_serialize_arg->replaceAllUsesWith(
ConstantInt::get(Type::getInt1Ty(codegen->context()), needs_serialize_));
+ // Replace prefetch_mode with constant so branches can be optimised out.
+ TPrefetchMode::type prefetch_mode = state_->query_options().prefetch_mode;
+ Value* prefetch_mode_arg = codegen->GetArgument(process_batch_streaming_fn, 3);
+ prefetch_mode_arg->replaceAllUsesWith(
+ ConstantInt::get(Type::getInt32Ty(codegen->context()), prefetch_mode));
+
Function* update_tuple_fn;
RETURN_IF_ERROR(CodegenUpdateTuple(&update_tuple_fn));
@@ -1893,11 +1911,14 @@ Status PartitionedAggregationNode::CodegenProcessBatchStreaming(Function** fn) {
DCHECK_EQ(replaced, 1);
DCHECK(process_batch_streaming_fn != NULL);
- *fn = codegen->FinalizeFunction(process_batch_streaming_fn);
- if (*fn == NULL) {
+ process_batch_streaming_fn = codegen->FinalizeFunction(process_batch_streaming_fn);
+ if (process_batch_streaming_fn == NULL) {
return Status("PartitionedAggregationNode::CodegenProcessBatchStreaming(): codegen'd "
"ProcessBatchStreaming() function failed verification, see log");
}
+
+ codegen->AddFunctionToJit(process_batch_streaming_fn,
+ reinterpret_cast<void**>(&process_batch_streaming_fn_));
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f4276ee/be/src/exec/partitioned-aggregation-node.h
----------------------------------------------------------------------
diff --git a/be/src/exec/partitioned-aggregation-node.h b/be/src/exec/partitioned-aggregation-node.h
index ab560c5..73976c2 100644
--- a/be/src/exec/partitioned-aggregation-node.h
+++ b/be/src/exec/partitioned-aggregation-node.h
@@ -228,13 +228,17 @@ class PartitionedAggregationNode : public ExecNode {
Partition* output_partition_;
HashTable::Iterator output_iterator_;
- typedef Status (*ProcessRowBatchFn)(
- PartitionedAggregationNode*, RowBatch*, HashTableCtx*);
- /// Jitted ProcessRowBatch function pointer. Null if codegen is disabled.
- ProcessRowBatchFn process_row_batch_fn_;
+ typedef Status (*ProcessBatchNoGroupingFn)(PartitionedAggregationNode*, RowBatch*);
+ /// Jitted ProcessBatchNoGrouping function pointer. Null if codegen is disabled.
+ ProcessBatchNoGroupingFn process_batch_no_grouping_fn_;
- typedef Status (*ProcessBatchStreamingFn)(PartitionedAggregationNode*, bool, RowBatch*,
- RowBatch*, HashTableCtx*, int[PARTITION_FANOUT]);
+ typedef Status (*ProcessBatchFn)(
+ PartitionedAggregationNode*, RowBatch*, TPrefetchMode::type, HashTableCtx*);
+ /// Jitted ProcessBatch function pointer. Null if codegen is disabled.
+ ProcessBatchFn process_batch_fn_;
+
+ typedef Status (*ProcessBatchStreamingFn)(PartitionedAggregationNode*, bool,
+ TPrefetchMode::type, RowBatch*, RowBatch*, HashTableCtx*, int[PARTITION_FANOUT]);
/// Jitted ProcessBatchStreaming function pointer. Null if codegen is disabled.
ProcessBatchStreamingFn process_batch_streaming_fn_;
@@ -312,6 +316,9 @@ class PartitionedAggregationNode : public ExecNode {
/// Current partitions we are partitioning into.
std::vector<Partition*> hash_partitions_;
+ /// Cache for hash tables in 'hash_partitions_'.
+ HashTable* hash_tbls_[PARTITION_FANOUT];
+
/// All partitions that have been spilled and need further processing.
std::list<Partition*> spilled_partitions_;
@@ -390,6 +397,13 @@ class PartitionedAggregationNode : public ExecNode {
/// a temporary buffer.
boost::scoped_ptr<BufferedTupleStream> serialize_stream_;
+ /// Accessor for 'hash_tbls_' that verifies consistency with the partitions.
+ HashTable* ALWAYS_INLINE GetHashTable(int partition_idx) {
+ HashTable* ht = hash_tbls_[partition_idx];
+ DCHECK_EQ(ht, hash_partitions_[partition_idx]->hash_tbl.get());
+ return ht;
+ }
+
/// Materializes 'row_batch' in either grouping or non-grouping case.
Status GetNextInternal(RuntimeState* state, RowBatch* row_batch, bool* eos);
@@ -465,19 +479,31 @@ class PartitionedAggregationNode : public ExecNode {
Tuple* tuple, MemPool* pool);
/// Do the aggregation for all tuple rows in the batch when there is no grouping.
- /// The HashTableCtx argument is unused, but included so the signature matches that of
- /// ProcessBatch() for codegen. This function is replaced by codegen.
- Status ProcessBatchNoGrouping(RowBatch* batch, const HashTableCtx* ht_ctx = NULL);
+ /// This function is replaced by codegen.
+ Status ProcessBatchNoGrouping(RowBatch* batch);
/// Processes a batch of rows. This is the core function of the algorithm. We partition
/// the rows into hash_partitions_, spilling as necessary.
/// If AGGREGATED_ROWS is true, it means that the rows in the batch are already
/// pre-aggregated.
+ /// 'prefetch_mode' specifies the prefetching mode in use. If it's not PREFETCH_NONE,
+ /// hash table buckets will be prefetched based on the hash values computed. Note
+ /// that 'prefetch_mode' will be substituted with constants during codegen time.
//
- /// This function is replaced by codegen. It's inlined into ProcessBatch_true/false in
- /// the IR module. We pass in ht_ctx_.get() as an argument for performance.
+ /// This function is replaced by codegen. We pass in ht_ctx_.get() as an argument for
+ /// performance.
+ template<bool AGGREGATED_ROWS>
+ Status IR_ALWAYS_INLINE ProcessBatch(RowBatch* batch,
+ TPrefetchMode::type prefetch_mode, HashTableCtx* ht_ctx);
+
+ /// Evaluates the rows in 'batch' starting at 'start_row_idx' and stores the results in
+ /// the expression values cache in 'ht_ctx'. The number of rows evaluated depends on
+ /// the capacity of the cache. 'prefetch_mode' specifies the prefetching mode in use.
+ /// If it's not PREFETCH_NONE, hash table buckets for the computed hashes will be
+ /// prefetched. Note that codegen replaces 'prefetch_mode' with a constant.
template<bool AGGREGATED_ROWS>
- Status IR_ALWAYS_INLINE ProcessBatch(RowBatch* batch, HashTableCtx* ht_ctx);
+ void EvalAndHashPrefetchGroup(RowBatch* batch, int start_row_idx,
+ TPrefetchMode::type prefetch_mode, HashTableCtx* ht_ctx);
/// This function processes each individual row in ProcessBatch(). Must be inlined into
/// ProcessBatch for codegen to substitute function calls with codegen'd versions.
@@ -532,25 +558,28 @@ class PartitionedAggregationNode : public ExecNode {
/// store all of the rows in 'in_batch'.
/// 'needs_serialize' is an argument so that codegen can replace it with a constant,
/// rather than using the member variable 'needs_serialize_'.
+ /// 'prefetch_mode' specifies the prefetching mode in use. If it's not PREFETCH_NONE,
+ /// hash table buckets will be prefetched based on the hash values computed. Note
+ /// that 'prefetch_mode' will be substituted with constants during codegen time.
/// 'remaining_capacity' is an array with PARTITION_FANOUT entries with the number of
/// additional rows that can be added to the hash table per partition. It is updated
/// by ProcessBatchStreaming() when it inserts new rows.
/// 'ht_ctx' is passed in as a way to avoid aliasing of 'this' confusing the optimiser.
- Status ProcessBatchStreaming(bool needs_serialize, RowBatch* in_batch,
- RowBatch* out_batch, HashTableCtx* ht_ctx,
+ Status ProcessBatchStreaming(bool needs_serialize, TPrefetchMode::type prefetch_mode,
+ RowBatch* in_batch, RowBatch* out_batch, HashTableCtx* ht_ctx,
int remaining_capacity[PARTITION_FANOUT]);
- /// Tries to add intermediate to the hash table of 'partition' for streaming aggregation.
- /// The input row must have been evaluated with 'ht_ctx', with 'hash' set to the
- /// corresponding hash. If the tuple already exists in the hash table, update the tuple
- /// and return true. Otherwise try to create a new entry in the hash table, returning
- /// true if successful or false if the table is full. 'remaining_capacity' keeps track
- /// of how many more entries can be added to the hash table so we can avoid retrying
- /// inserts. It is decremented if an insert succeeds and set to zero if an insert
- /// fails. If an error occurs, returns false and sets 'status'.
+ /// Tries to add intermediate to the hash table 'hash_tbl' of 'partition' for streaming
+ /// aggregation. The input row must have been evaluated with 'ht_ctx', with 'hash' set
+ /// to the corresponding hash. If the tuple already exists in the hash table, update
+ /// the tuple and return true. Otherwise try to create a new entry in the hash table,
+ /// returning true if successful or false if the table is full. 'remaining_capacity'
+ /// keeps track of how many more entries can be added to the hash table so we can avoid
+ /// retrying inserts. It is decremented if an insert succeeds and set to zero if an
+ /// insert fails. If an error occurs, returns false and sets 'status'.
bool IR_ALWAYS_INLINE TryAddToHashTable(HashTableCtx* ht_ctx,
- Partition* partition, TupleRow* in_row, uint32_t hash, int* remaining_capacity,
- Status* status);
+ Partition* partition, HashTable* hash_tbl, TupleRow* in_row, uint32_t hash,
+ int* remaining_capacity, Status* status);
/// Initializes hash_partitions_. 'level' is the level for the partitions to create.
/// Also sets ht_ctx_'s level to 'level'.
@@ -601,18 +630,15 @@ class PartitionedAggregationNode : public ExecNode {
/// Codegen the non-streaming process row batch loop. The loop has already been
/// compiled to IR and loaded into the codegen object. UpdateAggTuple has also been
/// codegen'd to IR. This function will modify the loop subsituting the statically
- /// compiled functions with codegen'd ones.
+ /// compiled functions with codegen'd ones. 'process_batch_fn_' or
+ /// 'process_batch_no_grouping_fn_' will be updated with the codegened function
+ /// depending on whether this is a grouping or non-grouping aggregation.
/// Assumes AGGREGATED_ROWS = false.
- Status CodegenProcessBatch(llvm::Function** fn);
+ Status CodegenProcessBatch();
/// Codegen the materialization loop for streaming preaggregations.
- Status CodegenProcessBatchStreaming(llvm::Function** fn);
-
- /// Functions to instantiate templated versions of ProcessBatch().
- /// The xcompiled versions of these functions are used in CodegenProcessBatch().
- /// TODO: is there a better way to do this?
- Status ProcessBatch_false(RowBatch* batch, HashTableCtx* ht_ctx);
- Status ProcessBatch_true(RowBatch* batch, HashTableCtx* ht_ctx);
+ /// 'process_batch_streaming_fn_' will be updated with the codegened function.
+ Status CodegenProcessBatchStreaming();
/// We need two buffers per partition, one for the aggregated stream and one
/// for the unaggregated stream. We need an additional buffer to read the stream
[3/4] incubator-impala git commit: IMPALA-3168: replace HashTable
parameters with constants
Posted by ta...@apache.org.
IMPALA-3168: replace HashTable parameters with constants
This addresses the regression for small-ndv aggs resulting from
prefetching. The idea is that for small-ndv aggs prefetching
increases the # of instructions and memory references, but
doesn't provide any compensating benefit. This change replaces
constant values in the hash-table code, which reduces the
instruction count and # of memory references in aggregations.
Preliminary perf results show that a low-NDV decimal agg is
around 20% faster (2.1s -> 1.7s) and a high-NDV decimal agg
is around 7% faster (15s -> 14s). I haven't investigated
how much of the speedup is reduced codegen time.
Change-Id: I483a19662c90ca54bc21d60fd6ba97dbed93eaef
Reviewed-on: http://gerrit.cloudera.org:8080/3088
Tested-by: Internal Jenkins
Reviewed-by: Dan Hecht <dh...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/265e39f8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/265e39f8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/265e39f8
Branch: refs/heads/master
Commit: 265e39f89a914541064555b764bfd8f7747bb294
Parents: 9f4276e
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Fri May 13 15:48:59 2016 -0700
Committer: Tim Armstrong <ta...@cloudera.com>
Committed: Tue May 17 10:09:06 2016 -0700
----------------------------------------------------------------------
be/src/codegen/llvm-codegen.cc | 52 ++++++++++++++++----
be/src/codegen/llvm-codegen.h | 16 +++++++
be/src/exec/hash-table-test.cc | 3 +-
be/src/exec/hash-table.cc | 32 ++++++++++---
be/src/exec/hash-table.h | 45 ++++++++++++++----
be/src/exec/hash-table.inline.h | 60 +++++++++++++-----------
be/src/exec/partitioned-aggregation-node.cc | 26 ++++++++--
be/src/exec/partitioned-hash-join-node.cc | 37 +++++++++++++++
8 files changed, 215 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/codegen/llvm-codegen.cc
----------------------------------------------------------------------
diff --git a/be/src/codegen/llvm-codegen.cc b/be/src/codegen/llvm-codegen.cc
index 0651bf7..b7f0601 100644
--- a/be/src/codegen/llvm-codegen.cc
+++ b/be/src/codegen/llvm-codegen.cc
@@ -622,27 +622,61 @@ Function* LlvmCodeGen::FnPrototype::GeneratePrototype(
}
int LlvmCodeGen::ReplaceCallSites(Function* caller, Function* new_fn,
- const string& replacee_name) {
+ const string& target_name) {
DCHECK(!is_compiled_);
DCHECK(caller->getParent() == module_);
DCHECK(caller != NULL);
DCHECK(new_fn != NULL);
+
+ vector<CallInst*> call_sites;
+ FindCallSites(caller, target_name, &call_sites);
int replaced = 0;
+ for (CallInst* call_instr: call_sites) {
+ // Replace the called function
+ call_instr->setCalledFunction(new_fn);
+ ++replaced;
+ }
+ return replaced;
+}
+
+int LlvmCodeGen::ReplaceCallSitesWithValue(Function* caller, Value* replacement,
+ const string& target_name) {
+ DCHECK(!is_compiled_);
+ DCHECK(caller->getParent() == module_);
+ DCHECK(caller != NULL);
+ DCHECK(replacement != NULL);
+
+ vector<CallInst*> call_sites;
+ FindCallSites(caller, target_name, &call_sites);
+ int replaced = 0;
+ for (CallInst* call_instr: call_sites) {
+ call_instr->replaceAllUsesWith(replacement);
+ ++replaced;
+ }
+ return replaced;
+}
+
+int LlvmCodeGen::ReplaceCallSitesWithBoolConst(llvm::Function* caller, bool constant,
+ const string& target_name) {
+ Value* replacement = ConstantInt::get(Type::getInt1Ty(context()), constant);
+ return ReplaceCallSitesWithValue(caller, replacement, target_name);
+}
+
+void LlvmCodeGen::FindCallSites(Function* caller, const string& target_name,
+ vector<CallInst*>* results) {
for (inst_iterator iter = inst_begin(caller); iter != inst_end(caller); ++iter) {
Instruction* instr = &*iter;
- // look for call instructions
+ // Look for call instructions. Note that we'll ignore invoke and other related
+ // instructions that are not a plain function call.
if (CallInst::classof(instr)) {
CallInst* call_instr = reinterpret_cast<CallInst*>(instr);
- Function* old_fn = call_instr->getCalledFunction();
- // look for call instruction that matches the name
- if (old_fn != NULL && old_fn->getName().find(replacee_name) != string::npos) {
- // Replace the called function
- call_instr->setCalledFunction(new_fn);
- ++replaced;
+ Function* callee = call_instr->getCalledFunction();
+ // Check for substring match.
+ if (callee != NULL && callee->getName().find(target_name) != string::npos) {
+ results->push_back(call_instr);
}
}
}
- return replaced;
}
Function* LlvmCodeGen::CloneFunction(Function* fn) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/codegen/llvm-codegen.h
----------------------------------------------------------------------
diff --git a/be/src/codegen/llvm-codegen.h b/be/src/codegen/llvm-codegen.h
index d4255a0..332e599 100644
--- a/be/src/codegen/llvm-codegen.h
+++ b/be/src/codegen/llvm-codegen.h
@@ -261,6 +261,11 @@ class LlvmCodeGen {
int ReplaceCallSites(llvm::Function* caller, llvm::Function* new_fn,
const std::string& target_name);
+ /// Same as ReplaceCallSites(), except replaces the function call instructions with the
+ /// boolean value 'constant'.
+ int ReplaceCallSitesWithBoolConst(llvm::Function* caller, bool constant,
+ const std::string& target_name);
+
/// Returns a copy of fn. The copy is added to the module.
llvm::Function* CloneFunction(llvm::Function* fn);
@@ -464,6 +469,17 @@ class LlvmCodeGen {
/// Clears generated hash fns. This is only used for testing.
void ClearHashFns();
+ /// Replace calls to functions in 'caller' where the callee's name has 'target_name'
+ /// as a substring. Calls to functions are replaced with the value 'replacement'. The
+ /// return value is the number of calls replaced.
+ int ReplaceCallSitesWithValue(llvm::Function* caller, llvm::Value* replacement,
+ const std::string& target_name);
+
+ /// Finds call instructions in 'caller' where 'target_name' is a substring of the
+ /// callee's name. Found instructions are appended to the 'results' vector.
+ static void FindCallSites(llvm::Function* caller, const std::string& target_name,
+ std::vector<llvm::CallInst*>* results);
+
/// Whether InitializeLlvm() has been called.
static bool llvm_initialized_;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/exec/hash-table-test.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hash-table-test.cc b/be/src/exec/hash-table-test.cc
index 7559473..d6cd196 100644
--- a/be/src/exec/hash-table-test.cc
+++ b/be/src/exec/hash-table-test.cc
@@ -189,7 +189,7 @@ class HashTableTest : public testing::Test {
// Initial_num_buckets must be a power of two.
EXPECT_EQ(initial_num_buckets, BitUtil::RoundUpToPowerOfTwo(initial_num_buckets));
int64_t max_num_buckets = 1L << 31;
- table->reset(new HashTable(quadratic, runtime_state_, client, 1, NULL,
+ table->reset(new HashTable(quadratic, runtime_state_, client, true, 1, NULL,
max_num_buckets, initial_num_buckets));
return (*table)->Init();
}
@@ -239,6 +239,7 @@ class HashTableTest : public testing::Test {
probe_expr_ctxs_, true /* stores_nulls_ */,
vector<bool>(build_expr_ctxs_.size(), false), 1, 0, 1, &tracker_, &ht_ctx);
EXPECT_OK(status);
+
for (int i = 0; i < 2; ++i) {
if (!ht_ctx->EvalAndHashBuild(build_rows[i])) continue;
BufferedTupleStream::RowIdx dummy_row_idx;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/exec/hash-table.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hash-table.cc b/be/src/exec/hash-table.cc
index 719b600..7ff2886 100644
--- a/be/src/exec/hash-table.cc
+++ b/be/src/exec/hash-table.cc
@@ -84,8 +84,8 @@ static const int NUM_SMALL_DATA_PAGES = sizeof(INITIAL_DATA_PAGE_SIZES) / sizeof
HashTableCtx::HashTableCtx(const std::vector<ExprContext*>& build_expr_ctxs,
const std::vector<ExprContext*>& probe_expr_ctxs, bool stores_nulls,
- const std::vector<bool>& finds_nulls, int32_t initial_seed, int max_levels,
- MemTracker* tracker)
+ const std::vector<bool>& finds_nulls, int32_t initial_seed,
+ int max_levels, MemTracker* tracker)
: build_expr_ctxs_(build_expr_ctxs),
probe_expr_ctxs_(probe_expr_ctxs),
stores_nulls_(stores_nulls),
@@ -341,20 +341,21 @@ void HashTableCtx::ExprValuesCache::ResetForRead() {
const double HashTable::MAX_FILL_FACTOR = 0.75f;
HashTable* HashTable::Create(RuntimeState* state,
- BufferedBlockMgr::Client* client, int num_build_tuples,
+ BufferedBlockMgr::Client* client, bool stores_duplicates, int num_build_tuples,
BufferedTupleStream* tuple_stream, int64_t max_num_buckets,
int64_t initial_num_buckets) {
- return new HashTable(FLAGS_enable_quadratic_probing, state, client,
+ return new HashTable(FLAGS_enable_quadratic_probing, state, client, stores_duplicates,
num_build_tuples, tuple_stream, max_num_buckets, initial_num_buckets);
}
HashTable::HashTable(bool quadratic_probing, RuntimeState* state,
- BufferedBlockMgr::Client* client, int num_build_tuples, BufferedTupleStream* stream,
- int64_t max_num_buckets, int64_t num_buckets)
+ BufferedBlockMgr::Client* client, bool stores_duplicates, int num_build_tuples,
+ BufferedTupleStream* stream, int64_t max_num_buckets, int64_t num_buckets)
: state_(state),
block_mgr_client_(client),
tuple_stream_(stream),
stores_tuples_(num_build_tuples == 1),
+ stores_duplicates_(stores_duplicates),
quadratic_probing_(quadratic_probing),
total_data_page_size_(0),
next_node_(NULL),
@@ -1154,3 +1155,22 @@ Status HashTableCtx::CodegenEquals(RuntimeState* state, bool force_null_equality
}
return Status::OK();
}
+
+Status HashTableCtx::ReplaceHashTableConstants(RuntimeState* state,
+ bool stores_duplicates, int num_build_tuples, Function* fn,
+ HashTableReplacedConstants* replacement_counts) {
+ LlvmCodeGen* codegen;
+ RETURN_IF_ERROR(state->GetCodegen(&codegen));
+
+ replacement_counts->stores_nulls = codegen->ReplaceCallSitesWithBoolConst(
+ fn, stores_nulls(), "stores_nulls");
+ replacement_counts->finds_some_nulls = codegen->ReplaceCallSitesWithBoolConst(
+ fn, finds_some_nulls(), "finds_some_nulls");
+ replacement_counts->stores_tuples = codegen->ReplaceCallSitesWithBoolConst(
+ fn, num_build_tuples == 1, "stores_tuples");
+ replacement_counts->stores_duplicates = codegen->ReplaceCallSitesWithBoolConst(
+ fn, stores_duplicates, "stores_duplicates");
+ replacement_counts->quadratic_probing = codegen->ReplaceCallSitesWithBoolConst(
+ fn, FLAGS_enable_quadratic_probing, "quadratic_probing");
+ return Status::OK();
+}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/exec/hash-table.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hash-table.h b/be/src/exec/hash-table.h
index 6fc4169..614917a 100644
--- a/be/src/exec/hash-table.h
+++ b/be/src/exec/hash-table.h
@@ -123,8 +123,8 @@ class HashTableCtx {
/// space by not storing some rows we know will never match.
HashTableCtx(const std::vector<ExprContext*>& build_expr_ctxs,
const std::vector<ExprContext*>& probe_expr_ctxs, bool stores_nulls,
- const std::vector<bool>& finds_nulls, int32_t initial_seed, int max_levels,
- MemTracker* tracker);
+ const std::vector<bool>& finds_nulls, int32_t initial_seed,
+ int max_levels, MemTracker* tracker);
/// Create a hash table context with the specified parameters, invoke Init() to
/// initialize the new hash table context and return it in 'ht_ctx'. Please see header
@@ -190,6 +190,22 @@ class HashTableCtx {
/// supports it (see hash-util.h).
Status CodegenHashCurrentRow(RuntimeState* state, bool use_murmur, llvm::Function** fn);
+ /// Struct that returns the number of constants replaced by ReplaceConstants().
+ struct HashTableReplacedConstants {
+ int stores_nulls;
+ int finds_some_nulls;
+ int stores_tuples;
+ int stores_duplicates;
+ int quadratic_probing;
+ };
+
+ /// Replace hash table parameters with constants in 'fn'. Updates 'replacement_counts'
+ /// with the number of replacements made. 'num_build_tuples' and 'stores_duplicates'
+ /// correspond to HashTable parameters with the same name.
+ Status ReplaceHashTableConstants(RuntimeState* state, bool stores_duplicates,
+ int num_build_tuples, llvm::Function* fn,
+ HashTableReplacedConstants* replacement_counts);
+
static const char* LLVM_CLASS_NAME;
/// To enable prefetching, the hash table building and probing are pipelined by the
@@ -417,13 +433,15 @@ class HashTableCtx {
/// Cross-compiled function to access member variables used in CodegenHashCurrentRow().
uint32_t GetHashSeed() const;
+ /// Functions to be replaced by codegen to specialize the hash table.
+ bool IR_NO_INLINE stores_nulls() const { return stores_nulls_; }
+ bool IR_NO_INLINE finds_some_nulls() const { return finds_some_nulls_; }
+
const std::vector<ExprContext*>& build_expr_ctxs_;
const std::vector<ExprContext*>& probe_expr_ctxs_;
/// Constants on how the hash table should behave. Joins and aggs have slightly
/// different behavior.
- /// TODO: these constants are an ideal candidate to be removed with codegen.
- /// TODO: ..or with template-ization
const bool stores_nulls_;
const std::vector<bool> finds_nulls_;
@@ -512,6 +530,8 @@ class HashTable {
/// Returns a newly allocated HashTable. The probing algorithm is set by the
/// FLAG_enable_quadratic_probing.
/// - client: block mgr client to allocate data pages from.
+ /// - stores_duplicates: true if rows with duplicate keys may be inserted into the
+ /// hash table.
/// - num_build_tuples: number of Tuples in the build tuple row.
/// - tuple_stream: the tuple stream which contains the tuple rows index by the
/// hash table. Can be NULL if the rows contain only a single tuple, in which
@@ -522,8 +542,8 @@ class HashTable {
/// - initial_num_buckets: number of buckets that the hash table should be initialized
/// with.
static HashTable* Create(RuntimeState* state, BufferedBlockMgr::Client* client,
- int num_build_tuples, BufferedTupleStream* tuple_stream, int64_t max_num_buckets,
- int64_t initial_num_buckets);
+ bool stores_duplicates, int num_build_tuples, BufferedTupleStream* tuple_stream,
+ int64_t max_num_buckets, int64_t initial_num_buckets);
/// Allocates the initial bucket structure. Returns false if OOM.
bool Init();
@@ -730,7 +750,7 @@ class HashTable {
/// - quadratic_probing: set to true when the probing algorithm is quadratic, as
/// opposed to linear.
HashTable(bool quadratic_probing, RuntimeState* state, BufferedBlockMgr::Client* client,
- int num_build_tuples, BufferedTupleStream* tuple_stream,
+ bool stores_duplicates, int num_build_tuples, BufferedTupleStream* tuple_stream,
int64_t max_num_buckets, int64_t initial_num_buckets);
/// Performs the probing operation according to the probing algorithm (linear or
@@ -802,6 +822,11 @@ class HashTable {
/// Grow the node array. Returns false on OOM.
bool GrowNodeArray();
+ /// Functions to be replaced by codegen to specialize the hash table.
+ bool IR_NO_INLINE stores_tuples() const { return stores_tuples_; }
+ bool IR_NO_INLINE stores_duplicates() const { return stores_duplicates_; }
+ bool IR_NO_INLINE quadratic_probing() const { return quadratic_probing_; }
+
/// Load factor that will trigger growing the hash table on insert. This is
/// defined as the number of non-empty buckets / total_buckets
static const double MAX_FILL_FACTOR;
@@ -818,10 +843,11 @@ class HashTable {
/// Constants on how the hash table should behave. Joins and aggs have slightly
/// different behavior.
- /// TODO: these constants are an ideal candidate to be removed with codegen.
- /// TODO: ..or with template-ization
const bool stores_tuples_;
+ /// True if duplicates may be inserted into hash table.
+ const bool stores_duplicates_;
+
/// Quadratic probing enabled (as opposed to linear).
const bool quadratic_probing_;
@@ -857,7 +883,6 @@ class HashTable {
int64_t num_buckets_with_duplicates_;
/// Number of build tuples, used for constructing temp row* for probes.
- /// TODO: We should remove it.
const int num_build_tuples_;
/// Flag used to disable spilling hash tables that already had matches in case of
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/exec/hash-table.inline.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hash-table.inline.h b/be/src/exec/hash-table.inline.h
index 6d33869..7499da3 100644
--- a/be/src/exec/hash-table.inline.h
+++ b/be/src/exec/hash-table.inline.h
@@ -25,14 +25,14 @@ namespace impala {
inline bool HashTableCtx::EvalAndHashBuild(TupleRow* row) {
bool has_null = EvalBuildRow(row);
- if (!stores_nulls_ && has_null) return false;
+ if (!stores_nulls() && has_null) return false;
expr_values_cache_.SetExprValuesHash(HashCurrentRow());
return true;
}
inline bool HashTableCtx::EvalAndHashProbe(TupleRow* row) {
bool has_null = EvalProbeRow(row);
- if (has_null && !(stores_nulls_ && finds_some_nulls_)) return false;
+ if (has_null && !(stores_nulls() && finds_some_nulls())) return false;
expr_values_cache_.SetExprValuesHash(HashCurrentRow());
return true;
}
@@ -72,7 +72,7 @@ inline int64_t HashTable::Probe(Bucket* buckets, int64_t num_buckets,
// Move to the next bucket.
++step;
++travel_length_;
- if (quadratic_probing_) {
+ if (quadratic_probing()) {
// The i-th probe location is idx = (hash + (step * (step + 1)) / 2) mod num_buckets.
// This gives num_buckets unique idxs (between 0 and N-1) when num_buckets is a power
// of 2.
@@ -82,7 +82,7 @@ inline int64_t HashTable::Probe(Bucket* buckets, int64_t num_buckets,
}
} while (LIKELY(step < num_buckets));
DCHECK_EQ(num_filled_buckets_, num_buckets) << "Probing of a non-full table "
- << "failed: " << quadratic_probing_ << " " << hash;
+ << "failed: " << quadratic_probing() << " " << hash;
return Iterator::BUCKET_NOT_FOUND;
}
@@ -108,7 +108,7 @@ inline bool HashTable::Insert(HashTableCtx* ht_ctx,
HtData* htdata = InsertInternal(ht_ctx);
// If successful insert, update the contents of the newly inserted entry with 'idx'.
if (LIKELY(htdata != NULL)) {
- if (stores_tuples_) {
+ if (stores_tuples()) {
htdata->tuple = row->GetTuple(0);
} else {
htdata->idx = idx;
@@ -136,7 +136,7 @@ inline HashTable::Iterator HashTable::FindProbeRow(HashTableCtx* ht_ctx) {
int64_t bucket_idx = Probe<false>(buckets_, num_buckets_, ht_ctx, hash, &found);
if (found) {
return Iterator(this, ht_ctx->scratch_row(), bucket_idx,
- buckets_[bucket_idx].bucketData.duplicates);
+ stores_duplicates() ? buckets_[bucket_idx].bucketData.duplicates : NULL);
}
return End();
}
@@ -147,8 +147,10 @@ inline HashTable::Iterator HashTable::FindBuildRowBucket(
++num_probes_;
uint32_t hash = ht_ctx->expr_values_cache()->ExprValuesHash();
int64_t bucket_idx = Probe<true>(buckets_, num_buckets_, ht_ctx, hash, found);
- DuplicateNode* duplicates = LIKELY(bucket_idx != Iterator::BUCKET_NOT_FOUND) ?
- buckets_[bucket_idx].bucketData.duplicates : NULL;
+ DuplicateNode* duplicates = NULL;
+ if (stores_duplicates() && LIKELY(bucket_idx != Iterator::BUCKET_NOT_FOUND)) {
+ duplicates = buckets_[bucket_idx].bucketData.duplicates;
+ }
return Iterator(this, ht_ctx->scratch_row(), bucket_idx, duplicates);
}
@@ -167,8 +169,8 @@ inline HashTable::Iterator HashTable::FirstUnmatched(HashTableCtx* ctx) {
// Check whether the bucket, or its first duplicate node, is matched. If it is not
// matched, then return. Otherwise, move to the first unmatched entry (node or bucket).
Bucket* bucket = &buckets_[bucket_idx];
- if ((!bucket->hasDuplicates && bucket->matched) ||
- (bucket->hasDuplicates && node->matched)) {
+ bool has_duplicates = stores_duplicates() && bucket->hasDuplicates;
+ if ((!has_duplicates && bucket->matched) || (has_duplicates && node->matched)) {
it.NextUnmatched();
}
return it;
@@ -178,7 +180,7 @@ inline void HashTable::NextFilledBucket(int64_t* bucket_idx, DuplicateNode** nod
++*bucket_idx;
for (; *bucket_idx < num_buckets_; ++*bucket_idx) {
if (buckets_[*bucket_idx].filled) {
- *node = buckets_[*bucket_idx].bucketData.duplicates;
+ *node = stores_duplicates() ? buckets_[*bucket_idx].bucketData.duplicates : NULL;
return;
}
}
@@ -212,6 +214,7 @@ inline HashTable::DuplicateNode* HashTable::InsertDuplicateNode(int64_t bucket_i
DCHECK_LT(bucket_idx, num_buckets_);
Bucket* bucket = &buckets_[bucket_idx];
DCHECK(bucket->filled);
+ DCHECK(stores_duplicates());
// Allocate one duplicate node for the new data and one for the preexisting data,
// if needed.
while (node_remaining_current_page_ < 1 + !bucket->hasDuplicates) {
@@ -234,18 +237,19 @@ inline HashTable::DuplicateNode* HashTable::InsertDuplicateNode(int64_t bucket_i
return AppendNextNode(bucket);
}
-inline TupleRow* HashTable::GetRow(HtData& htdata, TupleRow* row) const {
- if (stores_tuples_) {
+inline TupleRow* IR_ALWAYS_INLINE HashTable::GetRow(HtData& htdata, TupleRow* row) const {
+ if (stores_tuples()) {
return reinterpret_cast<TupleRow*>(&htdata.tuple);
} else {
+ // TODO: GetTupleRow() has interpreted code that iterates over the row's descriptor.
tuple_stream_->GetTupleRow(htdata.idx, row);
return row;
}
}
-inline TupleRow* HashTable::GetRow(Bucket* bucket, TupleRow* row) const {
+inline TupleRow* IR_ALWAYS_INLINE HashTable::GetRow(Bucket* bucket, TupleRow* row) const {
DCHECK(bucket != NULL);
- if (UNLIKELY(bucket->hasDuplicates)) {
+ if (UNLIKELY(stores_duplicates() && bucket->hasDuplicates)) {
DuplicateNode* duplicate = bucket->bucketData.duplicates;
DCHECK(duplicate != NULL);
return GetRow(duplicate->htdata, row);
@@ -254,12 +258,12 @@ inline TupleRow* HashTable::GetRow(Bucket* bucket, TupleRow* row) const {
}
}
-inline TupleRow* HashTable::Iterator::GetRow() const {
+inline TupleRow* IR_ALWAYS_INLINE HashTable::Iterator::GetRow() const {
DCHECK(!AtEnd());
DCHECK(table_ != NULL);
DCHECK(scratch_row_ != NULL);
Bucket* bucket = &table_->buckets_[bucket_idx_];
- if (UNLIKELY(bucket->hasDuplicates)) {
+ if (UNLIKELY(table_->stores_duplicates() && bucket->hasDuplicates)) {
DCHECK(node_ != NULL);
return table_->GetRow(node_->htdata, scratch_row_);
} else {
@@ -267,12 +271,12 @@ inline TupleRow* HashTable::Iterator::GetRow() const {
}
}
-inline Tuple* HashTable::Iterator::GetTuple() const {
+inline Tuple* IR_ALWAYS_INLINE HashTable::Iterator::GetTuple() const {
DCHECK(!AtEnd());
- DCHECK(table_->stores_tuples_);
+ DCHECK(table_->stores_tuples());
Bucket* bucket = &table_->buckets_[bucket_idx_];
// TODO: To avoid the hasDuplicates check, store the HtData* in the Iterator.
- if (UNLIKELY(bucket->hasDuplicates)) {
+ if (UNLIKELY(table_->stores_duplicates() && bucket->hasDuplicates)) {
DCHECK(node_ != NULL);
return node_->htdata.tuple;
} else {
@@ -282,7 +286,7 @@ inline Tuple* HashTable::Iterator::GetTuple() const {
inline void HashTable::Iterator::SetTuple(Tuple* tuple, uint32_t hash) {
DCHECK(!AtEnd());
- DCHECK(table_->stores_tuples_);
+ DCHECK(table_->stores_tuples());
table_->PrepareBucketForInsert(bucket_idx_, hash);
table_->buckets_[bucket_idx_].bucketData.htdata.tuple = tuple;
}
@@ -290,7 +294,7 @@ inline void HashTable::Iterator::SetTuple(Tuple* tuple, uint32_t hash) {
inline void HashTable::Iterator::SetMatched() {
DCHECK(!AtEnd());
Bucket* bucket = &table_->buckets_[bucket_idx_];
- if (bucket->hasDuplicates) {
+ if (table_->stores_duplicates() && bucket->hasDuplicates) {
node_->matched = true;
} else {
bucket->matched = true;
@@ -303,7 +307,7 @@ inline void HashTable::Iterator::SetMatched() {
inline bool HashTable::Iterator::IsMatched() const {
DCHECK(!AtEnd());
Bucket* bucket = &table_->buckets_[bucket_idx_];
- if (bucket->hasDuplicates) {
+ if (table_->stores_duplicates() && bucket->hasDuplicates) {
return node_->matched;
}
return bucket->matched;
@@ -326,7 +330,8 @@ inline void HashTable::Iterator::PrefetchBucket() {
inline void HashTable::Iterator::Next() {
DCHECK(!AtEnd());
- if (table_->buckets_[bucket_idx_].hasDuplicates && node_->next != NULL) {
+ if (table_->stores_duplicates() && table_->buckets_[bucket_idx_].hasDuplicates &&
+ node_->next != NULL) {
node_ = node_->next;
} else {
table_->NextFilledBucket(&bucket_idx_, &node_);
@@ -335,7 +340,8 @@ inline void HashTable::Iterator::Next() {
inline void HashTable::Iterator::NextDuplicate() {
DCHECK(!AtEnd());
- if (table_->buckets_[bucket_idx_].hasDuplicates && node_->next != NULL) {
+ if (table_->stores_duplicates() && table_->buckets_[bucket_idx_].hasDuplicates &&
+ node_->next != NULL) {
node_ = node_->next;
} else {
bucket_idx_ = BUCKET_NOT_FOUND;
@@ -347,7 +353,7 @@ inline void HashTable::Iterator::NextUnmatched() {
DCHECK(!AtEnd());
Bucket* bucket = &table_->buckets_[bucket_idx_];
// Check if there is any remaining unmatched duplicate node in the current bucket.
- if (bucket->hasDuplicates) {
+ if (table_->stores_duplicates() && bucket->hasDuplicates) {
while (node_->next != NULL) {
node_ = node_->next;
if (!node_->matched) return;
@@ -358,7 +364,7 @@ inline void HashTable::Iterator::NextUnmatched() {
table_->NextFilledBucket(&bucket_idx_, &node_);
while (bucket_idx_ != Iterator::BUCKET_NOT_FOUND) {
bucket = &table_->buckets_[bucket_idx_];
- if (!bucket->hasDuplicates) {
+ if (!table_->stores_duplicates() || !bucket->hasDuplicates) {
if (!bucket->matched) return;
} else {
while (node_->matched && node_->next != NULL) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/exec/partitioned-aggregation-node.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/partitioned-aggregation-node.cc b/be/src/exec/partitioned-aggregation-node.cc
index 0a6a811..c2c884a 100644
--- a/be/src/exec/partitioned-aggregation-node.cc
+++ b/be/src/exec/partitioned-aggregation-node.cc
@@ -763,9 +763,9 @@ bool PartitionedAggregationNode::Partition::InitHashTable() {
// though. Always start with small buffers.
// TODO: How many buckets? We currently use a default value, 1024.
static const int64_t PAGG_DEFAULT_HASH_TABLE_SZ = 1024;
-
- hash_tbl.reset(HashTable::Create(parent->state_, parent->block_mgr_client_, 1,
- NULL, 1L << (32 - NUM_PARTITIONING_BITS), PAGG_DEFAULT_HASH_TABLE_SZ));
+ hash_tbl.reset(HashTable::Create(parent->state_, parent->block_mgr_client_,
+ false, 1, NULL, 1L << (32 - NUM_PARTITIONING_BITS),
+ PAGG_DEFAULT_HASH_TABLE_SZ));
return hash_tbl->Init();
}
@@ -1841,6 +1841,16 @@ Status PartitionedAggregationNode::CodegenProcessBatch() {
replaced = codegen->ReplaceCallSites(process_batch_fn, build_equals_fn, "Equals");
DCHECK_EQ(replaced, 1);
+
+ HashTableCtx::HashTableReplacedConstants replaced_constants;
+ const bool stores_duplicates = false;
+ RETURN_IF_ERROR(ht_ctx_->ReplaceHashTableConstants(state_, stores_duplicates, 1,
+ process_batch_fn, &replaced_constants));
+ DCHECK_GE(replaced_constants.stores_nulls, 1);
+ DCHECK_GE(replaced_constants.finds_some_nulls, 1);
+ DCHECK_GE(replaced_constants.stores_duplicates, 1);
+ DCHECK_GE(replaced_constants.stores_tuples, 1);
+ DCHECK_GE(replaced_constants.quadratic_probing, 1);
}
replaced = codegen->ReplaceCallSites(process_batch_fn, update_tuple_fn, "UpdateTuple");
@@ -1910,6 +1920,16 @@ Status PartitionedAggregationNode::CodegenProcessBatchStreaming() {
replaced = codegen->ReplaceCallSites(process_batch_streaming_fn, equals_fn, "Equals");
DCHECK_EQ(replaced, 1);
+ HashTableCtx::HashTableReplacedConstants replaced_constants;
+ const bool stores_duplicates = false;
+ RETURN_IF_ERROR(ht_ctx_->ReplaceHashTableConstants(state_, stores_duplicates, 1,
+ process_batch_streaming_fn, &replaced_constants));
+ DCHECK_GE(replaced_constants.stores_nulls, 1);
+ DCHECK_GE(replaced_constants.finds_some_nulls, 1);
+ DCHECK_GE(replaced_constants.stores_duplicates, 1);
+ DCHECK_GE(replaced_constants.stores_tuples, 1);
+ DCHECK_GE(replaced_constants.quadratic_probing, 1);
+
DCHECK(process_batch_streaming_fn != NULL);
process_batch_streaming_fn = codegen->FinalizeFunction(process_batch_streaming_fn);
if (process_batch_streaming_fn == NULL) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/265e39f8/be/src/exec/partitioned-hash-join-node.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/partitioned-hash-join-node.cc b/be/src/exec/partitioned-hash-join-node.cc
index 1888e06..6f29008 100644
--- a/be/src/exec/partitioned-hash-join-node.cc
+++ b/be/src/exec/partitioned-hash-join-node.cc
@@ -455,6 +455,7 @@ Status PartitionedHashJoinNode::Partition::BuildHashTable(RuntimeState* state,
int64_t estimated_num_buckets = build_rows()->RowConsumesMemory() ?
HashTable::EstimateNumBuckets(build_rows()->num_rows()) : state->batch_size() * 2;
hash_tbl_.reset(HashTable::Create(state, parent_->block_mgr_client_,
+ true /* store_duplicates */,
parent_->child(1)->row_desc().tuple_descriptors().size(), build_rows(),
1 << (32 - NUM_PARTITIONING_BITS), estimated_num_buckets));
if (!hash_tbl_->Init()) goto not_built;
@@ -1660,6 +1661,18 @@ Status PartitionedHashJoinNode::CodegenProcessBuildBatch(RuntimeState* state,
"EvalBuildRow");
DCHECK_EQ(replaced, 1);
+ // Replace some hash table parameters with constants.
+ HashTableCtx::HashTableReplacedConstants replaced_constants;
+ const bool stores_duplicates = true;
+ const int num_build_tuples = child(1)->row_desc().tuple_descriptors().size();
+ RETURN_IF_ERROR(ht_ctx_->ReplaceHashTableConstants(state, stores_duplicates,
+ num_build_tuples, process_build_batch_fn, &replaced_constants));
+ DCHECK_GE(replaced_constants.stores_nulls, 1);
+ DCHECK_EQ(replaced_constants.finds_some_nulls, 0);
+ DCHECK_EQ(replaced_constants.stores_duplicates, 0);
+ DCHECK_EQ(replaced_constants.stores_tuples, 0);
+ DCHECK_EQ(replaced_constants.quadratic_probing, 0);
+
Function* process_build_batch_fn_level0 =
codegen->CloneFunction(process_build_batch_fn);
@@ -1836,6 +1849,18 @@ Status PartitionedHashJoinNode::CodegenProcessProbeBatch(
// TODO: switch statement
DCHECK(replaced == 1 || replaced == 2 || replaced == 3 || replaced == 4) << replaced;
+ // Replace hash-table parameters with constants.
+ HashTableCtx::HashTableReplacedConstants replaced_constants;
+ const bool stores_duplicates = true;
+ const int num_build_tuples = child(1)->row_desc().tuple_descriptors().size();
+ RETURN_IF_ERROR(ht_ctx_->ReplaceHashTableConstants(state, stores_duplicates,
+ num_build_tuples, process_probe_batch_fn, &replaced_constants));
+ DCHECK_GE(replaced_constants.stores_nulls, 1);
+ DCHECK_GE(replaced_constants.finds_some_nulls, 1);
+ DCHECK_GE(replaced_constants.stores_duplicates, 1);
+ DCHECK_GE(replaced_constants.stores_tuples, 1);
+ DCHECK_GE(replaced_constants.quadratic_probing, 1);
+
Function* process_probe_batch_fn_level0 =
codegen->CloneFunction(process_probe_batch_fn);
@@ -1895,6 +1920,18 @@ Status PartitionedHashJoinNode::CodegenInsertBatch(RuntimeState* state,
replaced = codegen->ReplaceCallSites(insert_batch_fn, build_equals_fn, "Equals");
DCHECK_EQ(replaced, 1);
+ // Replace hash-table parameters with constants.
+ HashTableCtx::HashTableReplacedConstants replaced_constants;
+ const bool stores_duplicates = true;
+ const int num_build_tuples = child(1)->row_desc().tuple_descriptors().size();
+ RETURN_IF_ERROR(ht_ctx_->ReplaceHashTableConstants(state, stores_duplicates,
+ num_build_tuples, insert_batch_fn, &replaced_constants));
+ DCHECK_GE(replaced_constants.stores_nulls, 1);
+ DCHECK_EQ(replaced_constants.finds_some_nulls, 0);
+ DCHECK_GE(replaced_constants.stores_duplicates, 1);
+ DCHECK_GE(replaced_constants.stores_tuples, 1);
+ DCHECK_GE(replaced_constants.quadratic_probing, 1);
+
Function* insert_batch_fn_level0 = codegen->CloneFunction(insert_batch_fn);
// Use codegen'd hash functions