You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by sa...@apache.org on 2016/11/22 03:05:53 UTC
[1/6] incubator-impala git commit: IMPALA-4497: Fix Kudu client crash
w/ SASL initialization
Repository: incubator-impala
Updated Branches:
refs/heads/master bb36433b1 -> 8ea21d099
IMPALA-4497: Fix Kudu client crash w/ SASL initialization
Kudu commit cd40d1c5126d0fd541d95370cdb2998b60f2f82c added a
call to sasl_set_mutex, which ends up in the client process
(Impala). That conflicts with Impala's SASL initialization code
resulting in a crash when the Kudu client is used to
open a connection to Kudu.
Kudu commit f2aeba6c059ea61f9cf8984d3e84d6c27b64d463
provided a method to disable the Kudu client's SASL
initialization to avoid this. This commit calls this method
on process startup.
Testing on kerberized cluster which was previously crashing.
Change-Id: Ic465c31406a6df2753f6cc7481b6b391ed5e671b
Reviewed-on: http://gerrit.cloudera.org:8080/5139
Reviewed-by: Matthew Jacobs <mj...@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/de88f0c4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/de88f0c4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/de88f0c4
Branch: refs/heads/master
Commit: de88f0c4af3a07ae6bd6b8c94edcb8748468f522
Parents: bb36433
Author: Matthew Jacobs <mj...@cloudera.com>
Authored: Fri Nov 18 07:15:58 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Sat Nov 19 09:55:51 2016 +0000
----------------------------------------------------------------------
be/src/rpc/authentication.cc | 8 ++++++++
1 file changed, 8 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/de88f0c4/be/src/rpc/authentication.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/authentication.cc b/be/src/rpc/authentication.cc
index 09ea5e4..cf6432a 100644
--- a/be/src/rpc/authentication.cc
+++ b/be/src/rpc/authentication.cc
@@ -36,6 +36,7 @@
#include <ldap.h>
+#include "exec/kudu-util.h"
#include "rpc/auth-provider.h"
#include "rpc/thrift-server.h"
#include "transport/TSaslClientTransport.h"
@@ -638,6 +639,13 @@ Status InitAuth(const string& appname) {
return Status(err_msg.str());
}
+ // Kudu client shouldn't attempt to initialize SASL which would conflict with
+ // Impala's SASL initialization. This must be called before any KuduClients are
+ // created to ensure that Kudu doesn't init SASL first, and this returns an error if
+ // Kudu has already initialized SASL.
+ KUDU_RETURN_IF_ERROR(kudu::client::DisableSaslInitialization(),
+ "Unable to disable Kudu SASL initialization.");
+
// Add our auxprop plugin, which gives us a hook before authentication
int rc = sasl_auxprop_add_plugin(IMPALA_AUXPROP_PLUGIN.c_str(), &ImpalaAuxpropInit);
if (rc != SASL_OK) {
[2/6] incubator-impala git commit: IMPALA-4440: lineage timestamps
can go backwards across daylight savings transitions
Posted by sa...@apache.org.
IMPALA-4440: lineage timestamps can go backwards across daylight savings transitions
Using TimestampValue (or equivalent string representation) for
timestamps that require a point in time doesn't work because the same
time can represent multiple point in times. For example, the timestamp:
'2016-11-13 01:01 AM' occurred twice last weekend.
Instead, we should use unix time directly rather than trying to derive
unix time from a (timezone-less) timestamp.
Note that there are other questionable uses of TimestampValue for
internal Impala service stuff, but I want to fix them separately as they
are not as important and fixing does add some risk.
While I'm here, remove a template TimestampValue constructor that was
unused and is confusing.
We don't have any end-to-end tests that exercise column lineage, so add
a simple custom cluster test that enables lineage and verifes the start
and end unix times are within appropriate bounds. The other column
lineage graph fields are at least tested via planner tests.
Automated regression testing for the specifc daylight savings issue is
difficult as we'd have to cross the daylight savings boundary at just
the right time during query execution in order to reproduce
reliably. But open to ideas.
Testing:
- loop the new test overnight without any failures.
- exhaustive run.
Change-Id: I34e435fc3511e65bc62906205cb558f2c116a8a9
Reviewed-on: http://gerrit.cloudera.org:8080/5129
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/035b775a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/035b775a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/035b775a
Branch: refs/heads/master
Commit: 035b775a6dca0609ffec049f765b311d4eda6ebd
Parents: de88f0c
Author: Dan Hecht <dh...@cloudera.com>
Authored: Thu Nov 17 10:06:28 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Mon Nov 21 22:18:37 2016 +0000
----------------------------------------------------------------------
be/src/runtime/timestamp-value.h | 18 -----
be/src/service/impala-server.cc | 8 +-
common/thrift/ImpalaInternalService.thrift | 3 +
.../impala/analysis/ColumnLineageGraph.java | 9 +--
.../org/apache/impala/testutil/TestUtils.java | 1 +
tests/custom_cluster/test_lineage.py | 77 ++++++++++++++++++++
6 files changed, 86 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/035b775a/be/src/runtime/timestamp-value.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/timestamp-value.h b/be/src/runtime/timestamp-value.h
index ab3f081..e722375 100644
--- a/be/src/runtime/timestamp-value.h
+++ b/be/src/runtime/timestamp-value.h
@@ -84,11 +84,6 @@ class TimestampValue {
/// Unix time (seconds since 1970-01-01 UTC by definition) constructors.
/// Conversion to local time will be done if
/// FLAGS_use_local_tz_for_unix_timestamp_conversions is true.
- template <typename Number>
- explicit TimestampValue(Number unix_time) {
- *this = UnixTimeToPtime(unix_time);
- }
-
TimestampValue(int64_t unix_time, int64_t nanos) {
boost::posix_time::ptime temp = UnixTimeToPtime(unix_time);
temp += boost::posix_time::nanoseconds(nanos);
@@ -171,19 +166,6 @@ class TimestampValue {
return true;
}
- /// Converts to Unix time (seconds since the Unix epoch) in UTC corresponding to this
- /// Timestamp instance.
- /// Returns false if the conversion failed (utc_time will be undefined), otherwise
- /// true.
- bool ToUnixTimeInUTC(time_t* utc_time) const {
- DCHECK(utc_time != NULL);
- if (UNLIKELY(!HasDateAndTime())) return false;
- const boost::posix_time::ptime temp(date_, time_);
- tm temp_tm = boost::posix_time::to_tm(temp);
- *utc_time = mktime(&temp_tm);
- return true;
- }
-
/// Converts to Unix time with fractional seconds.
/// Returns false if the conversion failed (unix_time will be undefined), otherwise
/// true.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/035b775a/be/src/service/impala-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index c210e28..49e81c8 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -375,10 +375,9 @@ Status ImpalaServer::LogLineageRecord(const QueryExecState& query_exec_state) {
} else {
return Status::OK();
}
- // Set the query end time in TLineageGraph
- time_t utc_end_time;
- query_exec_state.end_time().ToUnixTimeInUTC(&utc_end_time);
- lineage_graph.__set_ended(utc_end_time);
+ // Set the query end time in TLineageGraph. Must use UNIX time directly rather than
+ // e.g. converting from query_exec_state.end_time() (IMPALA-4440).
+ lineage_graph.__set_ended(UnixMillis() / 1000);
string lineage_record;
LineageUtil::TLineageToJSON(lineage_graph, &lineage_record);
const Status& status = lineage_logger_->AppendEntry(lineage_record);
@@ -836,6 +835,7 @@ Status ImpalaServer::ExecuteInternal(
void ImpalaServer::PrepareQueryContext(TQueryCtx* query_ctx) {
query_ctx->__set_pid(getpid());
query_ctx->__set_now_string(TimestampValue::LocalTime().DebugString());
+ query_ctx->__set_start_unix_millis(UnixMillis());
query_ctx->__set_coord_address(MakeNetworkAddress(FLAGS_hostname, FLAGS_be_port));
// Creating a random_generator every time is not free, but
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/035b775a/common/thrift/ImpalaInternalService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index e510331..25af712 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -311,6 +311,9 @@ struct TQueryCtx {
// Contains only the union of those descriptors referenced by list of fragments destined
// for a single host. Optional for frontend tests.
12: optional Descriptors.TDescriptorTable desc_tbl
+
+ // Milliseconds since UNIX epoch at the start of query execution.
+ 13: required i64 start_unix_millis
}
// Context to collect information, which is shared among all instances of that plan
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/035b775a/fe/src/main/java/org/apache/impala/analysis/ColumnLineageGraph.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/ColumnLineageGraph.java b/fe/src/main/java/org/apache/impala/analysis/ColumnLineageGraph.java
index 131d75a..d84cccc 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ColumnLineageGraph.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ColumnLineageGraph.java
@@ -375,14 +375,7 @@ public class ColumnLineageGraph {
queryStr_ = queryCtx.request.stmt;
}
Preconditions.checkNotNull(queryStr_);
- SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
- try {
- timestamp_ = df.parse(queryCtx.now_string).getTime() / 1000;
- } catch (java.text.ParseException e) {
- LOG.error("Error parsing timestamp value: " + queryCtx.now_string +
- " " + e.getMessage());
- timestamp_ = new Date().getTime() / 1000;
- }
+ timestamp_ = queryCtx.start_unix_millis / 1000;
descTbl_ = analyzer.getDescTbl();
user_ = analyzer.getUser().getName();
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/035b775a/fe/src/test/java/org/apache/impala/testutil/TestUtils.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/testutil/TestUtils.java b/fe/src/test/java/org/apache/impala/testutil/TestUtils.java
index 5454c7b..fd07971 100644
--- a/fe/src/test/java/org/apache/impala/testutil/TestUtils.java
+++ b/fe/src/test/java/org/apache/impala/testutil/TestUtils.java
@@ -256,6 +256,7 @@ public class TestUtils {
defaultDb, user, new TNetworkAddress("localhost", 0)));
SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSSSSSSS");
queryCtx.setNow_string(formatter.format(Calendar.getInstance().getTime()));
+ queryCtx.setStart_unix_millis(System.currentTimeMillis());
queryCtx.setPid(1000);
return queryCtx;
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/035b775a/tests/custom_cluster/test_lineage.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_lineage.py b/tests/custom_cluster/test_lineage.py
new file mode 100644
index 0000000..825e1b7
--- /dev/null
+++ b/tests/custom_cluster/test_lineage.py
@@ -0,0 +1,77 @@
+# 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.
+#
+# Tests for column lineage.
+# TODO: add verification for more fields.
+
+import json
+import logging
+import os
+import pytest
+import shutil
+import stat
+import tempfile
+import time
+
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+
+LOG = logging.getLogger(__name__)
+
+class TestLineage(CustomClusterTestSuite):
+
+ lineage_log_dir = tempfile.mkdtemp();
+
+ query = """
+ select count(*) from functional.alltypes
+ """
+
+ @classmethod
+ def setup_class(cls):
+ super(TestLineage, cls).setup_class()
+
+ @classmethod
+ def teardown_class(cls):
+ shutil.rmtree(cls.lineage_log_dir);
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args("--lineage_event_log_dir=%s" % lineage_log_dir)
+ def test_start_end_timestamp(self, vector):
+ """Test that 'timestamp' and 'endTime' in the lineage graph are populated with valid
+ UNIX times."""
+ LOG.info("lineage_event_log_dir is " + self.lineage_log_dir)
+ before_time = int(time.time())
+ self.execute_query_expect_success(self.client, self.query)
+ after_time = int(time.time())
+ LOG.info("before_time " + str(before_time) + " after_time " + str(after_time))
+
+ # Stop the cluster in order to flush the lineage log files.
+ self._stop_impala_cluster()
+
+ for log_filename in os.listdir(self.lineage_log_dir):
+ log_path = os.path.join(self.lineage_log_dir, log_filename)
+ # Only the coordinator's log file will be populated.
+ if os.path.getsize(log_path) > 0:
+ LOG.info("examining file: " + log_path)
+ with open(log_path) as log_file:
+ lineage_json = json.load(log_file)
+ timestamp = int(lineage_json["timestamp"])
+ end_time = int(lineage_json["endTime"])
+ assert before_time <= timestamp
+ assert timestamp <= end_time
+ assert end_time <= after_time
+ else:
+ LOG.info("empty file: " + log_path)
[5/6] incubator-impala git commit: IMPALA-4502:
test_partition_ddl_predicates breaks on non-HDFS filesystems
Posted by sa...@apache.org.
IMPALA-4502: test_partition_ddl_predicates breaks on non-HDFS filesystems
This is because that test uses 'set cached' and 'set uncached' which
are not supported on non-HDFS filesystems. This patch creates a
separate test file for non-HDFS filesystems with only supported
queries and invokes the right file based on the filesystem.
Change-Id: I8606aa427cb6e50be3395cdde246abb53db5172c
Reviewed-on: http://gerrit.cloudera.org:8080/5164
Reviewed-by: Sailesh Mukil <sa...@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/178fd591
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/178fd591
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/178fd591
Branch: refs/heads/master
Commit: 178fd59142db5f401d1426fa4daa11432e1e2f50
Parents: 7bfa6e3
Author: Sailesh Mukil <sa...@cloudera.com>
Authored: Mon Nov 21 08:27:21 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Nov 22 00:42:57 2016 +0000
----------------------------------------------------------------------
.../partition-ddl-predicates-all-fs.test | 135 ++++++++++++++++
.../partition-ddl-predicates-hdfs-only.test | 38 +++++
.../QueryTest/partition-ddl-predicates.test | 156 -------------------
tests/metadata/test_ddl.py | 7 +-
4 files changed, 178 insertions(+), 158 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/178fd591/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-all-fs.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-all-fs.test b/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-all-fs.test
new file mode 100644
index 0000000..36104c3
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-all-fs.test
@@ -0,0 +1,135 @@
+====
+---- QUERY
+# First create a partitioned table
+create table p1 (i int) partitioned by(j int, k string);
+# Add some partitions
+alter table p1 add partition (j=1,k="a");
+alter table p1 add partition (j=1,k="b");
+alter table p1 add partition (j=1,k="c");
+alter table p1 add partition (j=2,k="d");
+alter table p1 add partition (j=2,k="e");
+alter table p1 add partition (j=2,k="f");
+alter table p1 add partition (j=2,k=NULL);
+alter table p1 add partition (j=NULL,k="g");
+alter table p1 add partition (j=NULL,k=NULL);
+# Populate the table
+insert into p1 partition (j, k) values (100, 1, "a"), (200, 1, "b"), (300, 1, "c");
+====
+---- QUERY
+show partitions p1
+---- RESULTS
+'NULL','NULL',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=__HIVE_DEFAULT_PARTITION__/k=__HIVE_DEFAULT_PARTITION__
+'NULL','g',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=__HIVE_DEFAULT_PARTITION__/k=g
+'1','a',-1,1,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=1/k=a
+'1','b',-1,1,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=1/k=b
+'1','c',-1,1,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=1/k=c
+'2','NULL',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=__HIVE_DEFAULT_PARTITION__
+'2','d',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=d
+'2','e',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=e
+'2','f',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=f
+'Total','',-1,3,regex:.+,regex:.+,'','','',''
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
+====
+---- QUERY
+show files in p1 partition (j<2, k="a")
+---- RESULTS
+regex:.+,regex:.+,'j=1/k=a'
+---- TYPES
+STRING, STRING, STRING
+====
+---- QUERY
+alter table p1 partition (j=2) set fileformat parquet
+---- RESULTS
+'Updated 4 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 partition (k rlike ".*") set serdeproperties ('k'='v')
+---- RESULTS
+'Updated 7 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 partition (k="") set serdeproperties ('k'='v')
+---- RESULTS
+'Updated 2 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 partition (k=null) set serdeproperties ('k'='v')
+---- RESULTS
+'Updated 2 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 partition (j is not null) set tblproperties ('k'='v')
+---- RESULTS
+'Updated 7 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+compute incremental stats p1 partition(j<2)
+---- RESULTS
+'Updated 3 partition(s) and 1 column(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 drop partition (j<2)
+---- RESULTS
+'Dropped 3 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 drop partition (j<2)
+---- RESULTS
+'Dropped 0 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 drop if exists partition (j<2)
+---- RESULTS
+'Dropped 0 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 drop if exists partition (j=2, k="bla")
+---- RESULTS
+'Dropped 0 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 drop partition (j=2, k=NULL)
+---- RESULTS
+'Dropped 1 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1 drop partition (j IS NULL AND k IS NULL);
+---- RESULTS
+'Dropped 1 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+show partitions p1
+---- RESULTS
+'NULL','g',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=__HIVE_DEFAULT_PARTITION__/k=g
+'2','d',-1,0,regex:.+,regex:.+,regex:.+,'PARQUET',regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=d
+'2','e',-1,0,regex:.+,regex:.+,regex:.+,'PARQUET',regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=e
+'2','f',-1,0,regex:.+,regex:.+,regex:.+,'PARQUET',regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=f
+'Total','',0,0,regex:.+,regex:.+,'','','',''
+---- TYPES
+STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
+====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/178fd591/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-hdfs-only.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-hdfs-only.test b/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-hdfs-only.test
new file mode 100644
index 0000000..5ca2a06
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates-hdfs-only.test
@@ -0,0 +1,38 @@
+====
+---- QUERY
+# First create a partitioned table
+create table p1_hdfs (i int) partitioned by(j int, k string);
+# Add some partitions
+alter table p1_hdfs add partition (j=1,k="a");
+alter table p1_hdfs add partition (j=1,k="b");
+alter table p1_hdfs add partition (j=1,k="c");
+alter table p1_hdfs add partition (j=2,k="d");
+alter table p1_hdfs add partition (j=2,k="e");
+alter table p1_hdfs add partition (j=2,k="f");
+alter table p1_hdfs add partition (j=2,k=NULL);
+alter table p1_hdfs add partition (j=NULL,k="g");
+alter table p1_hdfs add partition (j=NULL,k=NULL);
+# Populate the table
+insert into p1_hdfs partition (j, k) values (100, 1, "a"), (200, 1, "b"), (300, 1, "c");
+====
+---- QUERY
+alter table p1_hdfs partition (j<2, k in ("b", "c")) set cached in 'testPool'
+---- RESULTS
+'Cached 2 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1_hdfs partition (j<2, j>0, k<>"d") set uncached
+---- RESULTS
+'Uncached 2 partition(s).'
+---- TYPES
+STRING
+====
+---- QUERY
+alter table p1_hdfs partition (j=3 or j=2, k like "%") set uncached
+---- RESULTS
+'Uncached 0 partition(s).'
+---- TYPES
+STRING
+====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/178fd591/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates.test b/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates.test
deleted file mode 100644
index cc7e865..0000000
--- a/testdata/workloads/functional-query/queries/QueryTest/partition-ddl-predicates.test
+++ /dev/null
@@ -1,156 +0,0 @@
-====
----- QUERY
-# First create a partitioned table
-create table p1 (i int) partitioned by(j int, k string);
-# Add some partitions
-alter table p1 add partition (j=1,k="a");
-alter table p1 add partition (j=1,k="b");
-alter table p1 add partition (j=1,k="c");
-alter table p1 add partition (j=2,k="d");
-alter table p1 add partition (j=2,k="e");
-alter table p1 add partition (j=2,k="f");
-alter table p1 add partition (j=2,k=NULL);
-alter table p1 add partition (j=NULL,k="g");
-alter table p1 add partition (j=NULL,k=NULL);
-# Populate the table
-insert into p1 partition (j, k) values (100, 1, "a"), (200, 1, "b"), (300, 1, "c");
-====
----- QUERY
-show partitions p1
----- RESULTS
-'NULL','NULL',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=__HIVE_DEFAULT_PARTITION__/k=__HIVE_DEFAULT_PARTITION__
-'NULL','g',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=__HIVE_DEFAULT_PARTITION__/k=g
-'1','a',-1,1,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=1/k=a
-'1','b',-1,1,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=1/k=b
-'1','c',-1,1,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=1/k=c
-'2','NULL',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=__HIVE_DEFAULT_PARTITION__
-'2','d',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=d
-'2','e',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=e
-'2','f',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=f
-'Total','',-1,3,regex:.+,regex:.+,'','','',''
----- TYPES
-STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
-====
----- QUERY
-show files in p1 partition (j<2, k="a")
----- RESULTS
-regex:.+,regex:.+,'j=1/k=a'
----- TYPES
-STRING, STRING, STRING
-====
----- QUERY
-alter table p1 partition (j<2, k in ("b", "c")) set cached in 'testPool'
----- RESULTS
-'Cached 2 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 partition (j<2, j>0, k<>"d") set uncached
----- RESULTS
-'Uncached 2 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 partition (j=3 or j=2, k like "%") set uncached
----- RESULTS
-'Uncached 0 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 partition (j=2) set fileformat parquet
----- RESULTS
-'Updated 4 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 partition (k rlike ".*") set serdeproperties ('k'='v')
----- RESULTS
-'Updated 7 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 partition (k="") set serdeproperties ('k'='v')
----- RESULTS
-'Updated 2 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 partition (k=null) set serdeproperties ('k'='v')
----- RESULTS
-'Updated 2 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 partition (j is not null) set tblproperties ('k'='v')
----- RESULTS
-'Updated 7 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-compute incremental stats p1 partition(j<2)
----- RESULTS
-'Updated 3 partition(s) and 1 column(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 drop partition (j<2)
----- RESULTS
-'Dropped 3 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 drop partition (j<2)
----- RESULTS
-'Dropped 0 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 drop if exists partition (j<2)
----- RESULTS
-'Dropped 0 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 drop if exists partition (j=2, k="bla")
----- RESULTS
-'Dropped 0 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 drop partition (j=2, k=NULL)
----- RESULTS
-'Dropped 1 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-alter table p1 drop partition (j IS NULL AND k IS NULL);
----- RESULTS
-'Dropped 1 partition(s).'
----- TYPES
-STRING
-====
----- QUERY
-show partitions p1
----- RESULTS
-'NULL','g',-1,0,regex:.+,regex:.+,regex:.+,regex:.+,regex:.+,regex:.*/test-warehouse/.+/p1/j=__HIVE_DEFAULT_PARTITION__/k=g
-'2','d',-1,0,regex:.+,regex:.+,regex:.+,'PARQUET',regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=d
-'2','e',-1,0,regex:.+,regex:.+,regex:.+,'PARQUET',regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=e
-'2','f',-1,0,regex:.+,regex:.+,regex:.+,'PARQUET',regex:.+,regex:.*/test-warehouse/.+/p1/j=2/k=f
-'Total','',0,0,regex:.+,regex:.+,'','','',''
----- TYPES
-STRING, STRING, BIGINT, BIGINT, STRING, STRING, STRING, STRING, STRING, STRING
-====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/178fd591/tests/metadata/test_ddl.py
----------------------------------------------------------------------
diff --git a/tests/metadata/test_ddl.py b/tests/metadata/test_ddl.py
index 15fe4c7..58f42eb 100644
--- a/tests/metadata/test_ddl.py
+++ b/tests/metadata/test_ddl.py
@@ -24,7 +24,7 @@ from tests.common.impala_test_suite import LOG
from tests.common.parametrize import UniqueDatabase
from tests.common.skip import SkipIf, SkipIfLocal, SkipIfOldAggsJoins
from tests.common.test_dimensions import create_single_exec_option_dimension
-from tests.util.filesystem_utils import WAREHOUSE, IS_LOCAL, IS_S3
+from tests.util.filesystem_utils import WAREHOUSE, IS_HDFS, IS_LOCAL, IS_S3
# Validates DDL statements (create, drop)
class TestDdlStatements(TestDdlBase):
@@ -393,8 +393,11 @@ class TestDdlStatements(TestDdlBase):
@UniqueDatabase.parametrize(sync_ddl=True)
def test_partition_ddl_predicates(self, vector, unique_database):
- self.run_test_case('QueryTest/partition-ddl-predicates', vector,
+ self.run_test_case('QueryTest/partition-ddl-predicates-all-fs', vector,
use_db=unique_database, multiple_impalad=self._use_multiple_impalad(vector))
+ if IS_HDFS:
+ self.run_test_case('QueryTest/partition-ddl-predicates-hdfs-only', vector,
+ use_db=unique_database, multiple_impalad=self._use_multiple_impalad(vector))
# IMPALA-2002: Tests repeated adding/dropping of .jar and .so in the lib cache.
class TestLibCache(TestDdlBase):
[4/6] incubator-impala git commit: IMPALA-4478: Initial Kudu client
mem tracking for sink
Posted by sa...@apache.org.
IMPALA-4478: Initial Kudu client mem tracking for sink
The Kudu client allocates memory which is not tracked by
Impala. There are several sources, but the most significant
is the memory allocated by the KuduSession on the write
path. This can be >100MB, so it is important to track to
avoid OOM.
Moving forward, we should have a better way to track Kudu
client memory, but for now we must at least handle this
potentially problematic case.
This changes the KuduTableSink to consume 200MB which should
be enough for the 100MB write mutation buffer as well as
100MB worth of errors buffered in the client before Impala
takes ownership of them (and deletes them). This is left as
a flag because it may turn out to be too high for some users
and too low for others. When we have better support from
Kudu (including KUDU-1752), we should simplify this.
TODO: Handle DML w/ small or known resource requirements
(e.g. VALUES specified or query has LIMIT) specially to
avoid over-consumption.
Testing: Have verified acceptable behavior in the stress
test with a simple workload containing DML statements of
moderate cardinality.
Change-Id: I47f17a81e4362ab490019382fedc66c25f07080a
Reviewed-on: http://gerrit.cloudera.org:8080/5152
Reviewed-by: Matthew Jacobs <mj...@cloudera.com>
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/7bfa6e3e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/7bfa6e3e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/7bfa6e3e
Branch: refs/heads/master
Commit: 7bfa6e3e3f20d741f36e713ea62f9324e5739ea0
Parents: af67b2f
Author: Matthew Jacobs <mj...@cloudera.com>
Authored: Mon Nov 14 15:31:38 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Nov 22 00:08:04 2016 +0000
----------------------------------------------------------------------
be/src/exec/kudu-table-sink.cc | 40 ++++++++++++++++++++++++++++++++++---
1 file changed, 37 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/7bfa6e3e/be/src/exec/kudu-table-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/kudu-table-sink.cc b/be/src/exec/kudu-table-sink.cc
index a195e84..20bbe69 100644
--- a/be/src/exec/kudu-table-sink.cc
+++ b/be/src/exec/kudu-table-sink.cc
@@ -31,8 +31,25 @@
#include "common/names.h"
-DEFINE_int32(kudu_mutation_buffer_size, 100 * 1024 * 1024, "The size (bytes) of the "
- "Kudu client buffer for mutations.");
+#define DEFAULT_KUDU_MUTATION_BUFFER_SIZE 100 * 1024 * 1024
+
+DEFINE_int32(kudu_mutation_buffer_size, DEFAULT_KUDU_MUTATION_BUFFER_SIZE,
+ "The size (bytes) of the Kudu client buffer for mutations.");
+
+// The memory (bytes) that this node needs to consume in order to operate. This is
+// necessary because the KuduClient allocates non-trivial amounts of untracked memory,
+// and is potentially unbounded due to how Kudu's async error reporting works.
+// Until Kudu's client memory usage can be bounded (KUDU-1752), we estimate that 2x the
+// mutation buffer size is enough memory, and that seems to provide acceptable results in
+// testing. This is still exposed as a flag for now though, because it may be possible
+// that in some cases this is always too high (in which case tracked mem >> RSS and the
+// memory is underutilized), or this may not be high enough (e.g. we underestimate the
+// size of error strings, and RSS grows until the process is killed).
+// TODO: Handle DML w/ small or known resource requirements (e.g. VALUES specified or
+// query has LIMIT) specially to avoid over-consumption.
+DEFINE_int32(kudu_sink_mem_required, 2 * DEFAULT_KUDU_MUTATION_BUFFER_SIZE,
+ "(Advanced) The memory required (bytes) for a KuduTableSink. The default value is "
+ " 2x the kudu_mutation_buffer_size. This flag is subject to change or removal.");
DECLARE_int32(kudu_operation_timeout_ms);
@@ -116,8 +133,21 @@ Status KuduTableSink::Prepare(RuntimeState* state, MemTracker* parent_mem_tracke
Status KuduTableSink::Open(RuntimeState* state) {
RETURN_IF_ERROR(Expr::Open(output_expr_ctxs_, state));
- RETURN_IF_ERROR(CreateKuduClient(table_desc_->kudu_master_addresses(), &client_));
+ int64_t required_mem = FLAGS_kudu_sink_mem_required;
+ if (!mem_tracker_->TryConsume(required_mem)) {
+ return mem_tracker_->MemLimitExceeded(state,
+ "Could not allocate memory for KuduTableSink", required_mem);
+ }
+
+ Status s = CreateKuduClient(table_desc_->kudu_master_addresses(), &client_);
+ if (!s.ok()) {
+ // Close() releases memory if client_ is not NULL, but since the memory was consumed
+ // and the client failed to be created, it must be released.
+ DCHECK(client_.get() == NULL);
+ mem_tracker_->Release(required_mem);
+ return s;
+ }
KUDU_RETURN_IF_ERROR(client_->OpenTable(table_desc_->table_name(), &table_),
"Unable to open Kudu table");
@@ -350,6 +380,10 @@ Status KuduTableSink::FlushFinal(RuntimeState* state) {
void KuduTableSink::Close(RuntimeState* state) {
if (closed_) return;
+ if (client_.get() != NULL) {
+ mem_tracker_->Release(FLAGS_kudu_sink_mem_required);
+ client_.reset();
+ }
SCOPED_TIMER(profile()->total_time_counter());
Expr::Close(output_expr_ctxs_, state);
DataSink::Close(state);
[3/6] incubator-impala git commit: IMPALA-4514: Fix broken exhaustive
builds caused by non-nullable columns
Posted by sa...@apache.org.
IMPALA-4514: Fix broken exhaustive builds caused by non-nullable columns
This commit fixes the broken exhaustive Impala builds. The issue was
caused by Kudu table that didn't properly specify nullability
constraints. Hence, some rows were rejected during data loading causing
some tests to fail.
Change-Id: Ib6f4b4c88ef18b1731b7c9789aad602880e18035
Reviewed-on: http://gerrit.cloudera.org:8080/5157
Reviewed-by: Alex Behm <al...@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/af67b2fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/af67b2fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/af67b2fe
Branch: refs/heads/master
Commit: af67b2fef4d397dc2a304ffda876e9484c24069e
Parents: 035b775
Author: Dimitris Tsirogiannis <dt...@cloudera.com>
Authored: Sun Nov 20 16:50:45 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Mon Nov 21 22:24:25 2016 +0000
----------------------------------------------------------------------
testdata/datasets/functional/functional_schema_template.sql | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/af67b2fe/testdata/datasets/functional/functional_schema_template.sql
----------------------------------------------------------------------
diff --git a/testdata/datasets/functional/functional_schema_template.sql b/testdata/datasets/functional/functional_schema_template.sql
index 1940c31..be7a6ac 100644
--- a/testdata/datasets/functional/functional_schema_template.sql
+++ b/testdata/datasets/functional/functional_schema_template.sql
@@ -1337,7 +1337,8 @@ OVERWRITE INTO TABLE {db_name}{db_suffix}.{table_name};
---- CREATE_KUDU
DROP TABLE IF EXISTS {db_name}{db_suffix}.{table_name};
create table {db_name}{db_suffix}.{table_name} (
- a string primary key, b string, c string, d int, e double, f string, g string
+ a string primary key, b string null, c string null, d int null, e double null,
+ f string null, g string null
)
distribute by hash(a) into 3 buckets stored as kudu;
====
[6/6] incubator-impala git commit: IMPALA-2523: Make HdfsTableSink
aware of clustered input
Posted by sa...@apache.org.
IMPALA-2523: Make HdfsTableSink aware of clustered input
IMPALA-2521 introduced clustering for insert statements. This change
makes the HdfsTableSink aware of clustered inputs, so that partitions
are opened, written, and closed one by one.
This change also adds/modifies tests in several ways:
- clustered insert tests switch from selecting all rows from
alltypessmall to alltypes. Together with varying settings for
batch_size, this results in a larger number of row batches being
written.
- clustered insert tests select from alltypes instead of
functional.alltypes to make sure we also select from various input
formats.
- clustered insert tests have been added to select from alltypestiny to
create inserts with 1 and 2 rows per partition respectively.
- exhaustive insert tests now use different values for batch_size: 1,
16, 0 (meaning default, 1024). This is limited to uncompressed parquet
files, to maintain a reasonable runtime. On my machine execution of
test.insert took 1778 seconds, compared to 1002 seconds with the just
default row batch size.
- There is additional testing in test_insert_behaviour.py to make sure
that insertion over several row batches only creates one file per
partition.
- It renames the test_insert method to make it unique in the file and
allow for effective filtering with -k.
- It adds tests to the Analyzer test suite.
Change-Id: Ibeda0bdabbfe44c8ac95bf7c982a75649e1b82d0
Reviewed-on: http://gerrit.cloudera.org:8080/4863
Reviewed-by: Lars Volker <lv...@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/8ea21d09
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/8ea21d09
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/8ea21d09
Branch: refs/heads/master
Commit: 8ea21d099fde57bd358cc073017bdcf80c8d74ca
Parents: 178fd59
Author: Lars Volker <lv...@cloudera.com>
Authored: Tue Oct 25 17:33:27 2016 -0700
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Nov 22 02:51:20 2016 +0000
----------------------------------------------------------------------
be/src/exec/hbase-table-sink.cc | 2 +-
be/src/exec/hbase-table-writer.cc | 8 +-
be/src/exec/hbase-table-writer.h | 4 +-
be/src/exec/hdfs-avro-table-writer.cc | 4 +-
be/src/exec/hdfs-avro-table-writer.h | 5 +-
be/src/exec/hdfs-parquet-table-writer.cc | 4 +-
be/src/exec/hdfs-parquet-table-writer.h | 5 +-
be/src/exec/hdfs-sequence-table-writer.cc | 5 +-
be/src/exec/hdfs-sequence-table-writer.h | 5 +-
be/src/exec/hdfs-table-sink.cc | 232 ++++++++++++-------
be/src/exec/hdfs-table-sink.h | 56 +++--
be/src/exec/hdfs-table-writer.cc | 1 +
be/src/exec/hdfs-table-writer.h | 21 +-
be/src/exec/hdfs-text-table-writer.cc | 5 +-
be/src/exec/hdfs-text-table-writer.h | 4 +-
bin/impala-config.sh | 4 +
common/thrift/DataSinks.thrift | 15 +-
.../org/apache/impala/analysis/DeleteStmt.java | 2 +-
.../org/apache/impala/analysis/InsertStmt.java | 2 +-
.../org/apache/impala/analysis/UpdateStmt.java | 2 +-
.../apache/impala/planner/HdfsTableSink.java | 9 +-
.../org/apache/impala/planner/TableSink.java | 4 +-
.../impala/analysis/AnalyzeStmtsTest.java | 16 ++
.../queries/QueryTest/insert.test | 85 ++++++-
tests/query_test/test_insert.py | 10 +-
tests/query_test/test_insert_behaviour.py | 108 ++++++++-
26 files changed, 457 insertions(+), 161 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hbase-table-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hbase-table-sink.cc b/be/src/exec/hbase-table-sink.cc
index a89384e..9c5998b 100644
--- a/be/src/exec/hbase-table-sink.cc
+++ b/be/src/exec/hbase-table-sink.cc
@@ -89,7 +89,7 @@ Status HBaseTableSink::Send(RuntimeState* state, RowBatch* batch) {
ExprContext::FreeLocalAllocations(output_expr_ctxs_);
RETURN_IF_ERROR(state->CheckQueryState());
// Since everything is set up just forward everything to the writer.
- RETURN_IF_ERROR(hbase_table_writer_->AppendRowBatch(batch));
+ RETURN_IF_ERROR(hbase_table_writer_->AppendRows(batch));
(*state->per_partition_status())[ROOT_PARTITION_KEY].num_modified_rows +=
batch->num_rows();
return Status::OK();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hbase-table-writer.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hbase-table-writer.cc b/be/src/exec/hbase-table-writer.cc
index 41ea184..7574adc 100644
--- a/be/src/exec/hbase-table-writer.cc
+++ b/be/src/exec/hbase-table-writer.cc
@@ -114,7 +114,7 @@ Status HBaseTableWriter::InitJNI() {
return Status::OK();
}
-Status HBaseTableWriter::AppendRowBatch(RowBatch* batch) {
+Status HBaseTableWriter::AppendRows(RowBatch* batch) {
JNIEnv* env = getJNIEnv();
if (env == NULL) return Status("Error getting JNIEnv.");
@@ -270,9 +270,9 @@ void HBaseTableWriter::Close(RuntimeState* state) {
table_.reset();
}
- // The jni should already have everything cleaned at this point
- // but try again just in case there was an error that caused
- // AppendRowBatch to exit out before calling CleanUpJni.
+ // The jni should already have everything cleaned at this point but try again just in
+ // case there was an error that caused AppendRows() to exit out before calling
+ // CleanUpJni.
Status status = CleanUpJni();
if (!status.ok()) {
stringstream ss;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hbase-table-writer.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hbase-table-writer.h b/be/src/exec/hbase-table-writer.h
index 94947c9..bc414ab 100644
--- a/be/src/exec/hbase-table-writer.h
+++ b/be/src/exec/hbase-table-writer.h
@@ -41,13 +41,13 @@ class RowBatch;
/// HBaseTableWriter::InitJni();
/// writer = new HBaseTableWriter(state, table_desc_, output_exprs_);
/// writer.Init(state);
-/// writer.AppendRowBatch(batch);
+/// writer.AppendRows(batch);
class HBaseTableWriter {
public:
HBaseTableWriter(HBaseTableDescriptor* table_desc,
const std::vector<ExprContext*>& output_expr_ctxs,
RuntimeProfile* profile);
- Status AppendRowBatch(RowBatch* batch);
+ Status AppendRows(RowBatch* batch);
/// Calls to Close release the HBaseTable.
void Close(RuntimeState* state);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-avro-table-writer.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-avro-table-writer.cc b/be/src/exec/hdfs-avro-table-writer.cc
index 1934304..ec0ee08 100644
--- a/be/src/exec/hdfs-avro-table-writer.cc
+++ b/be/src/exec/hdfs-avro-table-writer.cc
@@ -175,8 +175,8 @@ void HdfsAvroTableWriter::Close() {
mem_pool_->FreeAll();
}
-Status HdfsAvroTableWriter::AppendRowBatch(RowBatch* batch,
- const vector<int32_t>& row_group_indices, bool* new_file) {
+Status HdfsAvroTableWriter::AppendRows(
+ RowBatch* batch, const vector<int32_t>& row_group_indices, bool* new_file) {
int32_t limit;
bool all_rows = row_group_indices.empty();
if (all_rows) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-avro-table-writer.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-avro-table-writer.h b/be/src/exec/hdfs-avro-table-writer.h
index aeff0ed..38e820e 100644
--- a/be/src/exec/hdfs-avro-table-writer.h
+++ b/be/src/exec/hdfs-avro-table-writer.h
@@ -75,9 +75,8 @@ class HdfsAvroTableWriter : public HdfsTableWriter {
/// Outputs the given rows into an HDFS sequence file. The rows are buffered
/// to fill a sequence file block.
- virtual Status AppendRowBatch(RowBatch* rows,
- const std::vector<int32_t>& row_group_indices,
- bool* new_file);
+ virtual Status AppendRows(
+ RowBatch* rows, const std::vector<int32_t>& row_group_indices, bool* new_file);
private:
/// Processes a single row, appending to out_
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-parquet-table-writer.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-parquet-table-writer.cc b/be/src/exec/hdfs-parquet-table-writer.cc
index cc708ab..7ae55b7 100644
--- a/be/src/exec/hdfs-parquet-table-writer.cc
+++ b/be/src/exec/hdfs-parquet-table-writer.cc
@@ -872,8 +872,8 @@ Status HdfsParquetTableWriter::InitNewFile() {
return Status::OK();
}
-Status HdfsParquetTableWriter::AppendRowBatch(RowBatch* batch,
- const vector<int32_t>& row_group_indices, bool* new_file) {
+Status HdfsParquetTableWriter::AppendRows(
+ RowBatch* batch, const vector<int32_t>& row_group_indices, bool* new_file) {
SCOPED_TIMER(parent_->encode_timer());
*new_file = false;
int limit;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-parquet-table-writer.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-parquet-table-writer.h b/be/src/exec/hdfs-parquet-table-writer.h
index 581160d..4e16707 100644
--- a/be/src/exec/hdfs-parquet-table-writer.h
+++ b/be/src/exec/hdfs-parquet-table-writer.h
@@ -67,9 +67,8 @@ class HdfsParquetTableWriter : public HdfsTableWriter {
virtual Status InitNewFile();
/// Appends parquet representation of rows in the batch to the current file.
- virtual Status AppendRowBatch(RowBatch* batch,
- const std::vector<int32_t>& row_group_indices,
- bool* new_file);
+ virtual Status AppendRows(
+ RowBatch* batch, const std::vector<int32_t>& row_group_indices, bool* new_file);
/// Write out all the data.
virtual Status Finalize();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-sequence-table-writer.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-sequence-table-writer.cc b/be/src/exec/hdfs-sequence-table-writer.cc
index 99910e2..3c522ba 100644
--- a/be/src/exec/hdfs-sequence-table-writer.cc
+++ b/be/src/exec/hdfs-sequence-table-writer.cc
@@ -91,9 +91,8 @@ Status HdfsSequenceTableWriter::Init() {
return Status::OK();
}
-Status HdfsSequenceTableWriter::AppendRowBatch(RowBatch* batch,
- const vector<int32_t>& row_group_indices,
- bool* new_file) {
+Status HdfsSequenceTableWriter::AppendRows(
+ RowBatch* batch, const vector<int32_t>& row_group_indices, bool* new_file) {
int32_t limit;
if (row_group_indices.empty()) {
limit = batch->num_rows();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-sequence-table-writer.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-sequence-table-writer.h b/be/src/exec/hdfs-sequence-table-writer.h
index c94ab1c..7f6a888 100644
--- a/be/src/exec/hdfs-sequence-table-writer.h
+++ b/be/src/exec/hdfs-sequence-table-writer.h
@@ -57,9 +57,8 @@ class HdfsSequenceTableWriter : public HdfsTableWriter {
/// Outputs the given rows into an HDFS sequence file. The rows are buffered
/// to fill a sequence file block.
- virtual Status AppendRowBatch(RowBatch* rows,
- const std::vector<int32_t>& row_group_indices,
- bool* new_file);
+ virtual Status AppendRows(
+ RowBatch* rows, const std::vector<int32_t>& row_group_indices, bool* new_file);
private:
/// processes a single row, delegates to Compress or NoCompress ConsumeRow().
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-table-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-sink.cc b/be/src/exec/hdfs-table-sink.cc
index 8c74797..3a9725c 100644
--- a/be/src/exec/hdfs-table-sink.cc
+++ b/be/src/exec/hdfs-table-sink.cc
@@ -56,26 +56,30 @@ const static string& ROOT_PARTITION_KEY =
g_ImpalaInternalService_constants.ROOT_PARTITION_KEY;
HdfsTableSink::HdfsTableSink(const RowDescriptor& row_desc,
- const vector<TExpr>& select_list_texprs,
- const TDataSink& tsink)
- : DataSink(row_desc),
- table_desc_(NULL),
- default_partition_(NULL),
- current_row_(NULL),
- table_id_(tsink.table_sink.target_table_id),
- skip_header_line_count_(
- tsink.table_sink.hdfs_table_sink.__isset.skip_header_line_count
- ? tsink.table_sink.hdfs_table_sink.skip_header_line_count : 0),
- select_list_texprs_(select_list_texprs),
- partition_key_texprs_(tsink.table_sink.hdfs_table_sink.partition_key_exprs),
- overwrite_(tsink.table_sink.hdfs_table_sink.overwrite) {
+ const vector<TExpr>& select_list_texprs, const TDataSink& tsink)
+ : DataSink(row_desc),
+ table_desc_(nullptr),
+ default_partition_(nullptr),
+ table_id_(tsink.table_sink.target_table_id),
+ skip_header_line_count_(
+ tsink.table_sink.hdfs_table_sink.__isset.skip_header_line_count ?
+ tsink.table_sink.hdfs_table_sink.skip_header_line_count :
+ 0),
+ select_list_texprs_(select_list_texprs),
+ partition_key_texprs_(tsink.table_sink.hdfs_table_sink.partition_key_exprs),
+ overwrite_(tsink.table_sink.hdfs_table_sink.overwrite),
+ input_is_clustered_(tsink.table_sink.hdfs_table_sink.input_is_clustered),
+ current_clustered_partition_(nullptr) {
DCHECK(tsink.__isset.table_sink);
}
OutputPartition::OutputPartition()
- : hdfs_connection(NULL), tmp_hdfs_file(NULL), num_rows(0), num_files(0),
- partition_descriptor(NULL), block_size(0) {
-}
+ : hdfs_connection(nullptr),
+ tmp_hdfs_file(nullptr),
+ num_rows(0),
+ num_files(0),
+ partition_descriptor(nullptr),
+ block_size(0) {}
Status HdfsTableSink::PrepareExprs(RuntimeState* state) {
// Prepare select list expressions.
@@ -125,7 +129,7 @@ Status HdfsTableSink::Prepare(RuntimeState* state, MemTracker* parent_mem_tracke
table_desc_ = static_cast<const HdfsTableDescriptor*>(
state->desc_tbl().GetTableDescriptor(table_id_));
- if (table_desc_ == NULL) {
+ if (table_desc_ == nullptr) {
stringstream error_msg("Failed to get table descriptor for table id: ");
error_msg << table_id_;
return Status(error_msg.str());
@@ -192,14 +196,17 @@ Status HdfsTableSink::Open(RuntimeState* state) {
// partition keys. So only keep a reference to the partition which matches
// partition_key_values for constant values, since only that is written to.
void* table_partition_key_value =
- partition->partition_key_value_ctxs()[i]->GetValue(NULL);
- void* target_partition_key_value = partition_key_expr_ctxs_[i]->GetValue(NULL);
- if (table_partition_key_value == NULL && target_partition_key_value == NULL) {
+ partition->partition_key_value_ctxs()[i]->GetValue(nullptr);
+ void* target_partition_key_value =
+ partition_key_expr_ctxs_[i]->GetValue(nullptr);
+ if (table_partition_key_value == nullptr
+ && target_partition_key_value == nullptr) {
continue;
}
- if (table_partition_key_value == NULL || target_partition_key_value == NULL
+ if (table_partition_key_value == nullptr
+ || target_partition_key_value == nullptr
|| !RawValue::Eq(table_partition_key_value, target_partition_key_value,
- partition_key_expr_ctxs_[i]->root()->type())) {
+ partition_key_expr_ctxs_[i]->root()->type())) {
relevant_partition = false;
break;
}
@@ -207,17 +214,16 @@ Status HdfsTableSink::Open(RuntimeState* state) {
}
if (relevant_partition) {
string key;
- // It's ok if current_row_ is NULL (which it should be here), since all of these
- // expressions are constant, and can therefore be evaluated without a valid row
- // context.
- GetHashTblKey(dynamic_partition_key_value_ctxs, &key);
+ // Pass nullptr as row, since all of these expressions are constant, and can
+ // therefore be evaluated without a valid row context.
+ GetHashTblKey(nullptr, dynamic_partition_key_value_ctxs, &key);
DCHECK(partition_descriptor_map_.find(key) == partition_descriptor_map_.end())
<< "Partitions with duplicate 'static' keys found during INSERT";
partition_descriptor_map_[key] = partition;
}
}
}
- if (default_partition_ == NULL) {
+ if (default_partition_ == nullptr) {
return Status("No default partition found for HdfsTextTableSink");
}
return Status::OK();
@@ -262,6 +268,87 @@ void HdfsTableSink::BuildHdfsFileNames(
output_partition->num_files = 0;
}
+Status HdfsTableSink::WriteRowsToPartition(
+ RuntimeState* state, RowBatch* batch, PartitionPair* partition_pair) {
+ // The rows of this batch may span multiple files. We repeatedly pass the row batch to
+ // the writer until it sets new_file to false, indicating that all rows have been
+ // written. The writer tracks where it is in the batch when it returns with new_file
+ // set.
+ bool new_file;
+ while (true) {
+ OutputPartition* output_partition = partition_pair->first;
+ RETURN_IF_ERROR(
+ output_partition->writer->AppendRows(batch, partition_pair->second, &new_file));
+ if (!new_file) break;
+ RETURN_IF_ERROR(FinalizePartitionFile(state, output_partition));
+ RETURN_IF_ERROR(CreateNewTmpFile(state, output_partition));
+ }
+ partition_pair->second.clear();
+ return Status::OK();
+}
+
+Status HdfsTableSink::WriteClusteredRowBatch(RuntimeState* state, RowBatch* batch) {
+ DCHECK_GT(batch->num_rows(), 0);
+ DCHECK(!dynamic_partition_key_expr_ctxs_.empty());
+ DCHECK(input_is_clustered_);
+
+ // Initialize the clustered partition and key.
+ if (current_clustered_partition_ == nullptr) {
+ const TupleRow* current_row = batch->GetRow(0);
+ GetHashTblKey(
+ current_row, dynamic_partition_key_expr_ctxs_, ¤t_clustered_partition_key_);
+ RETURN_IF_ERROR(GetOutputPartition(state, current_row,
+ current_clustered_partition_key_, ¤t_clustered_partition_, false));
+ }
+
+ // Compare the last row of the batch to the last current partition key. If they match,
+ // then all the rows in the batch have the same key and can be written as a whole.
+ string last_row_key;
+ GetHashTblKey(batch->GetRow(batch->num_rows() - 1), dynamic_partition_key_expr_ctxs_,
+ &last_row_key);
+ if (last_row_key == current_clustered_partition_key_) {
+ DCHECK(current_clustered_partition_->second.empty());
+ RETURN_IF_ERROR(WriteRowsToPartition(state, batch, current_clustered_partition_));
+ return Status::OK();
+ }
+
+ // Not all rows in this batch match the previously written partition key, so we process
+ // them individually.
+ for (int i = 0; i < batch->num_rows(); ++i) {
+ const TupleRow* current_row = batch->GetRow(i);
+
+ string key;
+ GetHashTblKey(current_row, dynamic_partition_key_expr_ctxs_, &key);
+
+ if (current_clustered_partition_key_ != key) {
+ DCHECK(current_clustered_partition_ != nullptr);
+ // Done with previous partition - write rows and close.
+ if (!current_clustered_partition_->second.empty()) {
+ RETURN_IF_ERROR(WriteRowsToPartition(state, batch, current_clustered_partition_));
+ current_clustered_partition_->second.clear();
+ }
+ RETURN_IF_ERROR(FinalizePartitionFile(state, current_clustered_partition_->first));
+ if (current_clustered_partition_->first->writer.get() != nullptr) {
+ current_clustered_partition_->first->writer->Close();
+ }
+ partition_keys_to_output_partitions_.erase(current_clustered_partition_key_);
+ current_clustered_partition_key_ = std::move(key);
+ RETURN_IF_ERROR(GetOutputPartition(state, current_row,
+ current_clustered_partition_key_, ¤t_clustered_partition_, false));
+ }
+#ifdef DEBUG
+ string debug_row_key;
+ GetHashTblKey(current_row, dynamic_partition_key_expr_ctxs_, &debug_row_key);
+ DCHECK_EQ(current_clustered_partition_key_, debug_row_key);
+#endif
+ DCHECK(current_clustered_partition_ != nullptr);
+ current_clustered_partition_->second.push_back(i);
+ }
+ // Write final set of rows to the partition but keep its file open.
+ RETURN_IF_ERROR(WriteRowsToPartition(state, batch, current_clustered_partition_));
+ return Status::OK();
+}
+
Status HdfsTableSink::CreateNewTmpFile(RuntimeState* state,
OutputPartition* output_partition) {
SCOPED_TIMER(ADD_TIMER(profile(), "TmpFileCreateTimer"));
@@ -295,7 +382,7 @@ Status HdfsTableSink::CreateNewTmpFile(RuntimeState* state,
tmp_hdfs_file_name_cstr, O_WRONLY, 0, 0, block_size);
VLOG_FILE << "hdfsOpenFile() file=" << tmp_hdfs_file_name_cstr;
- if (output_partition->tmp_hdfs_file == NULL) {
+ if (output_partition->tmp_hdfs_file == nullptr) {
return Status(GetHdfsErrorMsg("Failed to open HDFS file for writing: ",
output_partition->current_file_name));
}
@@ -310,7 +397,7 @@ Status HdfsTableSink::CreateNewTmpFile(RuntimeState* state,
// files, we get the block size by stat-ing the file.
hdfsFileInfo* info = hdfsGetPathInfo(output_partition->hdfs_connection,
output_partition->current_file_name.c_str());
- if (info == NULL) {
+ if (info == nullptr) {
return Status(GetHdfsErrorMsg("Failed to get info on temporary HDFS file: ",
output_partition->current_file_name));
}
@@ -338,16 +425,16 @@ Status HdfsTableSink::CreateNewTmpFile(RuntimeState* state,
}
Status HdfsTableSink::InitOutputPartition(RuntimeState* state,
- const HdfsPartitionDescriptor& partition_descriptor,
+ const HdfsPartitionDescriptor& partition_descriptor, const TupleRow* row,
OutputPartition* output_partition, bool empty_partition) {
// Build the unique name for this partition from the partition keys, e.g. "j=1/f=foo/"
// etc.
stringstream partition_name_ss;
for (int j = 0; j < partition_key_expr_ctxs_.size(); ++j) {
partition_name_ss << table_desc_->col_descs()[j].name() << "=";
- void* value = partition_key_expr_ctxs_[j]->GetValue(current_row_);
- // NULL partition keys get a special value to be compatible with Hive.
- if (value == NULL) {
+ void* value = partition_key_expr_ctxs_[j]->GetValue(row);
+ // nullptr partition keys get a special value to be compatible with Hive.
+ if (value == nullptr) {
partition_name_ss << table_desc_->null_partition_key_value();
} else {
string value_str;
@@ -362,7 +449,7 @@ Status HdfsTableSink::InitOutputPartition(RuntimeState* state,
// decoded partition key value.
string encoded_str;
UrlEncode(value_str, &encoded_str, true);
- // If the string is empty, map it to NULL (mimicking Hive's behaviour)
+ // If the string is empty, map it to nullptr (mimicking Hive's behaviour)
partition_name_ss << (encoded_str.empty() ?
table_desc_->null_partition_key_value() : encoded_str);
}
@@ -459,24 +546,25 @@ Status HdfsTableSink::InitOutputPartition(RuntimeState* state,
return CreateNewTmpFile(state, output_partition);
}
-void HdfsTableSink::GetHashTblKey(const vector<ExprContext*>& ctxs, string* key) {
+void HdfsTableSink::GetHashTblKey(
+ const TupleRow* row, const vector<ExprContext*>& ctxs, string* key) {
stringstream hash_table_key;
for (int i = 0; i < ctxs.size(); ++i) {
RawValue::PrintValueAsBytes(
- ctxs[i]->GetValue(current_row_), ctxs[i]->root()->type(), &hash_table_key);
+ ctxs[i]->GetValue(row), ctxs[i]->root()->type(), &hash_table_key);
// Additionally append "/" to avoid accidental key collisions.
hash_table_key << "/";
}
*key = hash_table_key.str();
}
-inline Status HdfsTableSink::GetOutputPartition(RuntimeState* state,
+inline Status HdfsTableSink::GetOutputPartition(RuntimeState* state, const TupleRow* row,
const string& key, PartitionPair** partition_pair, bool no_more_rows) {
+ DCHECK(row != nullptr || key == ROOT_PARTITION_KEY);
PartitionMap::iterator existing_partition;
existing_partition = partition_keys_to_output_partitions_.find(key);
if (existing_partition == partition_keys_to_output_partitions_.end()) {
- // Create a new OutputPartition, and add it to
- // partition_keys_to_output_partitions.
+ // Create a new OutputPartition, and add it to partition_keys_to_output_partitions.
const HdfsPartitionDescriptor* partition_descriptor = default_partition_;
PartitionDescriptorMap::const_iterator it = partition_descriptor_map_.find(key);
if (it != partition_descriptor_map_.end()) {
@@ -484,13 +572,13 @@ inline Status HdfsTableSink::GetOutputPartition(RuntimeState* state,
}
OutputPartition* partition = state->obj_pool()->Add(new OutputPartition());
- Status status = InitOutputPartition(state, *partition_descriptor, partition,
- no_more_rows);
+ Status status =
+ InitOutputPartition(state, *partition_descriptor, row, partition, no_more_rows);
if (!status.ok()) {
// We failed to create the output partition successfully. Clean it up now
// as it is not added to partition_keys_to_output_partitions_ so won't be
// cleaned up in Close().
- if (partition->writer.get() != NULL) partition->writer->Close();
+ if (partition->writer.get() != nullptr) partition->writer->Close();
return status;
}
@@ -533,45 +621,25 @@ Status HdfsTableSink::Send(RuntimeState* state, RowBatch* batch) {
// If there are no dynamic keys just use an empty key.
PartitionPair* partition_pair;
RETURN_IF_ERROR(
- GetOutputPartition(state, ROOT_PARTITION_KEY, &partition_pair, false));
- // Pass the row batch to the writer. If new_file is returned true then the current
- // file is finalized and a new file is opened.
- // The writer tracks where it is in the batch when it returns with new_file set.
- OutputPartition* output_partition = partition_pair->first;
- bool new_file;
- do {
- RETURN_IF_ERROR(output_partition->writer->AppendRowBatch(
- batch, partition_pair->second, &new_file));
- if (new_file) {
- RETURN_IF_ERROR(FinalizePartitionFile(state, output_partition));
- RETURN_IF_ERROR(CreateNewTmpFile(state, output_partition));
- }
- } while (new_file);
+ GetOutputPartition(state, nullptr, ROOT_PARTITION_KEY, &partition_pair, false));
+ RETURN_IF_ERROR(WriteRowsToPartition(state, batch, partition_pair));
+ } else if (input_is_clustered_) {
+ WriteClusteredRowBatch(state, batch);
} else {
for (int i = 0; i < batch->num_rows(); ++i) {
- current_row_ = batch->GetRow(i);
+ const TupleRow* current_row = batch->GetRow(i);
string key;
- GetHashTblKey(dynamic_partition_key_expr_ctxs_, &key);
- PartitionPair* partition_pair = NULL;
- RETURN_IF_ERROR(GetOutputPartition(state, key, &partition_pair, false));
+ GetHashTblKey(current_row, dynamic_partition_key_expr_ctxs_, &key);
+ PartitionPair* partition_pair = nullptr;
+ RETURN_IF_ERROR(
+ GetOutputPartition(state, current_row, key, &partition_pair, false));
partition_pair->second.push_back(i);
}
- for (PartitionMap::iterator partition = partition_keys_to_output_partitions_.begin();
- partition != partition_keys_to_output_partitions_.end(); ++partition) {
- OutputPartition* output_partition = partition->second.first;
- if (partition->second.second.empty()) continue;
-
- bool new_file;
- do {
- RETURN_IF_ERROR(output_partition->writer->AppendRowBatch(
- batch, partition->second.second, &new_file));
- if (new_file) {
- RETURN_IF_ERROR(FinalizePartitionFile(state, output_partition));
- RETURN_IF_ERROR(CreateNewTmpFile(state, output_partition));
- }
- } while (new_file);
- partition->second.second.clear();
+ for (PartitionMap::value_type& partition : partition_keys_to_output_partitions_) {
+ if (!partition.second.second.empty()) {
+ RETURN_IF_ERROR(WriteRowsToPartition(state, batch, &partition.second));
+ }
}
}
return Status::OK();
@@ -579,11 +647,11 @@ Status HdfsTableSink::Send(RuntimeState* state, RowBatch* batch) {
Status HdfsTableSink::FinalizePartitionFile(RuntimeState* state,
OutputPartition* partition) {
- if (partition->tmp_hdfs_file == NULL && !overwrite_) return Status::OK();
+ if (partition->tmp_hdfs_file == nullptr && !overwrite_) return Status::OK();
SCOPED_TIMER(ADD_TIMER(profile(), "FinalizePartitionFileTimer"));
- // OutputPartition writer could be NULL if there is no row to output.
- if (partition->writer.get() != NULL) {
+ // OutputPartition writer could be nullptr if there is no row to output.
+ if (partition->writer.get() != nullptr) {
RETURN_IF_ERROR(partition->writer->Finalize());
// Track total number of appended rows per partition in runtime
@@ -604,10 +672,10 @@ Status HdfsTableSink::FinalizePartitionFile(RuntimeState* state,
Status HdfsTableSink::ClosePartitionFile(
RuntimeState* state, OutputPartition* partition) {
- if (partition->tmp_hdfs_file == NULL) return Status::OK();
+ if (partition->tmp_hdfs_file == nullptr) return Status::OK();
int hdfs_ret = hdfsCloseFile(partition->hdfs_connection, partition->tmp_hdfs_file);
VLOG_FILE << "hdfsCloseFile() file=" << partition->current_file_name;
- partition->tmp_hdfs_file = NULL;
+ partition->tmp_hdfs_file = nullptr;
ImpaladMetrics::NUM_FILES_OPEN_FOR_INSERT->Increment(-1);
if (hdfs_ret != 0) {
return Status(ErrorMsg(TErrorCode::GENERAL,
@@ -625,7 +693,7 @@ Status HdfsTableSink::FlushFinal(RuntimeState* state) {
// Make sure we create an output partition even if the input is empty because we need
// it to delete the existing data for 'insert overwrite'.
PartitionPair* dummy;
- RETURN_IF_ERROR(GetOutputPartition(state, ROOT_PARTITION_KEY, &dummy, true));
+ RETURN_IF_ERROR(GetOutputPartition(state, nullptr, ROOT_PARTITION_KEY, &dummy, true));
}
// Close Hdfs files, and update stats in runtime state.
@@ -646,7 +714,7 @@ void HdfsTableSink::Close(RuntimeState* state) {
partition_keys_to_output_partitions_.begin();
cur_partition != partition_keys_to_output_partitions_.end();
++cur_partition) {
- if (cur_partition->second.first->writer.get() != NULL) {
+ if (cur_partition->second.first->writer.get() != nullptr) {
cur_partition->second.first->writer->Close();
}
Status close_status = ClosePartitionFile(state, cur_partition->second.first);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-table-sink.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-sink.h b/be/src/exec/hdfs-table-sink.h
index ae32270..22b6a44 100644
--- a/be/src/exec/hdfs-table-sink.h
+++ b/be/src/exec/hdfs-table-sink.h
@@ -165,10 +165,11 @@ class HdfsTableSink : public DataSink {
private:
/// Initialises the filenames of a given output partition, and opens the temporary file.
- /// If the partition will not have any rows added to it, empty_partition must be true.
+ /// The partition key is derived from 'row'. If the partition will not have any rows
+ /// added to it, empty_partition must be true.
Status InitOutputPartition(RuntimeState* state,
- const HdfsPartitionDescriptor& partition_descriptor,
- OutputPartition* output_partition, bool empty_partition);
+ const HdfsPartitionDescriptor& partition_descriptor, const TupleRow* row,
+ OutputPartition* output_partition, bool empty_partition);
/// Add a temporary file to an output partition. Files are created in a
/// temporary directory and then moved to the real partition directory by the
@@ -177,24 +178,23 @@ class HdfsTableSink : public DataSink {
/// If this function fails, the tmp file is cleaned up.
Status CreateNewTmpFile(RuntimeState* state, OutputPartition* output_partition);
- /// Key is the concatenation of the evaluated
- /// dynamic_partition_key_exprs_ generated by GetHashTblKey().
- /// Maps to an OutputPartition, which are owned by the object pool and
- /// a vector of rows to insert into this partition from the current row batch.
+ /// Key is the concatenation of the evaluated dynamic_partition_key_exprs_ generated by
+ /// GetHashTblKey(). Maps to an OutputPartition, which are owned by the object pool, and
+ /// a vector of indices of the rows in the current batch to insert into the partition.
typedef std::pair<OutputPartition*, std::vector<int32_t>> PartitionPair;
typedef boost::unordered_map<std::string, PartitionPair> PartitionMap;
-
- /// Generates string key for hash_tbl_ as a concatenation
- /// of all evaluated exprs, evaluated against current_row_.
- /// The generated string is much shorter than the full Hdfs file name.
- void GetHashTblKey(const std::vector<ExprContext*>& ctxs, std::string* key);
+ /// Generates string key for hash_tbl_ as a concatenation of all evaluated exprs,
+ /// evaluated against 'row'. The generated string is much shorter than the full Hdfs
+ /// file name.
+ void GetHashTblKey(
+ const TupleRow* row, const std::vector<ExprContext*>& ctxs, std::string* key);
/// Given a hashed partition key, get the output partition structure from
- /// the partition_keys_to_output_partitions_.
- /// no_more_rows indicates that no more rows will be added to the partition.
- Status GetOutputPartition(RuntimeState* state, const std::string& key,
- PartitionPair** partition_pair, bool no_more_rows);
+ /// the 'partition_keys_to_output_partitions_'. 'no_more_rows' indicates that no more
+ /// rows will be added to the partition.
+ Status GetOutputPartition(RuntimeState* state, const TupleRow* row,
+ const std::string& key, PartitionPair** partition_pair, bool no_more_rows);
/// Initialise and prepare select and partition key expressions
Status PrepareExprs(RuntimeState* state);
@@ -206,6 +206,15 @@ class HdfsTableSink : public DataSink {
void BuildHdfsFileNames(const HdfsPartitionDescriptor& partition_descriptor,
OutputPartition* output);
+ /// Writes all rows referenced by the row index vector in 'partition_pair' to the
+ /// partition's writer and clears the row index vector afterwards.
+ Status WriteRowsToPartition(
+ RuntimeState* state, RowBatch* batch, PartitionPair* partition_pair);
+
+ /// Maps all rows in 'batch' to partitions and appends them to their temporary Hdfs
+ /// files. The input must be ordered by the partition key expressions.
+ Status WriteClusteredRowBatch(RuntimeState* state, RowBatch* batch);
+
/// Updates runtime stats of HDFS with rows written, then closes the file associated
/// with the partition by calling ClosePartitionFile()
Status FinalizePartitionFile(RuntimeState* state, OutputPartition* partition);
@@ -230,9 +239,6 @@ class HdfsTableSink : public DataSink {
/// Exprs that materialize output values
std::vector<ExprContext*> output_expr_ctxs_;
- /// Current row from the current RowBatch to output
- TupleRow* current_row_;
-
/// Table id resolved in Prepare() to set tuple_desc_;
TableId table_id_;
@@ -255,6 +261,18 @@ class HdfsTableSink : public DataSink {
/// Indicates whether the existing partitions should be overwritten.
bool overwrite_;
+ /// Indicates whether the input is ordered by the partition keys, meaning partitions can
+ /// be opened, written, and closed one by one.
+ bool input_is_clustered_;
+
+ /// Stores the current partition during clustered inserts across subsequent row batches.
+ /// Only set if 'input_is_clustered_' is true.
+ PartitionPair* current_clustered_partition_;
+
+ /// Stores the current partition key during clustered inserts across subsequent row
+ /// batches. Only set if 'input_is_clustered_' is true.
+ std::string current_clustered_partition_key_;
+
/// The directory in which to write intermediate results. Set to
/// <hdfs_table_base_dir>/_impala_insert_staging/ during Prepare()
std::string staging_dir_;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-table-writer.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-writer.cc b/be/src/exec/hdfs-table-writer.cc
index 71e9278..48349e3 100644
--- a/be/src/exec/hdfs-table-writer.cc
+++ b/be/src/exec/hdfs-table-writer.cc
@@ -39,6 +39,7 @@ HdfsTableWriter::HdfsTableWriter(HdfsTableSink* parent,
Status HdfsTableWriter::Write(const uint8_t* data, int32_t len) {
DCHECK_GE(len, 0);
+ DCHECK(output_->tmp_hdfs_file != nullptr);
int ret = hdfsWrite(output_->hdfs_connection, output_->tmp_hdfs_file, data, len);
if (ret == -1) {
string error_msg = GetHdfsErrorMsg("");
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-table-writer.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-writer.h b/be/src/exec/hdfs-table-writer.h
index b3d344f..8066315 100644
--- a/be/src/exec/hdfs-table-writer.h
+++ b/be/src/exec/hdfs-table-writer.h
@@ -52,7 +52,7 @@ class HdfsTableWriter {
/// The sequence of calls to this object are:
/// 1. Init()
/// 2. InitNewFile()
- /// 3. AppendRowBatch() - called repeatedly
+ /// 3. AppendRows() - called repeatedly
/// 4. Finalize()
/// For files formats that are splittable (and therefore can be written to an
/// arbitrarily large file), 1-4 is called once.
@@ -65,17 +65,14 @@ class HdfsTableWriter {
/// Called when a new file is started.
virtual Status InitNewFile() = 0;
- /// Appends the current batch of rows to the partition. If there are multiple
- /// partitions then row_group_indices will contain the rows that are for this
- /// partition, otherwise all rows in the batch are appended.
- /// If the current file is full, the writer stops appending and
- /// returns with *new_file == true. A new file will be opened and
- /// the same row batch will be passed again. The writer must track how
- /// much of the batch it had already processed asking for a new file.
- /// Otherwise the writer will return with *newfile == false.
- virtual Status AppendRowBatch(RowBatch* batch,
- const std::vector<int32_t>& row_group_indices,
- bool* new_file) = 0;
+ /// Appends rows of 'batch' to the partition that are selected via 'row_group_indices',
+ /// and if the latter is empty, appends every row.
+ /// If the current file is full, the writer stops appending and returns with
+ /// *new_file == true. A new file will be opened and the same row batch will be passed
+ /// again. The writer must track how much of the batch it had already processed asking
+ /// for a new file. Otherwise the writer will return with *newfile == false.
+ virtual Status AppendRows(
+ RowBatch* batch, const std::vector<int32_t>& row_group_indices, bool* new_file) = 0;
/// Finalize this partition. The writer needs to finish processing
/// all data have written out after the return from this call.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-text-table-writer.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-text-table-writer.cc b/be/src/exec/hdfs-text-table-writer.cc
index 053c821..cba4032 100644
--- a/be/src/exec/hdfs-text-table-writer.cc
+++ b/be/src/exec/hdfs-text-table-writer.cc
@@ -99,9 +99,8 @@ string HdfsTextTableWriter::file_extension() const {
return compressor_->file_extension();
}
-Status HdfsTextTableWriter::AppendRowBatch(RowBatch* batch,
- const vector<int32_t>& row_group_indices,
- bool* new_file) {
+Status HdfsTextTableWriter::AppendRows(
+ RowBatch* batch, const vector<int32_t>& row_group_indices, bool* new_file) {
int32_t limit;
if (row_group_indices.empty()) {
limit = batch->num_rows();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/be/src/exec/hdfs-text-table-writer.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-text-table-writer.h b/be/src/exec/hdfs-text-table-writer.h
index 1d65f11..2944f23 100644
--- a/be/src/exec/hdfs-text-table-writer.h
+++ b/be/src/exec/hdfs-text-table-writer.h
@@ -60,8 +60,8 @@ class HdfsTextTableWriter : public HdfsTableWriter {
/// Appends delimited string representation of the rows in the batch to output partition.
/// The resulting output is buffered until HDFS_FLUSH_WRITE_SIZE before being written
/// to HDFS.
- Status AppendRowBatch(RowBatch* current_row,
- const std::vector<int32_t>& row_group_indices, bool* new_file);
+ Status AppendRows(RowBatch* current_row, const std::vector<int32_t>& row_group_indices,
+ bool* new_file);
private:
/// Escapes occurrences of field_delim_ and escape_char_ with escape_char_ and
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/bin/impala-config.sh
----------------------------------------------------------------------
diff --git a/bin/impala-config.sh b/bin/impala-config.sh
index caae5e9..834f216 100755
--- a/bin/impala-config.sh
+++ b/bin/impala-config.sh
@@ -197,6 +197,10 @@ elif [ "${TARGET_FILESYSTEM}" = "isilon" ]; then
# isilon manages its own replication.
export HDFS_REPLICATION=1
elif [ "${TARGET_FILESYSTEM}" = "local" ]; then
+ if [[ "${WAREHOUSE_LOCATION_PREFIX}" = "" ]]; then
+ echo "WAREHOUSE_LOCATION_PREFIX cannot be an empty string for local filesystem"
+ return 1
+ fi
if [ ! -d "${WAREHOUSE_LOCATION_PREFIX}" ]; then
echo "'$WAREHOUSE_LOCATION_PREFIX' is not a directory on the local filesystem."
return 1
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/common/thrift/DataSinks.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/DataSinks.thrift b/common/thrift/DataSinks.thrift
index 2a57304..0b136b2 100644
--- a/common/thrift/DataSinks.thrift
+++ b/common/thrift/DataSinks.thrift
@@ -62,10 +62,14 @@ struct THdfsTableSink {
1: required list<Exprs.TExpr> partition_key_exprs
2: required bool overwrite
- /// The 'skip.header.line.count' property of the target Hdfs table. We will insert this
- /// many empty lines at the beginning of new text files, which will be skipped by the
- /// scanners while reading from the files.
+ // The 'skip.header.line.count' property of the target Hdfs table. We will insert this
+ // many empty lines at the beginning of new text files, which will be skipped by the
+ // scanners while reading from the files.
3: optional i32 skip_header_line_count
+
+ // This property indicates to the table sink whether the input is ordered by the
+ // partition keys, meaning partitions can be opened, written, and closed one by one.
+ 4: required bool input_is_clustered
}
// Structure to encapsulate specific options that are passed down to the KuduTableSink
@@ -73,7 +77,10 @@ struct TKuduTableSink {
// The position in this vector is equal to the position in the output expressions of the
// sink and holds the index of the corresponsding column in the Kudu schema,
// e.g. 'exprs[i]' references 'kudu_table.column(referenced_cols[i])'
- 1: optional list<i32> referenced_columns;
+ 1: optional list<i32> referenced_columns
+
+ // Defines if duplicate or not found keys should be ignored
+ 2: optional bool ignore_not_found_or_duplicate
}
// Sink to create the build side of a JoinNode.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java b/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java
index 9d84baa..2f7f670 100644
--- a/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java
@@ -56,7 +56,7 @@ public class DeleteStmt extends ModifyStmt {
// analyze() must have been called before.
Preconditions.checkState(table_ != null);
TableSink tableSink = TableSink.create(table_, TableSink.Op.DELETE,
- ImmutableList.<Expr>of(), referencedColumns_, false);
+ ImmutableList.<Expr>of(), referencedColumns_, false, false);
Preconditions.checkState(!referencedColumns_.isEmpty());
return tableSink;
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java b/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
index 87f7cef..5528da9 100644
--- a/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/InsertStmt.java
@@ -793,7 +793,7 @@ public class InsertStmt extends StatementBase {
// analyze() must have been called before.
Preconditions.checkState(table_ != null);
return TableSink.create(table_, isUpsert_ ? TableSink.Op.UPSERT : TableSink.Op.INSERT,
- partitionKeyExprs_, mentionedColumns_, overwrite_);
+ partitionKeyExprs_, mentionedColumns_, overwrite_, hasClusteredHint_);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java b/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java
index 9a1bc9e..de74bd8 100644
--- a/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java
@@ -65,7 +65,7 @@ public class UpdateStmt extends ModifyStmt {
// analyze() must have been called before.
Preconditions.checkState(table_ != null);
DataSink dataSink = TableSink.create(table_, TableSink.Op.UPDATE,
- ImmutableList.<Expr>of(), referencedColumns_, false);
+ ImmutableList.<Expr>of(), referencedColumns_, false, false);
Preconditions.checkState(!referencedColumns_.isEmpty());
return dataSink;
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java b/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
index bb06b5e..fc7f9b1 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsTableSink.java
@@ -46,12 +46,17 @@ public class HdfsTableSink extends TableSink {
// Whether to overwrite the existing partition(s).
protected final boolean overwrite_;
+ // Indicates whether the input is ordered by the partition keys, meaning partitions can
+ // be opened, written, and closed one by one.
+ protected final boolean inputIsClustered_;
+
public HdfsTableSink(Table targetTable, List<Expr> partitionKeyExprs,
- boolean overwrite) {
+ boolean overwrite, boolean inputIsClustered) {
super(targetTable, Op.INSERT);
Preconditions.checkState(targetTable instanceof HdfsTable);
partitionKeyExprs_ = partitionKeyExprs;
overwrite_ = overwrite;
+ inputIsClustered_ = inputIsClustered;
}
@Override
@@ -140,7 +145,7 @@ public class HdfsTableSink extends TableSink {
protected TDataSink toThrift() {
TDataSink result = new TDataSink(TDataSinkType.TABLE_SINK);
THdfsTableSink hdfsTableSink = new THdfsTableSink(
- Expr.treesToThrift(partitionKeyExprs_), overwrite_);
+ Expr.treesToThrift(partitionKeyExprs_), overwrite_, inputIsClustered_);
HdfsTable table = (HdfsTable) targetTable_;
StringBuilder error = new StringBuilder();
int skipHeaderLineCount = table.parseSkipHeaderLineCount(error);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/fe/src/main/java/org/apache/impala/planner/TableSink.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/TableSink.java b/fe/src/main/java/org/apache/impala/planner/TableSink.java
index 742e6c9..fb3cea2 100644
--- a/fe/src/main/java/org/apache/impala/planner/TableSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/TableSink.java
@@ -90,13 +90,13 @@ public abstract class TableSink extends DataSink {
*/
public static TableSink create(Table table, Op sinkAction,
List<Expr> partitionKeyExprs, List<Integer> referencedColumns,
- boolean overwrite) {
+ boolean overwrite, boolean inputIsClustered) {
if (table instanceof HdfsTable) {
// Hdfs only supports inserts.
Preconditions.checkState(sinkAction == Op.INSERT);
// Referenced columns don't make sense for an Hdfs table.
Preconditions.checkState(referencedColumns.isEmpty());
- return new HdfsTableSink(table, partitionKeyExprs, overwrite);
+ return new HdfsTableSink(table, partitionKeyExprs, overwrite, inputIsClustered);
} else if (table instanceof HBaseTable) {
// HBase only supports inserts.
Preconditions.checkState(sinkAction == Op.INSERT);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/fe/src/test/java/org/apache/impala/analysis/AnalyzeStmtsTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/analysis/AnalyzeStmtsTest.java b/fe/src/test/java/org/apache/impala/analysis/AnalyzeStmtsTest.java
index cacdd83..11f6842 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AnalyzeStmtsTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AnalyzeStmtsTest.java
@@ -1750,6 +1750,22 @@ public class AnalyzeStmtsTest extends AnalyzerTest {
"partition (year, month) /* +shuffle,noshuffle */ " +
"select * from functional.alltypes",
"Conflicting INSERT hints: shuffle and noshuffle");
+
+ // Test clustered hint.
+ AnalyzesOk(String.format(
+ "insert into functional.alltypessmall partition (year, month) %sclustered%s " +
+ "select * from functional.alltypes", prefix, suffix));
+ AnalyzesOk(String.format(
+ "insert into table functional.alltypesnopart %sclustered%s " +
+ "select * from functional.alltypesnopart", prefix, suffix));
+ AnalyzesOk(String.format(
+ "insert into table functional.alltypesnopart %snoclustered%s " +
+ "select * from functional.alltypesnopart", prefix, suffix));
+ // Conflicting clustered hints.
+ AnalysisError(String.format(
+ "insert into table functional.alltypessmall partition (year, month) " +
+ "/* +clustered,noclustered */ select * from functional.alltypes", prefix,
+ suffix), "Conflicting INSERT hints: clustered and noclustered");
}
// Multiple non-conflicting hints and case insensitivity of hints.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/testdata/workloads/functional-query/queries/QueryTest/insert.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/insert.test b/testdata/workloads/functional-query/queries/QueryTest/insert.test
index b0c5c7c..5601b35 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/insert.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/insert.test
@@ -861,14 +861,87 @@ Memory limit exceeded
---- QUERY
# IMPALA-2521: clustered insert into table
insert into table alltypesinsert
-partition (year, month) /*+ clustered */
-select * from functional.alltypessmall;
+partition (year, month) /*+ clustered,shuffle */
+select * from alltypes;
---- SETUP
DROP PARTITIONS alltypesinsert
RESET alltypesinsert
---- RESULTS
-year=2009/month=1/: 25
-year=2009/month=2/: 25
-year=2009/month=3/: 25
-year=2009/month=4/: 25
+year=2009/month=1/: 310
+year=2009/month=10/: 310
+year=2009/month=11/: 300
+year=2009/month=12/: 310
+year=2009/month=2/: 280
+year=2009/month=3/: 310
+year=2009/month=4/: 300
+year=2009/month=5/: 310
+year=2009/month=6/: 300
+year=2009/month=7/: 310
+year=2009/month=8/: 310
+year=2009/month=9/: 300
+year=2010/month=1/: 310
+year=2010/month=10/: 310
+year=2010/month=11/: 300
+year=2010/month=12/: 310
+year=2010/month=2/: 280
+year=2010/month=3/: 310
+year=2010/month=4/: 300
+year=2010/month=5/: 310
+year=2010/month=6/: 300
+year=2010/month=7/: 310
+year=2010/month=8/: 310
+year=2010/month=9/: 300
+====
+---- QUERY
+# IMPALA-2521: clustered insert into table
+insert into table alltypesinsert
+partition (year, month) /*+ clustered,shuffle */
+select * from alltypestiny;
+---- SETUP
+DROP PARTITIONS alltypesinsert
+RESET alltypesinsert
+---- RESULTS
+year=2009/month=1/: 2
+year=2009/month=2/: 2
+year=2009/month=3/: 2
+year=2009/month=4/: 2
+====
+---- QUERY
+# IMPALA-2521: clustered insert into table
+insert into table alltypesinsert
+partition (year, month) /*+ clustered,noshuffle */
+select * from alltypestiny;
+---- SETUP
+DROP PARTITIONS alltypesinsert
+RESET alltypesinsert
+---- RESULTS
+year=2009/month=1/: 2
+year=2009/month=2/: 2
+year=2009/month=3/: 2
+year=2009/month=4/: 2
+====
+---- QUERY
+# IMPALA-2521: clustered insert into table
+insert into table alltypesinsert
+partition (year, month) /*+ clustered,shuffle */
+select * from alltypestiny where int_col = 0;
+---- SETUP
+DROP PARTITIONS alltypesinsert
+RESET alltypesinsert
+---- RESULTS
+year=2009/month=1/: 1
+year=2009/month=2/: 1
+year=2009/month=3/: 1
+year=2009/month=4/: 1
+====
+---- QUERY
+# IMPALA-2521: clustered, unpartitioned insert into table
+insert into table alltypesnopart_insert
+ /*+ clustered,shuffle */
+select id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col,
+double_col, date_string_col, string_col, timestamp_col from alltypessmall;
+---- SETUP
+RESET alltypesnopart_insert
+---- RESULTS
+: 100
====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/tests/query_test/test_insert.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_insert.py b/tests/query_test/test_insert.py
index e5e74ed..d195050 100644
--- a/tests/query_test/test_insert.py
+++ b/tests/query_test/test_insert.py
@@ -55,7 +55,7 @@ class TestInsertQueries(ImpalaTestSuite):
cls.TestMatrix.add_dimension(create_uncompressed_text_dimension(cls.get_workload()))
else:
cls.TestMatrix.add_dimension(create_exec_option_dimension(
- cluster_sizes=[0], disable_codegen_options=[False], batch_sizes=[0],
+ cluster_sizes=[0], disable_codegen_options=[False], batch_sizes=[0, 1, 16],
sync_ddl=[0, 1]))
cls.TestMatrix.add_dimension(TestDimension("compression_codec", *PARQUET_CODECS));
# Insert is currently only supported for text and parquet
@@ -68,6 +68,12 @@ class TestInsertQueries(ImpalaTestSuite):
v.get_value('compression_codec') == 'none'))
cls.TestMatrix.add_constraint(lambda v:\
v.get_value('table_format').compression_codec == 'none')
+ # Only test other batch sizes for uncompressed parquet to keep the execution time
+ # within reasonable bounds.
+ cls.TestMatrix.add_constraint(lambda v:\
+ v.get_value('exec_option')['batch_size'] == 0 or \
+ (v.get_value('table_format').file_format == 'parquet' and \
+ v.get_value('compression_codec') == 'none'))
def test_insert_large_string(self, vector, unique_database):
"""Test handling of large strings in inserter and scanner."""
@@ -103,7 +109,7 @@ class TestInsertQueries(ImpalaTestSuite):
super(TestInsertQueries, cls).setup_class()
@pytest.mark.execute_serially
- def test_insert(self, vector):
+ def test_insert_test(self, vector):
if (vector.get_value('table_format').file_format == 'parquet'):
vector.get_value('exec_option')['COMPRESSION_CODEC'] = \
vector.get_value('compression_codec')
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8ea21d09/tests/query_test/test_insert_behaviour.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_insert_behaviour.py b/tests/query_test/test_insert_behaviour.py
index 53c9dbf..1584343 100644
--- a/tests/query_test/test_insert_behaviour.py
+++ b/tests/query_test/test_insert_behaviour.py
@@ -23,11 +23,14 @@ import pytest
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.parametrize import UniqueDatabase
from tests.common.skip import SkipIfS3, SkipIfIsilon, SkipIfLocal
-from tests.util.filesystem_utils import WAREHOUSE, get_fs_path
+from tests.util.filesystem_utils import WAREHOUSE, get_fs_path, IS_S3
@SkipIfLocal.hdfs_client
class TestInsertBehaviour(ImpalaTestSuite):
"""Tests for INSERT behaviour that isn't covered by checking query results"""
+ @classmethod
+ def get_workload(self):
+ return 'functional-query'
TEST_DB_NAME = "insert_empty_result_db"
@@ -473,3 +476,106 @@ class TestInsertBehaviour(ImpalaTestSuite):
"other::---".format(groups[0]))
self.execute_query_expect_success(self.client, "REFRESH " + table)
self.execute_query_expect_failure(self.client, insert_query)
+
+ def test_clustered_partition_single_file(self, unique_database):
+ """IMPALA-2523: Tests that clustered insert creates one file per partition, even when
+ inserting over multiple row batches."""
+ # On s3 this test takes about 220 seconds and we are unlikely to break it, so only run
+ # it in exhaustive strategy.
+ if self.exploration_strategy() != 'exhaustive' and IS_S3:
+ pytest.skip("only runs in exhaustive")
+ table = "{0}.insert_clustered".format(unique_database)
+ table_path = "test-warehouse/{0}.db/insert_clustered".format(unique_database)
+ table_location = get_fs_path("/" + table_path)
+
+ create_stmt = """create table {0} like functional.alltypes""".format(table)
+ self.execute_query_expect_success(self.client, create_stmt)
+
+ set_location_stmt = """alter table {0} set location '{1}'""".format(
+ table, table_location)
+ self.execute_query_expect_success(self.client, set_location_stmt)
+
+ # Setting a lower batch size will result in multiple row batches being written.
+ self.execute_query_expect_success(self.client, "set batch_size=10")
+
+ insert_stmt = """insert into {0} partition(year, month) /*+ clustered,shuffle */
+ select * from functional.alltypes""".format(table)
+ self.execute_query_expect_success(self.client, insert_stmt)
+
+ # We expect exactly one partition per year and month, since subsequent row batches of
+ # a partition will be written into the same file.
+ expected_partitions = \
+ ["year=%s/month=%s" % (y, m) for y in [2009, 2010] for m in range(1,13)]
+
+ for partition in expected_partitions:
+ partition_path = "{0}/{1}".format(table_path, partition)
+ files = self.filesystem_client.ls(partition_path)
+ assert len(files) == 1, "%s: %s" % (partition, files)
+
+ def test_clustered_partition_multiple_files(self, unique_database):
+ """IMPALA-2523: Tests that clustered insert creates the right number of files per
+ partition when inserting over multiple row batches."""
+ # This test takes about 30 seconds and we are unlikely to break it, so only run it in
+ # exhaustive strategy.
+ if self.exploration_strategy() != 'exhaustive':
+ pytest.skip("only runs in exhaustive")
+ table = "{0}.insert_clustered".format(unique_database)
+ table_path = "test-warehouse/{0}.db/insert_clustered".format(unique_database)
+ table_location = get_fs_path("/" + table_path)
+
+ create_stmt = """create table {0} (
+ l_orderkey BIGINT,
+ l_partkey BIGINT,
+ l_suppkey BIGINT,
+ l_linenumber INT,
+ l_quantity DECIMAL(12,2),
+ l_extendedprice DECIMAL(12,2),
+ l_discount DECIMAL(12,2),
+ l_tax DECIMAL(12,2),
+ l_linestatus STRING,
+ l_shipdate STRING,
+ l_commitdate STRING,
+ l_receiptdate STRING,
+ l_shipinstruct STRING,
+ l_shipmode STRING,
+ l_comment STRING)
+ partitioned by (l_returnflag STRING) stored as parquet
+ """.format(table)
+ self.execute_query_expect_success(self.client, create_stmt)
+
+ set_location_stmt = """alter table {0} set location '{1}'""".format(
+ table, table_location)
+ self.execute_query_expect_success(self.client, set_location_stmt)
+
+ # Setting a lower parquet file size will result in multiple files being written.
+ self.execute_query_expect_success(self.client, "set parquet_file_size=10485760")
+
+ insert_stmt = """insert into {0} partition(l_returnflag) /*+ clustered,shuffle */
+ select l_orderkey,
+ l_partkey,
+ l_suppkey,
+ l_linenumber,
+ l_quantity,
+ l_extendedprice,
+ l_discount,
+ l_tax,
+ l_linestatus,
+ l_shipdate,
+ l_commitdate,
+ l_receiptdate,
+ l_shipinstruct,
+ l_shipmode,
+ l_comment,
+ l_returnflag
+ from tpch_parquet.lineitem""".format(table)
+ self.execute_query_expect_success(self.client, insert_stmt)
+
+ expected_partition_files = [("l_returnflag=A", 3, 30),
+ ("l_returnflag=N", 3, 30),
+ ("l_returnflag=R", 3, 30)]
+
+ for partition, min_files, max_files in expected_partition_files:
+ partition_path = "{0}/{1}".format(table_path, partition)
+ files = self.filesystem_client.ls(partition_path)
+ assert min_files <= len(files) and len(files) <= max_files, \
+ "%s: %s" % (partition, files)