[improvement](lateral-view) Add number rows filtered in profile (#8251)

Add `RowsFiltered` counter in TableFunctionNode profile.
So that we can know the total number of rows that TableFunctionNode processed

be/src/exec/table_function_node.cpp

@@ -81,7 +81,9 @@ Status TableFunctionNode::_prepare_output_slot_ids(const TPlanNode& tnode) {
 
 Status TableFunctionNode::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(ExecNode::prepare(state));
-    
+
+    _num_rows_filtered_counter = ADD_COUNTER(_runtime_profile, "RowsFiltered", TUnit::UNIT);
+ 
     RETURN_IF_ERROR(Expr::prepare(_fn_ctxs, state, _row_descriptor, expr_mem_tracker()));
     for (auto fn : _fns) {
         RETURN_IF_ERROR(fn->prepare());
@@ -303,6 +305,7 @@ Status TableFunctionNode::get_next(RuntimeState* state, RowBatch* row_batch, boo
                 ++_num_rows_returned;
             } else {
                 tuple_ptr = pre_tuple_ptr;
+                ++_num_rows_filtered;
             }
 
             // Forward after write success.
@@ -340,6 +343,9 @@ Status TableFunctionNode::close(RuntimeState* state) {
     }
     RETURN_IF_ERROR(exec_debug_action(TExecNodePhase::CLOSE));
     Expr::close(_fn_ctxs, state);
+
+    COUNTER_SET(_num_rows_filtered_counter, static_cast<int64_t>(_num_rows_filtered));
+
     return ExecNode::close(state);
 }
 

be/src/exec/table_function_node.h

@@ -74,6 +74,9 @@ private:
     std::vector<int> _child_slot_sizes;
     // indicate if child node reach the end
     bool _child_eos = false;
+
+    RuntimeProfile::Counter* _num_rows_filtered_counter = nullptr;
+    uint64_t _num_rows_filtered = 0;
 };
 
 }; // namespace doris

be/src/exprs/table_function/explode_bitmap.cpp

@@ -22,7 +22,9 @@
 
 namespace doris {
 
-ExplodeBitmapTableFunction::ExplodeBitmapTableFunction() {}
+ExplodeBitmapTableFunction::ExplodeBitmapTableFunction() {
+    _fn_name = "explode_bitmap";
+}
 
 ExplodeBitmapTableFunction::~ExplodeBitmapTableFunction() {
     if (_cur_iter != nullptr) {

be/src/exprs/table_function/explode_json_array.cpp

@@ -128,6 +128,20 @@ int ParsedData::set_output(ExplodeJsonArrayType type, rapidjson::Document& docum
 ExplodeJsonArrayTableFunction::ExplodeJsonArrayTableFunction(ExplodeJsonArrayType type)
     : _type(type) {
     
+    switch (type) {
+        case ExplodeJsonArrayType::INT:
+            _fn_name = "explode_json_array_int";
+            break;
+        case ExplodeJsonArrayType::DOUBLE:
+            _fn_name = "explode_json_array_double";
+            break;
+        case ExplodeJsonArrayType::STRING:
+            _fn_name = "explode_json_array_string";
+            break;
+        default:
+            _fn_name = "unknown";
+            break;
+    }
 }
 
 ExplodeJsonArrayTableFunction::~ExplodeJsonArrayTableFunction() {

be/src/exprs/table_function/explode_split.cpp

@@ -24,6 +24,7 @@
 namespace doris {
 
 ExplodeSplitTableFunction::ExplodeSplitTableFunction() {
+    _fn_name = "explode_split";
 }
 
 ExplodeSplitTableFunction::~ExplodeSplitTableFunction() {

be/src/exprs/table_function/table_function.h

@@ -42,6 +42,7 @@ public:
     virtual Status forward(bool *eos) = 0;
 
 public:
+    std::string name() const { return _fn_name; }
     bool eos() const { return _eos; }
 
     void set_expr_context(ExprContext* expr_context) {