apache · morningman · Feb 28, 2019 · Mar 1, 2019 · Mar 1, 2019 · Mar 5, 2019
diff --git a/.gitignore b/.gitignore
@@ -8,3 +8,5 @@ gensrc/build
 fe/target
 thirdparty/src
 *.so.tmp
+.DS_Store
+*.iml
diff --git a/README.md b/README.md
@@ -23,7 +23,9 @@ The simplicity (of developing, deploying and using) and meeting many data servin
 
 Currently only supports Docker environment and Linux OS, such as Ubuntu and CentOS.
 
-### 4.1 For Docker 
+### 4.1 Compile in Docker environment (Recommended)
+
+We offer a docker images as a Doris compilation environment. You can compile Doris from source in it and run the output binaries in other Linux environment.
 
 Firstly, you must be install and start docker service.
 
@@ -45,15 +47,25 @@ apachedoris/doris-dev   build-env           f8bc5d4024e0        21 hours ago
 
 #### Step2: Run the Docker image 
 
-You can run image directyly:
+You can run the image directyly:
 
 ```
 $ docker run -it apachedoris/doris-dev:build-env
 ```
 
+Or if you want to compile the source located in your local host, you can map the local directory to the image by running:
+
+```
+$ docker run -it -v /your/local/path/incubator-doris-DORIS-x.x.x-release/:/root/incubator-doris-DORIS-x.x.x-release/ apachedoris/doris-dev:build-env
+```
+
 #### Step3: Download Doris source
+
+Now you should in docker environment.
+
 You can download Doris source by release package or by git clone in image.
-(If you have downloaded source and it is not in image, you can map its path to image in Step2.)
+
+(If you already downloaded the source in your local host and map it to the image in Step2, you can skip this step.)
 
 ```
 $ wget https://dist.apache.org/repos/dist/dev/incubator/doris/xxx.tar.gz
@@ -62,13 +74,14 @@ $ git clone https://github.com/apache/incubator-doris.git
 ```
 
 #### Step4: Build Doris
-Now you should in docker environment, and you can enter Doris source path and build Doris.
+
+Enter Doris source path and build Doris.
 
 ```
 $ sh build.sh
 ```
 
-After successfully building, it will install binary files in the directory output/.
+After successfully building, it will install binary files in the directory `output/`.
 
 ### 4.2 For Linux OS
 
