You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by kw...@apache.org on 2016/12/06 19:07:29 UTC
[1/8] incubator-impala git commit: Additional functional testing for
default values on Kudu tables
Repository: incubator-impala
Updated Branches:
refs/heads/master 1495b2007 -> 9b80224f9
Additional functional testing for default values on Kudu tables
This commit also fixes an issue where an error is thrown if a default
value is set for a boolean column on a Kudu table.
Change-Id: I25b66275d29d1cf21df14e78ab58f625a83b0725
Reviewed-on: http://gerrit.cloudera.org:8080/5337
Reviewed-by: Matthew Jacobs <mj...@cloudera.com>
Tested-by: Impala Public 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/867b2434
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/867b2434
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/867b2434
Branch: refs/heads/master
Commit: 867b2434ca678a66e8d7bdbd228c86a7d5d032cb
Parents: 1495b20
Author: Dimitris Tsirogiannis <dt...@cloudera.com>
Authored: Fri Dec 2 16:33:31 2016 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Tue Dec 6 00:04:43 2016 +0000
----------------------------------------------------------------------
.../java/org/apache/impala/util/KuduUtil.java | 3 +
.../queries/QueryTest/kudu_insert.test | 71 ++++++++++++++++++++
2 files changed, 74 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/867b2434/fe/src/main/java/org/apache/impala/util/KuduUtil.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/util/KuduUtil.java b/fe/src/main/java/org/apache/impala/util/KuduUtil.java
index 559c4a5..edd331b 100644
--- a/fe/src/main/java/org/apache/impala/util/KuduUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/KuduUtil.java
@@ -178,6 +178,9 @@ public class KuduUtil {
case STRING:
checkCorrectType(literal.isSetString_literal(), type, colName, literal);
return literal.getString_literal().getValue();
+ case BOOL:
+ checkCorrectType(literal.isSetBool_literal(), type, colName, literal);
+ return literal.getBool_literal().isValue();
default:
throw new ImpalaRuntimeException("Unsupported value for column type: " +
type.toString());
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/867b2434/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
index 759dc5e..e282567 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
@@ -331,3 +331,74 @@ insert into allkeytypes select cast(id as tinyint), smallint_col, int_col,
NumModifiedRows: 3
NumRowErrors: 6
====
+---- QUERY
+# Table with default values
+create table tbl_with_defaults (a int primary key, b int null default 10,
+ c int not null default 100, d int default 1000, e int null, f int not null,
+ g string default 'test', h boolean default true) distribute by hash (a) into 3
+ buckets stored as kudu
+---- RESULTS
+====
+---- QUERY
+insert into tbl_with_defaults (a, f) values (1, 1), (2, 2), (3, 3), (4, 4)
+---- RUNTIME_PROFILE
+NumModifiedRows: 4
+NumRowErrors: 0
+---- LABELS
+A, B, C, D, E, F, G, H
+---- DML_RESULTS: tbl_with_defaults
+1,10,100,1000,NULL,1,'test',true
+2,10,100,1000,NULL,2,'test',true
+3,10,100,1000,NULL,3,'test',true
+4,10,100,1000,NULL,4,'test',true
+---- TYPES
+INT,INT,INT,INT,INT,INT,STRING,BOOLEAN
+====
+---- QUERY
+insert into tbl_with_defaults values (5, 5, 5, 5, 5, 5, 'row', false)
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumRowErrors: 0
+---- LABELS
+A, B, C, D, E, F, G, H
+---- DML_RESULTS: tbl_with_defaults
+1,10,100,1000,NULL,1,'test',true
+2,10,100,1000,NULL,2,'test',true
+3,10,100,1000,NULL,3,'test',true
+4,10,100,1000,NULL,4,'test',true
+5,5,5,5,5,5,'row',false
+---- TYPES
+INT,INT,INT,INT,INT,INT,STRING,BOOLEAN
+====
+---- QUERY
+alter table tbl_with_defaults add columns (i int null, j int not null default 10000)
+---- RESULTS
+====
+---- QUERY
+select * from tbl_with_defaults
+---- RESULTS
+1,10,100,1000,NULL,1,'test',true,NULL,10000
+2,10,100,1000,NULL,2,'test',true,NULL,10000
+3,10,100,1000,NULL,3,'test',true,NULL,10000
+4,10,100,1000,NULL,4,'test',true,NULL,10000
+5,5,5,5,5,5,'row',false,NULL,10000
+---- TYPES
+INT,INT,INT,INT,INT,INT,STRING,BOOLEAN,INT,INT
+====
+---- QUERY
+insert into tbl_with_defaults values (6,6,6,6,6,6,'another row',false,6,6)
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumRowErrors: 0
+---- LABELS
+A, B, C, D, E, F, G, H, I, J
+---- DML_RESULTS: tbl_with_defaults
+1,10,100,1000,NULL,1,'test',true,NULL,10000
+2,10,100,1000,NULL,2,'test',true,NULL,10000
+3,10,100,1000,NULL,3,'test',true,NULL,10000
+4,10,100,1000,NULL,4,'test',true,NULL,10000
+5,5,5,5,5,5,'row',false,NULL,10000
+6,6,6,6,6,6,'another row',false,6,6
+---- TYPES
+INT,INT,INT,INT,INT,INT,STRING,BOOLEAN,INT,INT
+====
[3/8] incubator-impala git commit: IMPALA-4584: Make alter table
operations on Kudu tables synchronous
Posted by kw...@apache.org.
IMPALA-4584: Make alter table operations on Kudu tables synchronous
This commit changes the behavior of alter table operations on Kudu
tables from asynchronous to synchronous. With this change, alter table
operations return when either the operations complete successfully or
a timeout is reached.
Change-Id: I385bce66691ae9040e72f97557e1bba31009e36b
Reviewed-on: http://gerrit.cloudera.org:8080/5364
Reviewed-by: Dimitris Tsirogiannis <dt...@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/5bb9959f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/5bb9959f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/5bb9959f
Branch: refs/heads/master
Commit: 5bb9959fa4149f3b4976ab73c99b802b42e4fafc
Parents: 9f387c8
Author: Dimitris Tsirogiannis <dt...@cloudera.com>
Authored: Mon Dec 5 13:14:50 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Dec 6 03:53:15 2016 +0000
----------------------------------------------------------------------
.../impala/service/KuduCatalogOpExecutor.java | 61 +++++++++++---------
.../queries/QueryTest/kudu_alter.test | 3 +-
2 files changed, 36 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5bb9959f/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
index 8c978b4..0f8f8fd 100644
--- a/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
@@ -288,11 +288,15 @@ public class KuduCatalogOpExecutor {
Preconditions.checkState(!Strings.isNullOrEmpty(newName));
AlterTableOptions alterTableOptions = new AlterTableOptions();
alterTableOptions.renameTable(newName);
+ String errMsg = String.format("Error renaming Kudu table " +
+ "%s to %s", tbl.getKuduTableName(), newName);
try (KuduClient client = KuduUtil.createKuduClient(tbl.getKuduMasterHosts())) {
client.alterTable(tbl.getKuduTableName(), alterTableOptions);
+ if (!client.isAlterTableDone(newName)) {
+ throw new ImpalaRuntimeException(errMsg + ": Kudu operation timed out");
+ }
} catch (KuduException e) {
- throw new ImpalaRuntimeException(String.format("Error renaming Kudu table " +
- "%s to %s", tbl.getName(), newName), e);
+ throw new ImpalaRuntimeException(errMsg, e);
}
}
@@ -316,15 +320,13 @@ public class KuduCatalogOpExecutor {
alterTableOptions.dropRangePartition(lowerBound.first, upperBound.first,
lowerBound.second, upperBound.second);
}
- try (KuduClient client = KuduUtil.createKuduClient(tbl.getKuduMasterHosts())) {
- client.alterTable(tbl.getKuduTableName(), alterTableOptions);
- } catch (KuduException e) {
- if (!params.isIgnore_errors()) {
- throw new ImpalaRuntimeException(String.format("Error %s range partition in " +
- "table %s",
- (type == TRangePartitionOperationType.ADD ? "adding" : "dropping"),
- tbl.getName()), e);
- }
+ String errMsg = String.format("Error %s range partition in " +
+ "table %s", (type == TRangePartitionOperationType.ADD ? "adding" : "dropping"),
+ tbl.getName());
+ try {
+ alterKuduTable(tbl, alterTableOptions, errMsg);
+ } catch (ImpalaRuntimeException e) {
+ if (!params.isIgnore_errors()) throw e;
}
}
@@ -394,12 +396,8 @@ public class KuduCatalogOpExecutor {
}
}
}
- try (KuduClient client = KuduUtil.createKuduClient(tbl.getKuduMasterHosts())) {
- client.alterTable(tbl.getKuduTableName(), alterTableOptions);
- } catch (KuduException e) {
- throw new ImpalaRuntimeException("Error adding columns to Kudu table " +
- tbl.getKuduTableName(), e);
- }
+ String errMsg = "Error adding columns to Kudu table " + tbl.getName();
+ alterKuduTable(tbl, alterTableOptions, errMsg);
}
/**
@@ -410,12 +408,9 @@ public class KuduCatalogOpExecutor {
Preconditions.checkState(!Strings.isNullOrEmpty(colName));
AlterTableOptions alterTableOptions = new AlterTableOptions();
alterTableOptions.dropColumn(colName);
- try (KuduClient client = KuduUtil.createKuduClient(tbl.getKuduMasterHosts())) {
- client.alterTable(tbl.getKuduTableName(), alterTableOptions);
- } catch (KuduException e) {
- throw new ImpalaRuntimeException(String.format("Error dropping column %s from " +
- "Kudu table %s", colName, tbl.getName()), e);
- }
+ String errMsg = String.format("Error dropping column %s from " +
+ "Kudu table %s", colName, tbl.getName());
+ alterKuduTable(tbl, alterTableOptions, errMsg);
}
/**
@@ -427,11 +422,25 @@ public class KuduCatalogOpExecutor {
Preconditions.checkNotNull(newCol);
AlterTableOptions alterTableOptions = new AlterTableOptions();
alterTableOptions.renameColumn(oldName, newCol.getColumnName());
+ String errMsg = String.format("Error renaming column %s to %s " +
+ "for Kudu table %s", oldName, newCol.getColumnName(), tbl.getName());
+ alterKuduTable(tbl, alterTableOptions, errMsg);
+ }
+
+ /**
+ * Alters a Kudu table based on the specified AlterTableOptions params. Blocks until
+ * the alter table operation is finished or until the operation timeout is reached.
+ * Throws an ImpalaRuntimeException if the operation cannot be completed successfully.
+ */
+ public static void alterKuduTable(KuduTable tbl, AlterTableOptions ato, String errMsg)
+ throws ImpalaRuntimeException {
try (KuduClient client = KuduUtil.createKuduClient(tbl.getKuduMasterHosts())) {
- client.alterTable(tbl.getKuduTableName(), alterTableOptions);
+ client.alterTable(tbl.getKuduTableName(), ato);
+ if (!client.isAlterTableDone(tbl.getKuduTableName())) {
+ throw new ImpalaRuntimeException(errMsg + ": Kudu operation timed out");
+ }
} catch (KuduException e) {
- throw new ImpalaRuntimeException(String.format("Error renaming column %s to %s " +
- "for Kudu table %s", oldName, newCol.getColumnName(), tbl.getName()), e);
+ throw new ImpalaRuntimeException(errMsg, e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/5bb9959f/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
index 505e9fe..116a7e9 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
@@ -50,7 +50,6 @@ BIGINT
# Create a table with range distribution
create table tbl_to_alter (id int primary key, name string null, vali bigint not null)
distribute by range (id) (partition 1 < values <= 10) stored as kudu
- tblproperties('kudu.table_name'='tbl_to_alter')
---- RESULTS
====
---- QUERY
@@ -296,7 +295,7 @@ alter table tbl_to_alter set tblproperties('kudu.table_name'='kudu_tbl_to_alter'
---- QUERY
# Create a new table and try to rename to an existing kudu table
create table copy_of_tbl (a int primary key) distribute by hash (a) into 3 buckets
- stored as kudu;
+ stored as kudu tblproperties('kudu.table_name'='copy_of_tbl');
alter table copy_of_tbl set tblproperties('kudu.table_name'='kudu_tbl_to_alter')
---- CATCH
ImpalaRuntimeException: Error renaming Kudu table copy_of_tbl to kudu_tbl_to_alter
[7/8] incubator-impala git commit: IMPALA-4561: Replace DISTRIBUTE BY
with PARTITION BY in CREATE TABLE
Posted by kw...@apache.org.
IMPALA-4561: Replace DISTRIBUTE BY with PARTITION BY in CREATE TABLE
Change-Id: I0e07c41eabb4c8cb95754cf04293cbd9e03d6ab2
Reviewed-on: http://gerrit.cloudera.org:8080/5317
Reviewed-by: Dimitris Tsirogiannis <dt...@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/cba93f1a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/cba93f1a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/cba93f1a
Branch: refs/heads/master
Commit: cba93f1ac3d4c0219c6266924493ad19c8c10556
Parents: f837754
Author: Dimitris Tsirogiannis <dt...@cloudera.com>
Authored: Thu Dec 1 20:43:43 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Dec 6 10:41:53 2016 +0000
----------------------------------------------------------------------
common/thrift/CatalogObjects.thrift | 20 +-
common/thrift/JniCatalog.thrift | 4 +-
fe/src/main/cup/sql-parser.cup | 56 ++---
.../AlterTableAddDropRangePartitionStmt.java | 4 +-
.../analysis/CreateTableAsSelectStmt.java | 3 +-
.../apache/impala/analysis/CreateTableStmt.java | 38 ++--
.../apache/impala/analysis/DistributeParam.java | 211 -------------------
.../impala/analysis/KuduPartitionParam.java | 211 +++++++++++++++++++
.../apache/impala/analysis/RangePartition.java | 18 +-
.../apache/impala/analysis/TableDataLayout.java | 21 +-
.../org/apache/impala/analysis/TableDef.java | 6 +-
.../org/apache/impala/analysis/ToSqlUtils.java | 14 +-
.../org/apache/impala/catalog/KuduTable.java | 81 +++----
.../impala/service/KuduCatalogOpExecutor.java | 36 ++--
.../apache/impala/analysis/AnalyzeDDLTest.java | 164 +++++++-------
.../impala/analysis/AuthorizationTest.java | 2 +-
.../org/apache/impala/analysis/ParserTest.java | 58 ++---
testdata/bin/generate-schema-statements.py | 2 +-
.../functional/functional_schema_template.sql | 28 +--
testdata/datasets/tpcds/tpcds_kudu_template.sql | 48 ++---
testdata/datasets/tpch/tpch_kudu_template.sql | 16 +-
testdata/datasets/tpch/tpch_schema_template.sql | 16 +-
.../queries/PlannerTest/lineage.test | 4 +-
.../queries/QueryTest/kudu-scan-node.test | 6 +-
.../QueryTest/kudu-timeouts-catalogd.test | 2 +-
.../queries/QueryTest/kudu_alter.test | 6 +-
.../queries/QueryTest/kudu_create.test | 24 +--
.../queries/QueryTest/kudu_delete.test | 6 +-
.../queries/QueryTest/kudu_describe.test | 2 +-
.../queries/QueryTest/kudu_insert.test | 10 +-
.../queries/QueryTest/kudu_partition_ddl.test | 28 +--
.../queries/QueryTest/kudu_stats.test | 2 +-
.../queries/QueryTest/kudu_update.test | 2 +-
.../queries/QueryTest/kudu_upsert.test | 4 +-
tests/query_test/test_cancellation.py | 2 +-
tests/query_test/test_kudu.py | 38 ++--
tests/shell/test_shell_commandline.py | 2 +-
37 files changed, 598 insertions(+), 597 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/common/thrift/CatalogObjects.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 10cb777..de89b51 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -357,8 +357,8 @@ struct TDataSourceTable {
2: required string init_string
}
-// Parameters needed for hash distribution
-struct TDistributeByHashParam {
+// Parameters needed for hash partitioning
+struct TKuduPartitionByHashParam {
1: required list<string> columns
2: required i32 num_buckets
}
@@ -370,16 +370,16 @@ struct TRangePartition {
4: optional bool is_upper_bound_inclusive
}
-// A range distribution is identified by a list of columns and a list of range partitions.
-struct TDistributeByRangeParam {
+// A range partitioning is identified by a list of columns and a list of range partitions.
+struct TKuduPartitionByRangeParam {
1: required list<string> columns
2: optional list<TRangePartition> range_partitions
}
-// Parameters for the DISTRIBUTE BY clause.
-struct TDistributeParam {
- 1: optional TDistributeByHashParam by_hash_param;
- 2: optional TDistributeByRangeParam by_range_param;
+// Parameters for the PARTITION BY clause.
+struct TKuduPartitionParam {
+ 1: optional TKuduPartitionByHashParam by_hash_param;
+ 2: optional TKuduPartitionByRangeParam by_range_param;
}
// Represents a Kudu table
@@ -392,8 +392,8 @@ struct TKuduTable {
// Name of the key columns
3: required list<string> key_columns
- // Distribution schemes
- 4: required list<TDistributeParam> distribute_by
+ // Partitioning
+ 4: required list<TKuduPartitionParam> partition_by
}
// Represents a table or view.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/common/thrift/JniCatalog.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/JniCatalog.thrift b/common/thrift/JniCatalog.thrift
index 8bb07e6..d224658 100644
--- a/common/thrift/JniCatalog.thrift
+++ b/common/thrift/JniCatalog.thrift
@@ -415,9 +415,9 @@ struct TCreateTableParams {
// If set, the table will be cached after creation with details specified in cache_op.
13: optional THdfsCachingOp cache_op
- // If set, the table is automatically distributed according to this parameter.
+ // If set, the table is automatically partitioned according to this parameter.
// Kudu-only.
- 14: optional list<CatalogObjects.TDistributeParam> distribute_by
+ 14: optional list<CatalogObjects.TKuduPartitionParam> partition_by
// Primary key column names (Kudu-only)
15: optional list<string> primary_key_column_names;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/cup/sql-parser.cup
----------------------------------------------------------------------
diff --git a/fe/src/main/cup/sql-parser.cup b/fe/src/main/cup/sql-parser.cup
index a375c75..e09993b 100644
--- a/fe/src/main/cup/sql-parser.cup
+++ b/fe/src/main/cup/sql-parser.cup
@@ -401,21 +401,21 @@ nonterminal CreateTableAsSelectStmt create_tbl_as_select_stmt;
nonterminal CreateTableLikeStmt create_tbl_like_stmt;
nonterminal CreateTableStmt create_tbl_stmt;
nonterminal TableDef tbl_def_without_col_defs, tbl_def_with_col_defs;
-nonterminal TableDataLayout opt_tbl_data_layout, distributed_data_layout;
+nonterminal TableDataLayout opt_tbl_data_layout, partitioned_data_layout;
nonterminal TableDef.Options tbl_options;
nonterminal CreateViewStmt create_view_stmt;
nonterminal CreateDataSrcStmt create_data_src_stmt;
nonterminal DropDataSrcStmt drop_data_src_stmt;
nonterminal ShowDataSrcsStmt show_data_srcs_stmt;
nonterminal StructField struct_field_def;
-nonterminal DistributeParam distribute_hash_param;
+nonterminal KuduPartitionParam hash_partition_param;
nonterminal List<RangePartition> range_params_list;
nonterminal RangePartition range_param;
nonterminal Pair<Expr, Boolean> opt_lower_range_val,
opt_upper_range_val;
-nonterminal ArrayList<DistributeParam> distribute_hash_param_list;
-nonterminal ArrayList<DistributeParam> distribute_param_list;
-nonterminal DistributeParam distribute_range_param;
+nonterminal ArrayList<KuduPartitionParam> hash_partition_param_list;
+nonterminal ArrayList<KuduPartitionParam> partition_param_list;
+nonterminal KuduPartitionParam range_partition_param;
nonterminal ColumnDef column_def, view_column_def;
nonterminal ArrayList<ColumnDef> column_def_list, partition_column_defs,
view_column_def_list, view_column_defs;
@@ -1013,12 +1013,12 @@ create_tbl_as_select_stmt ::=
// An optional clause cannot be used directly below because it would conflict with
// the first rule in "create_tbl_stmt".
primary_keys:primary_keys
- distributed_data_layout:distribute_params
+ partitioned_data_layout:partition_params
tbl_options:options
KW_AS query_stmt:select_stmt
{:
tbl_def.getPrimaryKeyColumnNames().addAll(primary_keys);
- tbl_def.getDistributeParams().addAll(distribute_params.getDistributeParams());
+ tbl_def.getKuduPartitionParams().addAll(partition_params.getKuduPartitionParams());
tbl_def.setOptions(options);
RESULT = new CreateTableAsSelectStmt(new CreateTableStmt(tbl_def), select_stmt, null);
:}
@@ -1057,7 +1057,7 @@ create_tbl_stmt ::=
tbl_options:options
{:
tbl_def.getPartitionColumnDefs().addAll(data_layout.getPartitionColumnDefs());
- tbl_def.getDistributeParams().addAll(data_layout.getDistributeParams());
+ tbl_def.getKuduPartitionParams().addAll(data_layout.getKuduPartitionParams());
tbl_def.setOptions(options);
RESULT = new CreateTableStmt(tbl_def);
:}
@@ -1082,7 +1082,7 @@ create_tbl_stmt ::=
tbl_options:options
{:
tbl_def.getPartitionColumnDefs().addAll(data_layout.getPartitionColumnDefs());
- tbl_def.getDistributeParams().addAll(data_layout.getDistributeParams());
+ tbl_def.getKuduPartitionParams().addAll(data_layout.getKuduPartitionParams());
tbl_def.setOptions(options);
RESULT = new CreateTableLikeFileStmt(new CreateTableStmt(tbl_def),
schema_file_format, new HdfsUri(schema_location));
@@ -1148,13 +1148,13 @@ tbl_options ::=
opt_tbl_data_layout ::=
partition_column_defs:partition_column_defs
{: RESULT = TableDataLayout.createPartitionedLayout(partition_column_defs); :}
- | distributed_data_layout:data_layout
+ | partitioned_data_layout:data_layout
{: RESULT = data_layout; :}
;
-distributed_data_layout ::=
- distribute_param_list:distribute_params
- {: RESULT = TableDataLayout.createDistributedLayout(distribute_params); :}
+partitioned_data_layout ::=
+ partition_param_list:partition_params
+ {: RESULT = TableDataLayout.createKuduPartitionedLayout(partition_params); :}
| /* empty */
{: RESULT = TableDataLayout.createEmptyLayout(); :}
;
@@ -1164,25 +1164,25 @@ partition_column_defs ::=
{: RESULT = col_defs; :}
;
-// The DISTRIBUTE clause contains any number of HASH() clauses followed by exactly zero
+// The PARTITION BY clause contains any number of HASH() clauses followed by exactly zero
// or one RANGE clauses
-distribute_param_list ::=
- KW_DISTRIBUTE KW_BY distribute_hash_param_list:list
+partition_param_list ::=
+ KW_PARTITION KW_BY hash_partition_param_list:list
{: RESULT = list; :}
- | KW_DISTRIBUTE KW_BY distribute_range_param:rng
+ | KW_PARTITION KW_BY range_partition_param:rng
{: RESULT = Lists.newArrayList(rng); :}
- | KW_DISTRIBUTE KW_BY distribute_hash_param_list:list COMMA distribute_range_param:rng
+ | KW_PARTITION KW_BY hash_partition_param_list:list COMMA range_partition_param:rng
{:
list.add(rng);
RESULT = list;
:}
;
-// A list of HASH distribution clauses used for flexible partitioning
-distribute_hash_param_list ::=
- distribute_hash_param:dc
+// A list of HASH partitioning clauses used for flexible partitioning
+hash_partition_param_list ::=
+ hash_partition_param:dc
{: RESULT = Lists.newArrayList(dc); :}
- | distribute_hash_param_list:list COMMA distribute_hash_param:d
+ | hash_partition_param_list:list COMMA hash_partition_param:d
{:
list.add(d);
RESULT = list;
@@ -1190,26 +1190,26 @@ distribute_hash_param_list ::=
;
// The column list for a HASH clause is optional.
-distribute_hash_param ::=
+hash_partition_param ::=
KW_HASH LPAREN ident_list:cols RPAREN KW_INTO
INTEGER_LITERAL:buckets KW_BUCKETS
- {: RESULT = DistributeParam.createHashParam(cols, buckets.intValue()); :}
+ {: RESULT = KuduPartitionParam.createHashParam(cols, buckets.intValue()); :}
| KW_HASH KW_INTO INTEGER_LITERAL:buckets KW_BUCKETS
{:
- RESULT = DistributeParam.createHashParam(Lists.<String>newArrayList(),
+ RESULT = KuduPartitionParam.createHashParam(Lists.<String>newArrayList(),
buckets.intValue());
:}
;
// The column list for a RANGE clause is optional.
-distribute_range_param ::=
+range_partition_param ::=
KW_RANGE LPAREN ident_list:cols RPAREN LPAREN range_params_list:ranges RPAREN
{:
- RESULT = DistributeParam.createRangeParam(cols, ranges);
+ RESULT = KuduPartitionParam.createRangeParam(cols, ranges);
:}
| KW_RANGE LPAREN range_params_list:ranges RPAREN
{:
- RESULT = DistributeParam.createRangeParam(Collections.<String>emptyList(), ranges);
+ RESULT = KuduPartitionParam.createRangeParam(Collections.<String>emptyList(), ranges);
:}
;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/AlterTableAddDropRangePartitionStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableAddDropRangePartitionStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableAddDropRangePartitionStmt.java
index b1618e0..aee07f7 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableAddDropRangePartitionStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableAddDropRangePartitionStmt.java
@@ -93,10 +93,10 @@ public class AlterTableAddDropRangePartitionStmt extends AlterTableStmt {
"partitions: RANGE %s", table.getFullName(), rangePartitionSpec_.toSql()));
}
KuduTable kuduTable = (KuduTable) table;
- List<String> colNames = kuduTable.getRangeDistributionColNames();
+ List<String> colNames = kuduTable.getRangePartitioningColNames();
if (colNames.isEmpty()) {
throw new AnalysisException(String.format("Cannot add/drop partition %s: " +
- "Kudu table %s doesn't have a range-based distribution.",
+ "Kudu table %s doesn't have a range-based partitioning.",
rangePartitionSpec_.toSql(), kuduTable.getName()));
}
List<ColumnDef> rangeColDefs = Lists.newArrayListWithCapacity(colNames.size());
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/CreateTableAsSelectStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/CreateTableAsSelectStmt.java b/fe/src/main/java/org/apache/impala/analysis/CreateTableAsSelectStmt.java
index cd7b1c8..eb492c6 100644
--- a/fe/src/main/java/org/apache/impala/analysis/CreateTableAsSelectStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/CreateTableAsSelectStmt.java
@@ -193,7 +193,8 @@ public class CreateTableAsSelectStmt extends StatementBase {
Table tmpTable = null;
if (KuduTable.isKuduTable(msTbl)) {
tmpTable = KuduTable.createCtasTarget(db, msTbl, createStmt_.getColumnDefs(),
- createStmt_.getTblPrimaryKeyColumnNames(), createStmt_.getDistributeParams());
+ createStmt_.getTblPrimaryKeyColumnNames(),
+ createStmt_.getKuduPartitionParams());
} else {
// TODO: Creating a tmp table using load() is confusing.
// Refactor it to use a 'createCtasTarget()' function similar to Kudu table.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java b/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java
index f2db81b..1139005 100644
--- a/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/CreateTableStmt.java
@@ -88,8 +88,8 @@ public class CreateTableStmt extends StatementBase {
public List<ColumnDef> getPartitionColumnDefs() {
return tableDef_.getPartitionColumnDefs();
}
- public List<DistributeParam> getDistributeParams() {
- return tableDef_.getDistributeParams();
+ public List<KuduPartitionParam> getKuduPartitionParams() {
+ return tableDef_.getKuduPartitionParams();
}
public String getComment() { return tableDef_.getComment(); }
Map<String, String> getTblProperties() { return tableDef_.getTblProperties(); }
@@ -146,8 +146,8 @@ public class CreateTableStmt extends StatementBase {
params.setIf_not_exists(getIfNotExists());
params.setTable_properties(getTblProperties());
params.setSerde_properties(getSerdeProperties());
- for (DistributeParam d: getDistributeParams()) {
- params.addToDistribute_by(d.toThrift());
+ for (KuduPartitionParam d: getKuduPartitionParams()) {
+ params.addToPartition_by(d.toThrift());
}
for (ColumnDef pkColDef: getPrimaryKeyColumnDefs()) {
params.addToPrimary_key_column_names(pkColDef.getColName());
@@ -188,8 +188,8 @@ public class CreateTableStmt extends StatementBase {
getTblProperties().get(KuduTable.KEY_STORAGE_HANDLER))) {
throw new AnalysisException(KUDU_STORAGE_HANDLER_ERROR_MESSAGE);
}
- AnalysisUtils.throwIfNotEmpty(getDistributeParams(),
- "Only Kudu tables can use the DISTRIBUTE BY clause.");
+ AnalysisUtils.throwIfNotEmpty(getKuduPartitionParams(),
+ "Only Kudu tables can use the PARTITION BY clause.");
if (hasPrimaryKey()) {
throw new AnalysisException("Only Kudu tables can specify a PRIMARY KEY.");
}
@@ -263,8 +263,8 @@ public class CreateTableStmt extends StatementBase {
KuduTable.KEY_TABLET_REPLICAS));
AnalysisUtils.throwIfNotEmpty(getColumnDefs(),
"Columns cannot be specified with an external Kudu table.");
- AnalysisUtils.throwIfNotEmpty(getDistributeParams(),
- "DISTRIBUTE BY cannot be used with an external Kudu table.");
+ AnalysisUtils.throwIfNotEmpty(getKuduPartitionParams(),
+ "PARTITION BY cannot be used with an external Kudu table.");
}
/**
@@ -305,29 +305,29 @@ public class CreateTableStmt extends StatementBase {
}
}
- if (!getDistributeParams().isEmpty()) {
- analyzeDistributeParams(analyzer);
+ if (!getKuduPartitionParams().isEmpty()) {
+ analyzeKuduPartitionParams(analyzer);
} else {
- throw new AnalysisException("Table distribution must be specified for " +
+ throw new AnalysisException("Table partitioning must be specified for " +
"managed Kudu tables.");
}
}
/**
- * Analyzes the distribution schemes specified in the CREATE TABLE statement.
+ * Analyzes the partitioning schemes specified in the CREATE TABLE statement.
*/
- private void analyzeDistributeParams(Analyzer analyzer) throws AnalysisException {
+ private void analyzeKuduPartitionParams(Analyzer analyzer) throws AnalysisException {
Preconditions.checkState(getFileFormat() == THdfsFileFormat.KUDU);
Map<String, ColumnDef> pkColDefsByName =
ColumnDef.mapByColumnNames(getPrimaryKeyColumnDefs());
- for (DistributeParam distributeParam: getDistributeParams()) {
- // If no column names were specified in this distribution scheme, use all the
+ for (KuduPartitionParam partitionParam: getKuduPartitionParams()) {
+ // If no column names were specified in this partitioning scheme, use all the
// primary key columns.
- if (!distributeParam.hasColumnNames()) {
- distributeParam.setColumnNames(pkColDefsByName.keySet());
+ if (!partitionParam.hasColumnNames()) {
+ partitionParam.setColumnNames(pkColDefsByName.keySet());
}
- distributeParam.setPkColumnDefMap(pkColDefsByName);
- distributeParam.analyze(analyzer);
+ partitionParam.setPkColumnDefMap(pkColDefsByName);
+ partitionParam.analyze(analyzer);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/DistributeParam.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/DistributeParam.java b/fe/src/main/java/org/apache/impala/analysis/DistributeParam.java
deleted file mode 100644
index 0eb1329..0000000
--- a/fe/src/main/java/org/apache/impala/analysis/DistributeParam.java
+++ /dev/null
@@ -1,211 +0,0 @@
-// 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.
-
-package org.apache.impala.analysis;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.impala.common.AnalysisException;
-import org.apache.impala.thrift.TDistributeByHashParam;
-import org.apache.impala.thrift.TDistributeByRangeParam;
-import org.apache.impala.thrift.TDistributeParam;
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-
-/**
- * Represents the distribution of a Kudu table as defined in the DISTRIBUTE BY
- * clause of a CREATE TABLE statement. The distribution can be hash-based or
- * range-based or both. See RangePartition for details on the supported range partitions.
- *
- * Examples:
- * - Hash-based:
- * DISTRIBUTE BY HASH(id) INTO 10 BUCKETS
- * - Single column range-based:
- * DISTRIBUTE BY RANGE(age)
- * (
- * PARTITION VALUES < 10,
- * PARTITION 10 <= VALUES < 20,
- * PARTITION 20 <= VALUES < 30,
- * PARTITION VALUE = 100
- * )
- * - Combination of hash and range based:
- * DISTRIBUTE BY HASH (id) INTO 3 BUCKETS,
- * RANGE (age)
- * (
- * PARTITION 10 <= VALUES < 20,
- * PARTITION VALUE = 100
- * )
- * - Multi-column range based:
- * DISTRIBUTE BY RANGE (year, quarter)
- * (
- * PARTITION VALUE = (2001, 1),
- * PARTITION VALUE = (2001, 2),
- * PARTITION VALUE = (2002, 1)
- * )
- *
- */
-public class DistributeParam implements ParseNode {
-
- /**
- * Creates a hash-based DistributeParam.
- */
- public static DistributeParam createHashParam(List<String> cols, int buckets) {
- return new DistributeParam(Type.HASH, cols, buckets, null);
- }
-
- /**
- * Creates a range-based DistributeParam.
- */
- public static DistributeParam createRangeParam(List<String> cols,
- List<RangePartition> rangePartitions) {
- return new DistributeParam(Type.RANGE, cols, NO_BUCKETS, rangePartitions);
- }
-
- private static final int NO_BUCKETS = -1;
-
- /**
- * The distribution type.
- */
- public enum Type {
- HASH, RANGE
- }
-
- // Columns of this distribution. If no columns are specified, all
- // the primary key columns of the associated table are used.
- private final List<String> colNames_ = Lists.newArrayList();
-
- // Map of primary key column names to the associated column definitions. Must be set
- // before the call to analyze().
- private Map<String, ColumnDef> pkColumnDefByName_;
-
- // Distribution scheme type
- private final Type type_;
-
- // Only relevant for hash-based distribution, -1 otherwise
- private final int numBuckets_;
-
- // List of range partitions specified in a range-based distribution.
- private List<RangePartition> rangePartitions_;
-
- private DistributeParam(Type t, List<String> colNames, int buckets,
- List<RangePartition> partitions) {
- type_ = t;
- for (String name: colNames) colNames_.add(name.toLowerCase());
- rangePartitions_ = partitions;
- numBuckets_ = buckets;
- }
-
- @Override
- public void analyze(Analyzer analyzer) throws AnalysisException {
- Preconditions.checkState(!colNames_.isEmpty());
- Preconditions.checkNotNull(pkColumnDefByName_);
- Preconditions.checkState(!pkColumnDefByName_.isEmpty());
- // Validate that the columns specified in this distribution are primary key columns.
- for (String colName: colNames_) {
- if (!pkColumnDefByName_.containsKey(colName)) {
- throw new AnalysisException(String.format("Column '%s' in '%s' is not a key " +
- "column. Only key columns can be used in DISTRIBUTE BY.", colName, toSql()));
- }
- }
- if (type_ == Type.RANGE) analyzeRangeParam(analyzer);
- }
-
- /**
- * Analyzes a range-based distribution. This function does not check for overlapping
- * range partitions; these checks are performed by Kudu and an error is reported back
- * to the user.
- */
- public void analyzeRangeParam(Analyzer analyzer) throws AnalysisException {
- List<ColumnDef> pkColDefs = Lists.newArrayListWithCapacity(colNames_.size());
- for (String colName: colNames_) pkColDefs.add(pkColumnDefByName_.get(colName));
- for (RangePartition rangePartition: rangePartitions_) {
- rangePartition.analyze(analyzer, pkColDefs);
- }
- }
-
- @Override
- public String toSql() {
- StringBuilder builder = new StringBuilder(type_.toString());
- if (!colNames_.isEmpty()) {
- builder.append(" (");
- Joiner.on(", ").appendTo(builder, colNames_).append(")");
- }
- if (type_ == Type.HASH) {
- builder.append(" INTO ");
- Preconditions.checkState(numBuckets_ != NO_BUCKETS);
- builder.append(numBuckets_).append(" BUCKETS");
- } else {
- builder.append(" (");
- if (rangePartitions_ != null) {
- List<String> partsSql = Lists.newArrayList();
- for (RangePartition rangePartition: rangePartitions_) {
- partsSql.add(rangePartition.toSql());
- }
- builder.append(Joiner.on(", ").join(partsSql));
- } else {
- builder.append("...");
- }
- builder.append(")");
- }
- return builder.toString();
- }
-
- @Override
- public String toString() { return toSql(); }
-
- public TDistributeParam toThrift() {
- TDistributeParam result = new TDistributeParam();
- // TODO: Add a validate() function to ensure the validity of distribute params.
- if (type_ == Type.HASH) {
- TDistributeByHashParam hash = new TDistributeByHashParam();
- Preconditions.checkState(numBuckets_ != NO_BUCKETS);
- hash.setNum_buckets(numBuckets_);
- hash.setColumns(colNames_);
- result.setBy_hash_param(hash);
- } else {
- Preconditions.checkState(type_ == Type.RANGE);
- TDistributeByRangeParam rangeParam = new TDistributeByRangeParam();
- rangeParam.setColumns(colNames_);
- if (rangePartitions_ == null) {
- result.setBy_range_param(rangeParam);
- return result;
- }
- for (RangePartition rangePartition: rangePartitions_) {
- rangeParam.addToRange_partitions(rangePartition.toThrift());
- }
- result.setBy_range_param(rangeParam);
- }
- return result;
- }
-
- void setPkColumnDefMap(Map<String, ColumnDef> pkColumnDefByName) {
- pkColumnDefByName_ = pkColumnDefByName;
- }
-
- boolean hasColumnNames() { return !colNames_.isEmpty(); }
- public List<String> getColumnNames() { return ImmutableList.copyOf(colNames_); }
- void setColumnNames(Collection<String> colNames) {
- Preconditions.checkState(colNames_.isEmpty());
- colNames_.addAll(colNames);
- }
-
- public Type getType() { return type_; }
-}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/KuduPartitionParam.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/KuduPartitionParam.java b/fe/src/main/java/org/apache/impala/analysis/KuduPartitionParam.java
new file mode 100644
index 0000000..3f69fae
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/analysis/KuduPartitionParam.java
@@ -0,0 +1,211 @@
+// 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.
+
+package org.apache.impala.analysis;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.impala.common.AnalysisException;
+import org.apache.impala.thrift.TKuduPartitionByHashParam;
+import org.apache.impala.thrift.TKuduPartitionByRangeParam;
+import org.apache.impala.thrift.TKuduPartitionParam;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+/**
+ * Represents the partitioning of a Kudu table as defined in the PARTITION BY
+ * clause of a CREATE TABLE statement. The partitioning can be hash-based or
+ * range-based or both. See RangePartition for details on the supported range partitions.
+ *
+ * Examples:
+ * - Hash-based:
+ * PARTITION BY HASH(id) INTO 10 BUCKETS
+ * - Single column range-based:
+ * PARTITION BY RANGE(age)
+ * (
+ * PARTITION VALUES < 10,
+ * PARTITION 10 <= VALUES < 20,
+ * PARTITION 20 <= VALUES < 30,
+ * PARTITION VALUE = 100
+ * )
+ * - Combination of hash and range based:
+ * PARTITION BY HASH (id) INTO 3 BUCKETS,
+ * RANGE (age)
+ * (
+ * PARTITION 10 <= VALUES < 20,
+ * PARTITION VALUE = 100
+ * )
+ * - Multi-column range based:
+ * PARTITION BY RANGE (year, quarter)
+ * (
+ * PARTITION VALUE = (2001, 1),
+ * PARTITION VALUE = (2001, 2),
+ * PARTITION VALUE = (2002, 1)
+ * )
+ *
+ */
+public class KuduPartitionParam implements ParseNode {
+
+ /**
+ * Creates a hash-based KuduPartitionParam.
+ */
+ public static KuduPartitionParam createHashParam(List<String> cols, int buckets) {
+ return new KuduPartitionParam(Type.HASH, cols, buckets, null);
+ }
+
+ /**
+ * Creates a range-based KuduPartitionParam.
+ */
+ public static KuduPartitionParam createRangeParam(List<String> cols,
+ List<RangePartition> rangePartitions) {
+ return new KuduPartitionParam(Type.RANGE, cols, NO_BUCKETS, rangePartitions);
+ }
+
+ private static final int NO_BUCKETS = -1;
+
+ /**
+ * The partitioning type.
+ */
+ public enum Type {
+ HASH, RANGE
+ }
+
+ // Columns of this partitioning. If no columns are specified, all
+ // the primary key columns of the associated table are used.
+ private final List<String> colNames_ = Lists.newArrayList();
+
+ // Map of primary key column names to the associated column definitions. Must be set
+ // before the call to analyze().
+ private Map<String, ColumnDef> pkColumnDefByName_;
+
+ // partitioning scheme type
+ private final Type type_;
+
+ // Only relevant for hash-based partitioning, -1 otherwise
+ private final int numBuckets_;
+
+ // List of range partitions specified in a range-based partitioning.
+ private List<RangePartition> rangePartitions_;
+
+ private KuduPartitionParam(Type t, List<String> colNames, int buckets,
+ List<RangePartition> partitions) {
+ type_ = t;
+ for (String name: colNames) colNames_.add(name.toLowerCase());
+ rangePartitions_ = partitions;
+ numBuckets_ = buckets;
+ }
+
+ @Override
+ public void analyze(Analyzer analyzer) throws AnalysisException {
+ Preconditions.checkState(!colNames_.isEmpty());
+ Preconditions.checkNotNull(pkColumnDefByName_);
+ Preconditions.checkState(!pkColumnDefByName_.isEmpty());
+ // Validate that the columns specified in this partitioning are primary key columns.
+ for (String colName: colNames_) {
+ if (!pkColumnDefByName_.containsKey(colName)) {
+ throw new AnalysisException(String.format("Column '%s' in '%s' is not a key " +
+ "column. Only key columns can be used in PARTITION BY.", colName, toSql()));
+ }
+ }
+ if (type_ == Type.RANGE) analyzeRangeParam(analyzer);
+ }
+
+ /**
+ * Analyzes a range-based partitioning. This function does not check for overlapping
+ * range partitions; these checks are performed by Kudu and an error is reported back
+ * to the user.
+ */
+ public void analyzeRangeParam(Analyzer analyzer) throws AnalysisException {
+ List<ColumnDef> pkColDefs = Lists.newArrayListWithCapacity(colNames_.size());
+ for (String colName: colNames_) pkColDefs.add(pkColumnDefByName_.get(colName));
+ for (RangePartition rangePartition: rangePartitions_) {
+ rangePartition.analyze(analyzer, pkColDefs);
+ }
+ }
+
+ @Override
+ public String toSql() {
+ StringBuilder builder = new StringBuilder(type_.toString());
+ if (!colNames_.isEmpty()) {
+ builder.append(" (");
+ Joiner.on(", ").appendTo(builder, colNames_).append(")");
+ }
+ if (type_ == Type.HASH) {
+ builder.append(" INTO ");
+ Preconditions.checkState(numBuckets_ != NO_BUCKETS);
+ builder.append(numBuckets_).append(" BUCKETS");
+ } else {
+ builder.append(" (");
+ if (rangePartitions_ != null) {
+ List<String> partsSql = Lists.newArrayList();
+ for (RangePartition rangePartition: rangePartitions_) {
+ partsSql.add(rangePartition.toSql());
+ }
+ builder.append(Joiner.on(", ").join(partsSql));
+ } else {
+ builder.append("...");
+ }
+ builder.append(")");
+ }
+ return builder.toString();
+ }
+
+ @Override
+ public String toString() { return toSql(); }
+
+ public TKuduPartitionParam toThrift() {
+ TKuduPartitionParam result = new TKuduPartitionParam();
+ // TODO: Add a validate() function to ensure the validity of distribute params.
+ if (type_ == Type.HASH) {
+ TKuduPartitionByHashParam hash = new TKuduPartitionByHashParam();
+ Preconditions.checkState(numBuckets_ != NO_BUCKETS);
+ hash.setNum_buckets(numBuckets_);
+ hash.setColumns(colNames_);
+ result.setBy_hash_param(hash);
+ } else {
+ Preconditions.checkState(type_ == Type.RANGE);
+ TKuduPartitionByRangeParam rangeParam = new TKuduPartitionByRangeParam();
+ rangeParam.setColumns(colNames_);
+ if (rangePartitions_ == null) {
+ result.setBy_range_param(rangeParam);
+ return result;
+ }
+ for (RangePartition rangePartition: rangePartitions_) {
+ rangeParam.addToRange_partitions(rangePartition.toThrift());
+ }
+ result.setBy_range_param(rangeParam);
+ }
+ return result;
+ }
+
+ void setPkColumnDefMap(Map<String, ColumnDef> pkColumnDefByName) {
+ pkColumnDefByName_ = pkColumnDefByName;
+ }
+
+ boolean hasColumnNames() { return !colNames_.isEmpty(); }
+ public List<String> getColumnNames() { return ImmutableList.copyOf(colNames_); }
+ void setColumnNames(Collection<String> colNames) {
+ Preconditions.checkState(colNames_.isEmpty());
+ colNames_.addAll(colNames);
+ }
+
+ public Type getType() { return type_; }
+}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/RangePartition.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/RangePartition.java b/fe/src/main/java/org/apache/impala/analysis/RangePartition.java
index 3e41bc4..e2640a8 100644
--- a/fe/src/main/java/org/apache/impala/analysis/RangePartition.java
+++ b/fe/src/main/java/org/apache/impala/analysis/RangePartition.java
@@ -110,26 +110,26 @@ public class RangePartition implements ParseNode {
throw new IllegalStateException("Not implemented");
}
- public void analyze(Analyzer analyzer, List<ColumnDef> distributionColDefs)
+ public void analyze(Analyzer analyzer, List<ColumnDef> partColDefs)
throws AnalysisException {
- analyzeBoundaryValues(lowerBound_, distributionColDefs, analyzer);
+ analyzeBoundaryValues(lowerBound_, partColDefs, analyzer);
if (!isSingletonRange_) {
- analyzeBoundaryValues(upperBound_, distributionColDefs, analyzer);
+ analyzeBoundaryValues(upperBound_, partColDefs, analyzer);
}
}
private void analyzeBoundaryValues(List<Expr> boundaryValues,
- List<ColumnDef> distributionColDefs, Analyzer analyzer) throws AnalysisException {
+ List<ColumnDef> partColDefs, Analyzer analyzer) throws AnalysisException {
if (!boundaryValues.isEmpty()
- && boundaryValues.size() != distributionColDefs.size()) {
+ && boundaryValues.size() != partColDefs.size()) {
throw new AnalysisException(String.format("Number of specified range " +
- "partition values is different than the number of distribution " +
+ "partition values is different than the number of partitioning " +
"columns: (%d vs %d). Range partition: '%s'", boundaryValues.size(),
- distributionColDefs.size(), toSql()));
+ partColDefs.size(), toSql()));
}
for (int i = 0; i < boundaryValues.size(); ++i) {
LiteralExpr literal = analyzeBoundaryValue(boundaryValues.get(i),
- distributionColDefs.get(i), analyzer);
+ partColDefs.get(i), analyzer);
Preconditions.checkNotNull(literal);
boundaryValues.set(i, literal);
}
@@ -162,7 +162,7 @@ public class RangePartition implements ParseNode {
if (!org.apache.impala.catalog.Type.isImplicitlyCastable(literalType, colType,
true)) {
throw new AnalysisException(String.format("Range partition value %s " +
- "(type: %s) is not type compatible with distribution column '%s' (type: %s).",
+ "(type: %s) is not type compatible with partitioning column '%s' (type: %s).",
literal.toSql(), literalType, pkColumn.getColName(), colType.toSql()));
}
if (!literalType.equals(colType)) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/TableDataLayout.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/TableDataLayout.java b/fe/src/main/java/org/apache/impala/analysis/TableDataLayout.java
index 4d3ed80..aef5732 100644
--- a/fe/src/main/java/org/apache/impala/analysis/TableDataLayout.java
+++ b/fe/src/main/java/org/apache/impala/analysis/TableDataLayout.java
@@ -22,35 +22,34 @@ import com.google.common.collect.Lists;
import java.util.List;
/**
- * Represents the PARTITION BY and DISTRIBUTED BY clauses of a DDL statement.
- * TODO: Reconsider this class when we add support for new range partitioning syntax (see
- * IMPALA-3724).
+ * Represents the PARTITION BY and PARTITIONED BY clauses of a DDL statement.
*/
class TableDataLayout {
private final List<ColumnDef> partitionColDefs_;
- private final List<DistributeParam> distributeParams_;
+ private final List<KuduPartitionParam> kuduPartitionParams_;
private TableDataLayout(List<ColumnDef> partitionColumnDefs,
- List<DistributeParam> distributeParams) {
+ List<KuduPartitionParam> partitionParams) {
partitionColDefs_ = partitionColumnDefs;
- distributeParams_ = distributeParams;
+ kuduPartitionParams_ = partitionParams;
}
static TableDataLayout createPartitionedLayout(List<ColumnDef> partitionColumnDefs) {
return new TableDataLayout(partitionColumnDefs,
- Lists.<DistributeParam>newArrayList());
+ Lists.<KuduPartitionParam>newArrayList());
}
- static TableDataLayout createDistributedLayout(List<DistributeParam> distributeParams) {
- return new TableDataLayout(Lists.<ColumnDef>newArrayList(), distributeParams);
+ static TableDataLayout createKuduPartitionedLayout(
+ List<KuduPartitionParam> partitionParams) {
+ return new TableDataLayout(Lists.<ColumnDef>newArrayList(), partitionParams);
}
static TableDataLayout createEmptyLayout() {
return new TableDataLayout(Lists.<ColumnDef>newArrayList(),
- Lists.<DistributeParam>newArrayList());
+ Lists.<KuduPartitionParam>newArrayList());
}
List<ColumnDef> getPartitionColumnDefs() { return partitionColDefs_; }
- List<DistributeParam> getDistributeParams() { return distributeParams_; }
+ List<KuduPartitionParam> getKuduPartitionParams() { return kuduPartitionParams_; }
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/TableDef.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/TableDef.java b/fe/src/main/java/org/apache/impala/analysis/TableDef.java
index 1c16954..a40b4d2 100644
--- a/fe/src/main/java/org/apache/impala/analysis/TableDef.java
+++ b/fe/src/main/java/org/apache/impala/analysis/TableDef.java
@@ -74,7 +74,7 @@ class TableDef {
// If true, no errors are thrown if the table already exists.
private final boolean ifNotExists_;
- // Partitioned/distribute by parameters.
+ // Partitioning parameters.
private final TableDataLayout dataLayout_;
// True if analyze() has been called.
@@ -152,8 +152,8 @@ class TableDef {
List<ColumnDef> getPrimaryKeyColumnDefs() { return primaryKeyColDefs_; }
boolean isExternal() { return isExternal_; }
boolean getIfNotExists() { return ifNotExists_; }
- List<DistributeParam> getDistributeParams() {
- return dataLayout_.getDistributeParams();
+ List<KuduPartitionParam> getKuduPartitionParams() {
+ return dataLayout_.getKuduPartitionParams();
}
void setOptions(Options options) {
Preconditions.checkNotNull(options);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java b/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
index 4cd095c..be4ab6f 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ToSqlUtils.java
@@ -197,7 +197,7 @@ public class ToSqlUtils {
String storageHandlerClassName = table.getStorageHandlerClassName();
List<String> primaryKeySql = Lists.newArrayList();
- String kuduDistributeByParams = null;
+ String kuduPartitionByParams = null;
if (table instanceof KuduTable) {
KuduTable kuduTable = (KuduTable) table;
// Kudu tables don't use LOCATION syntax
@@ -219,10 +219,10 @@ public class ToSqlUtils {
primaryKeySql.addAll(kuduTable.getPrimaryKeyColumnNames());
List<String> paramsSql = Lists.newArrayList();
- for (DistributeParam param: kuduTable.getDistributeBy()) {
+ for (KuduPartitionParam param: kuduTable.getPartitionBy()) {
paramsSql.add(param.toSql());
}
- kuduDistributeByParams = Joiner.on(", ").join(paramsSql);
+ kuduPartitionByParams = Joiner.on(", ").join(paramsSql);
} else {
// We shouldn't output the columns for external tables
colsSql = null;
@@ -230,7 +230,7 @@ public class ToSqlUtils {
}
HdfsUri tableLocation = location == null ? null : new HdfsUri(location);
return getCreateTableSql(table.getDb().getName(), table.getName(), comment, colsSql,
- partitionColsSql, primaryKeySql, kuduDistributeByParams, properties,
+ partitionColsSql, primaryKeySql, kuduPartitionByParams, properties,
serdeParameters, isExternal, false, rowFormat, format, compression,
storageHandlerClassName, tableLocation);
}
@@ -242,7 +242,7 @@ public class ToSqlUtils {
*/
public static String getCreateTableSql(String dbName, String tableName,
String tableComment, List<String> columnsSql, List<String> partitionColumnsSql,
- List<String> primaryKeysSql, String kuduDistributeByParams,
+ List<String> primaryKeysSql, String kuduPartitionByParams,
Map<String, String> tblProperties, Map<String, String> serdeParameters,
boolean isExternal, boolean ifNotExists, RowFormat rowFormat,
HdfsFileFormat fileFormat, HdfsCompression compression, String storageHandlerClass,
@@ -271,8 +271,8 @@ public class ToSqlUtils {
Joiner.on(", \n ").join(partitionColumnsSql)));
}
- if (kuduDistributeByParams != null) {
- sb.append("DISTRIBUTE BY " + kuduDistributeByParams + "\n");
+ if (kuduPartitionByParams != null) {
+ sb.append("PARTITION BY " + kuduPartitionByParams + "\n");
}
if (rowFormat != null && !rowFormat.isDefault()) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
index a7f72c3..9bbcbd5 100644
--- a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
@@ -29,15 +29,15 @@ import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.impala.analysis.ColumnDef;
-import org.apache.impala.analysis.DistributeParam;
+import org.apache.impala.analysis.KuduPartitionParam;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.service.CatalogOpExecutor;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TColumn;
-import org.apache.impala.thrift.TDistributeByHashParam;
-import org.apache.impala.thrift.TDistributeByRangeParam;
-import org.apache.impala.thrift.TDistributeParam;
+import org.apache.impala.thrift.TKuduPartitionByHashParam;
+import org.apache.impala.thrift.TKuduPartitionByRangeParam;
+import org.apache.impala.thrift.TKuduPartitionParam;
import org.apache.impala.thrift.TKuduTable;
import org.apache.impala.thrift.TResultSet;
import org.apache.impala.thrift.TResultSetMetadata;
@@ -110,9 +110,9 @@ public class KuduTable extends Table {
// Primary key column names.
private final List<String> primaryKeyColumnNames_ = Lists.newArrayList();
- // Distribution schemes of this Kudu table. Both range and hash-based distributions are
+ // Partitioning schemes of this Kudu table. Both range and hash-based partitioning are
// supported.
- private final List<DistributeParam> distributeBy_ = Lists.newArrayList();
+ private final List<KuduPartitionParam> partitionBy_ = Lists.newArrayList();
// Schema of the underlying Kudu table.
private org.apache.kudu.Schema kuduSchema_;
@@ -148,36 +148,36 @@ public class KuduTable extends Table {
return ImmutableList.copyOf(primaryKeyColumnNames_);
}
- public List<DistributeParam> getDistributeBy() {
- return ImmutableList.copyOf(distributeBy_);
+ public List<KuduPartitionParam> getPartitionBy() {
+ return ImmutableList.copyOf(partitionBy_);
}
/**
- * Returns the range-based distribution of this table if it exists, null otherwise.
+ * Returns the range-based partitioning of this table if it exists, null otherwise.
*/
- private DistributeParam getRangeDistribution() {
- for (DistributeParam distributeParam: distributeBy_) {
- if (distributeParam.getType() == DistributeParam.Type.RANGE) {
- return distributeParam;
+ private KuduPartitionParam getRangePartitioning() {
+ for (KuduPartitionParam partitionParam: partitionBy_) {
+ if (partitionParam.getType() == KuduPartitionParam.Type.RANGE) {
+ return partitionParam;
}
}
return null;
}
/**
- * Returns the column names of the table's range-based distribution or an empty
- * list if the table doesn't have a range-based distribution.
+ * Returns the column names of the table's range-based partitioning or an empty
+ * list if the table doesn't have a range-based partitioning.
*/
- public List<String> getRangeDistributionColNames() {
- DistributeParam rangeDistribution = getRangeDistribution();
- if (rangeDistribution == null) return Collections.<String>emptyList();
- return rangeDistribution.getColumnNames();
+ public List<String> getRangePartitioningColNames() {
+ KuduPartitionParam rangePartitioning = getRangePartitioning();
+ if (rangePartitioning == null) return Collections.<String>emptyList();
+ return rangePartitioning.getColumnNames();
}
/**
* Loads the metadata of a Kudu table.
*
- * Schema and distribution schemes are loaded directly from Kudu whereas column stats
+ * Schema and partitioning schemes are loaded directly from Kudu whereas column stats
* are loaded from HMS. The function also updates the table schema in HMS in order to
* propagate alterations made to the Kudu table to HMS.
*/
@@ -209,7 +209,7 @@ public class KuduTable extends Table {
// Load metadata from Kudu and HMS
try {
loadSchema(kuduTable);
- loadDistributeByParams(kuduTable);
+ loadPartitionByParams(kuduTable);
loadAllColumnStats(msClient);
} catch (ImpalaRuntimeException e) {
throw new TableLoadingException("Error loading metadata for Kudu table " +
@@ -255,19 +255,19 @@ public class KuduTable extends Table {
}
}
- private void loadDistributeByParams(org.apache.kudu.client.KuduTable kuduTable) {
+ private void loadPartitionByParams(org.apache.kudu.client.KuduTable kuduTable) {
Preconditions.checkNotNull(kuduTable);
Schema tableSchema = kuduTable.getSchema();
PartitionSchema partitionSchema = kuduTable.getPartitionSchema();
Preconditions.checkState(!colsByPos_.isEmpty());
- distributeBy_.clear();
+ partitionBy_.clear();
for (HashBucketSchema hashBucketSchema: partitionSchema.getHashBucketSchemas()) {
List<String> columnNames = Lists.newArrayList();
for (int colId: hashBucketSchema.getColumnIds()) {
columnNames.add(getColumnNameById(tableSchema, colId));
}
- distributeBy_.add(
- DistributeParam.createHashParam(columnNames, hashBucketSchema.getNumBuckets()));
+ partitionBy_.add(KuduPartitionParam.createHashParam(columnNames,
+ hashBucketSchema.getNumBuckets()));
}
RangeSchema rangeSchema = partitionSchema.getRangeSchema();
List<Integer> columnIds = rangeSchema.getColumns();
@@ -277,7 +277,7 @@ public class KuduTable extends Table {
// We don't populate the split values because Kudu's API doesn't currently support
// retrieving the split values for range partitions.
// TODO: File a Kudu JIRA.
- distributeBy_.add(DistributeParam.createRangeParam(columnNames, null));
+ partitionBy_.add(KuduPartitionParam.createRangeParam(columnNames, null));
}
/**
@@ -297,14 +297,14 @@ public class KuduTable extends Table {
*/
public static KuduTable createCtasTarget(Db db,
org.apache.hadoop.hive.metastore.api.Table msTbl, List<ColumnDef> columnDefs,
- List<String> primaryKeyColumnNames, List<DistributeParam> distributeParams) {
+ List<String> primaryKeyColumnNames, List<KuduPartitionParam> partitionParams) {
KuduTable tmpTable = new KuduTable(msTbl, db, msTbl.getTableName(), msTbl.getOwner());
int pos = 0;
for (ColumnDef colDef: columnDefs) {
tmpTable.addColumn(new Column(colDef.getColName(), colDef.getType(), pos++));
}
tmpTable.primaryKeyColumnNames_.addAll(primaryKeyColumnNames);
- tmpTable.distributeBy_.addAll(distributeParams);
+ tmpTable.partitionBy_.addAll(partitionParams);
return tmpTable;
}
@@ -324,27 +324,28 @@ public class KuduTable extends Table {
kuduMasters_ = Joiner.on(',').join(tkudu.getMaster_addresses());
primaryKeyColumnNames_.clear();
primaryKeyColumnNames_.addAll(tkudu.getKey_columns());
- loadDistributeByParamsFromThrift(tkudu.getDistribute_by());
+ loadPartitionByParamsFromThrift(tkudu.getPartition_by());
}
- private void loadDistributeByParamsFromThrift(List<TDistributeParam> params) {
- distributeBy_.clear();
- for (TDistributeParam param: params) {
+ private void loadPartitionByParamsFromThrift(List<TKuduPartitionParam> params) {
+ partitionBy_.clear();
+ for (TKuduPartitionParam param: params) {
if (param.isSetBy_hash_param()) {
- TDistributeByHashParam hashParam = param.getBy_hash_param();
- distributeBy_.add(DistributeParam.createHashParam(
+ TKuduPartitionByHashParam hashParam = param.getBy_hash_param();
+ partitionBy_.add(KuduPartitionParam.createHashParam(
hashParam.getColumns(), hashParam.getNum_buckets()));
} else {
Preconditions.checkState(param.isSetBy_range_param());
- TDistributeByRangeParam rangeParam = param.getBy_range_param();
- distributeBy_.add(DistributeParam.createRangeParam(rangeParam.getColumns(),
+ TKuduPartitionByRangeParam rangeParam = param.getBy_range_param();
+ partitionBy_.add(KuduPartitionParam.createRangeParam(rangeParam.getColumns(),
null));
}
}
}
@Override
- public TTableDescriptor toThriftDescriptor(int tableId, Set<Long> referencedPartitions) {
+ public TTableDescriptor toThriftDescriptor(int tableId,
+ Set<Long> referencedPartitions) {
TTableDescriptor desc = new TTableDescriptor(tableId, TTableType.KUDU_TABLE,
getTColumnDescriptors(), numClusteringCols_, kuduTableName_, db_.getName());
desc.setKuduTable(getTKuduTable());
@@ -356,9 +357,9 @@ public class KuduTable extends Table {
tbl.setKey_columns(Preconditions.checkNotNull(primaryKeyColumnNames_));
tbl.setMaster_addresses(Lists.newArrayList(kuduMasters_.split(",")));
tbl.setTable_name(kuduTableName_);
- Preconditions.checkNotNull(distributeBy_);
- for (DistributeParam distributeParam: distributeBy_) {
- tbl.addToDistribute_by(distributeParam.toThrift());
+ Preconditions.checkNotNull(partitionBy_);
+ for (KuduPartitionParam partitionParam: partitionBy_) {
+ tbl.addToPartition_by(partitionParam.toThrift());
}
return tbl;
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
index 0f8f8fd..b0616c4 100644
--- a/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
@@ -34,7 +34,7 @@ import org.apache.impala.common.Pair;
import org.apache.impala.thrift.TAlterTableAddDropRangePartitionParams;
import org.apache.impala.thrift.TColumn;
import org.apache.impala.thrift.TCreateTableParams;
-import org.apache.impala.thrift.TDistributeParam;
+import org.apache.impala.thrift.TKuduPartitionParam;
import org.apache.impala.thrift.TRangePartition;
import org.apache.impala.thrift.TRangePartitionOperationType;
import org.apache.impala.util.KuduUtil;
@@ -138,22 +138,22 @@ public class KuduCatalogOpExecutor {
org.apache.hadoop.hive.metastore.api.Table msTbl,
TCreateTableParams params, Schema schema) throws ImpalaRuntimeException {
CreateTableOptions tableOpts = new CreateTableOptions();
- // Set the distribution schemes
- List<TDistributeParam> distributeParams = params.getDistribute_by();
- if (distributeParams != null) {
+ // Set the partitioning schemes
+ List<TKuduPartitionParam> partitionParams = params.getPartition_by();
+ if (partitionParams != null) {
boolean hasRangePartitioning = false;
- for (TDistributeParam distParam: distributeParams) {
- if (distParam.isSetBy_hash_param()) {
- Preconditions.checkState(!distParam.isSetBy_range_param());
- tableOpts.addHashPartitions(distParam.getBy_hash_param().getColumns(),
- distParam.getBy_hash_param().getNum_buckets());
+ for (TKuduPartitionParam partParam: partitionParams) {
+ if (partParam.isSetBy_hash_param()) {
+ Preconditions.checkState(!partParam.isSetBy_range_param());
+ tableOpts.addHashPartitions(partParam.getBy_hash_param().getColumns(),
+ partParam.getBy_hash_param().getNum_buckets());
} else {
- Preconditions.checkState(distParam.isSetBy_range_param());
+ Preconditions.checkState(partParam.isSetBy_range_param());
hasRangePartitioning = true;
- List<String> rangePartitionColumns = distParam.getBy_range_param().getColumns();
+ List<String> rangePartitionColumns = partParam.getBy_range_param().getColumns();
tableOpts.setRangePartitionColumns(rangePartitionColumns);
for (TRangePartition rangePartition:
- distParam.getBy_range_param().getRange_partitions()) {
+ partParam.getBy_range_param().getRange_partitions()) {
List<Pair<PartialRow, RangePartitionBound>> rangeBounds =
getRangePartitionBounds(rangePartition, schema, rangePartitionColumns);
Preconditions.checkState(rangeBounds.size() == 2);
@@ -164,7 +164,7 @@ public class KuduCatalogOpExecutor {
}
}
}
- // If no range-based distribution is specified in a CREATE TABLE statement, Kudu
+ // If no range-based partitioning is specified in a CREATE TABLE statement, Kudu
// generates one by default that includes all the primary key columns. To prevent
// this from happening, explicitly set the range partition columns to be
// an empty list.
@@ -333,7 +333,7 @@ public class KuduCatalogOpExecutor {
private static List<Pair<PartialRow, RangePartitionBound>> getRangePartitionBounds(
TRangePartition rangePartition, KuduTable tbl) throws ImpalaRuntimeException {
return getRangePartitionBounds(rangePartition, tbl.getKuduSchema(),
- tbl.getRangeDistributionColNames());
+ tbl.getRangePartitioningColNames());
}
/**
@@ -342,20 +342,20 @@ public class KuduCatalogOpExecutor {
*/
private static List<Pair<PartialRow, RangePartitionBound>> getRangePartitionBounds(
TRangePartition rangePartition, Schema schema,
- List<String> rangeDistributionColNames) throws ImpalaRuntimeException {
+ List<String> rangePartitioningColNames) throws ImpalaRuntimeException {
Preconditions.checkNotNull(schema);
- Preconditions.checkState(!rangeDistributionColNames.isEmpty());
+ Preconditions.checkState(!rangePartitioningColNames.isEmpty());
Preconditions.checkState(rangePartition.isSetLower_bound_values()
|| rangePartition.isSetUpper_bound_values());
List<Pair<PartialRow, RangePartitionBound>> rangeBounds =
Lists.newArrayListWithCapacity(2);
Pair<PartialRow, RangePartitionBound> lowerBound =
- KuduUtil.buildRangePartitionBound(schema, rangeDistributionColNames,
+ KuduUtil.buildRangePartitionBound(schema, rangePartitioningColNames,
rangePartition.getLower_bound_values(),
rangePartition.isIs_lower_bound_inclusive());
rangeBounds.add(lowerBound);
Pair<PartialRow, RangePartitionBound> upperBound =
- KuduUtil.buildRangePartitionBound(schema, rangeDistributionColNames,
+ KuduUtil.buildRangePartitionBound(schema, rangePartitioningColNames,
rangePartition.getUpper_bound_values(),
rangePartition.isIs_upper_bound_inclusive());
rangeBounds.add(upperBound);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java b/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
index 4aeb96d..d806a2f 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
@@ -1440,16 +1440,16 @@ public class AnalyzeDDLTest extends FrontendTestBase {
"Partition column name mismatch: tinyint_col != int_col");
// CTAS into managed Kudu tables
- AnalyzesOk("create table t primary key (id) distribute by hash (id) into 3 buckets" +
+ AnalyzesOk("create table t primary key (id) partition by hash (id) into 3 buckets" +
" stored as kudu as select id, bool_col, tinyint_col, smallint_col, int_col, " +
"bigint_col, float_col, double_col, date_string_col, string_col " +
"from functional.alltypestiny");
- AnalyzesOk("create table t primary key (id) distribute by range (id) " +
+ AnalyzesOk("create table t primary key (id) partition by range (id) " +
"(partition values < 10, partition 20 <= values < 30, partition value = 50) " +
"stored as kudu as select id, bool_col, tinyint_col, smallint_col, int_col, " +
"bigint_col, float_col, double_col, date_string_col, string_col " +
"from functional.alltypestiny");
- AnalyzesOk("create table t primary key (id) distribute by hash (id) into 3 buckets, "+
+ AnalyzesOk("create table t primary key (id) partition by hash (id) into 3 buckets, "+
"range (id) (partition values < 10, partition 10 <= values < 20, " +
"partition value = 30) stored as kudu as select id, bool_col, tinyint_col, " +
"smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, " +
@@ -1461,27 +1461,27 @@ public class AnalyzeDDLTest extends FrontendTestBase {
"external Kudu tables.");
// CTAS into Kudu tables with unsupported types
- AnalysisError("create table t primary key (id) distribute by hash into 3 buckets" +
+ AnalysisError("create table t primary key (id) partition by hash into 3 buckets" +
" stored as kudu as select id, timestamp_col from functional.alltypestiny",
"Cannot create table 't': Type TIMESTAMP is not supported in Kudu");
- AnalysisError("create table t primary key (cs) distribute by hash into 3 buckets" +
+ AnalysisError("create table t primary key (cs) partition by hash into 3 buckets" +
" stored as kudu as select cs from functional.chars_tiny",
"Cannot create table 't': Type CHAR(5) is not supported in Kudu");
- AnalysisError("create table t primary key (vc) distribute by hash into 3 buckets" +
+ AnalysisError("create table t primary key (vc) partition by hash into 3 buckets" +
" stored as kudu as select vc from functional.chars_tiny",
"Cannot create table 't': Type VARCHAR(32) is not supported in Kudu");
- AnalysisError("create table t primary key (id) distribute by hash into 3 buckets" +
+ AnalysisError("create table t primary key (id) partition by hash into 3 buckets" +
" stored as kudu as select c1 as id from functional.decimal_tiny",
"Cannot create table 't': Type DECIMAL(10,4) is not supported in Kudu");
- AnalysisError("create table t primary key (id) distribute by hash into 3 buckets" +
+ AnalysisError("create table t primary key (id) partition by hash into 3 buckets" +
" stored as kudu as select id, s from functional.complextypes_fileformat",
"Expr 's' in select list returns a complex type 'STRUCT<f1:STRING,f2:INT>'.\n" +
"Only scalar types are allowed in the select list.");
- AnalysisError("create table t primary key (id) distribute by hash into 3 buckets" +
+ AnalysisError("create table t primary key (id) partition by hash into 3 buckets" +
" stored as kudu as select id, m from functional.complextypes_fileformat",
"Expr 'm' in select list returns a complex type 'MAP<STRING,BIGINT>'.\n" +
"Only scalar types are allowed in the select list.");
- AnalysisError("create table t primary key (id) distribute by hash into 3 buckets" +
+ AnalysisError("create table t primary key (id) partition by hash into 3 buckets" +
" stored as kudu as select id, a from functional.complextypes_fileformat",
"Expr 'a' in select list returns a complex type 'ARRAY<INT>'.\n" +
"Only scalar types are allowed in the select list.");
@@ -1912,85 +1912,85 @@ public class AnalyzeDDLTest extends FrontendTestBase {
@Test
public void TestCreateManagedKuduTable() {
TestUtils.assumeKuduIsSupported();
- // Test primary keys and distribute by clauses
- AnalyzesOk("create table tab (x int primary key) distribute by hash(x) " +
+ // Test primary keys and partition by clauses
+ AnalyzesOk("create table tab (x int primary key) partition by hash(x) " +
"into 8 buckets stored as kudu");
- AnalyzesOk("create table tab (x int, primary key(x)) distribute by hash(x) " +
+ AnalyzesOk("create table tab (x int, primary key(x)) partition by hash(x) " +
"into 8 buckets stored as kudu");
AnalyzesOk("create table tab (x int, y int, primary key (x, y)) " +
- "distribute by hash(x, y) into 8 buckets stored as kudu");
+ "partition by hash(x, y) into 8 buckets stored as kudu");
AnalyzesOk("create table tab (x int, y int, primary key (x)) " +
- "distribute by hash(x) into 8 buckets stored as kudu");
+ "partition by hash(x) into 8 buckets stored as kudu");
AnalyzesOk("create table tab (x int, y int, primary key(x, y)) " +
- "distribute by hash(y) into 8 buckets stored as kudu");
- AnalyzesOk("create table tab (x int, y string, primary key (x)) distribute by " +
+ "partition by hash(y) into 8 buckets stored as kudu");
+ AnalyzesOk("create table tab (x int, y string, primary key (x)) partition by " +
"hash (x) into 3 buckets, range (x) (partition values < 1, partition " +
"1 <= values < 10, partition 10 <= values < 20, partition value = 30) " +
"stored as kudu");
- AnalyzesOk("create table tab (x int, y int, primary key (x, y)) distribute by " +
+ AnalyzesOk("create table tab (x int, y int, primary key (x, y)) partition by " +
"range (x, y) (partition value = (2001, 1), partition value = (2002, 1), " +
"partition value = (2003, 2)) stored as kudu");
// Non-literal boundary values in range partitions
- AnalyzesOk("create table tab (x int, y int, primary key (x)) distribute by " +
+ AnalyzesOk("create table tab (x int, y int, primary key (x)) partition by " +
"range (x) (partition values < 1 + 1, partition (1+3) + 2 < values < 10, " +
"partition factorial(4) < values < factorial(5), " +
"partition value = factorial(6)) stored as kudu");
- AnalyzesOk("create table tab (x int, y int, primary key(x, y)) distribute by " +
+ AnalyzesOk("create table tab (x int, y int, primary key(x, y)) partition by " +
"range(x, y) (partition value = (1+1, 2+2), partition value = ((1+1+1)+1, 10), " +
"partition value = (cast (30 as int), factorial(5))) stored as kudu");
- AnalysisError("create table tab (x int primary key) distribute by range (x) " +
+ AnalysisError("create table tab (x int primary key) partition by range (x) " +
"(partition values < x + 1) stored as kudu", "Only constant values are allowed " +
"for range-partition bounds: x + 1");
- AnalysisError("create table tab (x int primary key) distribute by range (x) " +
+ AnalysisError("create table tab (x int primary key) partition by range (x) " +
"(partition values <= isnull(null, null)) stored as kudu", "Range partition " +
"values cannot be NULL. Range partition: 'PARTITION VALUES <= " +
"isnull(NULL, NULL)'");
- AnalysisError("create table tab (x int primary key) distribute by range (x) " +
+ AnalysisError("create table tab (x int primary key) partition by range (x) " +
"(partition values <= (select count(*) from functional.alltypestiny)) " +
"stored as kudu", "Only constant values are allowed for range-partition " +
"bounds: (SELECT count(*) FROM functional.alltypestiny)");
// Multilevel partitioning. Data is split into 3 buckets based on 'x' and each
// bucket is partitioned into 4 tablets based on the range partitions of 'y'.
AnalyzesOk("create table tab (x int, y string, primary key(x, y)) " +
- "distribute by hash(x) into 3 buckets, range(y) " +
+ "partition by hash(x) into 3 buckets, range(y) " +
"(partition values < 'aa', partition 'aa' <= values < 'bb', " +
"partition 'bb' <= values < 'cc', partition 'cc' <= values) " +
"stored as kudu");
// Key column in upper case
AnalyzesOk("create table tab (x int, y int, primary key (X)) " +
- "distribute by hash (x) into 8 buckets stored as kudu");
+ "partition by hash (x) into 8 buckets stored as kudu");
// Flexible Partitioning
AnalyzesOk("create table tab (a int, b int, c int, d int, primary key (a, b, c))" +
- "distribute by hash (a, b) into 8 buckets, hash(c) into 2 buckets stored as " +
+ "partition by hash (a, b) into 8 buckets, hash(c) into 2 buckets stored as " +
"kudu");
- // No columns specified in the DISTRIBUTE BY HASH clause
+ // No columns specified in the PARTITION BY HASH clause
AnalyzesOk("create table tab (a int primary key, b int, c int, d int) " +
- "distribute by hash into 8 buckets stored as kudu");
+ "partition by hash into 8 buckets stored as kudu");
// Distribute range data types are picked up during analysis and forwarded to Kudu.
// Column names in distribute params should also be case-insensitive.
AnalyzesOk("create table tab (a int, b int, c int, d int, primary key(a, b, c, d))" +
- "distribute by hash (a, B, c) into 8 buckets, " +
+ "partition by hash (a, B, c) into 8 buckets, " +
"range (A) (partition values < 1, partition 1 <= values < 2, " +
"partition 2 <= values < 3, partition 3 <= values < 4, partition 4 <= values) " +
"stored as kudu");
- // Allowing range distribution on a subset of the primary keys
+ // Allowing range partitioning on a subset of the primary keys
AnalyzesOk("create table tab (id int, name string, valf float, vali bigint, " +
- "primary key (id, name)) distribute by range (name) " +
+ "primary key (id, name)) partition by range (name) " +
"(partition 'aa' < values <= 'bb') stored as kudu");
// Null values in range partition values
AnalysisError("create table tab (id int, name string, primary key(id, name)) " +
- "distribute by hash (id) into 3 buckets, range (name) " +
+ "partition by hash (id) into 3 buckets, range (name) " +
"(partition value = null, partition value = 1) stored as kudu",
"Range partition values cannot be NULL. Range partition: 'PARTITION " +
"VALUE = NULL'");
// Primary key specified in tblproperties
- AnalysisError(String.format("create table tab (x int) distribute by hash (x) " +
+ AnalysisError(String.format("create table tab (x int) partition by hash (x) " +
"into 8 buckets stored as kudu tblproperties ('%s' = 'x')",
KuduTable.KEY_KEY_COLUMNS), "PRIMARY KEY must be used instead of the table " +
"property");
// Primary key column that doesn't exist
AnalysisError("create table tab (x int, y int, primary key (z)) " +
- "distribute by hash (x) into 8 buckets stored as kudu",
+ "partition by hash (x) into 8 buckets stored as kudu",
"PRIMARY KEY column 'z' does not exist in the table");
// Invalid composite primary key
AnalysisError("create table tab (x int primary key, primary key(x)) stored " +
@@ -2002,65 +2002,65 @@ public class AnalyzeDDLTest extends FrontendTestBase {
"be specified using the PRIMARY KEY (col1, col2, ...) syntax at the end " +
"of the column definition.");
// Specifying the same primary key column multiple times
- AnalysisError("create table tab (x int, primary key (x, x)) distribute by hash (x) " +
+ AnalysisError("create table tab (x int, primary key (x, x)) partition by hash (x) " +
"into 8 buckets stored as kudu",
"Column 'x' is listed multiple times as a PRIMARY KEY.");
// Number of range partition boundary values should be equal to the number of range
// columns.
AnalysisError("create table tab (a int, b int, c int, d int, primary key(a, b, c)) " +
- "distribute by range(a) (partition value = (1, 2), " +
+ "partition by range(a) (partition value = (1, 2), " +
"partition value = 3, partition value = 4) stored as kudu",
"Number of specified range partition values is different than the number of " +
- "distribution columns: (2 vs 1). Range partition: 'PARTITION VALUE = (1,2)'");
+ "partitioning columns: (2 vs 1). Range partition: 'PARTITION VALUE = (1,2)'");
// Key ranges must match the column types.
AnalysisError("create table tab (a int, b int, c int, d int, primary key(a, b, c)) " +
- "distribute by hash (a, b, c) into 8 buckets, range (a) " +
+ "partition by hash (a, b, c) into 8 buckets, range (a) " +
"(partition value = 1, partition value = 'abc', partition 3 <= values) " +
"stored as kudu", "Range partition value 'abc' (type: STRING) is not type " +
- "compatible with distribution column 'a' (type: INT).");
- AnalysisError("create table tab (a tinyint primary key) distribute by range (a) " +
+ "compatible with partitioning column 'a' (type: INT).");
+ AnalysisError("create table tab (a tinyint primary key) partition by range (a) " +
"(partition value = 128) stored as kudu", "Range partition value 128 " +
- "(type: SMALLINT) is not type compatible with distribution column 'a' " +
+ "(type: SMALLINT) is not type compatible with partitioning column 'a' " +
"(type: TINYINT)");
- AnalysisError("create table tab (a smallint primary key) distribute by range (a) " +
+ AnalysisError("create table tab (a smallint primary key) partition by range (a) " +
"(partition value = 32768) stored as kudu", "Range partition value 32768 " +
- "(type: INT) is not type compatible with distribution column 'a' " +
+ "(type: INT) is not type compatible with partitioning column 'a' " +
"(type: SMALLINT)");
- AnalysisError("create table tab (a int primary key) distribute by range (a) " +
+ AnalysisError("create table tab (a int primary key) partition by range (a) " +
"(partition value = 2147483648) stored as kudu", "Range partition value " +
- "2147483648 (type: BIGINT) is not type compatible with distribution column 'a' " +
+ "2147483648 (type: BIGINT) is not type compatible with partitioning column 'a' " +
"(type: INT)");
- AnalysisError("create table tab (a bigint primary key) distribute by range (a) " +
+ AnalysisError("create table tab (a bigint primary key) partition by range (a) " +
"(partition value = 9223372036854775808) stored as kudu", "Range partition " +
"value 9223372036854775808 (type: DECIMAL(19,0)) is not type compatible with " +
- "distribution column 'a' (type: BIGINT)");
+ "partitioning column 'a' (type: BIGINT)");
// Test implicit casting/folding of partition values.
- AnalyzesOk("create table tab (a int primary key) distribute by range (a) " +
+ AnalyzesOk("create table tab (a int primary key) partition by range (a) " +
"(partition value = false, partition value = true) stored as kudu");
- // Non-key column used in DISTRIBUTE BY
+ // Non-key column used in PARTITION BY
AnalysisError("create table tab (a int, b string, c bigint, primary key (a)) " +
- "distribute by range (b) (partition value = 'abc') stored as kudu",
+ "partition by range (b) (partition value = 'abc') stored as kudu",
"Column 'b' in 'RANGE (b) (PARTITION VALUE = 'abc')' is not a key column. " +
- "Only key columns can be used in DISTRIBUTE BY.");
+ "Only key columns can be used in PARTITION BY.");
// No float range partition values
AnalysisError("create table tab (a int, b int, c int, d int, primary key (a, b, c))" +
- "distribute by hash (a, b, c) into 8 buckets, " +
+ "partition by hash (a, b, c) into 8 buckets, " +
"range (a) (partition value = 1.2, partition value = 2) stored as kudu",
"Range partition value 1.2 (type: DECIMAL(2,1)) is not type compatible with " +
- "distribution column 'a' (type: INT).");
- // Non-existing column used in DISTRIBUTE BY
+ "partitioning column 'a' (type: INT).");
+ // Non-existing column used in PARTITION BY
AnalysisError("create table tab (a int, b int, primary key (a, b)) " +
- "distribute by range(unknown_column) (partition value = 'abc') stored as kudu",
+ "partition by range(unknown_column) (partition value = 'abc') stored as kudu",
"Column 'unknown_column' in 'RANGE (unknown_column) (PARTITION VALUE = 'abc')' " +
- "is not a key column. Only key columns can be used in DISTRIBUTE BY");
+ "is not a key column. Only key columns can be used in PARTITION BY");
// Kudu table name is specified in tblproperties
- AnalyzesOk("create table tab (x int primary key) distribute by hash (x) " +
+ AnalyzesOk("create table tab (x int primary key) partition by hash (x) " +
"into 8 buckets stored as kudu tblproperties ('kudu.table_name'='tab_1'," +
"'kudu.num_tablet_replicas'='1'," +
"'kudu.master_addresses' = '127.0.0.1:8080, 127.0.0.1:8081')");
// No port is specified in kudu master address
AnalyzesOk("create table tdata_no_port (id int primary key, name string, " +
- "valf float, vali bigint) distribute by range(id) (partition values <= 10, " +
+ "valf float, vali bigint) partition by range(id) (partition values <= 10, " +
"partition 10 < values <= 30, partition 30 < values) " +
"stored as kudu tblproperties('kudu.master_addresses'='127.0.0.1')");
// Not using the STORED AS KUDU syntax to specify a Kudu table
@@ -2078,12 +2078,12 @@ public class AnalyzeDDLTest extends FrontendTestBase {
AnalysisError("create table tab (x int primary key) stored as kudu cached in " +
"'testPool'", "A Kudu table cannot be cached in HDFS.");
// LOCATION cannot be used with Kudu tables
- AnalysisError("create table tab (a int primary key) distribute by hash (a) " +
+ AnalysisError("create table tab (a int primary key) partition by hash (a) " +
"into 3 buckets stored as kudu location '/test-warehouse/'",
"LOCATION cannot be specified for a Kudu table.");
- // DISTRIBUTE BY is required for managed tables.
+ // PARTITION BY is required for managed tables.
AnalysisError("create table tab (a int, primary key (a)) stored as kudu",
- "Table distribution must be specified for managed Kudu tables.");
+ "Table partitioning must be specified for managed Kudu tables.");
AnalysisError("create table tab (a int) stored as kudu",
"A primary key is required for a Kudu table.");
// Using ROW FORMAT with a Kudu table
@@ -2105,12 +2105,12 @@ public class AnalyzeDDLTest extends FrontendTestBase {
// Unsupported type is PK and partition col
String stmt = String.format("create table tab (x %s primary key) " +
- "distribute by hash(x) into 3 buckets stored as kudu", t);
+ "partition by hash(x) into 3 buckets stored as kudu", t);
AnalysisError(stmt, expectedError);
// Unsupported type is not PK/partition col
stmt = String.format("create table tab (x int primary key, y %s) " +
- "distribute by hash(x) into 3 buckets stored as kudu", t);
+ "partition by hash(x) into 3 buckets stored as kudu", t);
AnalysisError(stmt, expectedError);
}
@@ -2125,7 +2125,7 @@ public class AnalyzeDDLTest extends FrontendTestBase {
for (String block: blockSize) {
AnalyzesOk(String.format("create table tab (x int primary key " +
"not null encoding %s compression %s %s %s, y int encoding %s " +
- "compression %s %s %s %s) distribute by hash (x) " +
+ "compression %s %s %s %s) partition by hash (x) " +
"into 3 buckets stored as kudu", enc, comp, def, block, enc,
comp, def, nul, block));
}
@@ -2136,23 +2136,23 @@ public class AnalyzeDDLTest extends FrontendTestBase {
// Primary key specified using the PRIMARY KEY clause
AnalyzesOk("create table tab (x int not null encoding plain_encoding " +
"compression snappy block_size 1, y int null encoding rle compression lz4 " +
- "default 1, primary key(x)) distribute by hash (x) into 3 buckets " +
+ "default 1, primary key(x)) partition by hash (x) into 3 buckets " +
"stored as kudu");
// Primary keys can't be null
AnalysisError("create table tab (x int primary key null, y int not null) " +
- "distribute by hash (x) into 3 buckets stored as kudu", "Primary key columns " +
+ "partition by hash (x) into 3 buckets stored as kudu", "Primary key columns " +
"cannot be nullable: x INT PRIMARY KEY NULL");
AnalysisError("create table tab (x int not null, y int null, primary key (x, y)) " +
- "distribute by hash (x) into 3 buckets stored as kudu", "Primary key columns " +
+ "partition by hash (x) into 3 buckets stored as kudu", "Primary key columns " +
"cannot be nullable: y INT NULL");
// Unsupported encoding value
AnalysisError("create table tab (x int primary key, y int encoding invalid_enc) " +
- "distribute by hash (x) into 3 buckets stored as kudu", "Unsupported encoding " +
+ "partition by hash (x) into 3 buckets stored as kudu", "Unsupported encoding " +
"value 'INVALID_ENC'. Supported encoding values are: " +
Joiner.on(", ").join(Encoding.values()));
// Unsupported compression algorithm
AnalysisError("create table tab (x int primary key, y int compression " +
- "invalid_comp) distribute by hash (x) into 3 buckets stored as kudu",
+ "invalid_comp) partition by hash (x) into 3 buckets stored as kudu",
"Unsupported compression algorithm 'INVALID_COMP'. Supported compression " +
"algorithms are: " + Joiner.on(", ").join(CompressionAlgorithm.values()));
// Default values
@@ -2160,38 +2160,38 @@ public class AnalyzeDDLTest extends FrontendTestBase {
"i3 int default 100, i4 bigint default 1000, vals string default 'test', " +
"valf float default cast(1.2 as float), vald double default " +
"cast(3.1452 as double), valb boolean default true, " +
- "primary key (i1, i2, i3, i4, vals)) distribute by hash (i1) into 3 " +
+ "primary key (i1, i2, i3, i4, vals)) partition by hash (i1) into 3 " +
"buckets stored as kudu");
AnalyzesOk("create table tab (i int primary key default 1+1+1) " +
- "distribute by hash (i) into 3 buckets stored as kudu");
+ "partition by hash (i) into 3 buckets stored as kudu");
AnalyzesOk("create table tab (i int primary key default factorial(5)) " +
- "distribute by hash (i) into 3 buckets stored as kudu");
+ "partition by hash (i) into 3 buckets stored as kudu");
AnalyzesOk("create table tab (i int primary key, x int null default " +
- "isnull(null, null)) distribute by hash (i) into 3 buckets stored as kudu");
+ "isnull(null, null)) partition by hash (i) into 3 buckets stored as kudu");
// Invalid default values
AnalysisError("create table tab (i int primary key default 'string_val') " +
- "distribute by hash (i) into 3 buckets stored as kudu", "Default value " +
+ "partition by hash (i) into 3 buckets stored as kudu", "Default value " +
"'string_val' (type: STRING) is not compatible with column 'i' (type: INT).");
AnalysisError("create table tab (i int primary key, x int default 1.1) " +
- "distribute by hash (i) into 3 buckets stored as kudu",
+ "partition by hash (i) into 3 buckets stored as kudu",
"Default value 1.1 (type: DECIMAL(2,1)) is not compatible with column " +
"'x' (type: INT).");
AnalysisError("create table tab (i tinyint primary key default 128) " +
- "distribute by hash (i) into 3 buckets stored as kudu", "Default value " +
+ "partition by hash (i) into 3 buckets stored as kudu", "Default value " +
"128 (type: SMALLINT) is not compatible with column 'i' (type: TINYINT).");
AnalysisError("create table tab (i int primary key default isnull(null, null)) " +
- "distribute by hash (i) into 3 buckets stored as kudu", "Default value of " +
+ "partition by hash (i) into 3 buckets stored as kudu", "Default value of " +
"NULL not allowed on non-nullable column: 'i'");
AnalysisError("create table tab (i int primary key, x int not null " +
- "default isnull(null, null)) distribute by hash (i) into 3 buckets " +
+ "default isnull(null, null)) partition by hash (i) into 3 buckets " +
"stored as kudu", "Default value of NULL not allowed on non-nullable column: " +
"'x'");
// Invalid block_size values
AnalysisError("create table tab (i int primary key block_size 1.1) " +
- "distribute by hash (i) into 3 buckets stored as kudu", "Invalid value " +
+ "partition by hash (i) into 3 buckets stored as kudu", "Invalid value " +
"for BLOCK_SIZE: 1.1. A positive INTEGER value is expected.");
AnalysisError("create table tab (i int primary key block_size 'val') " +
- "distribute by hash (i) into 3 buckets stored as kudu", "Invalid value " +
+ "partition by hash (i) into 3 buckets stored as kudu", "Invalid value " +
"for BLOCK_SIZE: 'val'. A positive INTEGER value is expected.");
}
@@ -2438,8 +2438,8 @@ public class AnalyzeDDLTest extends FrontendTestBase {
// Kudu specific clauses used in an Avro table.
AnalysisError("create table functional.new_table (i int) " +
- "distribute by hash(i) into 3 buckets stored as avro",
- "Only Kudu tables can use the DISTRIBUTE BY clause.");
+ "partition by hash(i) into 3 buckets stored as avro",
+ "Only Kudu tables can use the PARTITION BY clause.");
AnalysisError("create table functional.new_table (i int primary key) " +
"stored as avro", "Unsupported column options for file format 'AVRO': " +
"'i INT PRIMARY KEY'");
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/test/java/org/apache/impala/analysis/AuthorizationTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/analysis/AuthorizationTest.java b/fe/src/test/java/org/apache/impala/analysis/AuthorizationTest.java
index 431ac01..3d6bbec 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AuthorizationTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AuthorizationTest.java
@@ -911,7 +911,7 @@ public class AuthorizationTest {
// IMPALA-4000: ALL privileges on SERVER are not required to create managed tables.
AuthzOk("create table tpch.kudu_tbl (i int, j int, primary key (i))" +
- " DISTRIBUTE BY HASH (i) INTO 9 BUCKETS stored as kudu TBLPROPERTIES " +
+ " PARTITION BY HASH (i) INTO 9 BUCKETS stored as kudu TBLPROPERTIES " +
"('kudu.master_addresses'='127.0.0.1')");
// User does not have permission to create table at the specified location..
[2/8] incubator-impala git commit: IMPALA-4571: Push IN predicates to
Kudu
Posted by kw...@apache.org.
IMPALA-4571: Push IN predicates to Kudu
Fixes the KuduScanNode to convert InPredicates to
KuduPredicates and push them to the Kudu scan if possible.
An InPredicate can be pushed to the scan if expression is of
the exact form:
<SlotRef> IN (<LiteralExpr>, <LiteralExpr>, ...)
That means the InPredicate has the following properties:
1) It has a list of literal values (i.e. not a subquery);
All values are LiteralExprs (not SlotRefs).
2) Not negative, i.e. only 'IN' supported, not 'NOT IN'
3) The SlotRef is not wrapped in any casts
4) The types of all values match the type of the SlotRef
exactly.
A planner test was added exercising all supported types as
well as exprs where the values would not be supported.
TODO: perf testing
TODO: consider a limit on the number of list values before
keeping the predicate on the Impala scan node
(determine from testing)
Change-Id: I8988d4819d20d467b48e286917e347ca00f60cf0
Reviewed-on: http://gerrit.cloudera.org:8080/5316
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/9f387c85
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/9f387c85
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/9f387c85
Branch: refs/heads/master
Commit: 9f387c858354a5c7df5fd922e731f887aa7e51f7
Parents: 867b243
Author: Matthew Jacobs <mj...@cloudera.com>
Authored: Thu Dec 1 18:16:33 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Dec 6 03:24:39 2016 +0000
----------------------------------------------------------------------
.../org/apache/impala/planner/KuduScanNode.java | 63 +++++++++++++++++++-
.../queries/PlannerTest/kudu-selectivity.test | 33 ++++++++++
.../queries/PlannerTest/tpch-kudu.test | 10 ++--
3 files changed, 99 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f387c85/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
index 883a556..04fb244 100644
--- a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
@@ -26,6 +26,7 @@ import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.BinaryPredicate;
import org.apache.impala.analysis.BoolLiteral;
import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.InPredicate;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.analysis.NumericLiteral;
@@ -274,7 +275,11 @@ public class KuduScanNode extends ScanNode {
KuduClient client, org.apache.kudu.client.KuduTable rpcTable) {
ListIterator<Expr> it = conjuncts_.listIterator();
while (it.hasNext()) {
- if (tryConvertKuduPredicate(analyzer, rpcTable, it.next())) it.remove();
+ Expr predicate = it.next();
+ if (tryConvertBinaryKuduPredicate(analyzer, rpcTable, predicate) ||
+ tryConvertInListKuduPredicate(analyzer, rpcTable, predicate)) {
+ it.remove();
+ }
}
}
@@ -282,7 +287,7 @@ public class KuduScanNode extends ScanNode {
* If 'expr' can be converted to a KuduPredicate, returns true and updates
* kuduPredicates_ and kuduConjuncts_.
*/
- private boolean tryConvertKuduPredicate(Analyzer analyzer,
+ private boolean tryConvertBinaryKuduPredicate(Analyzer analyzer,
org.apache.kudu.client.KuduTable table, Expr expr) {
if (!(expr instanceof BinaryPredicate)) return false;
BinaryPredicate predicate = (BinaryPredicate) expr;
@@ -347,6 +352,60 @@ public class KuduScanNode extends ScanNode {
}
/**
+ * If the InList 'expr' can be converted to a KuduPredicate, returns true and updates
+ * kuduPredicates_ and kuduConjuncts_.
+ */
+ private boolean tryConvertInListKuduPredicate(Analyzer analyzer,
+ org.apache.kudu.client.KuduTable table, Expr expr) {
+ if (!(expr instanceof InPredicate)) return false;
+ InPredicate predicate = (InPredicate) expr;
+
+ // Only convert IN predicates, i.e. cannot convert NOT IN.
+ if (predicate.isNotIn()) return false;
+
+ // Do not convert if there is an implicit cast.
+ if (!(predicate.getChild(0) instanceof SlotRef)) return false;
+ SlotRef ref = (SlotRef) predicate.getChild(0);
+
+ // KuduPredicate takes a list of values as Objects.
+ List<Object> values = Lists.newArrayList();
+ for (int i = 1; i < predicate.getChildren().size(); ++i) {
+ if (!(predicate.getChild(i).isLiteral())) return false;
+ Object value = getKuduInListValue((LiteralExpr) predicate.getChild(i));
+ Preconditions.checkNotNull(value == null);
+ values.add(value);
+ }
+
+ String colName = ref.getDesc().getColumn().getName();
+ ColumnSchema column = table.getSchema().getColumn(colName);
+ kuduPredicates_.add(KuduPredicate.newInListPredicate(column, values));
+ kuduConjuncts_.add(predicate);
+ return true;
+ }
+
+ /**
+ * Return the value of the InList child expression 'e' as an Object that can be
+ * added to a KuduPredicate. If the Expr is not supported by Kudu or the type doesn't
+ * match the expected PrimitiveType 'type', null is returned.
+ */
+ private static Object getKuduInListValue(LiteralExpr e) {
+ switch (e.getType().getPrimitiveType()) {
+ case BOOLEAN: return ((BoolLiteral) e).getValue();
+ case TINYINT: return (byte) ((NumericLiteral) e).getLongValue();
+ case SMALLINT: return (short) ((NumericLiteral) e).getLongValue();
+ case INT: return (int) ((NumericLiteral) e).getLongValue();
+ case BIGINT: return ((NumericLiteral) e).getLongValue();
+ case FLOAT: return (float) ((NumericLiteral) e).getDoubleValue();
+ case DOUBLE: return ((NumericLiteral) e).getDoubleValue();
+ case STRING: return ((StringLiteral) e).getValue();
+ default:
+ Preconditions.checkState(false,
+ "Unsupported Kudu type considered for predicate: %s", e.getType().toSql());
+ }
+ return null;
+ }
+
+ /**
* Returns a Kudu comparison operator for the BinaryPredicate operator, or null if
* the operation is not supported by Kudu.
*/
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f387c85/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
index a187ed8..eba7741 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
@@ -97,3 +97,36 @@ F00:PLAN FRAGMENT [RANDOM]
hosts=3 per-host-mem=0B
tuple-ids=0 row-size=124B cardinality=2
====
+select * from functional_kudu.alltypes where
+-- predicates that can be pushed
+tinyint_col in (1, 2) and
+smallint_col in (false, 2) and
+int_col in (true, 2) and
+bigint_col in (1, 2) and
+bool_col in (true) and
+float_col in (0.0) and
+double_col in (0.0) and
+string_col in ("foo", cast("foo" as char(10))) and
+
+-- predicates that cannot be pushed because the SlotRef is wrapped in a cast
+tinyint_col in (1, 999) and
+smallint_col in (99999, 2) and
+int_col in (9999999999) and
+bigint_col in (9999999999999999999) and
+bool_col in (1) and
+float_col in (cast('NaN' as float)) and
+double_col in (cast('inf' as double)) and
+
+-- 'NOT IN' and lists containing slotrefs cannot be pushed
+string_col not in ("bar") and
+id in (int_col)
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED]
+ PLAN-ROOT SINK
+ |
+ 00:SCAN KUDU [functional_kudu.alltypes]
+ predicates: id IN (int_col), string_col NOT IN ('bar'), bigint_col IN (9999999999999999999), double_col IN (CAST('inf' AS DOUBLE)), float_col IN (CAST('NaN' AS FLOAT)), int_col IN (9999999999), smallint_col IN (99999, 2), tinyint_col IN (1, 999), bool_col IN (1)
+ kudu predicates: double_col IN (0.0), float_col IN (0.0), bigint_col IN (1, 2), int_col IN (1, 2), smallint_col IN (0, 2), string_col IN ('foo', 'foo '), tinyint_col IN (1, 2), bool_col IN (TRUE)
+ hosts=3 per-host-mem=unavailable
+ tuple-ids=0 row-size=126B cardinality=4
+====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f387c85/testdata/workloads/functional-planner/queries/PlannerTest/tpch-kudu.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpch-kudu.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpch-kudu.test
index a5f2511..950fb90 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpch-kudu.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpch-kudu.test
@@ -725,8 +725,8 @@ PLAN-ROOT SINK
| hash predicates: o_orderkey = l_orderkey
|
|--01:SCAN KUDU [tpch_kudu.lineitem]
-| predicates: l_shipmode IN ('MAIL', 'SHIP'), l_commitdate < l_receiptdate, l_shipdate < l_commitdate
-| kudu predicates: l_receiptdate < '1995-01-01', l_receiptdate >= '1994-01-01'
+| predicates: l_commitdate < l_receiptdate, l_shipdate < l_commitdate
+| kudu predicates: l_shipmode IN ('MAIL', 'SHIP'), l_receiptdate < '1995-01-01', l_receiptdate >= '1994-01-01'
|
00:SCAN KUDU [tpch_kudu.orders]
====
@@ -917,7 +917,8 @@ PLAN-ROOT SINK
| hash predicates: ps_partkey = p_partkey
|
|--01:SCAN KUDU [tpch_kudu.part]
-| predicates: p_size IN (49, 14, 23, 45, 19, 3, 36, 9), p_brand != 'Brand#45', NOT p_type LIKE 'MEDIUM POLISHED%'
+| predicates: p_brand != 'Brand#45', NOT p_type LIKE 'MEDIUM POLISHED%'
+| kudu predicates: p_size IN (49, 14, 23, 45, 19, 3, 36, 9)
|
00:SCAN KUDU [tpch_kudu.partsupp]
====
@@ -1080,8 +1081,7 @@ PLAN-ROOT SINK
| kudu predicates: p_size >= 1
|
00:SCAN KUDU [tpch_kudu.lineitem]
- predicates: l_shipmode IN ('AIR', 'AIR REG')
- kudu predicates: l_shipinstruct = 'DELIVER IN PERSON'
+ kudu predicates: l_shipmode IN ('AIR', 'AIR REG'), l_shipinstruct = 'DELIVER IN PERSON'
====
# Q20 - Potential Part Promotion Query
select
[5/8] incubator-impala git commit: IMPALA-3167: Fix assignment of
WHERE conjunct through grouping agg + OJ.
Posted by kw...@apache.org.
IMPALA-3167: Fix assignment of WHERE conjunct through grouping agg + OJ.
Background: We generally allow the assignment of predicates below the
nullable side of a left/right outer join, explained as follows using an
example:
SELECT * FROM t1 LEFT OUTER JOIN t2 ON t1.id = t2.id
WHERE t2.int_col < 10
The scan of 't2' picks up 't2.int_col < 10' via
Analyzer.getBoundPredicates() and recognizes that the predicate must
also be evaluated by a join later, so the predicate is not marked as
assigned. The join then picks up the unassigned predicate via
Analyzer.getUnassignedConjuncts().
The bug was that our logic for detecting whether a bound predicate must
also be evaluated at a join node was flawed because it only considered
whether the tuples of the source or destination predicate were outer
joined (plus other conditions).
The underlying assumption is that either the source or destination tuple
are bound by a tuple produced by a TableRef, but in the buggy query the
source predicate is bound by an aggregation tuple, so we incorrectly
marked the bound predicate as assigned in Analyzer.getBoundPredicates().
The fix is to conservatively not mark bound predicates as assigned if
the slots referenced by the predicate have equivalent slots that
belong to an outer-joined tuple. As a result, a plan node may pick up
the same predicate multiple times, once via
Analyzer.getBoundPredicates() and another time via
Analyzer.getUnassignedConjuncts(). Those are deduped now.
The following example explains the duplicate predicate assignment:
SELECT * FROM (SELECT * FROM t t1) a LEFT OUTER JOIN t b ON a.id = b.id
WHERE a.id < 10
1. The predicate 'a.id < 10' gets migrated into the inline view.
'a.id < 10' is marked as assigned but is still registered as
a single-tid conjunct in the Analyzer for potential propagation
2. The scan node of 't1' calls Analyzer.getBoundPredicates() and
generates 't1.id < 10' based on the source predicate 'a.id < 10'.
3. The scan node of 't1' picks up the migrated conjunct 't1.id < 10'
via Analyzer.getUnassignedConjuncts().
Change-Id: I774d13a13ad1e8fe82512df98dc29983bdd232eb
Reviewed-on: http://gerrit.cloudera.org:8080/4960
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/f8377543
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/f8377543
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/f8377543
Branch: refs/heads/master
Commit: f8377543778b654336c978a4bb97efa3c1847441
Parents: b656f57
Author: Alex Behm <al...@cloudera.com>
Authored: Fri Nov 4 10:41:25 2016 -0700
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Dec 6 07:24:01 2016 +0000
----------------------------------------------------------------------
.../org/apache/impala/analysis/Analyzer.java | 29 ++++++++++++--------
.../java/org/apache/impala/analysis/Expr.java | 19 ++-----------
.../org/apache/impala/analysis/SelectStmt.java | 2 +-
.../org/apache/impala/planner/HdfsScanNode.java | 2 +-
.../impala/planner/SingleNodePlanner.java | 2 +-
.../queries/PlannerTest/outer-joins.test | 29 ++++++++++++++++++++
6 files changed, 52 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f8377543/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
index 4819342..6bba436 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
@@ -1223,10 +1223,11 @@ public class Analyzer {
}
/**
- * Returns true if e must be evaluated by a join node. Note that it may still be
- * safe to evaluate e elsewhere as well, but in any case the join must evaluate e.
+ * Returns true if 'e' must be evaluated after or by a join node. Note that it may
+ * still be safe to evaluate 'e' elsewhere as well, but in any case 'e' must be
+ * evaluated again by or after a join.
*/
- public boolean evalByJoin(Expr e) {
+ public boolean evalAfterJoin(Expr e) {
List<TupleId> tids = Lists.newArrayList();
e.getIds(tids, null);
if (tids.isEmpty()) return false;
@@ -1555,18 +1556,22 @@ public class Analyzer {
}
}
- // Check if either srcConjunct or the generated predicate needs to be evaluated
- // at a join node (IMPALA-2018).
- boolean evalByJoin =
- (evalByJoin(srcConjunct)
- && (globalState_.ojClauseByConjunct.get(srcConjunct.getId())
- != globalState_.outerJoinedTupleIds.get(srcTid)))
- || (evalByJoin(p)
+ // IMPALA-2018/4379: Check if srcConjunct or the generated predicate need to
+ // be evaluated again at a later point in the plan, e.g., by a join that makes
+ // referenced tuples nullable. The first condition is conservative but takes
+ // into account that On-clause conjuncts can sometimes be legitimately assigned
+ // below their originating join.
+ boolean evalAfterJoin =
+ (hasOuterJoinedTuple && !srcConjunct.isOnClauseConjunct_)
+ || (evalAfterJoin(srcConjunct)
+ && (globalState_.ojClauseByConjunct.get(srcConjunct.getId())
+ != globalState_.outerJoinedTupleIds.get(srcTid)))
+ || (evalAfterJoin(p)
&& (globalState_.ojClauseByConjunct.get(p.getId())
- != globalState_.outerJoinedTupleIds.get(destTid)));
+ != globalState_.outerJoinedTupleIds.get(destTid)));
// mark all bound predicates including duplicate ones
- if (reverseValueTransfer && !evalByJoin) markConjunctAssigned(srcConjunct);
+ if (reverseValueTransfer && !evalAfterJoin) markConjunctAssigned(srcConjunct);
}
// check if we already created this predicate
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f8377543/fe/src/main/java/org/apache/impala/analysis/Expr.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/Expr.java b/fe/src/main/java/org/apache/impala/analysis/Expr.java
index 779e252..87a2a12 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Expr.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Expr.java
@@ -851,22 +851,9 @@ abstract public class Expr extends TreeNode<Expr> implements ParseNode, Cloneabl
*/
public static <C extends Expr> void removeDuplicates(List<C> l) {
if (l == null) return;
- ListIterator<C> it1 = l.listIterator();
- while (it1.hasNext()) {
- C e1 = it1.next();
- ListIterator<C> it2 = l.listIterator();
- boolean duplicate = false;
- while (it2.hasNext()) {
- C e2 = it2.next();
- // only check up to but excluding e1
- if (e1 == e2) break;
- if (e1.equals(e2)) {
- duplicate = true;
- break;
- }
- }
- if (duplicate) it1.remove();
- }
+ List<C> origList = Lists.newArrayList(l);
+ l.clear();
+ for (C expr: origList) if (!l.contains(expr)) l.add(expr);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f8377543/fe/src/main/java/org/apache/impala/analysis/SelectStmt.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/analysis/SelectStmt.java b/fe/src/main/java/org/apache/impala/analysis/SelectStmt.java
index 80ffde5..74399d0 100644
--- a/fe/src/main/java/org/apache/impala/analysis/SelectStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/SelectStmt.java
@@ -333,7 +333,7 @@ public class SelectStmt extends QueryStmt {
analyzer.getUnassignedConjuncts(getTableRefIds(), true);
List<Expr> unassignedJoinConjuncts = Lists.newArrayList();
for (Expr e: unassigned) {
- if (analyzer.evalByJoin(e)) unassignedJoinConjuncts.add(e);
+ if (analyzer.evalAfterJoin(e)) unassignedJoinConjuncts.add(e);
}
List<Expr> baseTblJoinConjuncts =
Expr.substituteList(unassignedJoinConjuncts, baseTblSmap_, analyzer, false);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f8377543/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
index 9642b97..d2eff19 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -292,7 +292,7 @@ public class HdfsScanNode extends ScanNode {
// Mark those conjuncts as assigned that do not also need to be evaluated by a
// subsequent semi or outer join.
for (Expr conjunct: collectionConjuncts) {
- if (!analyzer.evalByJoin(conjunct)) analyzer.markConjunctAssigned(conjunct);
+ if (!analyzer.evalAfterJoin(conjunct)) analyzer.markConjunctAssigned(conjunct);
}
if (!collectionConjuncts.isEmpty()) {
analyzer.materializeSlots(collectionConjuncts);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f8377543/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 4bc8a88..3687a17 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -1195,7 +1195,6 @@ public class SingleNodePlanner {
*/
private PlanNode createHdfsScanPlan(TableRef hdfsTblRef, boolean fastPartitionKeyScans,
Analyzer analyzer) throws ImpalaException {
- HdfsTable hdfsTable = (HdfsTable)hdfsTblRef.getTable();
TupleDescriptor tupleDesc = hdfsTblRef.getDesc();
// Get all predicates bound by the tuple.
@@ -1208,6 +1207,7 @@ public class SingleNodePlanner {
analyzer.markConjunctsAssigned(unassigned);
analyzer.createEquivConjuncts(tupleDesc.getId(), conjuncts);
+ Expr.removeDuplicates(conjuncts);
// Do partition pruning before deciding which slots to materialize,
// We might end up removing some predicates.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/f8377543/testdata/workloads/functional-planner/queries/PlannerTest/outer-joins.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/outer-joins.test b/testdata/workloads/functional-planner/queries/PlannerTest/outer-joins.test
index 2d5d6cd..95d16f8 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/outer-joins.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/outer-joins.test
@@ -919,3 +919,32 @@ PLAN-ROOT SINK
partitions=24/24 files=24 size=478.45KB
runtime filters: RF000 -> b.int_col
====
+# IMPALA-3167: Test correct assignment of a WHERE-clause predicate through an inline view
+# that has a grouping aggregation and an outer join. The predicate can be assigned at the
+# scan on the nullable side of the outer join, but it must also be evaluated after the join.
+select v2.id, v2.s
+from (select v1.id, sum(bigint_col) s
+ from functional.alltypes t1
+ left outer join (select t2.int_col, t2.id
+ from functional.alltypessmall t2) v1
+ on t1.int_col = v1.int_col
+ group by v1.id) v2
+where v2.id < 10
+---- PLAN
+PLAN-ROOT SINK
+|
+03:AGGREGATE [FINALIZE]
+| output: sum(bigint_col)
+| group by: t2.id
+| having: v1.id < 10
+|
+02:HASH JOIN [LEFT OUTER JOIN]
+| hash predicates: t1.int_col = t2.int_col
+|
+|--01:SCAN HDFS [functional.alltypessmall t2]
+| partitions=4/4 files=4 size=6.32KB
+| predicates: t2.id < 10
+|
+00:SCAN HDFS [functional.alltypes t1]
+ partitions=24/24 files=24 size=478.45KB
+====
[8/8] incubator-impala git commit: IMPALA-2925: Mark
test_alloc_update as xfail.
Posted by kw...@apache.org.
IMPALA-2925: Mark test_alloc_update as xfail.
test_alloc_update.py is flaky and the expected failure sometimes
doesn't occur. Mark this test as xfail for now to unblock the build.
Change-Id: If4e86e7b9c064bc78b672814cd3569453ecc268d
Reviewed-on: http://gerrit.cloudera.org:8080/5366
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/9b80224f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/9b80224f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/9b80224f
Branch: refs/heads/master
Commit: 9b80224f9f782cd9a870c4a70d651392bb07c86c
Parents: cba93f1
Author: Michael Ho <kw...@cloudera.com>
Authored: Mon Dec 5 14:13:33 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Dec 6 12:37:17 2016 +0000
----------------------------------------------------------------------
tests/custom_cluster/test_alloc_fail.py | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9b80224f/tests/custom_cluster/test_alloc_fail.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_alloc_fail.py b/tests/custom_cluster/test_alloc_fail.py
index 9e08adb..28cdc19 100644
--- a/tests/custom_cluster/test_alloc_fail.py
+++ b/tests/custom_cluster/test_alloc_fail.py
@@ -36,6 +36,9 @@ class TestAllocFail(CustomClusterTestSuite):
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args("--stress_free_pool_alloc=3")
def test_alloc_fail_update(self, vector):
+ # TODO: Rewrite or remove the non-deterministic test.
+ pytest.xfail("IMPALA-2925: the execution is not deterministic so some "
+ "tests sometimes don't fail as expected")
# Note that this test relies on pre-aggregation to exercise the Serialize() path so
# query option 'num_nodes' must not be 1. CustomClusterTestSuite.add_test_dimensions()
# already filters out vectors with 'num_nodes' != 0 so just assert it here.
[6/8] incubator-impala git commit: IMPALA-4561: Replace DISTRIBUTE BY
with PARTITION BY in CREATE TABLE
Posted by kw...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
index 6833ee9..a73adf3 100644
--- a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
@@ -2445,52 +2445,52 @@ public class ParserTest extends FrontendTestBase {
// Flexible partitioning
- ParsesOk("CREATE TABLE Foo (i int) DISTRIBUTE BY HASH(i) INTO 4 BUCKETS");
- ParsesOk("CREATE TABLE Foo (i int) DISTRIBUTE BY HASH(i) INTO 4 BUCKETS, " +
+ ParsesOk("CREATE TABLE Foo (i int) PARTITION BY HASH(i) INTO 4 BUCKETS");
+ ParsesOk("CREATE TABLE Foo (i int) PARTITION BY HASH(i) INTO 4 BUCKETS, " +
"HASH(a) INTO 2 BUCKETS");
- ParsesOk("CREATE TABLE Foo (i int) DISTRIBUTE BY HASH INTO 4 BUCKETS");
- ParsesOk("CREATE TABLE Foo (i int, k int) DISTRIBUTE BY HASH INTO 4 BUCKETS," +
+ ParsesOk("CREATE TABLE Foo (i int) PARTITION BY HASH INTO 4 BUCKETS");
+ ParsesOk("CREATE TABLE Foo (i int, k int) PARTITION BY HASH INTO 4 BUCKETS," +
" HASH(k) INTO 4 BUCKETS");
- ParserError("CREATE TABLE Foo (i int) DISTRIBUTE BY HASH(i)");
- ParserError("CREATE EXTERNAL TABLE Foo DISTRIBUTE BY HASH INTO 4 BUCKETS");
+ ParserError("CREATE TABLE Foo (i int) PARTITION BY HASH(i)");
+ ParserError("CREATE EXTERNAL TABLE Foo PARTITION BY HASH INTO 4 BUCKETS");
// Range partitioning
- ParsesOk("CREATE TABLE Foo (i int) DISTRIBUTE BY RANGE (PARTITION VALUE = 10)");
- ParsesOk("CREATE TABLE Foo (i int) DISTRIBUTE BY RANGE(i) " +
+ ParsesOk("CREATE TABLE Foo (i int) PARTITION BY RANGE (PARTITION VALUE = 10)");
+ ParsesOk("CREATE TABLE Foo (i int) PARTITION BY RANGE(i) " +
"(PARTITION 1 <= VALUES < 10, PARTITION 10 <= VALUES < 20, " +
"PARTITION 21 < VALUES <= 30, PARTITION VALUE = 50)");
- ParsesOk("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE(a) " +
+ ParsesOk("CREATE TABLE Foo (a int) PARTITION BY RANGE(a) " +
"(PARTITION 10 <= VALUES)");
- ParsesOk("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE(a) " +
+ ParsesOk("CREATE TABLE Foo (a int) PARTITION BY RANGE(a) " +
"(PARTITION VALUES < 10)");
- ParsesOk("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) " +
+ ParsesOk("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) " +
"(PARTITION VALUE = 10, PARTITION VALUE = 20)");
- ParsesOk("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE(a) " +
+ ParsesOk("CREATE TABLE Foo (a int) PARTITION BY RANGE(a) " +
"(PARTITION VALUES <= 10, PARTITION VALUE = 20)");
- ParsesOk("CREATE TABLE Foo (a int, b int) DISTRIBUTE BY RANGE(a, b) " +
+ ParsesOk("CREATE TABLE Foo (a int, b int) PARTITION BY RANGE(a, b) " +
"(PARTITION VALUE = (2001, 1), PARTITION VALUE = (2001, 2), " +
"PARTITION VALUE = (2002, 1))");
- ParsesOk("CREATE TABLE Foo (a int, b string) DISTRIBUTE BY " +
+ ParsesOk("CREATE TABLE Foo (a int, b string) PARTITION BY " +
"HASH (a) INTO 3 BUCKETS, RANGE (a, b) (PARTITION VALUE = (1, 'abc'), " +
"PARTITION VALUE = (2, 'def'))");
- ParsesOk("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) " +
+ ParsesOk("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) " +
"(PARTITION VALUE = 1 + 1) STORED AS KUDU");
- ParsesOk("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) " +
+ ParsesOk("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) " +
"(PARTITION 1 + 1 < VALUES) STORED AS KUDU");
- ParsesOk("CREATE TABLE Foo (a int, b int) DISTRIBUTE BY RANGE (a) " +
+ ParsesOk("CREATE TABLE Foo (a int, b int) PARTITION BY RANGE (a) " +
"(PARTITION b < VALUES <= a) STORED AS KUDU");
- ParsesOk("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) " +
+ ParsesOk("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) " +
"(PARTITION now() <= VALUES, PARTITION VALUE = add_months(now(), 2)) " +
"STORED AS KUDU");
- ParserError("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) ()");
- ParserError("CREATE TABLE Foo (a int) DISTRIBUTE BY HASH (a) INTO 4 BUCKETS, " +
+ ParserError("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) ()");
+ ParserError("CREATE TABLE Foo (a int) PARTITION BY HASH (a) INTO 4 BUCKETS, " +
"RANGE (a) (PARTITION VALUE = 10), RANGE (a) (PARTITION VALUES < 10)");
- ParserError("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) " +
+ ParserError("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) " +
"(PARTITION VALUE = 10), HASH (a) INTO 3 BUCKETS");
- ParserError("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) " +
+ ParserError("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) " +
"(PARTITION VALUES = 10) STORED AS KUDU");
- ParserError("CREATE TABLE Foo (a int) DISTRIBUTE BY RANGE (a) " +
+ ParserError("CREATE TABLE Foo (a int) PARTITION BY RANGE (a) " +
"(PARTITION 10 < VALUE < 20) STORED AS KUDU");
// Column options for Kudu tables
@@ -2676,9 +2676,9 @@ public class ParserTest extends FrontendTestBase {
ParsesOk("CREATE TABLE Foo ROW FORMAT DELIMITED STORED AS PARQUETFILE AS SELECT 1");
ParsesOk("CREATE TABLE Foo TBLPROPERTIES ('a'='b', 'c'='d') AS SELECT * from bar");
ParsesOk("CREATE TABLE Foo PRIMARY KEY (a, b) AS SELECT * from bar");
- ParsesOk("CREATE TABLE Foo PRIMARY KEY (a, b) DISTRIBUTE BY HASH INTO 2 BUCKETS " +
+ ParsesOk("CREATE TABLE Foo PRIMARY KEY (a, b) PARTITION BY HASH INTO 2 BUCKETS " +
"AS SELECT * from bar");
- ParsesOk("CREATE TABLE Foo PRIMARY KEY (a, b) DISTRIBUTE BY HASH (b) INTO 2 " +
+ ParsesOk("CREATE TABLE Foo PRIMARY KEY (a, b) PARTITION BY HASH (b) INTO 2 " +
"BUCKETS AS SELECT * from bar");
// With clause works
@@ -2708,12 +2708,12 @@ public class ParserTest extends FrontendTestBase {
ParserError("CREATE TABLE Foo PARTITIONED BY (a, b=2) AS SELECT * from Bar");
// Flexible partitioning
- ParsesOk("CREATE TABLE Foo PRIMARY KEY (i) DISTRIBUTE BY HASH(i) INTO 4 BUCKETS AS " +
+ ParsesOk("CREATE TABLE Foo PRIMARY KEY (i) PARTITION BY HASH(i) INTO 4 BUCKETS AS " +
"SELECT 1");
- ParserError("CREATE TABLE Foo DISTRIBUTE BY HASH(i) INTO 4 BUCKETS AS SELECT 1");
- ParsesOk("CREATE TABLE Foo PRIMARY KEY (a) DISTRIBUTE BY HASH(a) INTO 4 BUCKETS " +
+ ParserError("CREATE TABLE Foo PARTITION BY HASH(i) INTO 4 BUCKETS AS SELECT 1");
+ ParsesOk("CREATE TABLE Foo PRIMARY KEY (a) PARTITION BY HASH(a) INTO 4 BUCKETS " +
"TBLPROPERTIES ('a'='b', 'c'='d') AS SELECT * from bar");
- ParsesOk("CREATE TABLE Foo PRIMARY KEY (a) DISTRIBUTE BY RANGE(a) " +
+ ParsesOk("CREATE TABLE Foo PRIMARY KEY (a) PARTITION BY RANGE(a) " +
"(PARTITION 1 < VALUES < 10, PARTITION 10 <= VALUES < 20, PARTITION VALUE = 30) " +
"STORED AS KUDU AS SELECT * FROM Bar");
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/bin/generate-schema-statements.py
----------------------------------------------------------------------
diff --git a/testdata/bin/generate-schema-statements.py b/testdata/bin/generate-schema-statements.py
index 1fe6a01..5d5da1b 100755
--- a/testdata/bin/generate-schema-statements.py
+++ b/testdata/bin/generate-schema-statements.py
@@ -223,7 +223,7 @@ def build_table_template(file_format, columns, partition_columns, row_format,
elif file_format == 'kudu':
# Use partitioned_by to set a trivial hash distribution
assert not partitioned_by, "Kudu table shouldn't have partition cols defined"
- partitioned_by = "distribute by hash into 3 buckets"
+ partitioned_by = "partition by hash into 3 buckets"
# Fetch KUDU host and port from environment
kudu_master = os.getenv("KUDU_MASTER_ADDRESS", "127.0.0.1")
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/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 be7a6ac..d6859eb 100644
--- a/testdata/datasets/functional/functional_schema_template.sql
+++ b/testdata/datasets/functional/functional_schema_template.sql
@@ -93,7 +93,7 @@ CREATE TABLE {db_name}{db_suffix}.{table_name} (
year INT,
month INT
)
-DISTRIBUTE BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
+PARTITION BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
---- DEPENDENT_LOAD_KUDU
INSERT into TABLE {db_name}{db_suffix}.{table_name}
SELECT id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col,
@@ -171,7 +171,7 @@ CREATE TABLE {db_name}{db_suffix}.{table_name} (
year INT,
month INT
)
-DISTRIBUTE BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
+PARTITION BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
---- DEPENDENT_LOAD_KUDU
INSERT into TABLE {db_name}{db_suffix}.{table_name}
SELECT id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col,
@@ -230,7 +230,7 @@ CREATE TABLE {db_name}{db_suffix}.{table_name} (
year INT,
month INT
)
-DISTRIBUTE BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
+PARTITION BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
---- DEPENDENT_LOAD_KUDU
INSERT INTO TABLE {db_name}{db_suffix}.{table_name}
SELECT id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col,
@@ -566,7 +566,7 @@ CREATE TABLE {db_name}{db_suffix}.{table_name}_idx (
month INT NULL,
day INT NULL
)
-DISTRIBUTE BY HASH (kudu_idx) INTO 3 BUCKETS STORED AS KUDU;
+PARTITION BY HASH (kudu_idx) INTO 3 BUCKETS STORED AS KUDU;
CREATE VIEW {db_name}{db_suffix}.{table_name} AS
SELECT id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col,
double_col, date_string_col, string_col, timestamp_col, year, month, day
@@ -645,7 +645,7 @@ CREATE TABLE {db_name}{db_suffix}.{table_name} (
month INT,
day INT
)
-DISTRIBUTE BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
+PARTITION BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
---- DEPENDENT_LOAD_KUDU
INSERT into TABLE {db_name}{db_suffix}.{table_name}
SELECT id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col,
@@ -766,7 +766,7 @@ create table {db_name}{db_suffix}.{table_name} (
name string null,
zip int null
)
-distribute by range(id) (partition values <= 1003, partition 1003 < values <= 1007,
+partition by range(id) (partition values <= 1003, partition 1003 < values <= 1007,
partition 1007 < values) stored as kudu;
====
---- DATASET
@@ -790,7 +790,7 @@ create table {db_name}{db_suffix}.{table_name} (
name string,
zip int
)
-distribute by range(id) (partition values <= 1003, partition 1003 < values <= 1007,
+partition by range(id) (partition values <= 1003, partition 1003 < values <= 1007,
partition 1007 < values) stored as kudu;
====
---- DATASET
@@ -817,7 +817,7 @@ create table {db_name}{db_suffix}.{table_name} (
alltypes_id int,
primary key (test_id, test_name, test_zip, alltypes_id)
)
-distribute by range(test_id) (partition values <= 1003, partition 1003 < values <= 1007,
+partition by range(test_id) (partition values <= 1003, partition 1003 < values <= 1007,
partition 1007 < values) stored as kudu;
====
---- DATASET
@@ -1157,7 +1157,7 @@ CREATE TABLE {db_name}{db_suffix}.{table_name} (
field STRING PRIMARY KEY,
f2 INT
)
-DISTRIBUTE BY HASH (field) INTO 3 BUCKETS STORED AS KUDU;
+PARTITION BY HASH (field) INTO 3 BUCKETS STORED AS KUDU;
====
---- DATASET
functional
@@ -1264,7 +1264,7 @@ create table {db_name}{db_suffix}.{table_name} (
a string primary key,
b string
)
-distribute by range(a) (partition values <= 'b', partition 'b' < values <= 'd',
+partition by range(a) (partition values <= 'b', partition 'b' < values <= 'd',
partition 'd' < values) stored as kudu;
====
---- DATASET
@@ -1284,7 +1284,7 @@ DROP TABLE IF EXISTS {db_name}{db_suffix}.{table_name};
create table {db_name}{db_suffix}.{table_name} (
int_col int primary key
)
-distribute by range(int_col) (partition values <= 2, partition 2 < values <= 4,
+partition by range(int_col) (partition values <= 2, partition 2 < values <= 4,
partition 4 < values <= 6, partition 6 < values <= 8, partition 8 < values)
stored as kudu;
====
@@ -1313,7 +1313,7 @@ create table {db_name}{db_suffix}.{table_name} (
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;
+partition by hash(a) into 3 buckets stored as kudu;
====
---- DATASET
functional
@@ -1340,7 +1340,7 @@ create table {db_name}{db_suffix}.{table_name} (
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;
+partition by hash(a) into 3 buckets stored as kudu;
====
---- DATASET
functional
@@ -1418,7 +1418,7 @@ create table {db_name}{db_suffix}.{table_name} (
description1 string null,
description2 string null,
income int null)
-distribute by range(id)
+partition by range(id)
(partition values <= '8600000US01475',
partition '8600000US01475' < values <= '8600000US63121',
partition '8600000US63121' < values <= '8600000US84712',
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/datasets/tpcds/tpcds_kudu_template.sql
----------------------------------------------------------------------
diff --git a/testdata/datasets/tpcds/tpcds_kudu_template.sql b/testdata/datasets/tpcds/tpcds_kudu_template.sql
index 6c0b994..021fe87 100644
--- a/testdata/datasets/tpcds/tpcds_kudu_template.sql
+++ b/testdata/datasets/tpcds/tpcds_kudu_template.sql
@@ -49,7 +49,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.store_sales (
ss_net_profit DOUBLE,
PRIMARY KEY (ss_ticket_number, ss_item_sk)
)
-DISTRIBUTE BY HASH (ss_ticket_number,ss_item_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (ss_ticket_number,ss_item_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -117,7 +117,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.web_sales (
ws_net_profit DOUBLE,
PRIMARY KEY (ws_order_number, ws_item_sk)
)
-DISTRIBUTE BY HASH (ws_order_number,ws_item_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (ws_order_number,ws_item_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -197,7 +197,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.catalog_sales (
cs_net_profit DOUBLE,
PRIMARY KEY (cs_order_number, cs_item_sk)
)
-DISTRIBUTE BY HASH (cs_order_number,cs_item_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (cs_order_number,cs_item_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -263,7 +263,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.store_returns (
sr_net_loss DOUBLE,
PRIMARY KEY (sr_ticket_number, sr_item_sk)
)
-DISTRIBUTE BY HASH (sr_ticket_number,sr_item_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (sr_ticket_number,sr_item_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -319,7 +319,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.web_returns (
wr_net_loss DOUBLE,
PRIMARY KEY (wr_order_number, wr_item_sk)
)
-DISTRIBUTE BY HASH (wr_order_number,wr_item_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (wr_order_number,wr_item_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -382,7 +382,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.catalog_returns (
cr_net_loss DOUBLE,
PRIMARY KEY (cr_order_number, cr_item_sk)
)
-DISTRIBUTE BY HASH (cr_order_number,cr_item_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (cr_order_number,cr_item_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -425,7 +425,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.inventory (
inv_quantity_on_hand BIGINT,
PRIMARY KEY (inv_date_sk, inv_item_sk, inv_warehouse_sk)
)
-DISTRIBUTE BY HASH (inv_item_sk,inv_date_sk,inv_warehouse_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (inv_item_sk,inv_date_sk,inv_warehouse_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -453,7 +453,7 @@ CREATE TABLE {target_db_name}.customer (
c_email_address STRING,
c_last_review_date BIGINT
)
-DISTRIBUTE BY HASH (c_customer_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (c_customer_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -475,7 +475,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.customer_address (
ca_gmt_offset DOUBLE,
ca_location_type STRING
)
-DISTRIBUTE BY HASH (ca_address_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (ca_address_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -494,7 +494,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.customer_demographics (
cd_dep_employed_count BIGINT,
cd_dep_college_count BIGINT
)
-DISTRIBUTE BY HASH (cd_demo_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (cd_demo_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -532,7 +532,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.date_dim (
d_current_quarter STRING,
d_current_year STRING
)
-DISTRIBUTE BY HASH (d_date_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (d_date_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -546,7 +546,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.household_demographics (
hd_dep_count BIGINT,
hd_vehicle_count BIGINT
)
-DISTRIBUTE BY HASH (hd_demo_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (hd_demo_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -578,7 +578,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.item (
i_manager_id BIGINT,
i_product_name STRING
)
-DISTRIBUTE BY HASH (i_item_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (i_item_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -606,7 +606,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.promotion (
p_purpose STRING,
p_discount_active STRING
)
-DISTRIBUTE BY HASH (p_promo_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (p_promo_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -665,7 +665,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.store (
s_gmt_offset DOUBLE,
s_tax_precentage DOUBLE
)
-DISTRIBUTE BY HASH (s_store_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (s_store_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -684,7 +684,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.time_dim (
t_sub_shift STRING,
t_meal_time STRING
)
-DISTRIBUTE BY HASH (t_time_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (t_time_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -724,7 +724,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.call_center (
cc_gmt_offset DOUBLE,
cc_tax_percentage DOUBLE
)
-DISTRIBUTE BY HASH (cc_call_center_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (cc_call_center_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -742,7 +742,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.catalog_page (
cp_description STRING,
cp_type STRING
)
-DISTRIBUTE BY HASH (cp_catalog_page_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (cp_catalog_page_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -754,7 +754,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.income_band (
ib_lower_bound BIGINT,
ib_upper_bound BIGINT
)
-DISTRIBUTE BY HASH (ib_income_band_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (ib_income_band_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -766,7 +766,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.reason (
r_reason_id STRING,
r_reason_desc STRING
)
-DISTRIBUTE BY HASH (r_reason_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (r_reason_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -781,7 +781,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.ship_mode (
sm_carrier STRING,
sm_contract STRING
)
-DISTRIBUTE BY HASH (sm_ship_mode_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (sm_ship_mode_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -804,7 +804,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.warehouse (
w_country STRING,
w_gmt_offset DOUBLE
)
-DISTRIBUTE BY HASH (w_warehouse_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (w_warehouse_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -827,7 +827,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.web_page (
wp_image_count BIGINT,
wp_max_ad_count BIGINT
)
-DISTRIBUTE BY HASH (wp_web_page_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (wp_web_page_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
@@ -862,7 +862,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.web_site (
web_gmt_offset DOUBLE,
web_tax_percentage DOUBLE
)
-DISTRIBUTE BY HASH (web_site_sk) INTO {buckets} BUCKETS
+PARTITION BY HASH (web_site_sk) INTO {buckets} BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_master}:7051');
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/datasets/tpch/tpch_kudu_template.sql
----------------------------------------------------------------------
diff --git a/testdata/datasets/tpch/tpch_kudu_template.sql b/testdata/datasets/tpch/tpch_kudu_template.sql
index 26dbb7a..ed9765d 100644
--- a/testdata/datasets/tpch/tpch_kudu_template.sql
+++ b/testdata/datasets/tpch/tpch_kudu_template.sql
@@ -42,7 +42,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.lineitem (
L_COMMENT STRING,
PRIMARY KEY (L_ORDERKEY, L_LINENUMBER)
)
-distribute by hash (l_orderkey) into {buckets} buckets
+partition by hash (l_orderkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -78,7 +78,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.part (
P_RETAILPRICE DOUBLE,
P_COMMENT STRING
)
-distribute by hash (p_partkey) into {buckets} buckets
+partition by hash (p_partkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -93,7 +93,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.partsupp (
PS_COMMENT STRING,
PRIMARY KEY (PS_PARTKEY, PS_SUPPKEY)
)
-distribute by hash (ps_partkey, ps_suppkey) into {buckets} buckets
+partition by hash (ps_partkey, ps_suppkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -109,7 +109,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.supplier (
S_ACCTBAL DOUBLE,
S_COMMENT STRING
)
-distribute by hash (s_suppkey) into {buckets} buckets
+partition by hash (s_suppkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -122,7 +122,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.nation (
N_REGIONKEY BIGINT,
N_COMMENT STRING
)
-distribute by hash (n_nationkey) into {buckets} buckets
+partition by hash (n_nationkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -134,7 +134,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.region (
R_NAME STRING,
R_COMMENT STRING
)
-distribute by hash (r_regionkey) into {buckets} buckets
+partition by hash (r_regionkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -152,7 +152,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.orders (
O_SHIPPRIORITY BIGINT,
O_COMMENT STRING
)
-distribute by hash (o_orderkey) into {buckets} buckets
+partition by hash (o_orderkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
@@ -169,7 +169,7 @@ CREATE TABLE IF NOT EXISTS {target_db_name}.customer (
C_MKTSEGMENT STRING,
C_COMMENT STRING
)
-distribute by hash (c_custkey) into {buckets} buckets
+partition by hash (c_custkey) into {buckets} buckets
STORED AS KUDU
tblproperties ('kudu.master_addresses' = '{kudu_master}:7051');
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/datasets/tpch/tpch_schema_template.sql
----------------------------------------------------------------------
diff --git a/testdata/datasets/tpch/tpch_schema_template.sql b/testdata/datasets/tpch/tpch_schema_template.sql
index 0512f6a..dfa2305 100644
--- a/testdata/datasets/tpch/tpch_schema_template.sql
+++ b/testdata/datasets/tpch/tpch_schema_template.sql
@@ -60,7 +60,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
L_COMMENT STRING,
PRIMARY KEY(L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER)
)
-distribute by hash (l_orderkey) into 9 buckets stored as kudu;
+partition by hash (l_orderkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
@@ -95,7 +95,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
P_RETAILPRICE DOUBLE,
P_COMMENT STRING
)
-distribute by hash (p_partkey) into 9 buckets stored as kudu;
+partition by hash (p_partkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
@@ -123,7 +123,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
PS_COMMENT STRING,
PRIMARY KEY(PS_PARTKEY, PS_SUPPKEY)
)
-distribute by hash (ps_partkey, ps_suppkey) into 9 buckets stored as kudu;
+partition by hash (ps_partkey, ps_suppkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
@@ -154,7 +154,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
S_ACCTBAL DOUBLE,
S_COMMENT STRING
)
-distribute by hash (s_suppkey) into 9 buckets stored as kudu;
+partition by hash (s_suppkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
@@ -179,7 +179,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
N_REGIONKEY SMALLINT,
N_COMMENT STRING
)
-distribute by hash (n_nationkey) into 9 buckets stored as kudu;
+partition by hash (n_nationkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
@@ -202,7 +202,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
R_NAME STRING,
R_COMMENT STRING
)
-distribute by hash (r_regionkey) into 9 buckets stored as kudu;
+partition by hash (r_regionkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
@@ -237,7 +237,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
O_SHIPPRIORITY INT,
O_COMMENT STRING
)
-distribute by hash (o_orderkey) into 9 buckets stored as kudu;
+partition by hash (o_orderkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
@@ -270,7 +270,7 @@ create table if not exists {db_name}{db_suffix}.{table_name} (
C_MKTSEGMENT STRING,
C_COMMENT STRING
)
-distribute by hash (c_custkey) into 9 buckets stored as kudu;
+partition by hash (c_custkey) into 9 buckets stored as kudu;
---- DEPENDENT_LOAD
INSERT OVERWRITE TABLE {db_name}{db_suffix}.{table_name} SELECT * FROM {db_name}.{table_name};
---- LOAD
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-planner/queries/PlannerTest/lineage.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/lineage.test b/testdata/workloads/functional-planner/queries/PlannerTest/lineage.test
index c2d916e..5dfa96a 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/lineage.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/lineage.test
@@ -4717,13 +4717,13 @@ functional.alltypes where id < 10
}
====
# CTAS a Kudu table
-create table kudu_ctas primary key (id) distribute by hash (id) into 3 buckets
+create table kudu_ctas primary key (id) partition by hash (id) into 3 buckets
stored as kudu as select id, bool_col, tinyint_col, smallint_col, int_col,
bigint_col, float_col, double_col, date_string_col, string_col
from functional.alltypestiny
---- LINEAGE
{
- "queryText":"create table kudu_ctas primary key (id) distribute by hash (id) into 3 buckets\nstored as kudu as select id, bool_col, tinyint_col, smallint_col, int_col,\nbigint_col, float_col, double_col, date_string_col, string_col\nfrom functional.alltypestiny",
+ "queryText":"create table kudu_ctas primary key (id) partition by hash (id) into 3 buckets\nstored as kudu as select id, bool_col, tinyint_col, smallint_col, int_col,\nbigint_col, float_col, double_col, date_string_col, string_col\nfrom functional.alltypestiny",
"hash":"6e3e192c7fb8bb6b22674a9b7b488b55",
"user":"dev",
"timestamp":1479933751,
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test b/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
index 1017658..21b1b95 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
@@ -21,7 +21,7 @@ INT, BIGINT
# Regression test for IMPALA-2740, a NULL value from a previously filtered row would
# carry over into the next unfiltered row (the result below would incorrectly be 2,NULL).
CREATE TABLE impala_2740 (key INT PRIMARY KEY, value INT)
- DISTRIBUTE BY HASH (key) INTO 3 BUCKETS STORED AS KUDU;
+ PARTITION BY HASH (key) INTO 3 BUCKETS STORED AS KUDU;
INSERT INTO impala_2740 VALUES (1, NULL), (2, -2);
SELECT * FROM impala_2740 WHERE key != 1;
---- RESULTS
@@ -35,9 +35,9 @@ INT, INT
# is run on all impalads. However, for the t1 table there is only as single scan range,
# so two of the scan instances get empty scan ranges.
CREATE TABLE impala_2635_t1 (id BIGINT PRIMARY KEY, name STRING)
- DISTRIBUTE BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
+ PARTITION BY HASH (id) INTO 3 BUCKETS STORED AS KUDU;
CREATE TABLE impala_2635_t2 (id BIGINT PRIMARY KEY, name STRING)
- DISTRIBUTE BY HASH(id) INTO 16 BUCKETS STORED AS KUDU;
+ PARTITION BY HASH(id) INTO 16 BUCKETS STORED AS KUDU;
INSERT INTO impala_2635_t1 VALUES (0, 'Foo');
INSERT INTO impala_2635_t2 VALUES (1, 'Blah');
SELECT * FROM impala_2635_t1 UNION ALL SELECT * FROM impala_2635_t2;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu-timeouts-catalogd.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu-timeouts-catalogd.test b/testdata/workloads/functional-query/queries/QueryTest/kudu-timeouts-catalogd.test
index 253c139..ac57812 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu-timeouts-catalogd.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu-timeouts-catalogd.test
@@ -12,7 +12,7 @@ Error opening Kudu table 'impala::functional_kudu.alltypes'
====
---- QUERY
create table test_kudu (x int primary key)
-distribute by hash(x) into 3 buckets stored as kudu
+partition by hash(x) into 3 buckets stored as kudu
---- CATCH
Error creating Kudu table
====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
index 116a7e9..8722f8a 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_alter.test
@@ -1,7 +1,7 @@
====
---- QUERY
create table simple (id int primary key, name string, valf float, vali bigint)
- distribute by hash (id) into 3 buckets stored as kudu
+ partition by hash (id) into 3 buckets stored as kudu
---- RESULTS
====
---- QUERY
@@ -49,7 +49,7 @@ BIGINT
---- QUERY
# Create a table with range distribution
create table tbl_to_alter (id int primary key, name string null, vali bigint not null)
- distribute by range (id) (partition 1 < values <= 10) stored as kudu
+ partition by range (id) (partition 1 < values <= 10) stored as kudu
---- RESULTS
====
---- QUERY
@@ -294,7 +294,7 @@ alter table tbl_to_alter set tblproperties('kudu.table_name'='kudu_tbl_to_alter'
====
---- QUERY
# Create a new table and try to rename to an existing kudu table
-create table copy_of_tbl (a int primary key) distribute by hash (a) into 3 buckets
+create table copy_of_tbl (a int primary key) partition by hash (a) into 3 buckets
stored as kudu tblproperties('kudu.table_name'='copy_of_tbl');
alter table copy_of_tbl set tblproperties('kudu.table_name'='kudu_tbl_to_alter')
---- CATCH
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_create.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_create.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_create.test
index d04efd6..651979d 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_create.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_create.test
@@ -1,6 +1,6 @@
====
---- QUERY
-create table t primary key (id) distribute by hash (id) into 3 buckets
+create table t primary key (id) partition by hash (id) into 3 buckets
stored as kudu
as select id, int_col from functional.alltypestiny;
select * from t;
@@ -19,14 +19,14 @@ INT,INT
---- QUERY
# Boolean primary key column
create table tab (x int, y boolean, primary key(x, y))
- distribute by hash (x) into 3 buckets stored as kudu
+ partition by hash (x) into 3 buckets stored as kudu
---- CATCH
NonRecoverableException: Key column may not have type of BOOL, FLOAT, or DOUBLE
====
---- QUERY
# Float primary key column
create table tab (x int, y float, primary key(x, y))
- distribute by hash (x) into 3 buckets stored as kudu
+ partition by hash (x) into 3 buckets stored as kudu
---- CATCH
NonRecoverableException: Key column may not have type of BOOL, FLOAT, or DOUBLE
====
@@ -34,27 +34,27 @@ NonRecoverableException: Key column may not have type of BOOL, FLOAT, or DOUBLE
# Primary keys should be declared first
# TODO: See KUDU-1709 for improving Kudu error messages.
create table tab (x int, y int, primary key(y))
- distribute by hash (y) into 3 buckets stored as kudu
+ partition by hash (y) into 3 buckets stored as kudu
---- CATCH
NonRecoverableException: Got out-of-order key column: name: "y" type: INT32 is_key: true is_nullable: false cfile_block_size: 0
====
---- QUERY
# Small number of hash buckets
create table tab (a int, b int, c int, d int, primary key(a, b, c))
- distribute by hash(a,b) into 8 buckets, hash(c) into 1 buckets stored as kudu
+ partition by hash(a,b) into 8 buckets, hash(c) into 1 buckets stored as kudu
---- CATCH
NonRecoverableException: must have at least two hash buckets
====
---- QUERY
# Same column in multiple hash based distributions
create table tab (a int, b int, primary key (a))
- distribute by hash (a) into 3 buckets, hash (a) into 2 buckets stored as kudu
+ partition by hash (a) into 3 buckets, hash (a) into 2 buckets stored as kudu
---- CATCH
NonRecoverableException: hash bucket schema components must not contain columns in common
====
---- QUERY
# Same column referenced multiple times in the same hash-based distribution
-create table tab (a int primary key) distribute by hash (a, a, a) into 3 buckets
+create table tab (a int primary key) partition by hash (a, a, a) into 3 buckets
stored as kudu
---- CATCH
NonRecoverableException: hash bucket schema components must not contain columns in common
@@ -62,7 +62,7 @@ NonRecoverableException: hash bucket schema components must not contain columns
---- QUERY
# Kudu table that uses Impala keywords as table name and column names
create table `add`(`analytic` int, `function` int, primary key(`analytic`, `function`))
-distribute by hash (`analytic`) into 4 buckets, range (`function`)
+partition by hash (`analytic`) into 4 buckets, range (`function`)
(partition values <= 1, partition 1 < values <= 10, partition 10 < values) stored as kudu;
insert into `add` select id, int_col from functional.alltypestiny;
select * from `add`
@@ -80,14 +80,14 @@ INT,INT
====
---- QUERY
# Test implicit casting/folding of partition values.
-create table tab (a int primary key) distribute by range (a) (partition value = false)
+create table tab (a int primary key) partition by range (a) (partition value = false)
stored as kudu
---- RESULTS
====
---- QUERY
# Invalid hostname
create table tdata_bogus_host (id int primary key, name string, valf float, vali bigint)
- DISTRIBUTE BY RANGE (PARTITION 10 <= VALUES <= 30) STORED AS KUDU
+ PARTITION BY RANGE (PARTITION 10 <= VALUES <= 30) STORED AS KUDU
TBLPROPERTIES('kudu.master_addresses' = 'bogus host name')
---- CATCH
Couldn't resolve this master's address bogus host name:7051
@@ -96,7 +96,7 @@ Couldn't resolve this master's address bogus host name:7051
# Non-existing host
create table tdata_non_existing_host
(id int primary key, name string, valf float, vali bigint)
- DISTRIBUTE BY RANGE (PARTITION 10 <= VALUES <= 30) STORED AS KUDU
+ PARTITION BY RANGE (PARTITION 10 <= VALUES <= 30) STORED AS KUDU
TBLPROPERTIES('kudu.master_addresses' = 'bogus.host.name')
---- CATCH
Couldn't resolve this master's address bogus.host.name:7051
@@ -104,7 +104,7 @@ Couldn't resolve this master's address bogus.host.name:7051
---- QUERY
# Test that string case is ignored
create table ignore_column_case (Id int, NAME string, vAlf float, vali bigint,
- primary key (Id, NAME)) DISTRIBUTE BY RANGE (PARTITION VALUE = (1, 'Martin'))
+ primary key (Id, NAME)) PARTITION BY RANGE (PARTITION VALUE = (1, 'Martin'))
STORED AS KUDU
---- RESULTS
====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_delete.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_delete.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_delete.test
index 9e2a924..4dd3377 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_delete.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_delete.test
@@ -3,7 +3,7 @@
create table tdata
(id int primary key, valf float null, vali bigint null, valv string null,
valb boolean null, valt tinyint null, vals smallint null, vald double null)
- DISTRIBUTE BY RANGE (PARTITION VALUES < 100, PARTITION 100 <= VALUES < 1000,
+ PARTITION BY RANGE (PARTITION VALUES < 100, PARTITION 100 <= VALUES < 1000,
PARTITION 1000 <= VALUES <= 10000) STORED AS KUDU
---- RESULTS
====
@@ -301,7 +301,7 @@ create table multiple_key_cols
smallint_col smallint, bool_col boolean null, int_col int null,
double_col double null, float_col float null,
primary key (string_col, bigint_col, tinyint_col, smallint_col))
- DISTRIBUTE BY HASH (string_col) INTO 16 BUCKETS STORED AS KUDU
+ PARTITION BY HASH (string_col) INTO 16 BUCKETS STORED AS KUDU
====
---- QUERY
insert into multiple_key_cols values
@@ -343,7 +343,7 @@ STRING,BIGINT,TINYINT,SMALLINT,BOOLEAN,INT,DOUBLE,FLOAT
# IMPALA-3454: A delete that requires a rewrite may not get the Kudu column order correct
# if the Kudu columns are of different types.
create table impala_3454 (key_1 tinyint, key_2 bigint, PRIMARY KEY (key_1, key_2))
- DISTRIBUTE BY HASH INTO 3 BUCKETS STORED AS KUDU
+ PARTITION BY HASH INTO 3 BUCKETS STORED AS KUDU
---- RESULTS
====
---- QUERY
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_describe.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_describe.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_describe.test
index 3d8ac2c..fade3ce 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_describe.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_describe.test
@@ -30,7 +30,7 @@ create table describe_test
c2 int not null default 100 encoding plain_encoding compression snappy,
c3 int null block_size 8388608,
primary key (pk1, pk2, pk3))
-distribute by hash (pk1) into 3 buckets
+partition by hash (pk1) into 3 buckets
stored as kudu;
describe describe_test;
---- LABELS
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
index e282567..d67b75f 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_insert.test
@@ -3,7 +3,7 @@
create table tdata
(id int primary key, valf float null, vali bigint null, valv string null,
valb boolean null, valt tinyint null, vals smallint null, vald double null)
- DISTRIBUTE BY RANGE (PARTITION VALUES < 10, PARTITION 10 <= VALUES < 30,
+ PARTITION BY RANGE (PARTITION VALUES < 10, PARTITION 10 <= VALUES < 30,
PARTITION 30 <= VALUES) STORED AS KUDU
---- RESULTS
====
@@ -238,7 +238,7 @@ NumRowErrors: 1
====
---- QUERY
create table kudu_test_tbl primary key(id)
-distribute by range(id) (partition values < 100, partition 100 <= values <= 10000)
+partition by range(id) (partition values < 100, partition 100 <= values <= 10000)
stored as kudu as
select * from functional_kudu.alltypes where id < 100;
---- RESULTS
@@ -281,7 +281,7 @@ NumRowErrors: 7299
# IMPALA-2521: clustered insert into table.
create table impala_2521
(id bigint primary key, name string, zip int)
-distribute by hash into 3 buckets stored as kudu
+partition by hash into 3 buckets stored as kudu
---- RESULTS
====
---- QUERY
@@ -316,7 +316,7 @@ BIGINT,STRING,INT
---- QUERY
# Table with all supported types as primary key and distribution columns
create table allkeytypes (i1 tinyint, i2 smallint, i3 int, i4 bigint, name string,
- valf float, vald double, primary key (i1, i2, i3, i4, name)) distribute by
+ valf float, vald double, primary key (i1, i2, i3, i4, name)) partition by
hash into 3 buckets, range (partition value = (1,1,1,1,'1'),
partition value = (2,2,2,2,'2'), partition value = (3,3,3,3,'3')) stored as kudu
---- RESULTS
@@ -335,7 +335,7 @@ NumRowErrors: 6
# Table with default values
create table tbl_with_defaults (a int primary key, b int null default 10,
c int not null default 100, d int default 1000, e int null, f int not null,
- g string default 'test', h boolean default true) distribute by hash (a) into 3
+ g string default 'test', h boolean default true) partition by hash (a) into 3
buckets stored as kudu
---- RESULTS
====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_partition_ddl.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_partition_ddl.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_partition_ddl.test
index 13eec9d..5e9c477 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_partition_ddl.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_partition_ddl.test
@@ -2,7 +2,7 @@
---- QUERY
-- Test hash partitioning
create table simple_hash (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by hash(id) INTO 4 buckets,
+ primary key (id, name)) partition by hash(id) INTO 4 buckets,
hash(name) INTO 2 buckets stored as kudu
---- RESULTS
====
@@ -25,7 +25,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test single column range partitioning with bounded and unbounded partitions
create table range_part_bounds (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by range (id)
+ primary key (id, name)) partition by range (id)
(partition values <= 10, partition 10 < values <= 20, partition 20 < values)
stored as kudu
---- RESULTS
@@ -44,7 +44,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test single column range partitioning with single value partitions
create table range_part_single (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by range (id)
+ primary key (id, name)) partition by range (id)
(partition value = 1, partition value = 10, partition value = 100)
stored as kudu
---- RESULTS
@@ -64,7 +64,7 @@ INT,STRING,STRING,STRING,INT
-- Test single column range partitioning with bounded, unbounded and single
-- value partitions
create table range_part_multiple_bounds (id int, name string, valf float,
- primary key (id, name)) distribute by range (id)
+ primary key (id, name)) partition by range (id)
(partition values <= 10, partition 10 < values <= 20, partition 20 < values <= 30,
partition value = 40, partition value = 50) stored as kudu
---- RESULTS
@@ -85,7 +85,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test multiple column range partitioning
create table range_part_multiple_cols (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by range (id, name)
+ primary key (id, name)) partition by range (id, name)
(partition value = (10, 'martin'), partition value = (20, 'dimitris'),
partition value = (30, 'matthew')) stored as kudu
---- RESULTS
@@ -104,7 +104,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test single column range partitioning with string partition column
create table range_part_single_string_col (id int, name string, valf float,
- primary key (id, name)) distribute by range(name)
+ primary key (id, name)) partition by range(name)
(partition values <= 'aaa', partition 'aaa' < values <= 'bbb',
partition 'bbb' < values <= 'ccc', partition value = 'ddd') stored as kudu
---- RESULTS
@@ -124,7 +124,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test hash and range partitioning
create table simple_hash_range (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by hash(id) into 4 buckets, range(id, name)
+ primary key (id, name)) partition by hash(id) into 4 buckets, range(id, name)
(partition value = (10, 'martin'), partition value = (20, 'alex')) stored as kudu
---- RESULTS
====
@@ -146,7 +146,7 @@ INT,STRING,STRING,STRING,INT
====
---- QUERY
create table simple_hash_range_ctas
- primary key (id, name) distribute by hash(id) into 4 buckets,
+ primary key (id, name) partition by hash(id) into 4 buckets,
range(id, name) (partition value = (10, 'casey'), partition value = (20, 'marcel'))
stored as kudu
as select * from simple_hash
@@ -172,7 +172,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test hash defaults to all columns
create table simple_hash_all_columns (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by hash into 4 buckets stored as kudu
+ primary key (id, name)) partition by hash into 4 buckets stored as kudu
---- RESULTS
====
---- QUERY
@@ -190,7 +190,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test range defaults to all columns
create table simple_range_all_columns (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by range
+ primary key (id, name)) partition by range
(partition value = (1, 'a'), partition value = (2, 'b'))
stored as kudu
---- RESULTS
@@ -208,7 +208,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test using non-literal constant values in range-partition bounds
create table range_complex_const_boundary_vals (x int, y int, primary key (x))
- distribute by range (x) (partition values < 1 + 1, partition (1+3) + 2 < values < 10,
+ partition by range (x) (partition values < 1 + 1, partition (1+3) + 2 < values < 10,
partition factorial(4) < values < factorial(5), partition value = factorial(6))
stored as kudu
---- RESULTS
@@ -228,7 +228,7 @@ INT,STRING,STRING,STRING,INT
---- QUERY
-- Test range partitioning with overlapping partitions
create table simple_range_with_overlapping (id int, name string, valf float, vali bigint,
- primary key (id, name)) distribute by range (id)
+ primary key (id, name)) partition by range (id)
(partition values <= 10, partition values < 20, partition value = 5) stored as kudu
---- CATCH
NonRecoverableException: overlapping range partitions: first range partition: [<start>, (int32 id=11)), second range partition: [<start>, (int32 id=20))
@@ -236,7 +236,7 @@ NonRecoverableException: overlapping range partitions: first range partition: [<
---- QUERY
-- Test range partitioning with the same partition specified multiple times
create table simple_range_duplicate_parts (id int, name string, valf float, vali bigint,
- primary key(id, name)) distribute by range (id)
+ primary key(id, name)) partition by range (id)
(partition 10 < values <= 20, partition 10 < values <= 20) stored as kudu
---- CATCH
NonRecoverableException: overlapping range partitions: first range partition: [(int32 id=11), (int32 id=21)), second range partition: [(int32 id=11), (int32 id=21))
@@ -244,7 +244,7 @@ NonRecoverableException: overlapping range partitions: first range partition: [(
---- QUERY
-- Test multi-column range partitioning with the same partition specified multiple times
create table range_multi_col_duplicate_parts (id int, name string, valf float,
- vali bigint, primary key (id, name)) distribute by range (id, name)
+ vali bigint, primary key (id, name)) partition by range (id, name)
(partition value = (10, 'dimitris'), partition value = (10, 'dimitris')) stored as kudu
---- CATCH
NonRecoverableException: overlapping range partitions: first range partition: [(int32 id=10, string name=dimitris), (int32 id=10, string name=dimitris\000)), second range partition: [(int32 id=10, string name=dimitris), (int32 id=10, string name=dimitris\000))
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_stats.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_stats.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_stats.test
index dc8f052..d3b7bd3 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_stats.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_stats.test
@@ -1,7 +1,7 @@
====
---- QUERY
create table simple (id int primary key, name string, valf float, vali bigint)
- distribute by range (partition values < 10, partition 10 <= values < 30,
+ partition by range (partition values < 10, partition 10 <= values < 30,
partition 30 <= values) stored as kudu tblproperties('kudu.num_tablet_replicas' = '2')
---- RESULTS
====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_update.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_update.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_update.test
index 27120f5..8d0d149 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_update.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_update.test
@@ -4,7 +4,7 @@ create table tdata
(id int primary key, name string null, valf float null, vali bigint null,
valv string null, valb boolean null, valt tinyint null, vals smallint null,
vald double null)
- DISTRIBUTE BY RANGE (PARTITION VALUES < 10, PARTITION 10 <= VALUES < 30,
+ PARTITION BY RANGE (PARTITION VALUES < 10, PARTITION 10 <= VALUES < 30,
PARTITION 30 <= VALUES <= 10000) STORED AS KUDU
---- RESULTS
====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/testdata/workloads/functional-query/queries/QueryTest/kudu_upsert.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu_upsert.test b/testdata/workloads/functional-query/queries/QueryTest/kudu_upsert.test
index 0f117f1..b0adb48 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu_upsert.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu_upsert.test
@@ -4,7 +4,7 @@ create table tdata
(id int primary key, name string null, valf float null, vali bigint null,
valv string null, valb boolean null, valt tinyint null, vals smallint null,
vald double null)
- DISTRIBUTE BY RANGE (PARTITION VALUES < 10, PARTITION 10 <= VALUES < 30,
+ PARTITION BY RANGE (PARTITION VALUES < 10, PARTITION 10 <= VALUES < 30,
PARTITION 30 <= VALUES) STORED AS KUDU
---- RESULTS
====
@@ -392,7 +392,7 @@ create table multiple_key_cols
(string_col string, bigint_col bigint, tinyint_col tinyint, smallint_col smallint,
bool_col boolean null, int_col int null, double_col double null,
float_col float null, primary key (string_col, bigint_col, tinyint_col, smallint_col))
- DISTRIBUTE BY HASH (string_col) INTO 16 BUCKETS STORED AS KUDU
+ PARTITION BY HASH (string_col) INTO 16 BUCKETS STORED AS KUDU
====
---- QUERY
insert into multiple_key_cols values
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/tests/query_test/test_cancellation.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_cancellation.py b/tests/query_test/test_cancellation.py
index 91e81dc..b660c61 100644
--- a/tests/query_test/test_cancellation.py
+++ b/tests/query_test/test_cancellation.py
@@ -102,7 +102,7 @@ class TestCancellation(ImpalaTestSuite):
assert QUERIES.has_key(query) and QUERIES[query] is not None,\
"PRIMARY KEY for query %s not specified" % query
query = "create table ctas_cancel primary key (%s) "\
- "distribute by hash into 3 buckets stored as kudu as %s" %\
+ "partition by hash into 3 buckets stored as kudu as %s" %\
(QUERIES[query], query)
else:
query = "create table ctas_cancel stored as %sfile as %s" %\
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/tests/query_test/test_kudu.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_kudu.py b/tests/query_test/test_kudu.py
index 9c9112e..602b5f4 100644
--- a/tests/query_test/test_kudu.py
+++ b/tests/query_test/test_kudu.py
@@ -83,7 +83,7 @@ class TestKuduOperations(KuduTestSuite):
for nullable in nullability:
impala_tbl_name = "test_column_options_%s" % str(indx)
cursor.execute("""CREATE TABLE %s.%s (a INT PRIMARY KEY
- %s %s %s %s, b INT %s %s %s %s %s) DISTRIBUTE BY HASH (a) INTO 3
+ %s %s %s %s, b INT %s %s %s %s %s) PARTITION BY HASH (a) INTO 3
BUCKETS STORED AS KUDU""" % (unique_database, impala_tbl_name,
encoding, compression, default, blocksize, nullable, encoding,
compression, default, blocksize))
@@ -93,7 +93,7 @@ class TestKuduOperations(KuduTestSuite):
def test_kudu_rename_table(self, cursor, kudu_client, unique_database):
"""Test Kudu table rename"""
- cursor.execute("""CREATE TABLE %s.foo (a INT PRIMARY KEY) DISTRIBUTE BY HASH(a)
+ cursor.execute("""CREATE TABLE %s.foo (a INT PRIMARY KEY) PARTITION BY HASH(a)
INTO 3 BUCKETS STORED AS KUDU""" % unique_database)
kudu_tbl_name = "impala::%s.foo" % unique_database
assert kudu_client.table_exists(kudu_tbl_name)
@@ -272,20 +272,20 @@ class TestShowCreateTable(KuduTestSuite):
self.assert_show_create_equals(cursor,
"""
CREATE TABLE {table} (c INT PRIMARY KEY)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS STORED AS KUDU""",
+ PARTITION BY HASH (c) INTO 3 BUCKETS STORED AS KUDU""",
"""
CREATE TABLE {db}.{{table}} (
c INT NOT NULL ENCODING AUTO_ENCODING COMPRESSION DEFAULT_COMPRESSION,
PRIMARY KEY (c)
)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS
+ PARTITION BY HASH (c) INTO 3 BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_addr}')""".format(
db=cursor.conn.db_name, kudu_addr=KUDU_MASTER_HOSTS))
self.assert_show_create_equals(cursor,
"""
CREATE TABLE {table} (c INT PRIMARY KEY, d STRING NULL)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS, RANGE (c)
+ PARTITION BY HASH (c) INTO 3 BUCKETS, RANGE (c)
(PARTITION VALUES <= 1, PARTITION 1 < VALUES <= 2,
PARTITION 2 < VALUES) STORED AS KUDU""",
"""
@@ -294,27 +294,27 @@ class TestShowCreateTable(KuduTestSuite):
d STRING NULL ENCODING AUTO_ENCODING COMPRESSION DEFAULT_COMPRESSION,
PRIMARY KEY (c)
)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS, RANGE (c) (...)
+ PARTITION BY HASH (c) INTO 3 BUCKETS, RANGE (c) (...)
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_addr}')""".format(
db=cursor.conn.db_name, kudu_addr=KUDU_MASTER_HOSTS))
self.assert_show_create_equals(cursor,
"""
CREATE TABLE {table} (c INT ENCODING PLAIN_ENCODING, PRIMARY KEY (c))
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS STORED AS KUDU""",
+ PARTITION BY HASH (c) INTO 3 BUCKETS STORED AS KUDU""",
"""
CREATE TABLE {db}.{{table}} (
c INT NOT NULL ENCODING PLAIN_ENCODING COMPRESSION DEFAULT_COMPRESSION,
PRIMARY KEY (c)
)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS
+ PARTITION BY HASH (c) INTO 3 BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_addr}')""".format(
db=cursor.conn.db_name, kudu_addr=KUDU_MASTER_HOSTS))
self.assert_show_create_equals(cursor,
"""
CREATE TABLE {table} (c INT COMPRESSION LZ4, d STRING, PRIMARY KEY(c, d))
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS, HASH (d) INTO 3 BUCKETS,
+ PARTITION BY HASH (c) INTO 3 BUCKETS, HASH (d) INTO 3 BUCKETS,
RANGE (c, d) (PARTITION VALUE = (1, 'aaa'), PARTITION VALUE = (2, 'bbb'))
STORED AS KUDU""",
"""
@@ -323,14 +323,14 @@ class TestShowCreateTable(KuduTestSuite):
d STRING NOT NULL ENCODING AUTO_ENCODING COMPRESSION DEFAULT_COMPRESSION,
PRIMARY KEY (c, d)
)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS, HASH (d) INTO 3 BUCKETS, RANGE (c, d) (...)
+ PARTITION BY HASH (c) INTO 3 BUCKETS, HASH (d) INTO 3 BUCKETS, RANGE (c, d) (...)
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_addr}')""".format(
db=cursor.conn.db_name, kudu_addr=KUDU_MASTER_HOSTS))
self.assert_show_create_equals(cursor,
"""
CREATE TABLE {table} (c INT, d STRING, e INT NULL DEFAULT 10, PRIMARY KEY(c, d))
- DISTRIBUTE BY RANGE (c) (PARTITION VALUES <= 1, PARTITION 1 < VALUES <= 2,
+ PARTITION BY RANGE (c) (PARTITION VALUES <= 1, PARTITION 1 < VALUES <= 2,
PARTITION 2 < VALUES <= 3, PARTITION 3 < VALUES) STORED AS KUDU""",
"""
CREATE TABLE {db}.{{table}} (
@@ -339,7 +339,7 @@ class TestShowCreateTable(KuduTestSuite):
e INT NULL ENCODING AUTO_ENCODING COMPRESSION DEFAULT_COMPRESSION DEFAULT 10,
PRIMARY KEY (c, d)
)
- DISTRIBUTE BY RANGE (c) (...)
+ PARTITION BY RANGE (c) (...)
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_addr}')""".format(
db=cursor.conn.db_name, kudu_addr=KUDU_MASTER_HOSTS))
@@ -352,7 +352,7 @@ class TestShowCreateTable(KuduTestSuite):
self.assert_show_create_equals(cursor,
"""
CREATE TABLE {{table}} (c INT PRIMARY KEY)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS
+ PARTITION BY HASH (c) INTO 3 BUCKETS
STORED AS KUDU
TBLPROPERTIES ({props})""".format(props=props),
"""
@@ -360,7 +360,7 @@ class TestShowCreateTable(KuduTestSuite):
c INT NOT NULL ENCODING AUTO_ENCODING COMPRESSION DEFAULT_COMPRESSION,
PRIMARY KEY (c)
)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS
+ PARTITION BY HASH (c) INTO 3 BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_addr}', {props})""".format(
db=cursor.conn.db_name, kudu_addr=KUDU_MASTER_HOSTS, props=props))
@@ -371,7 +371,7 @@ class TestShowCreateTable(KuduTestSuite):
self.assert_show_create_equals(cursor,
"""
CREATE TABLE {{table}} (c INT PRIMARY KEY)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS
+ PARTITION BY HASH (c) INTO 3 BUCKETS
STORED AS KUDU
TBLPROPERTIES ({props})""".format(props=props),
"""
@@ -379,7 +379,7 @@ class TestShowCreateTable(KuduTestSuite):
c INT NOT NULL ENCODING AUTO_ENCODING COMPRESSION DEFAULT_COMPRESSION,
PRIMARY KEY (c)
)
- DISTRIBUTE BY HASH (c) INTO 3 BUCKETS
+ PARTITION BY HASH (c) INTO 3 BUCKETS
STORED AS KUDU
TBLPROPERTIES ('kudu.master_addresses'='{kudu_addr}')""".format(
db=cursor.conn.db_name, kudu_addr=KUDU_MASTER_HOSTS))
@@ -421,7 +421,7 @@ class TestDropDb(KuduTestSuite):
# Create a managed Kudu table
managed_table_name = self.random_table_name()
unique_cursor.execute("""
- CREATE TABLE %s (a INT PRIMARY KEY) DISTRIBUTE BY HASH (a) INTO 3 BUCKETS
+ CREATE TABLE %s (a INT PRIMARY KEY) PARTITION BY HASH (a) INTO 3 BUCKETS
STORED AS KUDU TBLPROPERTIES ('kudu.table_name' = '%s')"""
% (managed_table_name, managed_table_name))
assert kudu_client.table_exists(managed_table_name)
@@ -506,7 +506,7 @@ class TestImpalaKuduIntegration(KuduTestSuite):
"""Check that dropping a managed Kudu table works even if the underlying Kudu table
has been dropped externally."""
impala_tbl_name = "foo"
- cursor.execute("""CREATE TABLE %s.%s (a INT PRIMARY KEY) DISTRIBUTE BY HASH (a)
+ cursor.execute("""CREATE TABLE %s.%s (a INT PRIMARY KEY) PARTITION BY HASH (a)
INTO 3 BUCKETS STORED AS KUDU""" % (unique_database, impala_tbl_name))
kudu_tbl_name = "impala::%s.%s" % (unique_database, impala_tbl_name)
assert kudu_client.table_exists(kudu_tbl_name)
@@ -545,7 +545,7 @@ class TestKuduMemLimits(KuduTestSuite):
l_shipmode STRING,
l_comment STRING,
PRIMARY KEY (l_orderkey, l_linenumber))
- DISTRIBUTE BY HASH (l_orderkey, l_linenumber) INTO 3 BUCKETS
+ PARTITION BY HASH (l_orderkey, l_linenumber) INTO 3 BUCKETS
STORED AS KUDU"""
LOAD = """
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/cba93f1a/tests/shell/test_shell_commandline.py
----------------------------------------------------------------------
diff --git a/tests/shell/test_shell_commandline.py b/tests/shell/test_shell_commandline.py
index 5b9859c..88deed0 100644
--- a/tests/shell/test_shell_commandline.py
+++ b/tests/shell/test_shell_commandline.py
@@ -478,7 +478,7 @@ class TestImpalaShell(ImpalaTestSuite):
def test_kudu_dml_reporting(self, unique_database):
db = unique_database
run_impala_shell_cmd('--query="create table %s.dml_test (id int primary key, '\
- 'age int null) distribute by hash(id) into 2 buckets stored as kudu"' % db)
+ 'age int null) partition by hash(id) into 2 buckets stored as kudu"' % db)
self._validate_dml_stmt("insert into %s.dml_test (id) values (7), (7)" % db, 1, 1)
self._validate_dml_stmt("insert into %s.dml_test (id) values (7)" % db, 0, 1)
[4/8] incubator-impala git commit: IMPALA-4578: Pick up bound
predicates for Kudu scan nodes.
Posted by kw...@apache.org.
IMPALA-4578: Pick up bound predicates for Kudu scan nodes.
The bug was a simple oversight. In KuduScanNiode.init()
we forgot to call Analyzer.getBoundPredicates().
Change-Id: I19a38d6ea8cc0d2b0ddc3808d1f9ffef5ce306a8
Reviewed-on: http://gerrit.cloudera.org:8080/5365
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/b656f570
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/b656f570
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/b656f570
Branch: refs/heads/master
Commit: b656f570e9d395129b820506a5e92ffb92b452f4
Parents: 5bb9959
Author: Alex Behm <al...@cloudera.com>
Authored: Mon Dec 5 13:43:36 2016 -0800
Committer: Internal Jenkins <cl...@gerrit.cloudera.org>
Committed: Tue Dec 6 06:19:03 2016 +0000
----------------------------------------------------------------------
.../org/apache/impala/planner/KuduScanNode.java | 10 +++++-
.../queries/PlannerTest/kudu.test | 33 ++++++++++++++++++++
2 files changed, 42 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b656f570/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
index 04fb244..580f5e0 100644
--- a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
@@ -102,8 +102,16 @@ public class KuduScanNode extends ScanNode {
@Override
public void init(Analyzer analyzer) throws ImpalaRuntimeException {
- assignConjuncts(analyzer);
+ conjuncts_.clear();
+ // Add bound predicates.
+ conjuncts_.addAll(analyzer.getBoundPredicates(desc_.getId()));
+ // Add unassigned predicates.
+ List<Expr> unassigned = analyzer.getUnassignedConjuncts(this);
+ conjuncts_.addAll(unassigned);
+ analyzer.markConjunctsAssigned(unassigned);
+ // Add equivalence predicates.
analyzer.createEquivConjuncts(tupleIds_.get(0), conjuncts_);
+ Expr.removeDuplicates(conjuncts_);
conjuncts_ = orderConjunctsByCost(conjuncts_);
try (KuduClient client = KuduUtil.createKuduClient(kuduTable_.getKuduMasterHosts())) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b656f570/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
index fa54ace..776882d 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
@@ -281,3 +281,36 @@ PLAN-ROOT SINK
00:SCAN KUDU [functional_kudu.alltypes]
kudu predicates: bool_col = FALSE
====
+# IMPALA-4578: Test predicate propagation for Kudu scans.
+select /* +straight_join */ count(*)
+from functional_kudu.alltypes a
+inner join functional_kudu.alltypessmall b
+ on a.id = b.id
+left outer join functional_kudu.alltypestiny c
+ on b.int_col = c.int_col
+where a.id > 10 and c.int_col > 20
+ and cast(b.id as string) > '123'
+---- PLAN
+PLAN-ROOT SINK
+|
+05:AGGREGATE [FINALIZE]
+| output: count(*)
+|
+04:HASH JOIN [LEFT OUTER JOIN]
+| hash predicates: b.int_col = c.int_col
+| other predicates: c.int_col > 20
+|
+|--02:SCAN KUDU [functional_kudu.alltypestiny c]
+| kudu predicates: c.int_col > 20
+|
+03:HASH JOIN [INNER JOIN]
+| hash predicates: a.id = b.id
+|
+|--01:SCAN KUDU [functional_kudu.alltypessmall b]
+| predicates: CAST(b.id AS STRING) > '123'
+| kudu predicates: b.id > 10
+|
+00:SCAN KUDU [functional_kudu.alltypes a]
+ predicates: CAST(a.id AS STRING) > '123'
+ kudu predicates: a.id > 10
+====