Refactor alter job

be/src/agent/agent_server.cpp

@@ -329,8 +329,8 @@ void AgentServer::submit_tasks(
             break;
         case TTaskType::ROLLUP:
         case TTaskType::SCHEMA_CHANGE:
-        case TTaskType::ALTER_TASK:
-            if (task.__isset.alter_tablet_req) {
+        case TTaskType::ALTER:
+            if (task.__isset.alter_tablet_req || task.__isset.alter_tablet_req_v2) {
                 _alter_tablet_workers->submit_task(task);
             } else {
                 status_code = TStatusCode::ANALYSIS_ERROR;

be/src/agent/task_worker_pool.cpp

@@ -554,7 +554,7 @@ void* TaskWorkerPool::_alter_tablet_worker_thread_callback(void* arg_this) {
             int64_t time_elapsed = time(nullptr) - agent_task_req.recv_time;
             if (time_elapsed > config::report_task_interval_seconds * 20) {
                 LOG(INFO) << "task elapsed " << time_elapsed 
-                          << " since it is inserted to queue, it is timeout";
+                          << " seconds since it is inserted to queue, it is timeout";
                 is_task_timeout = true;
             }
         }
@@ -564,7 +564,7 @@ void* TaskWorkerPool::_alter_tablet_worker_thread_callback(void* arg_this) {
             switch (task_type) {
             case TTaskType::SCHEMA_CHANGE:
             case TTaskType::ROLLUP:
-            case TTaskType::ALTER_TASK:
+            case TTaskType::ALTER:
                 worker_pool_this->_alter_tablet(worker_pool_this,
                                             agent_task_req,
                                             signatrue,
@@ -602,8 +602,8 @@ void TaskWorkerPool::_alter_tablet(
     case TTaskType::SCHEMA_CHANGE:
         process_name = "schema change";
         break;
-    case TTaskType::ALTER_TASK:
-        process_name = "alter table";
+    case TTaskType::ALTER:
+        process_name = "alter";
         break;
     default:
         std::string task_name;
@@ -621,7 +621,7 @@ void TaskWorkerPool::_alter_tablet(
     TSchemaHash new_schema_hash = 0;
     if (status == DORIS_SUCCESS) {
         OLAPStatus sc_status = OLAP_SUCCESS;
-        if (task_type == TTaskType::ALTER_TASK) {
+        if (task_type == TTaskType::ALTER) {
             new_tablet_id = agent_task_req.alter_tablet_req_v2.new_tablet_id;
             new_schema_hash = agent_task_req.alter_tablet_req_v2.new_schema_hash;
             EngineAlterTabletTask engine_task(agent_task_req.alter_tablet_req_v2, signature, task_type, &error_msgs, process_name);

be/src/exec/tablet_sink.h

@@ -219,7 +219,7 @@ class OlapTableSink : public DataSink {
     int64_t _db_id = -1;
     int64_t _table_id = -1;
     int _num_repicas = -1;
-    bool _need_gen_rollup = true;
+    bool _need_gen_rollup = false;
     std::string _db_name;
     std::string _table_name;
     int _tuple_desc_id = -1;

be/src/exprs/timestamp_functions.cpp

@@ -391,6 +391,7 @@ DateTimeVal TimestampFunctions::timestamp(
     return val;
 }
 
+// FROM_UNIXTIME()
 StringVal TimestampFunctions::from_unix(
         FunctionContext* context, const IntVal& unix_time) {
     if (unix_time.is_null) {
@@ -405,6 +406,7 @@ StringVal TimestampFunctions::from_unix(
     return AnyValUtil::from_string_temp(context, buf);
 }
 
+// FROM_UNIXTIME()
 StringVal TimestampFunctions::from_unix(
             FunctionContext* context, const IntVal& unix_time, const StringVal& fmt) {
     if (unix_time.is_null || fmt.is_null) {
@@ -422,10 +424,12 @@ StringVal TimestampFunctions::from_unix(
     return AnyValUtil::from_string_temp(context, buf);
 }
 
+// UNIX_TIMESTAMP()
 IntVal TimestampFunctions::to_unix(FunctionContext* context) {
     return IntVal(context->impl()->state()->timestamp_ms() / 1000);
 }
 
+// UNIX_TIMESTAMP()
 IntVal TimestampFunctions::to_unix(
             FunctionContext* context, const StringVal& string_val, const StringVal& fmt) {
     if (string_val.is_null || fmt.is_null) {

be/src/olap/data_dir.cpp

@@ -1026,10 +1026,10 @@ void DataDir::perform_path_scan() {
 
 void DataDir::_process_garbage_path(const std::string& path) {
     if (check_dir_existed(path)) {
-        LOG(INFO) << "collect garbage dir path:" << path;
+        LOG(INFO) << "collect garbage dir path: " << path;
         OLAPStatus status = remove_all_dir(path);
         if (status != OLAP_SUCCESS) {
-            LOG(WARNING) << "remove garbage dir path:" << path << " failed";
+            LOG(WARNING) << "remove garbage dir path: " << path << " failed";
         }
     }
 }

be/src/olap/delta_writer.cpp

@@ -130,19 +130,9 @@ OLAPStatus DeltaWriter::init() {
     writer_context.load_id = _req.load_id;
     RETURN_NOT_OK(RowsetFactory::create_rowset_writer(writer_context, &_rowset_writer));
 
-    const std::vector<SlotDescriptor*>& slots = _req.tuple_desc->slots();
-    const TabletSchema& schema = _tablet->tablet_schema();
-    for (size_t col_id = 0; col_id < schema.num_columns(); ++col_id) {
-        const TabletColumn& column = schema.column(col_id);
-        for (size_t i = 0; i < slots.size(); ++i) {
-            if (slots[i]->col_name() == column.name()) {
-                _col_ids.push_back(i);
-            }
-        }
-    }
     _tablet_schema = &(_tablet->tablet_schema());
     _schema = new Schema(*_tablet_schema);
-    _mem_table = new MemTable(_schema, _tablet_schema, &_col_ids,
+    _mem_table = new MemTable(_schema, _tablet_schema, _req.slots,
                               _req.tuple_desc, _tablet->keys_type());
     _is_init = true;
     return OLAP_SUCCESS;
@@ -161,7 +151,7 @@ OLAPStatus DeltaWriter::write(Tuple* tuple) {
         RETURN_NOT_OK(_mem_table->flush(_rowset_writer.get()));
 
         SAFE_DELETE(_mem_table);
-        _mem_table = new MemTable(_schema, _tablet_schema, &_col_ids,
+        _mem_table = new MemTable(_schema, _tablet_schema, _req.slots,
                                   _req.tuple_desc, _tablet->keys_type());
     }
     return OLAP_SUCCESS;

be/src/olap/delta_writer.h

@@ -47,6 +47,8 @@ struct WriteRequest {
     PUniqueId load_id;
     bool need_gen_rollup;
     TupleDescriptor* tuple_desc;
+    // slots are in order of tablet's schema
+    const std::vector<SlotDescriptor*>* slots;
 };
 
 class DeltaWriter {
@@ -76,7 +78,6 @@ class DeltaWriter {
     MemTable* _mem_table;
     Schema* _schema;
     const TabletSchema* _tablet_schema;
-    std::vector<uint32_t> _col_ids;
     bool _delta_written_success;
 };
 

be/src/olap/memtable.cpp

@@ -27,12 +27,12 @@
 namespace doris {
 
 MemTable::MemTable(Schema* schema, const TabletSchema* tablet_schema,
-                   std::vector<uint32_t>* col_ids, TupleDescriptor* tuple_desc,
+                   const std::vector<SlotDescriptor*>* slot_descs, TupleDescriptor* tuple_desc,
                    KeysType keys_type)
     : _schema(schema),
       _tablet_schema(tablet_schema),
       _tuple_desc(tuple_desc),
-      _col_ids(col_ids),
+      _slot_descs(slot_descs),
       _keys_type(keys_type),
       _row_comparator(_schema) {
     _schema_size = _schema->schema_size();
@@ -58,11 +58,10 @@ size_t MemTable::memory_usage() {
 }
 
 void MemTable::insert(Tuple* tuple) {
-    const std::vector<SlotDescriptor*>& slots = _tuple_desc->slots();
     ContiguousRow row(_schema, _tuple_buf);
-    for (size_t i = 0; i < _col_ids->size(); ++i) {
+    for (size_t i = 0; i < _slot_descs->size(); ++i) {
         auto cell = row.cell(i);
-        const SlotDescriptor* slot = slots[(*_col_ids)[i]];
+        const SlotDescriptor* slot = (*_slot_descs)[i];
 
         bool is_null = tuple->is_null(slot->null_indicator_offset());
         void* value = tuple->get_slot(slot->tuple_offset());

be/src/olap/memtable.h

@@ -32,7 +32,7 @@ class RowCursor;
 class MemTable {
 public:
     MemTable(Schema* schema, const TabletSchema* tablet_schema,
-             std::vector<uint32_t>* col_ids, TupleDescriptor* tuple_desc,
+             const std::vector<SlotDescriptor*>* slot_descs, TupleDescriptor* tuple_desc,
              KeysType keys_type);
     ~MemTable();
     size_t memory_usage();
@@ -43,7 +43,8 @@ class MemTable {
     Schema* _schema;
     const TabletSchema* _tablet_schema;
     TupleDescriptor* _tuple_desc;
-    std::vector<uint32_t>* _col_ids;
+    // the slot in _slot_descs are in order of tablet's schema
+    const std::vector<SlotDescriptor*>* _slot_descs;
     KeysType _keys_type;
 
     struct RowCursorComparator {

be/src/olap/schema_change.cpp

@@ -981,6 +981,7 @@ bool SchemaChangeWithSorting::process(
 
             row_block_arr.push_back(new_row_block);
         } else {
+            LOG(INFO) << "new block num rows is: " << new_row_block->row_block_info().row_num;
             _row_block_allocator->release(new_row_block);
             new_row_block = nullptr;
         }
@@ -1181,7 +1182,7 @@ OLAPStatus SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletRe
             request.new_tablet_id, request.new_schema_hash);
     if (new_tablet == nullptr) {
         LOG(WARNING) << "fail to find new tablet."
-                     << ", new_tablet=" << request.new_tablet_id
+                     << " new_tablet=" << request.new_tablet_id
                      << ", new_schema_hash=" << request.new_schema_hash;
         return OLAP_ERR_TABLE_NOT_FOUND;
     }
@@ -1335,15 +1336,19 @@ OLAPStatus SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletRe
         if (res != OLAP_SUCCESS) {
             break;
         }
-        res = _validate_alter_result(new_tablet, request);
     } while(0);
 
+    if (res == OLAP_SUCCESS) {
+        // _validate_alter_result should be outside the above while loop.
+        // to avoid requiring the header lock twice.
+        res = _validate_alter_result(new_tablet, request);
+    }
+
     // if failed convert history data, then just remove the new tablet
     if (res != OLAP_SUCCESS) {
         LOG(WARNING) << "failed to alter tablet. base_tablet=" << base_tablet->full_name()
                      << ", drop new_tablet=" << new_tablet->full_name();
         // do not drop the new tablet and its data. GC thread will 
-        // StorageEngine::instance()->tablet_manager()->drop_tablet(new_tablet->tablet_id(), new_tablet->schema_hash());
     }
 
     return res;
@@ -2134,7 +2139,7 @@ OLAPStatus SchemaChangeHandler::_validate_alter_result(TabletSharedPtr new_table
     Version max_continuous_version = {-1, 0};
     VersionHash max_continuous_version_hash = 0;
     new_tablet->max_continuous_version_from_begining(&max_continuous_version, &max_continuous_version_hash);
-    LOG(INFO) << "find max continuous version "
+    LOG(INFO) << "find max continuous version of tablet=" << new_tablet->full_name()
               << ", start_version=" << max_continuous_version.first
               << ", end_version=" << max_continuous_version.second
               << ", version_hash=" << max_continuous_version_hash;

be/src/olap/tablet.cpp

@@ -282,7 +282,8 @@ OLAPStatus Tablet::modify_rowsets(const vector<RowsetSharedPtr>& to_add,
 const RowsetSharedPtr Tablet::get_rowset_by_version(const Version& version) const {
     auto iter = _rs_version_map.find(version);
     if (iter == _rs_version_map.end()) {
-        LOG(INFO) << "no rowset for version:" << version.first << "-" << version.second;
+        LOG(INFO) << "no rowset for version:" << version.first << "-" << version.second
+                << ", tablet: " << full_name();
         return nullptr;
     }
     RowsetSharedPtr rowset = iter->second;
@@ -294,7 +295,8 @@ size_t Tablet::get_rowset_size_by_version(const Version& version) {
             << "invalid version:" << version.first << "-" << version.second;
     auto iter = _rs_version_map.find(version);
     if (iter == _rs_version_map.end()) {
-        LOG(WARNING) << "no rowset for version:" << version.first << "-" << version.second;
+        LOG(WARNING) << "no rowset for version:" << version.first << "-" << version.second
+                << ", tablet: " << full_name();
         return -1;
     }
     RowsetSharedPtr rowset = iter->second;

be/src/olap/task/engine_alter_tablet_task.cpp

@@ -33,18 +33,13 @@ EngineAlterTabletTask::EngineAlterTabletTask(const TAlterTabletReqV2& request,
         _process_name(process_name) { }
 
 OLAPStatus EngineAlterTabletTask::execute() {
-    LOG(INFO) << "begin to create new alter tablet. base_tablet_id=" << _alter_tablet_req.base_tablet_id
-              << ", base_schema_hash=" << _alter_tablet_req.base_schema_hash
-              << ", new_tablet_id=" << _alter_tablet_req.new_tablet_id
-              << ", new_schema_hash=" << _alter_tablet_req.new_schema_hash;
-
     DorisMetrics::create_rollup_requests_total.increment(1);
 
     SchemaChangeHandler handler;
     OLAPStatus res = handler.process_alter_tablet_v2(_alter_tablet_req);
 
     if (res != OLAP_SUCCESS) {
-        LOG(WARNING) << "failed to do rollup. res=" << res
+        LOG(WARNING) << "failed to do alter task. res=" << res
                      << " base_tablet_id=" << _alter_tablet_req.base_tablet_id
                      << ", base_schema_hash=" << _alter_tablet_req.base_schema_hash
                      << ", new_tablet_id=" << _alter_tablet_req.new_tablet_id

be/src/olap/txn_manager.cpp

@@ -201,7 +201,8 @@ OLAPStatus TxnManager::commit_txn(
                 << " partition_id: " << key.first
                 << ", transaction_id: " << key.second
                 << ", tablet: " << tablet_info.to_string()
-                << ", rowsetid: " << rowset_ptr->rowset_id();
+                << ", rowsetid: " << rowset_ptr->rowset_id()
+                << ", version: " << rowset_ptr->version().first;
     }
     return OLAP_SUCCESS;
 }

be/src/runtime/tablet_writer_mgr.cpp

@@ -207,16 +207,16 @@ Status TabletsChannel::close(int sender_id, bool* finished,
 }
 
 Status TabletsChannel::_open_all_writers(const PTabletWriterOpenRequest& params) {
-    std::vector<SlotDescriptor*>* columns = nullptr;
+    std::vector<SlotDescriptor*>* index_slots = nullptr;
     int32_t schema_hash = 0;
     for (auto& index : _schema->indexes()) {
         if (index->index_id == _index_id) {
-            columns = &index->slots;
+            index_slots = &index->slots;
             schema_hash = index->schema_hash;
             break;
         }
     }
-    if (columns == nullptr) {
+    if (index_slots == nullptr) {
         std::stringstream ss;
         ss << "unknown index id, key=" << _key;
         return Status::InternalError(ss.str());
@@ -231,6 +231,7 @@ Status TabletsChannel::_open_all_writers(const PTabletWriterOpenRequest& params)
         request.load_id = params.id();
         request.need_gen_rollup = params.need_gen_rollup();
         request.tuple_desc = _tuple_desc;
+        request.slots = index_slots;
 
         DeltaWriter* writer = nullptr;
         auto st = DeltaWriter::open(&request, &writer);

be/test/olap/delta_writer_test.cpp

@@ -323,17 +323,18 @@ TEST_F(TestDeltaWriter, write) {
     DescriptorTbl* desc_tbl = nullptr;
     DescriptorTbl::create(&obj_pool, tdesc_tbl, &desc_tbl);
     TupleDescriptor* tuple_desc = desc_tbl->get_tuple_descriptor(0);
+    const std::vector<SlotDescriptor*>& slots = tuple_desc->slots();
 
     PUniqueId load_id;
     load_id.set_hi(0);
     load_id.set_lo(0);
     WriteRequest write_req = {10004, 270068376, WriteType::LOAD,
-                              20002, 30002, load_id, false, tuple_desc};
+                              20002, 30002, load_id, false, tuple_desc,
+                              &(tuple_desc->slots())};
     DeltaWriter* delta_writer = nullptr;
     DeltaWriter::open(&write_req, &delta_writer); 
     ASSERT_NE(delta_writer, nullptr);
 
-    const std::vector<SlotDescriptor*>& slots = tuple_desc->slots();
     Arena arena;
     // Tuple 1
     {

docs/documentation/cn/administrator-guide/operation/tablet-meta-tool.md

@@ -72,7 +72,7 @@ api：
 命令:
 
 ```
-./lib/meta_tool --operation=delete_header --root_path=/path/to/root_path --tablet_id=xxx --schema_hash=xxx`
+./lib/meta_tool --operation=delete_header --root_path=/path/to/root_path --tablet_id=xxx --schema_hash=xxx
 ```
 
 ### 展示 pb 格式的 TabletMeta

docs/documentation/cn/administrator-guide/time-zone.md

@@ -37,7 +37,15 @@ Doris 内部存在多个时区相关参数
 
 包括NOW()或CURTIME()等时间函数显示的值，也包括show load, show backends中的时间值。
 
-但不会影响create table 中时间类型分区列的less than值，也不会影响存储为date/datetime类型的值的显示。
+但不会影响 create table 中时间类型分区列的 less than 值，也不会影响存储为 date/datetime 类型的值的显示。
+
+受时区影响的函数：
+
+* `FROM_UNIXTIME`：给定一个 UTC 时间戳，返回指定时区的日期时间：如 `FROM_UNIXTIME(0)`， 返回 CST 时区：`1970-01-01 08:00:00`。
+* `UNIX_TIMESTAMP`：给定一个指定时区日期时间，返回 UTC 时间戳：如 CST 时区 `UNIX_TIMESTAMP('1970-01-01 08:00:00')`，返回 `0`。
+* `CURTIME`：返回指定时区时间。
+* `NOW`：返指定地时区日期时间。
+* `CONVERT_TZ`：将一个日期时间从一个指定时区转换到另一个指定时区。
 
 ## 使用限制
 
@@ -51,3 +59,7 @@ Doris 内部存在多个时区相关参数
 
 * 为了兼容Doris，支持CST缩写时区，内部会将CST转移为"Asia/Shanghai"的中国标准时区
 
+## 时区格式列表
+
+[List of tz database time zones](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones)
+

docs/documentation/cn/sql-reference/sql-statements/Data Manipulation/BROKER LOAD.md

@@ -133,7 +133,8 @@
         max_filter_ratio：最大容忍可过滤（数据不规范等原因）的数据比例。默认零容忍。
         exec_mem_limit:   设置导入使用的内存上限。默认为2G，单位字节。这里是指单个 BE 节点的内存上限。
                           一个导入可能分布于多个BE。我们假设 1GB 数据在单个节点处理需要最大5GB内存。那么假设1GB文件分布在2个节点处理，那么理论上，每个节点需要内存为2.5GB。则该参数可以设置为 2684354560，即2.5GB
-	strict mode：     是否对数据进行严格限制。默认为true。
+        strict mode：     是否对数据进行严格限制。默认为true。
+        timezone:         指定某些受时区影响的函数的时区，如 strftime/alignment_timestamp/from_unixtime 等等，具体请查阅 [时区] 文档。如果不指定，则使用 "Asia/Shanghai" 时区。
 
     5. 导入数据格式样例
 

docs/documentation/cn/sql-reference/sql-statements/Data Manipulation/ROUTINE LOAD.md

@@ -258,5 +258,6 @@
         );
 
 ## keyword
+
     CREATE,ROUTINE,LOAD