@@ -98,7 +111,7 @@ Run following script, it will compile thirdparty libraries and build whole Doris
 sh build.sh
 ```
 
-After successfully building, it will install binary files in the directory output/.
+After successfully building, it will install binary files in the directory `output/`.
 
 ## 5. Reporting Issues
 
@@ -116,3 +129,4 @@ If you find any bugs, please file a [GitHub issue](https://github.com/apache/inc
 * Deploy and Upgrade - <https://github.com/apache/incubator-doris/wiki/Doris-Deploy-%26-Upgrade>
 * User Manual - <https://github.com/apache/incubator-doris/wiki/Doris-Create%2C-Load-and-Delete>
 * FAQs - <https://github.com/apache/incubator-doris/wiki/Doris-FAQ>
+
diff --git a/be/CMakeLists.txt b/be/CMakeLists.txt
@@ -178,12 +178,12 @@ set_target_properties(brpc PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib64/
 add_library(rocksdb STATIC IMPORTED)
 set_target_properties(rocksdb PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/librocksdb.a)
 
-add_library(librdkafka STATIC IMPORTED)
-set_target_properties(librdkafka PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/librdkafka.a)
-
 add_library(librdkafka_cpp STATIC IMPORTED)
 set_target_properties(librdkafka_cpp PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/librdkafka++.a)
 
+add_library(librdkafka STATIC IMPORTED)
+set_target_properties(librdkafka PROPERTIES IMPORTED_LOCATION ${THIRDPARTY_DIR}/lib/librdkafka.a)
+
 find_program(THRIFT_COMPILER thrift ${CMAKE_SOURCE_DIR}/bin)
 
 # llvm-config
@@ -440,9 +440,10 @@ set(DORIS_LINK_LIBS
 
 # Set thirdparty libraries
 set(DORIS_DEPENDENCIES
+    ${WL_START_GROUP}
     rocksdb
-    librdkafka
     librdkafka_cpp
+    librdkafka
     lzo
     snappy
     ${Boost_LIBRARIES}
@@ -456,16 +457,15 @@ set(DORIS_DEPENDENCIES
     libevent
     mysql
     curl
-    ${WL_START_GROUP}
     ${LIBZ}
     ${LIBBZ2}
     gflags
     brpc
     protobuf
     openssl
     crypto
-    ${WL_START_GROUP}
     leveldb
+    ${WL_END_GROUP}
 )
 
 # Add all external dependencies. They should come after the palo libs.

diff --git a/be/src/common/config.h b/be/src/common/config.h
@@ -265,6 +265,7 @@ namespace config {
     CONF_Int32(number_tablet_writer_threads, "16");
 
     CONF_Int64(streaming_load_max_mb, "10240");
+    CONF_Int32(streaming_load_rpc_max_alive_time_sec, "600");
 
     // Fragment thread pool
     CONF_Int32(fragment_pool_thread_num, "64");

diff --git a/be/src/common/daemon.cpp b/be/src/common/daemon.cpp
@@ -45,6 +45,7 @@
 #include "exprs/es_functions.h"
 #include "exprs/timestamp_functions.h"
 #include "exprs/decimal_operators.h"
+#include "exprs/decimalv2_operators.h"
 #include "exprs/utility_functions.h"
 #include "exprs/json_functions.h"
 #include "exprs/hll_hash_function.h"
@@ -182,6 +183,7 @@ void init_daemon(int argc, char** argv, const std::vector<StorePath>& paths) {
     EncryptionFunctions::init();
     TimestampFunctions::init();
     DecimalOperators::init();
+    DecimalV2Operators::init();
     UtilityFunctions::init();
     CompoundPredicate::init();
     JsonFunctions::init();

diff --git a/be/src/common/status.cpp b/be/src/common/status.cpp
@@ -34,6 +34,9 @@ const Status Status::MEM_LIMIT_EXCEEDED(
 const Status Status::THRIFT_RPC_ERROR(
     TStatusCode::THRIFT_RPC_ERROR, "Thrift RPC failed", true);
 
+const Status Status::TIMEOUT(
+    TStatusCode::TIMEOUT, "timeout", true);
+
 Status::ErrorDetail::ErrorDetail(const TStatus& status) : 
         error_code(status.status_code),
         error_msgs(status.error_msgs) {

diff --git a/be/src/common/status.h b/be/src/common/status.h
@@ -54,6 +54,7 @@ class Status {
     static const Status CANCELLED;
     static const Status MEM_LIMIT_EXCEEDED;
     static const Status THRIFT_RPC_ERROR;
+    static const Status TIMEOUT;
 
     // copy c'tor makes copy of error detail so Status can be returned by value
     Status(const Status& status) : _error_detail(

diff --git a/be/src/common/utils.h b/be/src/common/utils.h
@@ -0,0 +1,52 @@
+// 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.
+
+#pragma once
+
+#include <string>
+
+namespace doris {
+
+struct AuthInfo {
+    std::string user;
+    std::string passwd;
+    std::string cluster;
+    std::string user_ip;
+    // -1 as unset
+    int64_t auth_code = -1;
+};
+
+template<class T>
+void set_request_auth(T* req, const AuthInfo& auth) {
+    if (auth.auth_code != -1) {
+        // if auth_code is set, no need to set other info
+        req->__set_auth_code(auth.auth_code);
+        // user name and passwd is unused, but they are required field.
+        // so they have to be set.
+        req->user = "";
+        req->passwd = "";
+    } else {
+        req->user = auth.user;
+        req->passwd = auth.passwd;
+        if (!auth.cluster.empty()) {
+            req->__set_cluster(auth.cluster);
+        }
+        req->__set_user_ip(auth.user_ip);
+    }
+}
+
+}
diff --git a/be/src/exec/CMakeLists.txt b/be/src/exec/CMakeLists.txt
@@ -75,7 +75,7 @@ set(EXEC_FILES
     schema_scanner/schema_columns_scanner.cpp
     schema_scanner/schema_charsets_scanner.cpp
     schema_scanner/schema_collations_scanner.cpp
-    schema_scanner/frontend_helper.cpp
+    schema_scanner/schema_helper.cpp
     partitioned_hash_table.cc
     partitioned_hash_table_ir.cc
     partitioned_aggregation_node.cc

diff --git a/be/src/exec/broker_scan_node.cpp b/be/src/exec/broker_scan_node.cpp
@@ -344,7 +344,7 @@ Status BrokerScanNode::scanner_scan(
                 tuple = reinterpret_cast<Tuple*>(new_tuple);
                 counter->num_rows_returned++;
             } else {
-                counter->num_rows_filtered++;
+                counter->num_rows_unselected++;
             }
         }
 
@@ -409,6 +409,7 @@ void BrokerScanNode::scanner_worker(int start_idx, int length) {
     // Update stats
     _runtime_state->update_num_rows_load_success(counter.num_rows_returned);
     _runtime_state->update_num_rows_load_filtered(counter.num_rows_filtered);
+    _runtime_state->update_num_rows_load_unselected(counter.num_rows_unselected);
 
     // scanner is going to finish 
     {

diff --git a/be/src/exec/broker_scanner.cpp b/be/src/exec/broker_scanner.cpp
@@ -24,8 +24,8 @@
 #include "runtime/exec_env.h"
 #include "runtime/mem_tracker.h"
 #include "runtime/raw_value.h"
-#include "runtime/load_stream_mgr.h"
-#include "runtime/stream_load_pipe.h"
+#include "runtime/stream_load/load_stream_mgr.h"
+#include "runtime/stream_load/stream_load_pipe.h"
 #include "runtime/tuple.h"
 #include "exprs/expr.h"
 #include "exec/text_converter.h"
@@ -237,6 +237,7 @@ Status BrokerScanner::open_file_reader() {
     case TFileType::FILE_STREAM: {
         _stream_load_pipe = _state->exec_env()->load_stream_mgr()->get(range.load_id);
         if (_stream_load_pipe == nullptr) {
+            VLOG(3) << "unknown stream load id: " << UniqueId(range.load_id);
             return Status("unknown stream load id");
         }
         _cur_file_reader = _stream_load_pipe.get();

diff --git a/be/src/exec/broker_scanner.h b/be/src/exec/broker_scanner.h
@@ -49,11 +49,15 @@ class RuntimeProfile;
 class StreamLoadPipe;
 
 struct BrokerScanCounter {
-    BrokerScanCounter() : num_rows_returned(0), num_rows_filtered(0) {
+    BrokerScanCounter() :
+        num_rows_returned(0),
+        num_rows_filtered(0),
+        num_rows_unselected(0) {
     }
 
-    int64_t num_rows_returned;
-    int64_t num_rows_filtered;
+    int64_t num_rows_returned;  // qualified rows
+    int64_t num_rows_filtered;  // unqualified rows
+    int64_t num_rows_unselected; // rows filterd by predicates
 };
 
 // Broker scanner convert the data read from broker to doris's tuple.