Add validate for rowset writer (#1104)

Author: kangpinghuang

be/src/olap/rowset/alpha_rowset_writer.cpp

@@ -151,7 +151,7 @@ OLAPStatus AlphaRowsetWriter::flush() {
 
 RowsetSharedPtr AlphaRowsetWriter::build() {
     if (_writer_state != WRITER_FLUSHED) {
-        LOG(WARNING) << "invalid writer state before build:" << _writer_state;
+        LOG(WARNING) << "invalid writer state before build, state:" << _writer_state;
         return nullptr;
     }
     for (auto& segment_group : _segment_groups) {
@@ -200,6 +200,13 @@ RowsetSharedPtr AlphaRowsetWriter::build() {
     _current_rowset_meta->set_num_rows(_num_rows_written);
     _current_rowset_meta->set_creation_time(time(nullptr));
 
+    // validate rowset arguments before create rowset
+    bool ret = _validate_rowset();
+    if (!ret) {
+        LOG(WARNING) << "valiate rowset arguments failed";
+        return nullptr;
+    }
+
     RowsetSharedPtr rowset(new(std::nothrow) AlphaRowset(_rowset_writer_context.tablet_schema,
                                     _rowset_writer_context.rowset_path_prefix,
                                     _rowset_writer_context.data_dir, _current_rowset_meta));
@@ -289,4 +296,18 @@ OLAPStatus AlphaRowsetWriter::_init() {
     return OLAP_SUCCESS;
 }
 
+bool AlphaRowsetWriter::_validate_rowset() {
+    if (_is_pending_rowset) {
+        int64_t partition_id = _current_rowset_meta->partition_id();
+        if (partition_id <= 0) {
+            LOG(WARNING) << "invalid partition id:" << partition_id << " for pending rowset."
+                         << ", rowset_id:" << _current_rowset_meta->rowset_id()
+                         << ", tablet_id:" << _current_rowset_meta->tablet_id()
+                         << ", schema_hash:" << _current_rowset_meta->tablet_schema_hash();
+            return false;
+        }
+    }
+    return true;
+}
+
 } // namespace doris

be/src/olap/rowset/alpha_rowset_writer.h

@@ -70,6 +70,9 @@ class AlphaRowsetWriter : public RowsetWriter {
 
 private:
     OLAPStatus _init();
+    
+    // validate rowset build arguments before create rowset to make sure correctness
+    bool _validate_rowset();
 
 private:
     int32_t _segment_group_id;

be/src/olap/schema_change.cpp

@@ -498,7 +498,7 @@ OLAPStatus RowBlockAllocator::allocate(RowBlock** row_block,
 
 void RowBlockAllocator::release(RowBlock* row_block) {
     if (row_block == nullptr) {
-        LOG(FATAL) << "null row block released.";
+        LOG(INFO) << "null row block released.";
         return;
     }
 
@@ -811,7 +811,10 @@ bool SchemaChangeDirectly::process(RowsetReaderSharedPtr rowset_reader, RowsetWr
     }
 
 DIRECTLY_PROCESS_ERR:
-    _row_block_allocator->release(new_row_block);
+    if (new_row_block) {
+        _row_block_allocator->release(new_row_block);
+        new_row_block = nullptr;
+    }
     return result;
 }