You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by li...@apache.org on 2021/11/16 03:59:18 UTC
[incubator-doris] branch master updated: [Feature] Support query hive table (#6569)
This is an automated email from the ASF dual-hosted git repository.
lingmiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new 5b01f7b [Feature] Support query hive table (#6569)
5b01f7b is described below
commit 5b01f7bba2462daf818fc956f650527d51e8be97
Author: qiye <ji...@gmail.com>
AuthorDate: Tue Nov 16 11:59:07 2021 +0800
[Feature] Support query hive table (#6569)
Users can directly query the data in the hive table in Doris, and can use join to perform complex queries without laboriously importing data from hive.
Main changes list below:
FE:
Extend HiveScanNode from BrokerScanNode
HiveMetaStoreClientHelper communicate with HIVE and HDFS.
BE:
Treate HiveScanNode as BrokerScanNode, treate HiveTable as BrokerTable.
broker_scanner.cpp: suppot read column from HDFS path.
orc_scanner.cpp: support read hdfs file.
POM:
Add hive.version=2.3.7, hive-metastore and hive-exec
Add hadoop.version=2.8.0, hadoop-hdfs
Upgrade commons-lang to fix incompatiblity of Java 9 and later.
Thrift:
Add THiveTable
Add read_by_column_def in TBrokerRangeDesc
---
be/src/exec/broker_scanner.cpp | 68 ++-
be/src/exec/orc_scanner.cpp | 13 +
docs/.vuepress/sidebar/en.js | 1 +
docs/.vuepress/sidebar/zh-CN.js | 1 +
docs/en/extending-doris/hive-of-doris.md | 117 +++++
.../sql-statements/Data Definition/CREATE TABLE.md | 2 +-
docs/zh-CN/extending-doris/hive-of-doris.md | 117 +++++
.../sql-statements/Data Definition/CREATE TABLE.md | 1 -
fe/fe-core/pom.xml | 35 +-
.../java/org/apache/doris/analysis/FromClause.java | 3 -
.../java/org/apache/doris/catalog/Catalog.java | 8 +
.../doris/catalog/HiveMetaStoreClientHelper.java | 552 +++++++++++++++++++++
.../java/org/apache/doris/catalog/HiveTable.java | 14 +-
.../org/apache/doris/load/BrokerFileGroup.java | 18 +
.../org/apache/doris/planner/BrokerScanNode.java | 108 ++--
.../org/apache/doris/planner/HiveScanNode.java | 210 ++++++++
.../apache/doris/planner/SingleNodePlanner.java | 6 +-
fe/pom.xml | 84 +++-
gensrc/thrift/Descriptors.thrift | 7 +
gensrc/thrift/PlanNodes.thrift | 2 +
20 files changed, 1278 insertions(+), 89 deletions(-)
diff --git a/be/src/exec/broker_scanner.cpp b/be/src/exec/broker_scanner.cpp
index a260035..e99bf7e 100644
--- a/be/src/exec/broker_scanner.cpp
+++ b/be/src/exec/broker_scanner.cpp
@@ -478,35 +478,51 @@ bool BrokerScanner::line_to_src_tuple(const Slice& line) {
// range of current file
const TBrokerRangeDesc& range = _ranges.at(_next_range - 1);
+ bool read_by_column_def = false;
+ if (range.__isset.read_by_column_def) {
+ read_by_column_def = range.read_by_column_def;
+ }
const std::vector<std::string>& columns_from_path = range.columns_from_path;
- if (_split_values.size() + columns_from_path.size() < _src_slot_descs.size()) {
- std::stringstream error_msg;
- error_msg << "actual column number is less than schema column number. "
- << "actual number: " << _split_values.size() << " column separator: ["
- << _value_separator << "], "
- << "line delimiter: [" << _line_delimiter << "], "
- << "schema number: " << _src_slot_descs.size() << "; ";
- if (_file_format_type == TFileFormatType::FORMAT_PROTO) {
- _state->append_error_msg_to_file("", error_msg.str());
- } else {
- _state->append_error_msg_to_file(std::string(line.data, line.size), error_msg.str());
+ // read data by column defination, resize _split_values to _src_solt_size
+ if (read_by_column_def) {
+ // fill solts by NULL
+ while (_split_values.size() + columns_from_path.size() < _src_slot_descs.size()) {
+ _split_values.emplace_back(_split_values.back().get_data(), 0);
}
- _counter->num_rows_filtered++;
- return false;
- } else if (_split_values.size() + columns_from_path.size() > _src_slot_descs.size()) {
- std::stringstream error_msg;
- error_msg << "actual column number is more than schema column number. "
- << "actual number: " << _split_values.size() << " column separator: ["
- << _value_separator << "], "
- << "line delimiter: [" << _line_delimiter << "], "
- << "schema number: " << _src_slot_descs.size() << "; ";
- if (_file_format_type == TFileFormatType::FORMAT_PROTO) {
- _state->append_error_msg_to_file("", error_msg.str());
- } else {
- _state->append_error_msg_to_file(std::string(line.data, line.size), error_msg.str());
+ // remove redundant slots
+ while (_split_values.size() + columns_from_path.size() > _src_slot_descs.size()) {
+ _split_values.pop_back();
+ }
+ } else {
+ if (_split_values.size() + columns_from_path.size() < _src_slot_descs.size()) {
+ std::stringstream error_msg;
+ error_msg << "actual column number is less than schema column number. "
+ << "actual number: " << _split_values.size() << " column separator: ["
+ << _value_separator << "], "
+ << "line delimiter: [" << _line_delimiter << "], "
+ << "schema number: " << _src_slot_descs.size() << "; ";
+ if (_file_format_type == TFileFormatType::FORMAT_PROTO) {
+ _state->append_error_msg_to_file("", error_msg.str());
+ } else {
+ _state->append_error_msg_to_file(std::string(line.data, line.size), error_msg.str());
+ }
+ _counter->num_rows_filtered++;
+ return false;
+ } else if (_split_values.size() + columns_from_path.size() > _src_slot_descs.size()) {
+ std::stringstream error_msg;
+ error_msg << "actual column number is more than schema column number. "
+ << "actual number: " << _split_values.size() << " column separator: ["
+ << _value_separator << "], "
+ << "line delimiter: [" << _line_delimiter << "], "
+ << "schema number: " << _src_slot_descs.size() << "; ";
+ if (_file_format_type == TFileFormatType::FORMAT_PROTO) {
+ _state->append_error_msg_to_file("", error_msg.str());
+ } else {
+ _state->append_error_msg_to_file(std::string(line.data, line.size), error_msg.str());
+ }
+ _counter->num_rows_filtered++;
+ return false;
}
- _counter->num_rows_filtered++;
- return false;
}
for (int i = 0; i < _split_values.size(); ++i) {
diff --git a/be/src/exec/orc_scanner.cpp b/be/src/exec/orc_scanner.cpp
index fafb9e2..8732b7a 100644
--- a/be/src/exec/orc_scanner.cpp
+++ b/be/src/exec/orc_scanner.cpp
@@ -29,6 +29,10 @@
#include "runtime/runtime_state.h"
#include "runtime/tuple.h"
+#if defined(__x86_64__)
+ #include "exec/hdfs_file_reader.h"
+#endif
+
// orc include file didn't expose orc::TimezoneError
// we have to declare it by hand, following is the source code in orc link
// https://github.com/apache/orc/blob/84353fbfc447b06e0924024a8e03c1aaebd3e7a5/c%2B%2B/src/Timezone.hh#L104-L109
@@ -406,6 +410,15 @@ Status ORCScanner::open_next_reader() {
new S3Reader(_params.properties, range.path, range.start_offset)));
break;
}
+ case TFileType::FILE_HDFS: {
+#if defined(__x86_64__)
+ file_reader.reset(new HdfsFileReader(
+ range.hdfs_params, range.path, range.start_offset));
+ break;
+#else
+ return Status::InternalError("HdfsFileReader do not support on non x86 platform");
+#endif
+ }
default: {
std::stringstream ss;
ss << "Unknown file type, type=" << range.file_type;
diff --git a/docs/.vuepress/sidebar/en.js b/docs/.vuepress/sidebar/en.js
index 6692464..71f3853 100644
--- a/docs/.vuepress/sidebar/en.js
+++ b/docs/.vuepress/sidebar/en.js
@@ -232,6 +232,7 @@ module.exports = [
"doris-on-es",
"logstash",
"odbc-of-doris",
+ "hive-of-doris",
"plugin-development-manual",
"spark-doris-connector",
"flink-doris-connector",
diff --git a/docs/.vuepress/sidebar/zh-CN.js b/docs/.vuepress/sidebar/zh-CN.js
index 4ae4f06..8f065e9 100644
--- a/docs/.vuepress/sidebar/zh-CN.js
+++ b/docs/.vuepress/sidebar/zh-CN.js
@@ -233,6 +233,7 @@ module.exports = [
"doris-on-es",
"logstash",
"odbc-of-doris",
+ "hive-of-doris",
"plugin-development-manual",
"spark-doris-connector",
"flink-doris-connector",
diff --git a/docs/en/extending-doris/hive-of-doris.md b/docs/en/extending-doris/hive-of-doris.md
new file mode 100644
index 0000000..ed730dc
--- /dev/null
+++ b/docs/en/extending-doris/hive-of-doris.md
@@ -0,0 +1,117 @@
+---
+{
+ "title": "Hive of Doris",
+ "language": "en"
+}
+---
+
+<!--
+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.
+-->
+
+# Hive External Table of Doris
+
+Hive External Table of Doris provides Doris with direct access to Hive external tables, which eliminates the need for cumbersome data import and solves the problem of analyzing Hive tables with the help of Doris' OLAP capabilities:
+
+ 1. support for Hive data sources to access Doris
+ 2. Support joint queries between Doris and Hive data sources to perform more complex analysis operations
+
+This document introduces how to use this feature and the considerations.
+
+## Glossary
+
+### Noun in Doris
+
+* FE: Frontend, the front-end node of Doris, responsible for metadata management and request access.
+* BE: Backend, the backend node of Doris, responsible for query execution and data storage
+
+## How To Use
+
+### Create Hive External Table
+
+```sql
+-- Syntax
+CREATE [EXTERNAL] TABLE table_name (
+ col_name col_type [NULL | NOT NULL] [COMMENT "comment"]
+) ENGINE=HIVE
+[COMMENT "comment"] )
+PROPERTIES (
+ 'property_name'='property_value',
+ ...
+);
+
+-- Example: Create the hive_table table under hive_db in a Hive cluster
+CREATE TABLE `t_hive` (
+ `k1` int NOT NULL COMMENT "",
+ `k2` char(10) NOT NULL COMMENT "",
+ `k3` datetime NOT NULL COMMENT "",
+ `k5` varchar(20) NOT NULL COMMENT "",
+ `k6` double NOT NULL COMMENT ""
+) ENGINE=HIVE
+COMMENT "HIVE"
+PROPERTIES (
+'hive.metastore.uris' = 'thrift://192.168.0.1:9083',
+'database' = 'hive_db',
+'table' = 'hive_table'
+);
+```
+
+#### Parameter Description
+
+- External Table Columns
+ - Column names should correspond to the Hive table
+ - The order of the columns should be the same as the Hive table
+ - Must contain all the columns in the Hive table
+ - Hive table partition columns do not need to be specified, they can be defined as normal columns.
+- ENGINE should be specified as HIVE
+- PROPERTIES attribute.
+ - `hive.metastore.uris`: Hive Metastore service address
+ - `database`: the name of the database to which Hive is mounted
+ - `table`: the name of the table to which Hive is mounted
+
+## Data Type Matching
+
+The supported Hive column types correspond to Doris in the following table.
+
+| Hive | Doris | 描述 |
+| :------: | :----: | :-------------------------------: |
+| BOOLEAN | BOOLEAN | |
+| CHAR | CHAR | Only UTF8 encoding is supported |
+| VARCHAR | VARCHAR | Only UTF8 encoding is supported |
+| TINYINT | TINYINT | |
+| SMALLINT | SMALLINT | |
+| INT | INT | |
+| BIGINT | BIGINT | |
+| FLOAT | FLOAT | |
+| DOUBLE | DOUBLE | |
+| DECIMAL | DECIMAL | |
+| DATE | DATE | |
+| DATETIME | TIMESTAMP | Timestamp to Datetime will lose precision |
+
+**Note:**
+- Hive table Schema changes **are not automatically synchronized** and require rebuilding the Hive external table in Doris.
+- The current Hive storage format only supports Text, Parquet and ORC types
+- The Hive version currently supported by default is 2.3.7, which has not been tested in other versions. More versions will be supported in the future.
+
+### Query Usage
+
+After you finish building the Hive external table in Doris, it is no different from a normal Doris OLAP table except that you cannot use the data model in Doris (rollup, preaggregation, materialized view, etc.)
+
+```sql
+select * from t_hive where k1 > 1000 and k3 = 'term' or k4 like '%doris';
+```
diff --git a/docs/en/sql-reference/sql-statements/Data Definition/CREATE TABLE.md b/docs/en/sql-reference/sql-statements/Data Definition/CREATE TABLE.md
index fe7fbbd..329b053 100644
--- a/docs/en/sql-reference/sql-statements/Data Definition/CREATE TABLE.md
+++ b/docs/en/sql-reference/sql-statements/Data Definition/CREATE TABLE.md
@@ -156,7 +156,7 @@ Syntax:
)
```
"database" is the name of the database corresponding to the hive table, "table" is the name of the hive table, and "hive.metastore.uris" is the hive metastore service address.
- Notice: At present, hive external tables are only used for Spark Load and query is not supported.
+
4. key_desc
Syntax:
key_type(k1[,k2 ...])
diff --git a/docs/zh-CN/extending-doris/hive-of-doris.md b/docs/zh-CN/extending-doris/hive-of-doris.md
new file mode 100644
index 0000000..759e689
--- /dev/null
+++ b/docs/zh-CN/extending-doris/hive-of-doris.md
@@ -0,0 +1,117 @@
+---
+{
+ "title": "Hive of Doris",
+ "language": "zh-CN"
+}
+---
+
+<!--
+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.
+-->
+
+# Hive External Table of Doris
+
+Hive External Table of Doris 提供了 Doris 直接访问 Hive 外部表的能力，外部表省去了繁琐的数据导入工作，并借助 Doris 本身的 OLAP 的能力来解决 Hive 表的数据分析问题：
+
+ 1. 支持 Hive 数据源接入Doris
+ 2. 支持 Doris 与 Hive 数据源中的表联合查询，进行更加复杂的分析操作
+
+本文档主要介绍该功能的使用方式和注意事项等。
+
+## 名词解释
+
+### Doris 相关
+
+* FE：Frontend，Doris 的前端节点,负责元数据管理和请求接入
+* BE：Backend，Doris 的后端节点,负责查询执行和数据存储
+
+## 使用方法
+
+### Doris 中创建 Hive 的外表
+
+```sql
+-- 语法
+CREATE [EXTERNAL] TABLE table_name (
+ col_name col_type [NULL | NOT NULL] [COMMENT "comment"]
+) ENGINE=HIVE
+[COMMENT "comment"]
+PROPERTIES (
+ 'property_name'='property_value',
+ ...
+);
+
+-- 例子：创建 Hive 集群中 hive_db 下的 hive_table 表
+CREATE TABLE `t_hive` (
+ `k1` int NOT NULL COMMENT "",
+ `k2` char(10) NOT NULL COMMENT "",
+ `k3` datetime NOT NULL COMMENT "",
+ `k5` varchar(20) NOT NULL COMMENT "",
+ `k6` double NOT NULL COMMENT ""
+) ENGINE=HIVE
+COMMENT "HIVE"
+PROPERTIES (
+'hive.metastore.uris' = 'thrift://192.168.0.1:9083',
+'database' = 'hive_db',
+'table' = 'hive_table'
+);
+```
+
+#### 参数说明：
+
+- 外表列
+ - 列名要于 Hive 表一一对应
+ - 列的顺序需要与 Hive 表一致
+ - 必须包含 Hive 表中的全部列
+ - Hive 表分区列无需指定，与普通列一样定义即可。
+- ENGINE 需要指定为 HIVE
+- PROPERTIES 属性：
+ - `hive.metastore.uris`：Hive Metastore 服务地址
+ - `database`：挂载 Hive 对应的数据库名
+ - `table`：挂载 Hive 对应的表名
+
+## 类型匹配
+
+支持的 Hive 列类型与 Doris 对应关系如下表：
+
+| Hive | Doris | 描述 |
+| :------: | :----: | :-------------------------------: |
+| BOOLEAN | BOOLEAN | |
+| CHAR | CHAR | 当前仅支持UTF8编码 |
+| VARCHAR | VARCHAR | 当前仅支持UTF8编码 |
+| TINYINT | TINYINT | |
+| SMALLINT | SMALLINT | |
+| INT | INT | |
+| BIGINT | BIGINT | |
+| FLOAT | FLOAT | |
+| DOUBLE | DOUBLE | |
+| DECIMAL | DECIMAL | |
+| DATE | DATE | |
+| DATETIME | TIMESTAMP | Timestamp 转成 Datetime 会损失精度 |
+
+**注意：**
+- Hive 表 Schema 变更**不会自动同步**，需要在 Doris 中重建 Hive 外表。
+- 当前 Hive 的存储格式仅支持 Text，Parquet 和 ORC 类型
+- 当前默认支持的 Hive 版本为 2.3.7，未在其他版本进行测试。后续后支持更多版本。
+
+### 查询用法
+
+完成在 Doris 中建立 Hive 外表后，除了无法使用 Doris 中的数据模型(rollup、预聚合、物化视图等)外，与普通的 Doris OLAP 表并无区别
+
+```sql
+select * from t_hive where k1 > 1000 and k3 ='term' or k4 like '%doris';
+```
diff --git a/docs/zh-CN/sql-reference/sql-statements/Data Definition/CREATE TABLE.md b/docs/zh-CN/sql-reference/sql-statements/Data Definition/CREATE TABLE.md
index 07b8f9e..bcbd152 100644
--- a/docs/zh-CN/sql-reference/sql-statements/Data Definition/CREATE TABLE.md
+++ b/docs/zh-CN/sql-reference/sql-statements/Data Definition/CREATE TABLE.md
@@ -165,7 +165,6 @@ under the License.
```
其中 database 是 hive 表对应的库名字，table 是 hive 表的名字，hive.metastore.uris 是 hive metastore 服务地址。
- 注意：目前hive外部表仅用于Spark Load使用，不支持查询。
4. key_desc
语法：
diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml
index e40f29a..51f50ab 100644
--- a/fe/fe-core/pom.xml
+++ b/fe/fe-core/pom.xml
@@ -514,21 +514,6 @@ under the License.
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-aws</artifactId>
- <version>2.8.0</version>
- <exclusions>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- <exclusion>
- <artifactId>servlet-api</artifactId>
- <groupId>javax.servlet</groupId>
- </exclusion>
- </exclusions>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-web -->
@@ -650,6 +635,26 @@ under the License.
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-metastore</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-exec</artifactId>
+ <classifier>core</classifier>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-hdfs -->
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/FromClause.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/FromClause.java
index c3f953c..fba2085 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/FromClause.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/FromClause.java
@@ -82,9 +82,6 @@ public class FromClause implements ParseNode, Iterable<TableRef> {
Database db = analyzer.getCatalog().getDbOrAnalysisException(dbName);
String tblName = tableName.getTbl();
Table table = db.getTableOrAnalysisException(tblName);
- if (table.getType() == Table.TableType.HIVE) {
- throw new AnalysisException("Query from hive table is not supported, table: " + tblName);
- }
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java
index 5c42ea3..ec315ce 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java
@@ -258,6 +258,7 @@ import com.sleepycat.je.rep.NetworkRestore;
import com.sleepycat.je.rep.NetworkRestoreConfig;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.codehaus.jackson.map.ObjectMapper;
@@ -4015,6 +4016,13 @@ public class Catalog {
long tableId = getNextId();
HiveTable hiveTable = new HiveTable(tableId, tableName, columns, stmt.getProperties());
hiveTable.setComment(stmt.getComment());
+ // check hive table if exists in hive database
+ HiveMetaStoreClient hiveMetaStoreClient =
+ HiveMetaStoreClientHelper.getClient(hiveTable.getHiveProperties().get(HiveTable.HIVE_METASTORE_URIS));
+ if (!HiveMetaStoreClientHelper.tableExists(hiveMetaStoreClient, hiveTable.getHiveDb(), hiveTable.getHiveTable())) {
+ throw new DdlException("Table is not exists in hive: " + hiveTable.getHiveDbTable());
+ }
+ // check hive table if exists in doris database
if (!db.createTableWithLock(hiveTable, false, stmt.isSetIfNotExists()).first) {
ErrorReport.reportDdlException(ErrorCode.ERR_CANT_CREATE_TABLE, tableName, "table already exist");
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/HiveMetaStoreClientHelper.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/HiveMetaStoreClientHelper.java
new file mode 100644
index 0000000..5dcfd12
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/HiveMetaStoreClientHelper.java
@@ -0,0 +1,552 @@
+// 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.doris.catalog;
+
+import org.apache.doris.analysis.BinaryPredicate;
+import org.apache.doris.analysis.BoolLiteral;
+import org.apache.doris.analysis.CastExpr;
+import org.apache.doris.analysis.CompoundPredicate;
+import org.apache.doris.analysis.DateLiteral;
+import org.apache.doris.analysis.DecimalLiteral;
+import org.apache.doris.analysis.Expr;
+import org.apache.doris.analysis.FloatLiteral;
+import org.apache.doris.analysis.IntLiteral;
+import org.apache.doris.analysis.LiteralExpr;
+import org.apache.doris.analysis.NullLiteral;
+import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.analysis.StringLiteral;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TExprOpcode;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
+import org.apache.hadoop.hive.ql.exec.SerializationUtilities;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.thrift.TException;
+
+import com.google.common.base.Strings;
+
+import java.io.IOException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Stack;
+
+/**
+ * Helper class for HiveMetaStoreClient
+ */
+public class HiveMetaStoreClientHelper {
+ private static final Logger LOG = LogManager.getLogger(HiveMetaStoreClientHelper.class);
+
+ public enum HiveFileFormat {
+ TEXT_FILE(0, "text"),
+ PARQUET(1, "parquet"),
+ ORC(2, "orc");
+
+ private int index;
+ private String desc;
+
+ HiveFileFormat(int index, String desc) {
+ this.index = index;
+ this.desc = desc;
+ }
+
+ public int getIndex() {
+ return index;
+ }
+
+ public String getDesc() {
+ return desc;
+ }
+
+ /**
+ * convert Hive table inputFormat to file format
+ * @param input inputFormat of Hive file
+ * @return
+ * @throws DdlException
+ */
+ public static String getFormat(String input) throws DdlException {
+ String formatDesc = "";
+ for (HiveFileFormat format : HiveFileFormat.values()) {
+ String lowerCaseInput = input.toLowerCase();
+ if (lowerCaseInput.contains(format.getDesc())) {
+ formatDesc = format.getDesc();
+ break;
+ }
+ }
+ if (Strings.isNullOrEmpty(formatDesc)) {
+ LOG.warn("Not supported Hive file format [{}].", input);
+ throw new DdlException("Not supported Hive file format " + input);
+ }
+ return formatDesc;
+ }
+ }
+
+ public static HiveMetaStoreClient getClient(String metaStoreUris) throws DdlException {
+ HiveConf hiveConf = new HiveConf();
+ hiveConf.setVar(HiveConf.ConfVars.METASTOREURIS, metaStoreUris);
+ HiveMetaStoreClient hivemetastoreclient = null;
+ try {
+ hivemetastoreclient = new HiveMetaStoreClient(hiveConf);
+ } catch (MetaException e) {
+ LOG.warn("Create HiveMetaStoreClient failed: {}", e.getMessage());
+ throw new DdlException("Create HiveMetaStoreClient failed");
+ }
+ return hivemetastoreclient;
+ }
+
+ /**
+ * Check to see if the specified table exists in the specified database.
+ * @param client HiveMetaStoreClient
+ * @param dbName the specified database name
+ * @param tblName the specified table name
+ * @return TRUE if specified.tableName exists, FALSE otherwise.
+ * @throws DdlException
+ */
+ public static boolean tableExists(HiveMetaStoreClient client, String dbName, String tblName) throws DdlException {
+ try {
+ return client.tableExists(dbName, tblName);
+ } catch (TException e) {
+ LOG.warn("Hive metastore thrift exception: {}", e.getMessage());
+ throw new DdlException("Connect hive metastore failed.");
+ } finally {
+ dropClient(client);
+ }
+ }
+
+ /**
+ * close connection to meta store
+ */
+ public static void dropClient(HiveMetaStoreClient client) {
+ client.close();
+ }
+
+ /**
+ * Get data files of partitions in hive table, filter by partition predicate
+ * @param hiveTable
+ * @param hivePartitionPredicate
+ * @param fileStatuses
+ * @param remoteHiveTbl
+ * @return
+ * @throws DdlException
+ */
+ public static String getHiveDataFiles(HiveTable hiveTable, ExprNodeGenericFuncDesc hivePartitionPredicate,
+ List<TBrokerFileStatus> fileStatuses, Table remoteHiveTbl) throws DdlException {
+ HiveMetaStoreClient client = getClient(hiveTable.getHiveProperties().get(HiveTable.HIVE_METASTORE_URIS));
+
+ List<RemoteIterator<LocatedFileStatus>> remoteIterators;
+ if (remoteHiveTbl.getPartitionKeys().size() > 0) {
+ // hive partitioned table, get file iterator from table partition sd info
+ List<Partition> hivePartitions = new ArrayList<>();
+ try {
+ client.listPartitionsByExpr(hiveTable.getHiveDb(), hiveTable.getHiveTable(),
+ SerializationUtilities.serializeExpressionToKryo(hivePartitionPredicate), null, (short) -1, hivePartitions);
+ } catch (TException e) {
+ LOG.warn("Hive metastore thrift exception: {}", e.getMessage());
+ throw new DdlException("Connect hive metastore failed.");
+ } finally {
+ client.close();
+ }
+ remoteIterators = getRemoteIterator(hivePartitions);
+ } else {
+ // hive non-partitioned table, get file iterator from table sd info
+ remoteIterators = getRemoteIterator(remoteHiveTbl);
+ }
+
+ String hdfsUrl = "";
+ for (RemoteIterator<LocatedFileStatus> iterator : remoteIterators) {
+ try {
+ while (iterator.hasNext()) {
+ LocatedFileStatus fileStatus = iterator.next();
+ TBrokerFileStatus brokerFileStatus = new TBrokerFileStatus();
+ brokerFileStatus.setIsDir(fileStatus.isDirectory());
+ brokerFileStatus.setIsSplitable(true);
+ brokerFileStatus.setSize(fileStatus.getLen());
+ // path = "/path/to/partition/file_name"
+ // eg: /home/work/dev/hive/apache-hive-2.3.7-bin/data/warehouse/dae.db/customer/state=CA/city=SanJose/000000_0
+ String path = fileStatus.getPath().toUri().getPath();
+ brokerFileStatus.setPath(path);
+ fileStatuses.add(brokerFileStatus);
+ if (StringUtils.isEmpty(hdfsUrl)) {
+ // hdfs://host:port/path/to/partition/file_name
+ String fullUri = fileStatus.getPath().toString();
+ // hdfs://host:port
+ hdfsUrl = fullUri.replace(path, "");
+ }
+ }
+ } catch (IOException e) {
+ LOG.warn("List HDFS file IOException: {}", e.getMessage());
+ throw new DdlException("List HDFS file failed.");
+ }
+ }
+
+ return hdfsUrl;
+ }
+
+ private static List<RemoteIterator<LocatedFileStatus>> getRemoteIterator(List<Partition> partitions) throws DdlException {
+ List<RemoteIterator<LocatedFileStatus>> iterators = new ArrayList<>();
+ Configuration configuration = new Configuration(false);
+ for (Partition p : partitions) {
+ String location = p.getSd().getLocation();
+ org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(location);
+ try {
+ FileSystem fileSystem = path.getFileSystem(configuration);
+ iterators.add(fileSystem.listLocatedStatus(path));
+ } catch (IOException e) {
+ LOG.warn("Get HDFS file remote iterator failed. {}", e.getMessage());
+ throw new DdlException("Get HDFS file remote iterator failed.");
+ }
+ }
+ return iterators;
+ }
+
+ private static List<RemoteIterator<LocatedFileStatus>> getRemoteIterator(Table table) throws DdlException {
+ List<RemoteIterator<LocatedFileStatus>> iterators = new ArrayList<>();
+ Configuration configuration = new Configuration(false);
+ String location = table.getSd().getLocation();
+ org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(location);
+ try {
+ FileSystem fileSystem = path.getFileSystem(configuration);
+ iterators.add(fileSystem.listLocatedStatus(path));
+ } catch (IOException e) {
+ LOG.warn("Get HDFS file remote iterator failed. {}" + e.getMessage());
+ throw new DdlException("Get HDFS file remote iterator failed.");
+ }
+ return iterators;
+ }
+
+ public static List<String> getPartitionNames(HiveTable hiveTable) throws DdlException {
+ HiveMetaStoreClient client = getClient(hiveTable.getHiveProperties().get(HiveTable.HIVE_METASTORE_URIS));
+ List<String> partitionNames = new ArrayList<>();
+ try {
+ partitionNames = client.listPartitionNames(hiveTable.getHiveDb(), hiveTable.getHiveTable(), (short) -1);
+ } catch (TException e) {
+ LOG.warn("Hive metastore thrift exception: {}", e.getMessage());
+ throw new DdlException("Connect hive metastore failed.");
+ }
+
+ return partitionNames;
+ }
+
+ public static Table getTable(HiveTable hiveTable) throws DdlException {
+ HiveMetaStoreClient client = getClient(hiveTable.getHiveProperties().get(HiveTable.HIVE_METASTORE_URIS));
+ Table table;
+ try {
+ table = client.getTable(hiveTable.getHiveDb(), hiveTable.getHiveTable());
+ } catch (TException e) {
+ LOG.warn("Hive metastore thrift exception: {}", e.getMessage());
+ throw new DdlException("Connect hive metastore failed.");
+ }
+ return table;
+ }
+
+ /**
+ * Convert Doris expr to Hive expr, only for partition column
+ * @param dorisExpr
+ * @param partitions
+ * @param tblName
+ * @return
+ * @throws DdlException
+ * @throws SemanticException
+ */
+ public static ExprNodeGenericFuncDesc convertToHivePartitionExpr(Expr dorisExpr, List<String> partitions, String tblName) throws DdlException {
+ if (dorisExpr == null) {
+ return null;
+ }
+
+ if (dorisExpr instanceof CompoundPredicate) {
+ CompoundPredicate compoundPredicate = (CompoundPredicate) dorisExpr;
+ switch (compoundPredicate.getOp()) {
+ case AND: {
+ ExprNodeGenericFuncDesc left = convertToHivePartitionExpr(compoundPredicate.getChild(0), partitions, tblName);
+ ExprNodeGenericFuncDesc right = convertToHivePartitionExpr(compoundPredicate.getChild(0), partitions, tblName);
+ if (left != null && right != null) {
+ List<ExprNodeDesc> andArgs = new ArrayList<>();
+ andArgs.add(left);
+ andArgs.add(right);
+ return getCompoundExpr(andArgs, "and");
+ } else if (left != null && right == null) {
+ return left;
+ } else if (left == null && right != null) {
+ return right;
+ }
+ return null;
+ }
+ case OR: {
+ ExprNodeGenericFuncDesc left = convertToHivePartitionExpr(compoundPredicate.getChild(0), partitions, tblName);
+ ExprNodeGenericFuncDesc right = convertToHivePartitionExpr(compoundPredicate.getChild(0), partitions, tblName);
+ if (left != null && right != null) {
+ List<ExprNodeDesc> orArgs = new ArrayList<>();
+ orArgs.add(left);
+ orArgs.add(right);
+ return getCompoundExpr(orArgs, "or");
+ } else if (left != null && right == null) {
+ return left;
+ } else if (left == null && right != null) {
+ return right;
+ }
+ return null;
+ }
+ default:
+ return null;
+ }
+ }
+
+ TExprOpcode opcode = dorisExpr.getOpcode();
+ switch (opcode) {
+ case EQ:
+ case NE:
+ case GE:
+ case GT:
+ case LE:
+ case LT:
+ case EQ_FOR_NULL:
+ BinaryPredicate eq = (BinaryPredicate) dorisExpr;
+ SlotRef slotRef = convertDorisExprToSlotRef(eq.getChild(0));
+ LiteralExpr literalExpr = null;
+ if (slotRef == null && eq.getChild(0).isLiteral()) {
+ literalExpr = (LiteralExpr) eq.getChild(0);
+ slotRef = convertDorisExprToSlotRef(eq.getChild(1));
+ } else if (eq.getChild(1).isLiteral()) {
+ literalExpr = (LiteralExpr) eq.getChild(1);
+ }
+ if (slotRef == null || literalExpr == null) {
+ return null;
+ }
+ String colName = slotRef.getColumnName();
+ // check whether colName is partition column or not
+ if (!partitions.contains(colName)) {
+ return null;
+ }
+ PrimitiveType dorisPrimitiveType = slotRef.getType().getPrimitiveType();
+ PrimitiveTypeInfo hivePrimitiveType = convertToHiveColType(dorisPrimitiveType);
+ Object value = extractDorisLiteral(literalExpr);
+ ExprBuilder exprBuilder = new ExprBuilder(tblName);
+ if (value == null) {
+ if (opcode == TExprOpcode.EQ_FOR_NULL && literalExpr instanceof NullLiteral) {
+ return exprBuilder.col(hivePrimitiveType, colName)
+ .val(hivePrimitiveType, "NULL")
+ .pred("=", 2).build();
+ } else {
+ return null;
+ }
+ }
+ switch (opcode) {
+ case EQ:
+ case EQ_FOR_NULL:
+ return exprBuilder.col(hivePrimitiveType, colName)
+ .val(hivePrimitiveType, value)
+ .pred("=", 2).build();
+ case NE:
+ return exprBuilder.col(hivePrimitiveType, colName)
+ .val(hivePrimitiveType, value)
+ .pred("!=", 2).build();
+ case GE:
+ return exprBuilder.col(hivePrimitiveType, colName)
+ .val(hivePrimitiveType, value)
+ .pred(">=", 2).build();
+ case GT:
+ return exprBuilder.col(hivePrimitiveType, colName)
+ .val(hivePrimitiveType, value)
+ .pred(">", 2).build();
+ case LE:
+ return exprBuilder.col(hivePrimitiveType, colName)
+ .val(hivePrimitiveType, value)
+ .pred("<=", 2).build();
+ case LT:
+ return exprBuilder.col(hivePrimitiveType, colName)
+ .val(hivePrimitiveType, value)
+ .pred("<", 2).build();
+ default:
+ return null;
+ }
+ default:
+ return null;
+ }
+
+ }
+
+ public static ExprNodeGenericFuncDesc getCompoundExpr(List<ExprNodeDesc> args, String op) throws DdlException {
+ ExprNodeGenericFuncDesc compoundExpr;
+ try {
+ compoundExpr = ExprNodeGenericFuncDesc.newInstance(
+ FunctionRegistry.getFunctionInfo(op).getGenericUDF(), args);
+ } catch (SemanticException e) {
+ LOG.warn("Convert to Hive expr failed: {}", e.getMessage());
+ throw new DdlException("Convert to Hive expr failed.");
+ }
+ return compoundExpr;
+ }
+
+ private static SlotRef convertDorisExprToSlotRef(Expr expr) {
+ SlotRef slotRef = null;
+ if (expr instanceof SlotRef) {
+ slotRef = (SlotRef) expr;
+ } else if (expr instanceof CastExpr) {
+ if (expr.getChild(0) instanceof SlotRef) {
+ slotRef = (SlotRef) expr.getChild(0);
+ }
+ }
+ return slotRef;
+ }
+
+ private static Object extractDorisLiteral(Expr expr) {
+ if (!expr.isLiteral()) {
+ return null;
+ }
+ if (expr instanceof BoolLiteral) {
+ BoolLiteral boolLiteral = (BoolLiteral) expr;
+ return boolLiteral.getValue();
+ } else if (expr instanceof DateLiteral) {
+ DateLiteral dateLiteral = (DateLiteral) expr;
+ SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmmss");
+ StringBuilder sb = new StringBuilder();
+ sb.append(dateLiteral.getYear())
+ .append(dateLiteral.getMonth())
+ .append(dateLiteral.getDay())
+ .append(dateLiteral.getHour())
+ .append(dateLiteral.getMinute())
+ .append(dateLiteral.getSecond());
+ Date date;
+ try {
+ date = formatter.parse(sb.toString());
+ } catch (ParseException e) {
+ return null;
+ }
+ return date.getTime();
+ } else if (expr instanceof DecimalLiteral) {
+ DecimalLiteral decimalLiteral = (DecimalLiteral) expr;
+ return decimalLiteral.getValue();
+ } else if (expr instanceof FloatLiteral) {
+ FloatLiteral floatLiteral = (FloatLiteral) expr;
+ return floatLiteral.getValue();
+ } else if (expr instanceof IntLiteral) {
+ IntLiteral intLiteral = (IntLiteral) expr;
+ return intLiteral.getValue();
+ } else if (expr instanceof StringLiteral) {
+ StringLiteral stringLiteral = (StringLiteral) expr;
+ return stringLiteral.getStringValue();
+ }
+ return null;
+ }
+ /**
+ * Convert from Doris column type to Hive column type
+ * @param dorisType
+ * @return hive primitive type info
+ * @throws DdlException
+ */
+ private static PrimitiveTypeInfo convertToHiveColType(PrimitiveType dorisType) throws DdlException {
+ switch (dorisType) {
+ case BOOLEAN:
+ return TypeInfoFactory.booleanTypeInfo;
+ case TINYINT:
+ return TypeInfoFactory.byteTypeInfo;
+ case SMALLINT:
+ return TypeInfoFactory.shortTypeInfo;
+ case INT:
+ return TypeInfoFactory.intTypeInfo;
+ case BIGINT:
+ return TypeInfoFactory.longTypeInfo;
+ case FLOAT:
+ return TypeInfoFactory.floatTypeInfo;
+ case DOUBLE:
+ return TypeInfoFactory.doubleTypeInfo;
+ case DECIMALV2:
+ return TypeInfoFactory.decimalTypeInfo;
+ case DATE:
+ return TypeInfoFactory.dateTypeInfo;
+ case DATETIME:
+ return TypeInfoFactory.timestampTypeInfo;
+ case CHAR:
+ return TypeInfoFactory.charTypeInfo;
+ case VARCHAR:
+ return TypeInfoFactory.varcharTypeInfo;
+ default:
+ throw new DdlException("Unsupported column type: " + dorisType);
+ }
+ }
+
+ /**
+ * Helper class for building a Hive expression.
+ */
+ public static class ExprBuilder {
+ private final String tblName;
+ private final Stack<ExprNodeDesc> stack = new Stack<>();
+
+ public ExprBuilder(String tblName) {
+ this.tblName = tblName;
+ }
+
+ public ExprNodeGenericFuncDesc build() throws DdlException {
+ if (stack.size() != 1) {
+ throw new DdlException("Build Hive expression Failed: " + stack.size());
+ }
+ return (ExprNodeGenericFuncDesc)stack.pop();
+ }
+
+ public ExprBuilder pred(String name, int args) throws DdlException {
+ return fn(name, TypeInfoFactory.booleanTypeInfo, args);
+ }
+
+ private ExprBuilder fn(String name, TypeInfo ti, int args) throws DdlException {
+ List<ExprNodeDesc> children = new ArrayList<>();
+ for (int i = 0; i < args; ++i) {
+ children.add(stack.pop());
+ }
+ try {
+ stack.push(new ExprNodeGenericFuncDesc(ti,
+ FunctionRegistry.getFunctionInfo(name).getGenericUDF(), children));
+ } catch (SemanticException e) {
+ LOG.warn("Build Hive expression failed: semantic analyze exception: {}", e.getMessage());
+ throw new DdlException("Build Hive expression Failed");
+ }
+ return this;
+ }
+
+ public ExprBuilder col(TypeInfo ti, String col) {
+ stack.push(new ExprNodeColumnDesc(ti, col, tblName, true));
+ return this;
+ }
+
+ public ExprBuilder val(TypeInfo ti, Object val) {
+ stack.push(new ExprNodeConstantDesc(ti, val));
+ return this;
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/HiveTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/HiveTable.java
index 919ebfc..236d266 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/HiveTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/HiveTable.java
@@ -19,6 +19,9 @@ package org.apache.doris.catalog;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.io.Text;
+import org.apache.doris.thrift.THiveTable;
+import org.apache.doris.thrift.TTableDescriptor;
+import org.apache.doris.thrift.TTableType;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
@@ -38,7 +41,7 @@ public class HiveTable extends Table {
private static final String HIVE_DB = "database";
private static final String HIVE_TABLE = "table";
- private static final String HIVE_METASTORE_URIS = "hive.metastore.uris";
+ public static final String HIVE_METASTORE_URIS = "hive.metastore.uris";
private String hiveDb;
private String hiveTable;
@@ -127,4 +130,13 @@ public class HiveTable extends Table {
hiveProperties.put(key, val);
}
}
+
+ @Override
+ public TTableDescriptor toThrift() {
+ THiveTable tHiveTable = new THiveTable(getHiveDb(), getHiveTable(), getHiveProperties());
+ TTableDescriptor tTableDescriptor = new TTableDescriptor(getId(), TTableType.BROKER_TABLE,
+ fullSchema.size(), 0, getName(), "");
+ tTableDescriptor.setHiveTable(tHiveTable);
+ return tTableDescriptor;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java b/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java
index b19d312..95dbc12 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java
@@ -119,6 +119,24 @@ public class BrokerFileGroup implements Writable {
this.fileFormat = table.getFileFormat();
}
+ // Used for hive table, no need to parse
+ public BrokerFileGroup(HiveTable table,
+ String columnSeparator,
+ String lineDelimiter,
+ String filePath,
+ String fileFormat,
+ List<String> columnsFromPath,
+ List<ImportColumnDesc> columnExprList) throws AnalysisException {
+ this.tableId = table.getId();
+ this.valueSeparator = Separator.convertSeparator(columnSeparator);
+ this.lineDelimiter = Separator.convertSeparator(lineDelimiter);
+ this.isNegative = false;
+ this.filePaths = Lists.newArrayList(filePath);
+ this.fileFormat = fileFormat;
+ this.columnsFromPath = columnsFromPath;
+ this.columnExprList = columnExprList;
+ }
+
public BrokerFileGroup(DataDescription dataDescription) {
this.fileFieldNames = dataDescription.getFileFieldNames();
this.columnsFromPath = dataDescription.getColumnsFromPath();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java
index 269f570..1aa9a38 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java
@@ -47,6 +47,7 @@ import org.apache.doris.thrift.TBrokerScanRange;
import org.apache.doris.thrift.TBrokerScanRangeParams;
import org.apache.doris.thrift.TExplainLevel;
import org.apache.doris.thrift.TFileFormatType;
+import org.apache.doris.thrift.THdfsParams;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TScanRange;
import org.apache.doris.thrift.TScanRangeLocation;
@@ -69,7 +70,15 @@ import java.util.Map;
import java.util.Random;
import java.util.stream.Collectors;
-// Broker scan node
+/**
+ * Broker scan node
+ *
+ * Since https://github.com/apache/incubator-doris/pull/5686, Doris can read data from HDFS without broker by
+ * broker scan node.
+ * Broker scan node is more likely a file scan node for now.
+ * With this feature, we can extend BrokerScanNode to query external table which data is stored in HDFS, such as
+ * Hive and Iceberg, etc.
+ */
public class BrokerScanNode extends LoadScanNode {
private static final Logger LOG = LogManager.getLogger(BrokerScanNode.class);
private static final TBrokerFileStatusComparator T_BROKER_FILE_STATUS_COMPARATOR
@@ -99,15 +108,15 @@ public class BrokerScanNode extends LoadScanNode {
// Parameters need to process
private long loadJobId = -1; // -1 means this scan node is not for a load job
private long txnId = -1;
- private Table targetTable;
- private BrokerDesc brokerDesc;
- private List<BrokerFileGroup> fileGroups;
+ protected Table targetTable;
+ protected BrokerDesc brokerDesc;
+ protected List<BrokerFileGroup> fileGroups;
private boolean strictMode = false;
private int loadParallelism = 1;
- private List<List<TBrokerFileStatus>> fileStatusesList;
+ protected List<List<TBrokerFileStatus>> fileStatusesList;
// file num
- private int filesAdded;
+ protected int filesAdded;
// Only used for external table in select statement
private List<Backend> backends;
@@ -139,14 +148,7 @@ public class BrokerScanNode extends LoadScanNode {
this.analyzer = analyzer;
if (desc.getTable() != null) {
- BrokerTable brokerTable = (BrokerTable) desc.getTable();
- try {
- fileGroups = Lists.newArrayList(new BrokerFileGroup(brokerTable));
- } catch (AnalysisException e) {
- throw new UserException(e.getMessage());
- }
- brokerDesc = new BrokerDesc(brokerTable.getBrokerName(), brokerTable.getBrokerProperties());
- targetTable = brokerTable;
+ this.initFileGroup();
}
// Get all broker file status
@@ -163,7 +165,18 @@ public class BrokerScanNode extends LoadScanNode {
}
}
- private boolean isLoad() {
+ protected void initFileGroup() throws UserException {
+ BrokerTable brokerTable = (BrokerTable) desc.getTable();
+ try {
+ fileGroups = Lists.newArrayList(new BrokerFileGroup(brokerTable));
+ } catch (AnalysisException e) {
+ throw new UserException(e.getMessage());
+ }
+ brokerDesc = new BrokerDesc(brokerTable.getBrokerName(), brokerTable.getBrokerProperties());
+ targetTable = brokerTable;
+ }
+
+ protected boolean isLoad() {
return desc.getTable() == null;
}
@@ -312,30 +325,7 @@ public class BrokerScanNode extends LoadScanNode {
// This will be fixed later.
fileStatusesList = Lists.newArrayList();
filesAdded = 0;
- for (BrokerFileGroup fileGroup : fileGroups) {
- boolean isBinaryFileFormat = fileGroup.isBinaryFileFormat();
- List<TBrokerFileStatus> fileStatuses = Lists.newArrayList();
- for (int i = 0; i < fileGroup.getFilePaths().size(); i++) {
- if (brokerDesc.isMultiLoadBroker()) {
- TBrokerFileStatus fileStatus = new TBrokerFileStatus(fileGroup.getFilePaths().get(i),
- false, fileGroup.getFileSize().get(i), false);
- fileStatuses.add(fileStatus);
- } else {
- BrokerUtil.parseFile(fileGroup.getFilePaths().get(i), brokerDesc, fileStatuses);
- }
- }
-
- // only get non-empty file or non-binary file
- fileStatuses = fileStatuses.stream().filter(f -> {
- return f.getSize() > 0 || !isBinaryFileFormat;
- }).collect(Collectors.toList());
-
- fileStatusesList.add(fileStatuses);
- filesAdded += fileStatuses.size();
- for (TBrokerFileStatus fstatus : fileStatuses) {
- LOG.info("Add file status is {}", fstatus);
- }
- }
+ this.getFileStatus();
}
Preconditions.checkState(fileStatusesList.size() == fileGroups.size());
@@ -370,6 +360,33 @@ public class BrokerScanNode extends LoadScanNode {
LOG.info("number instance of broker scan node is: {}, bytes per instance: {}", numInstances, bytesPerInstance);
}
+ protected void getFileStatus() throws UserException {
+ for (BrokerFileGroup fileGroup : fileGroups) {
+ boolean isBinaryFileFormat = fileGroup.isBinaryFileFormat();
+ List<TBrokerFileStatus> fileStatuses = Lists.newArrayList();
+ for (int i = 0; i < fileGroup.getFilePaths().size(); i++) {
+ if (brokerDesc.isMultiLoadBroker()) {
+ TBrokerFileStatus fileStatus = new TBrokerFileStatus(fileGroup.getFilePaths().get(i),
+ false, fileGroup.getFileSize().get(i), false);
+ fileStatuses.add(fileStatus);
+ } else {
+ BrokerUtil.parseFile(fileGroup.getFilePaths().get(i), brokerDesc, fileStatuses);
+ }
+ }
+
+ // only get non-empty file or non-binary file
+ fileStatuses = fileStatuses.stream().filter(f -> {
+ return f.getSize() > 0 || !isBinaryFileFormat;
+ }).collect(Collectors.toList());
+
+ fileStatusesList.add(fileStatuses);
+ filesAdded += fileStatuses.size();
+ for (TBrokerFileStatus fstatus : fileStatuses) {
+ LOG.info("Add file status is {}", fstatus);
+ }
+ }
+ }
+
private void assignBackends() throws UserException {
backends = Lists.newArrayList();
for (Backend be : Catalog.getCurrentSystemInfo().getIdToBackend().values()) {
@@ -422,6 +439,11 @@ public class BrokerScanNode extends LoadScanNode {
if (fileStatuses == null || fileStatuses.isEmpty()) {
return;
}
+ THdfsParams tHdfsParams = new THdfsParams();
+ if (this instanceof HiveScanNode) {
+ String fsName = ((HiveScanNode) this).getHdfsUri();
+ tHdfsParams.setFsName(fsName);
+ }
TScanRangeLocations curLocations = newLocations(context.params, brokerDesc);
long curInstanceBytes = 0;
long curFileOffset = 0;
@@ -454,6 +476,10 @@ public class BrokerScanNode extends LoadScanNode {
} else {
TBrokerRangeDesc rangeDesc = createBrokerRangeDesc(curFileOffset, fileStatus, formatType,
leftBytes, columnsFromPath, numberOfColumnsFromFile, brokerDesc);
+ if (this instanceof HiveScanNode) {
+ rangeDesc.setHdfsParams(tHdfsParams);
+ rangeDesc.setReadByColumnDef(true);
+ }
brokerScanRange(curLocations).addToRanges(rangeDesc);
curFileOffset = 0;
i++;
@@ -475,6 +501,10 @@ public class BrokerScanNode extends LoadScanNode {
rangeDesc.setNumAsString(context.fileGroup.isNumAsString());
rangeDesc.setReadJsonByLine(context.fileGroup.isReadJsonByLine());
}
+ if (this instanceof HiveScanNode) {
+ rangeDesc.setHdfsParams(tHdfsParams);
+ rangeDesc.setReadByColumnDef(true);
+ }
brokerScanRange(curLocations).addToRanges(rangeDesc);
curFileOffset = 0;
curInstanceBytes += leftBytes;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/HiveScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/HiveScanNode.java
new file mode 100644
index 0000000..2619079
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/HiveScanNode.java
@@ -0,0 +1,210 @@
+// 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.doris.planner;
+
+import org.apache.doris.analysis.Analyzer;
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.Expr;
+import org.apache.doris.analysis.ImportColumnDesc;
+import org.apache.doris.analysis.StorageBackend;
+import org.apache.doris.analysis.TupleDescriptor;
+import org.apache.doris.catalog.HiveMetaStoreClientHelper;
+import org.apache.doris.catalog.HiveTable;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.load.BrokerFileGroup;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TExplainLevel;
+
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+
+public class HiveScanNode extends BrokerScanNode {
+ private static final Logger LOG = LogManager.getLogger(HiveScanNode.class);
+
+ private static final String HIVE_DEFAULT_COLUMN_SEPARATOR = "\001";
+ private static final String HIVE_DEFAULT_LINE_DELIMITER = "\n";
+
+ private HiveTable hiveTable;
+ // partition column predicates of hive table
+ private List<ExprNodeDesc> hivePredicates = new ArrayList<>();
+ private ExprNodeGenericFuncDesc hivePartitionPredicate;
+ private List<ImportColumnDesc> parsedColumnExprList = new ArrayList<>();
+ private String hdfsUri;
+
+ private Table remoteHiveTable;
+
+ /* hive table properties */
+ private String columnSeparator;
+ private String lineDelimiter;
+ private String fileFormat;
+ private String path;
+ private List<String> partitionKeys = new ArrayList<>();
+ /* hive table properties */
+
+ public String getHdfsUri() {
+ return hdfsUri;
+ }
+
+ public List<ImportColumnDesc> getParsedColumnExprList() {
+ return parsedColumnExprList;
+ }
+
+ public String getColumnSeparator() {
+ return columnSeparator;
+ }
+
+ public String getLineDelimiter() {
+ return lineDelimiter;
+ }
+
+ public String getFileFormat() {
+ return fileFormat;
+ }
+
+ public String getPath() {
+ return path;
+ }
+
+ public List<String> getPartitionKeys() {
+ return partitionKeys;
+ }
+
+ public HiveScanNode(PlanNodeId id, TupleDescriptor destTupleDesc, String planNodeName,
+ List<List<TBrokerFileStatus>> fileStatusesList, int filesAdded) {
+ super(id, destTupleDesc, planNodeName, fileStatusesList, filesAdded);
+ this.hiveTable = (HiveTable) destTupleDesc.getTable();
+ }
+
+ @Override
+ public void init(Analyzer analyzer) throws UserException {
+ super.init(analyzer);
+ }
+
+ @Override
+ protected void initFileGroup() throws UserException {
+ initHiveTblProperties();
+ analyzeColumnFromPath();
+
+ HiveTable hiveTable = (HiveTable) desc.getTable();
+ fileGroups = Lists.newArrayList(
+ new BrokerFileGroup(hiveTable,
+ getColumnSeparator(),
+ getLineDelimiter(),
+ getPath(),
+ getFileFormat(),
+ getPartitionKeys(),
+ getParsedColumnExprList()));
+ brokerDesc = new BrokerDesc("HiveTableDesc", StorageBackend.StorageType.HDFS, hiveTable.getHiveProperties());
+ targetTable = hiveTable;
+ }
+
+ private void initHiveTblProperties() throws DdlException {
+ this.remoteHiveTable = HiveMetaStoreClientHelper.getTable(hiveTable);
+ this.fileFormat = HiveMetaStoreClientHelper.HiveFileFormat.getFormat(remoteHiveTable.getSd().getInputFormat());
+
+ Map<String, String> serDeInfoParams = remoteHiveTable.getSd().getSerdeInfo().getParameters();
+ this.columnSeparator = Strings.isNullOrEmpty(serDeInfoParams.get("field.delim")) ?
+ HIVE_DEFAULT_COLUMN_SEPARATOR : serDeInfoParams.get("field.delim");
+ this.lineDelimiter = Strings.isNullOrEmpty(serDeInfoParams.get("line.delim")) ?
+ HIVE_DEFAULT_LINE_DELIMITER : serDeInfoParams.get("line.delim");
+ this.path = remoteHiveTable.getSd().getLocation();
+ for (FieldSchema fieldSchema : remoteHiveTable.getPartitionKeys()) {
+ this.partitionKeys.add(fieldSchema.getName());
+ }
+ }
+
+ /**
+ * Extracts partition predicate from SelectStmt.whereClause that can be pushed down to Hive
+ */
+ private void extractHivePartitionPredicate() throws DdlException {
+ ListIterator<Expr> it = conjuncts.listIterator();
+ while (it.hasNext()) {
+ ExprNodeGenericFuncDesc hiveExpr = HiveMetaStoreClientHelper.convertToHivePartitionExpr(
+ it.next(), partitionKeys, hiveTable.getName());
+ if (hiveExpr != null) {
+ hivePredicates.add(hiveExpr);
+ }
+ }
+ int count = hivePredicates.size();
+ // combine all predicate by `and`
+ // compoundExprs must have at least 2 predicates
+ if (count >= 2) {
+ hivePartitionPredicate = HiveMetaStoreClientHelper.getCompoundExpr(hivePredicates, "and");
+ } else if (count == 1) {
+ // only one predicate
+ hivePartitionPredicate = (ExprNodeGenericFuncDesc) hivePredicates.get(0);
+ } else {
+ // have no predicate, make a dummy predicate "1=1" to get all partitions
+ HiveMetaStoreClientHelper.ExprBuilder exprBuilder =
+ new HiveMetaStoreClientHelper.ExprBuilder(hiveTable.getName());
+ hivePartitionPredicate = exprBuilder.val(TypeInfoFactory.intTypeInfo, 1)
+ .val(TypeInfoFactory.intTypeInfo, 1)
+ .pred("=", 2).build();
+ }
+ }
+
+ @Override
+ protected void getFileStatus() throws UserException {
+ if (partitionKeys.size() > 0) {
+ extractHivePartitionPredicate();
+ }
+ List<TBrokerFileStatus> fileStatuses = new ArrayList<>();
+ this.hdfsUri = HiveMetaStoreClientHelper.getHiveDataFiles(hiveTable, hivePartitionPredicate,
+ fileStatuses, remoteHiveTable);
+ fileStatusesList.add(fileStatuses);
+ filesAdded += fileStatuses.size();
+ for (TBrokerFileStatus fstatus : fileStatuses) {
+ LOG.info("Add file status is {}", fstatus);
+ }
+ }
+
+ @Override
+ public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
+ StringBuilder output = new StringBuilder();
+ if (!isLoad()) {
+ output.append(prefix).append("TABLE: ").append(hiveTable.getName()).append("\n");
+ output.append(prefix).append("PATH: ")
+ .append(hiveTable.getHiveProperties().get(HiveTable.HIVE_METASTORE_URIS)).append("\n");
+ }
+ return output.toString();
+ }
+
+ /**
+ * Analyze columns from path, the partition columns
+ */
+ private void analyzeColumnFromPath() {
+ for (String colName : partitionKeys) {
+ ImportColumnDesc importColumnDesc = new ImportColumnDesc(colName, null);
+ parsedColumnExprList.add(importColumnDesc);
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java
index 3de5692..31c0b3f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java
@@ -1692,10 +1692,14 @@ public class SingleNodePlanner {
case ELASTICSEARCH:
scanNode = new EsScanNode(ctx_.getNextNodeId(), tblRef.getDesc(), "EsScanNode");
break;
+ case HIVE:
+ scanNode = new HiveScanNode(ctx_.getNextNodeId(), tblRef.getDesc(), "HiveScanNode",
+ null, -1);
+ break;
default:
break;
}
- if (scanNode instanceof OlapScanNode || scanNode instanceof EsScanNode) {
+ if (scanNode instanceof OlapScanNode || scanNode instanceof EsScanNode || scanNode instanceof HiveScanNode) {
PredicatePushDown.visitScanNode(scanNode, tblRef.getJoinOp(), analyzer);
scanNode.setSortColumn(tblRef.getSortColumn());
}
diff --git a/fe/pom.xml b/fe/pom.xml
index 31d14d8..253c095 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -51,6 +51,8 @@ under the License.
<grpc.version>1.30.0</grpc.version>
<protobuf.version>3.14.0</protobuf.version>
<skip.plugin>false</skip.plugin>
+ <hive.version>2.3.7</hive.version>
+ <hadoop.version>2.8.0</hadoop.version>
</properties>
<profiles>
@@ -155,10 +157,11 @@ under the License.
</dependency>
<!-- https://mvnrepository.com/artifact/commons-lang/commons-lang -->
+ <!-- upgrade commons-lang from 2.4 to 2.6 to fix incompatibility of with versioning scheme of Java 9 and later -->
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
- <version>2.4</version>
+ <version>2.6</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
@@ -588,9 +591,86 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-metastore</artifactId>
+ <version>${hive.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ <exclusion>
+ <groupId>jdk.tools</groupId>
+ <artifactId>jdk.tools</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-compiler</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-exec</artifactId>
+ <classifier>core</classifier>
+ <version>${hive.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.codehaus.groovy</groupId>
+ <artifactId>groovy-all</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-aws</artifactId>
+ <version>${hadoop.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <artifactId>servlet-api</artifactId>
+ <groupId>javax.servlet</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+
+ <dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
- <version>2.8.0</version>
+ <version>${hadoop.version}</version>
<scope>provided</scope>
<exclusions>
<exclusion>
diff --git a/gensrc/thrift/Descriptors.thrift b/gensrc/thrift/Descriptors.thrift
index 4224e86..467c4ba 100644
--- a/gensrc/thrift/Descriptors.thrift
+++ b/gensrc/thrift/Descriptors.thrift
@@ -237,6 +237,12 @@ struct TSchemaTable {
struct TBrokerTable {
}
+struct THiveTable {
+ 1: required string db_name
+ 2: required string table_name
+ 3: required map<string, string> properties
+}
+
// "Union" of all table types.
struct TTableDescriptor {
1: required Types.TTableId id
@@ -255,6 +261,7 @@ struct TTableDescriptor {
14: optional TBrokerTable BrokerTable
15: optional TEsTable esTable
16: optional TOdbcTable odbcTable
+ 17: optional THiveTable hiveTable
}
struct TDescriptorTable {
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index a6549fe..6a862ef 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -152,6 +152,8 @@ struct TBrokerRangeDesc {
15: optional bool fuzzy_parse;
16: optional THdfsParams hdfs_params
17: optional bool read_json_by_line;
+ // Whether read line by column defination, only for Hive
+ 18: optional bool read_by_column_def;
}
struct TBrokerScanRangeParams {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org