Merge branch 'master' into feature-explode-json-object

Author: amorynan

be/src/cloud/cloud_meta_mgr.cpp

@@ -839,8 +839,12 @@ Status CloudMetaMgr::abort_txn(const StreamLoadContext& ctx) {
     if (ctx.db_id > 0 && !ctx.label.empty()) {
         req.set_db_id(ctx.db_id);
         req.set_label(ctx.label);
-    } else {
+    } else if (ctx.txn_id > 0) {
         req.set_txn_id(ctx.txn_id);
+    } else {
+        LOG(WARNING) << "failed abort txn, with illegal input, db_id=" << ctx.db_id
+                     << " txn_id=" << ctx.txn_id << " label=" << ctx.label;
+        return Status::InternalError<false>("failed to abort txn");
     }
     return retry_rpc("abort txn", req, &res, &MetaService_Stub::abort_txn);
 }

be/src/cloud/cloud_stream_load_executor.cpp

@@ -26,6 +26,12 @@
 
 namespace doris {
 
+enum class TxnOpParamType : int {
+    ILLEGAL,
+    WITH_TXN_ID,
+    WITH_LABEL,
+};
+
 CloudStreamLoadExecutor::CloudStreamLoadExecutor(ExecEnv* exec_env)
         : StreamLoadExecutor(exec_env) {}
 
@@ -42,13 +48,48 @@ Status CloudStreamLoadExecutor::pre_commit_txn(StreamLoadContext* ctx) {
 }
 
 Status CloudStreamLoadExecutor::operate_txn_2pc(StreamLoadContext* ctx) {
-    VLOG_DEBUG << "operate_txn_2pc, op: " << ctx->txn_operation;
+    std::stringstream ss;
+    ss << "db_id=" << ctx->db_id << " txn_id=" << ctx->txn_id << " label=" << ctx->label
+       << " txn_2pc_op=" << ctx->txn_operation;
+    std::string op_info = ss.str();
+    VLOG_DEBUG << "operate_txn_2pc " << op_info;
+    TxnOpParamType topt = ctx->txn_id > 0       ? TxnOpParamType::WITH_TXN_ID
+                          : !ctx->label.empty() ? TxnOpParamType::WITH_LABEL
+                                                : TxnOpParamType::ILLEGAL;
+
+    Status st = Status::InternalError<false>("impossible branch reached, " + op_info);
+
     if (ctx->txn_operation.compare("commit") == 0) {
-        return _exec_env->storage_engine().to_cloud().meta_mgr().commit_txn(*ctx, true);
+        if (topt == TxnOpParamType::WITH_TXN_ID) {
+            VLOG_DEBUG << "2pc commit stream load txn directly: " << op_info;
+            st = _exec_env->storage_engine().to_cloud().meta_mgr().commit_txn(*ctx, true);
+        } else if (topt == TxnOpParamType::WITH_LABEL) {
+            VLOG_DEBUG << "2pc commit stream load txn with FE support: " << op_info;
+            st = StreamLoadExecutor::operate_txn_2pc(ctx);
+        } else {
+            st = Status::InternalError<false>(
+                    "failed to 2pc commit txn, with TxnOpParamType::illegal input, " + op_info);
+        }
+    } else if (ctx->txn_operation.compare("abort") == 0) {
+        if (topt == TxnOpParamType::WITH_TXN_ID) {
+            LOG(INFO) << "2pc abort stream load txn directly: " << op_info;
+            st = _exec_env->storage_engine().to_cloud().meta_mgr().abort_txn(*ctx);
+            WARN_IF_ERROR(st, "failed to rollback txn " + op_info);
+        } else if (topt == TxnOpParamType::WITH_LABEL) { // maybe a label send to FE to abort
+            VLOG_DEBUG << "2pc abort stream load txn with FE support: " << op_info;
+            StreamLoadExecutor::rollback_txn(ctx);
+            st = Status::OK();
+        } else {
+            st = Status::InternalError<false>("failed abort txn, with illegal input, " + op_info);
+        }
     } else {
-        // 2pc abort
-        return _exec_env->storage_engine().to_cloud().meta_mgr().abort_txn(*ctx);
+        std::string msg =
+                "failed to operate_txn_2pc, unrecognized operation: " + ctx->txn_operation;
+        LOG(WARNING) << msg << " " << op_info;
+        st = Status::InternalError<false>(msg + " " + op_info);
     }
+    WARN_IF_ERROR(st, "failed to operate_txn_2pc " + op_info)
+    return st;
 }
 
 Status CloudStreamLoadExecutor::commit_txn(StreamLoadContext* ctx) {
@@ -85,8 +126,24 @@ Status CloudStreamLoadExecutor::commit_txn(StreamLoadContext* ctx) {
 }
 
 void CloudStreamLoadExecutor::rollback_txn(StreamLoadContext* ctx) {
-    WARN_IF_ERROR(_exec_env->storage_engine().to_cloud().meta_mgr().abort_txn(*ctx),
-                  "Failed to rollback txn");
+    std::stringstream ss;
+    ss << "db_id=" << ctx->db_id << " txn_id=" << ctx->txn_id << " label=" << ctx->label;
+    std::string op_info = ss.str();
+    LOG(INFO) << "rollback stream laod txn " << op_info;
+    TxnOpParamType topt = ctx->txn_id > 0       ? TxnOpParamType::WITH_TXN_ID
+                          : !ctx->label.empty() ? TxnOpParamType::WITH_LABEL
+                                                : TxnOpParamType::ILLEGAL;
+
+    if (topt == TxnOpParamType::WITH_TXN_ID) {
+        VLOG_DEBUG << "abort stream load txn directly: " << op_info;
+        WARN_IF_ERROR(_exec_env->storage_engine().to_cloud().meta_mgr().abort_txn(*ctx),
+                      "failed to rollback txn " + op_info);
+    } else { // maybe a label send to FE to abort
+        // does not care about the return status
+        // ctx->db_id > 0 && !ctx->label.empty()
+        VLOG_DEBUG << "abort stream load txn with FE support: " << op_info;
+        StreamLoadExecutor::rollback_txn(ctx);
+    }
 }
 
 } // namespace doris

be/src/common/config.cpp

@@ -139,7 +139,7 @@ DEFINE_mBool(enable_stacktrace_in_allocator_check_failed, "false");
 
 DEFINE_mInt64(large_memory_check_bytes, "2147483648");
 
-DEFINE_mBool(enable_memory_orphan_check, "true");
+DEFINE_mBool(enable_memory_orphan_check, "false");
 
 // The maximum time a thread waits for full GC. Currently only query will wait for full gc.
 DEFINE_mInt32(thread_wait_gc_max_milliseconds, "1000");
@@ -1068,7 +1068,7 @@ DEFINE_mInt32(schema_cache_sweep_time_sec, "100");
 
 // max number of segment cache, default -1 for backward compatibility fd_number*2/5
 DEFINE_mInt32(segment_cache_capacity, "-1");
-DEFINE_mInt32(estimated_num_columns_per_segment, "30");
+DEFINE_mInt32(estimated_num_columns_per_segment, "200");
 DEFINE_mInt32(estimated_mem_per_column_reader, "1024");
 // The value is calculate by storage_page_cache_limit * index_page_cache_percentage
 DEFINE_mInt32(segment_cache_memory_percentage, "2");
@@ -1313,6 +1313,11 @@ DEFINE_Bool(enable_file_logger, "true");
 // The minimum row group size when exporting Parquet files. default 128MB
 DEFINE_Int64(min_row_group_size, "134217728");
 
+// If set to false, the parquet reader will not use page index to filter data.
+// This is only for debug purpose, in case sometimes the page index
+// filter wrong data.
+DEFINE_mBool(enable_parquet_page_index, "true");
+
 // clang-format off
 #ifdef BE_TEST
 // test s3

be/src/common/config.h

@@ -1399,6 +1399,8 @@ DECLARE_Bool(enable_file_logger);
 // The minimum row group size when exporting Parquet files.
 DECLARE_Int64(min_row_group_size);
 
+DECLARE_mBool(enable_parquet_page_index);
+
 #ifdef BE_TEST
 // test s3
 DECLARE_String(test_s3_resource);

be/src/olap/lru_cache.cpp

@@ -22,6 +22,7 @@ namespace doris {
 
 DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(cache_capacity, MetricUnit::BYTES);
 DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(cache_usage, MetricUnit::BYTES);
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(cache_element_count, MetricUnit::NOUNIT);
 DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(cache_usage_ratio, MetricUnit::NOUNIT);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(cache_lookup_count, MetricUnit::OPERATIONS);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(cache_hit_count, MetricUnit::OPERATIONS);
@@ -542,6 +543,7 @@ ShardedLRUCache::ShardedLRUCache(const std::string& name, size_t total_capacity,
     _entity->register_hook(name, std::bind(&ShardedLRUCache::update_cache_metrics, this));
     INT_GAUGE_METRIC_REGISTER(_entity, cache_capacity);
     INT_GAUGE_METRIC_REGISTER(_entity, cache_usage);
+    INT_GAUGE_METRIC_REGISTER(_entity, cache_element_count);
     INT_DOUBLE_METRIC_REGISTER(_entity, cache_usage_ratio);
     INT_ATOMIC_COUNTER_METRIC_REGISTER(_entity, cache_lookup_count);
     INT_ATOMIC_COUNTER_METRIC_REGISTER(_entity, cache_hit_count);
@@ -640,15 +642,18 @@ void ShardedLRUCache::update_cache_metrics() const {
     size_t total_usage = 0;
     size_t total_lookup_count = 0;
     size_t total_hit_count = 0;
+    size_t total_element_count = 0;
     for (int i = 0; i < _num_shards; i++) {
         total_capacity += _shards[i]->get_capacity();
         total_usage += _shards[i]->get_usage();
         total_lookup_count += _shards[i]->get_lookup_count();
         total_hit_count += _shards[i]->get_hit_count();
+        total_element_count += _shards[i]->get_element_count();
     }
 
     cache_capacity->set_value(total_capacity);
     cache_usage->set_value(total_usage);
+    cache_element_count->set_value(total_element_count);
     cache_lookup_count->set_value(total_lookup_count);
     cache_hit_count->set_value(total_hit_count);
     cache_usage_ratio->set_value(total_capacity == 0 ? 0 : ((double)total_usage / total_capacity));

be/src/olap/lru_cache.h

@@ -60,7 +60,7 @@ enum LRUCacheType {
 };
 
 static constexpr LRUCacheType DEFAULT_LRU_CACHE_TYPE = LRUCacheType::SIZE;
-static constexpr uint32_t DEFAULT_LRU_CACHE_NUM_SHARDS = 16;
+static constexpr uint32_t DEFAULT_LRU_CACHE_NUM_SHARDS = 32;
 static constexpr size_t DEFAULT_LRU_CACHE_ELEMENT_COUNT_CAPACITY = 0;
 
 class CacheKey {
@@ -349,6 +349,7 @@ class LRUCache {
     uint64_t get_hit_count() const { return _hit_count; }
     size_t get_usage() const { return _usage; }
     size_t get_capacity() const { return _capacity; }
+    size_t get_element_count() const { return _table.element_count(); }
 
 private:
     void _lru_remove(LRUHandle* e);
@@ -433,6 +434,7 @@ class ShardedLRUCache : public Cache {
     std::shared_ptr<MetricEntity> _entity;
     IntGauge* cache_capacity = nullptr;
     IntGauge* cache_usage = nullptr;
+    IntGauge* cache_element_count = nullptr;
     DoubleGauge* cache_usage_ratio = nullptr;
     IntAtomicCounter* cache_lookup_count = nullptr;
     IntAtomicCounter* cache_hit_count = nullptr;

be/src/olap/rowset/segment_v2/column_reader.cpp

@@ -128,6 +128,7 @@ Status ColumnReader::create_map(const ColumnReaderOptions& opts, const ColumnMet
                                 const io::FileReaderSPtr& file_reader,
                                 std::unique_ptr<ColumnReader>* reader) {
     // map reader now has 3 sub readers for key, value, offsets(scalar), null(scala)
+    DCHECK(meta.children_columns_size() == 3 || meta.children_columns_size() == 4);
     std::unique_ptr<ColumnReader> key_reader;
     RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(0),
                                          meta.children_columns(0).num_rows(), file_reader,

be/src/olap/rowset/segment_v2/column_writer.cpp

@@ -206,6 +206,11 @@ Status ColumnWriter::create_map_writer(const ColumnWriterOptions& opts, const Ta
                                        io::FileWriter* file_writer,
                                        std::unique_ptr<ColumnWriter>* writer) {
     DCHECK(column->get_subtype_count() == 2);
+    if (column->get_subtype_count() < 2) {
+        return Status::InternalError(
+                "If you upgraded from version 1.2.*, please DROP the MAP columns and then "
+                "ADD the MAP columns back.");
+    }
     // create key & value writer
     std::vector<std::unique_ptr<ColumnWriter>> inner_writer_list;
     for (int i = 0; i < 2; ++i) {
@@ -1141,4 +1146,4 @@ Status MapColumnWriter::write_inverted_index() {
     return Status::OK();
 }
 
-} // namespace doris::segment_v2
+} // namespace doris::segment_v2

be/src/olap/segment_loader.h

@@ -80,10 +80,11 @@ class SegmentCache : public LRUCachePolicyTrackingManual {
         segment_v2::SegmentSharedPtr segment;
     };
 
-    SegmentCache(size_t capacity)
-            : LRUCachePolicyTrackingManual(CachePolicy::CacheType::SEGMENT_CACHE, capacity,
-                                           LRUCacheType::SIZE,
-                                           config::tablet_rowset_stale_sweep_time_sec) {}
+    SegmentCache(size_t memory_bytes_limit, size_t segment_num_limit)
+            : LRUCachePolicyTrackingManual(CachePolicy::CacheType::SEGMENT_CACHE,
+                                           memory_bytes_limit, LRUCacheType::SIZE,
+                                           config::tablet_rowset_stale_sweep_time_sec,
+                                           DEFAULT_LRU_CACHE_NUM_SHARDS * 2, segment_num_limit) {}
 
     // Lookup the given segment in the cache.
     // If the segment is found, the cache entry will be written into handle.
@@ -110,7 +111,9 @@ class SegmentLoader {
     // After the estimation of segment memory usage is provided later, it is recommended
     // to use Memory as the capacity limit of the cache.
 
-    SegmentLoader(size_t capacity) { _segment_cache = std::make_unique<SegmentCache>(capacity); }
+    SegmentLoader(size_t memory_limit_bytes, size_t segment_num_count) {
+        _segment_cache = std::make_unique<SegmentCache>(memory_limit_bytes, segment_num_count);
+    }
 
     // Load segments of "rowset", return the "cache_handle" which contains segments.
     // If use_cache is true, it will be loaded from _cache.

be/src/olap/tablet_schema.cpp

@@ -553,7 +553,8 @@ void TabletColumn::init_from_pb(const ColumnPB& column) {
         CHECK(column.children_columns_size() == 1) << "ARRAY type has more than 1 children types.";
     }
     if (_type == FieldType::OLAP_FIELD_TYPE_MAP) {
-        CHECK(column.children_columns_size() == 2) << "MAP type has more than 2 children types.";
+        DCHECK(column.children_columns_size() == 2) << "MAP type has more than 2 children types.";
+        LOG(WARNING) << "MAP type has more than 2 children types.";
     }
     for (size_t i = 0; i < column.children_columns_size(); i++) {
         TabletColumn child_column;
@@ -623,7 +624,8 @@ void TabletColumn::to_schema_pb(ColumnPB* column) const {
         CHECK(_sub_columns.size() == 1) << "ARRAY type has more than 1 children types.";
     }
     if (_type == FieldType::OLAP_FIELD_TYPE_MAP) {
-        CHECK(_sub_columns.size() == 2) << "MAP type has more than 2 children types.";
+        DCHECK(_sub_columns.size() == 2) << "MAP type has more than 2 children types.";
+        LOG(WARNING) << "MAP type has more than 2 children types.";
     }
 
     for (size_t i = 0; i < _sub_columns.size(); i++) {