[improvement](be,fe) Address adaptive batch size review comments

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #61535

Problem Summary: Address review comments from velodb-robyang on the adaptive
batch size PR. Changes include:

1. Move block_budget.h from common/ to util/ and update 8 include paths
2. Add doc comments to block.h explaining bytes()/allocated_bytes()/columns_byte_size()
3. Remove unused default parameter values from join_hash_table.h and vdata_stream_recvr.h
4. Add per-operator output block byte profile counters (OutputBlockBytes,
   MaxOutputBlockBytes, MinOutputBlockBytes) in the operator base class
5. Add comment to new_json_reader.h explaining row-based reader batch size
6. Fix preferred_block_size_bytes() to return 8MB default instead of 0;
   simplify block_max_rows() condition; add relationship documentation
7. Add bounds checking for session variables: preferredBlockSizeBytes [1MB,512MB],
   preferredBlockSizeRows [1,1000000], preferredMaxColumnInBlockSizeBytes [256KB,128MB]
8. Simplify file_scanner.cpp adaptive batch size condition

### Release note

Added per-operator output block byte size profile counters (OutputBlockBytes,
MaxOutputBlockBytes, MinOutputBlockBytes) for better observability of block
sizes flowing through the query pipeline. Added bounds checking for adaptive
batch size session variables.

### Check List (For Author)

- Test: Regression test / Unit Test / Manual test
    - FE UT: JniScannerTest#testSetBatchSize passed
    - BE+FE build succeeded
- Behavior changed: No
- Does this need documentation: No

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: mrhhsg

be/src/core/block/block.h

@@ -183,12 +183,17 @@ class Block {
 
     Status check_type_and_column() const;
 
-    /// Approximate number of bytes in memory - for profiling and limits.
+    /// Approximate number of bytes used by column data in memory.
+    /// This reflects the actual data footprint (e.g. string contents, numeric arrays)
+    /// and is the metric used by adaptive batch size byte budgets.
     size_t bytes() const;
 
+    /// Returns per-column byte sizes as a comma-separated string (for debugging).
     std::string columns_bytes() const;
 
-    /// Approximate number of allocated bytes in memory - for profiling and limits.
+    /// Approximate number of allocated (reserved) bytes in memory.
+    /// This may be larger than bytes() due to pre-allocated capacity in vectors/arenas.
+    /// Used for memory tracking and profiling.
     MOCK_FUNCTION size_t allocated_bytes() const;
 
     /** Get a list of column names separated by commas. */
@@ -353,6 +358,8 @@ class Block {
                                    bool need_keep_first);
 
     // Helper: sum byte_size() of all mutable columns.
+    // Unlike Block::bytes() which operates on immutable ColumnPtr,
+    // this works on MutableColumns during block construction (e.g. in BlockReader).
     static inline size_t columns_byte_size(const MutableColumns& cols) {
         size_t total = 0;
         for (const auto& col : cols) {

be/src/exec/common/hash_table/join_hash_table.h

@@ -107,7 +107,7 @@ class JoinHashTable {
                     uint32_t* __restrict probe_idxs, bool& probe_visited,
                     uint32_t* __restrict build_idxs, const uint8_t* null_map,
                     bool with_other_conjuncts, bool is_mark_join, bool has_mark_join_conjunct,
-                    int batch_size_limit = 0) {
+                    int batch_size_limit) {
         const int effective_batch_size = batch_size_limit > 0 ? batch_size_limit : max_batch_size;
 
         if ((JoinOpType == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN ||

be/src/exec/exchange/local_exchanger.cpp

@@ -150,8 +150,8 @@ Status ShuffleExchanger::get_block(RuntimeState* state, Block* block, bool* eos,
     PartitionedBlock partitioned_block;
     MutableBlock mutable_block;
 
-    const auto max_batch_size = state->block_max_rows();
-    const auto max_block_bytes = state->block_max_bytes();
+    const auto max_batch_size = state->batch_size();
+    const auto max_block_bytes = state->preferred_block_size_bytes();
 
     auto get_data = [&]() -> Status {
         do {

be/src/exec/exchange/vdata_stream_recvr.h

@@ -79,7 +79,7 @@ class VDataStreamRecvr : public HasTaskExecutionCtx {
     MOCK_FUNCTION Status create_merger(const VExprContextSPtrs& ordering_expr,
                                        const std::vector<bool>& is_asc_order,
                                        const std::vector<bool>& nulls_first, size_t batch_size,
-                                       int64_t limit, size_t offset, size_t block_max_bytes = 0);
+                                       int64_t limit, size_t offset, size_t block_max_bytes);
 
     std::vector<SenderQueue*> sender_queues() const { return _sender_queues; }
 

be/src/exec/exchange/vdata_stream_sender.cpp

@@ -291,8 +291,8 @@ Status Channel::close(RuntimeState* state) {
 BlockSerializer::BlockSerializer(ExchangeSinkLocalState* parent, bool is_local)
         : _parent(parent),
           _is_local(is_local),
-          _batch_size(parent->state()->block_max_rows()),
-          _block_max_bytes(parent->state()->block_max_bytes()) {}
+          _batch_size(parent->state()->batch_size()),
+          _block_max_bytes(parent->state()->preferred_block_size_bytes()) {}
 
 Status BlockSerializer::next_serialized_block(Block* block, PBlock* dest, size_t num_receivers,
                                               bool* serialized, bool eos, const uint32_t* data,
@@ -315,7 +315,7 @@ Status BlockSerializer::next_serialized_block(Block* block, PBlock* dest, size_t
 
     if (_mutable_block->rows() >= _batch_size || eos ||
         (_mutable_block->rows() > 0 && _mutable_block->allocated_bytes() > _buffer_mem_limit) ||
-        (_block_max_bytes > 0 && _mutable_block->bytes() >= _block_max_bytes)) {
+        _mutable_block->bytes() >= _block_max_bytes) {
         if (!_is_local) {
             RETURN_IF_ERROR(_serialize_block(dest, num_receivers));
         }

be/src/exec/operator/aggregation_sink_operator.cpp

@@ -814,7 +814,7 @@ size_t AggSinkLocalState::get_reserve_mem_size(RuntimeState* state, bool eos) co
                 if constexpr (std::is_same_v<HashTableCtxType, std::monostate>) {
                     return 0;
                 } else {
-                    return arg.hash_table->estimate_memory(state->block_max_rows());
+                    return arg.hash_table->estimate_memory(state->batch_size());
                 }
             },
             _agg_data->method_variant);

be/src/exec/operator/aggregation_source_operator.cpp

@@ -20,7 +20,7 @@
 #include <memory>
 #include <string>
 
-#include "common/block_budget.h"
+#include "util/block_budget.h"
 #include "common/exception.h"
 #include "core/column/column_fixed_length_object.h"
 #include "exec/operator/operator.h"
@@ -123,7 +123,7 @@ Status AggLocalState::_get_results_with_serialized_key(RuntimeState* state, Bloc
     }
 
     // Compute effective max rows based on estimated bytes per row.
-    const BlockBudget budget(state->block_max_rows(), state->block_max_bytes());
+    const BlockBudget budget(state->batch_size(), state->preferred_block_size_bytes());
     const size_t effective_max_rows = budget.effective_max_rows(_estimated_row_bytes);
 
     std::visit(
@@ -319,7 +319,7 @@ Status AggLocalState::_get_with_serialized_key_result(RuntimeState* state, Block
     SCOPED_TIMER(_get_results_timer);
 
     // Compute effective max rows based on estimated bytes per row.
-    const BlockBudget budget(state->block_max_rows(), state->block_max_bytes());
+    const BlockBudget budget(state->batch_size(), state->preferred_block_size_bytes());
     const size_t effective_max_rows = budget.effective_max_rows(_estimated_row_bytes);
 
     std::visit(

be/src/exec/operator/distinct_streaming_aggregation_operator.cpp

@@ -22,7 +22,7 @@
 #include <memory>
 #include <utility>
 
-#include "common/block_budget.h"
+#include "util/block_budget.h"
 #include "common/compiler_util.h" // IWYU pragma: keep
 #include "exec/operator/streaming_agg_min_reduction.h"
 #include "exprs/vectorized_agg_fn.h"
@@ -38,8 +38,8 @@ namespace doris {
 DistinctStreamingAggLocalState::DistinctStreamingAggLocalState(RuntimeState* state,
                                                                OperatorXBase* parent)
         : PipelineXLocalState<FakeSharedState>(state, parent),
-          batch_size(state->block_max_rows()),
-          block_max_bytes(state->block_max_bytes()),
+          batch_size(state->batch_size()),
+          block_max_bytes(state->preferred_block_size_bytes()),
           _agg_data(std::make_unique<DistinctDataVariants>()),
           _child_block(Block::create_unique()),
           _aggregated_block(Block::create_unique()),
@@ -397,7 +397,7 @@ Status DistinctStreamingAggOperatorX::pull(RuntimeState* state, Block* block, bo
 
 bool DistinctStreamingAggOperatorX::need_more_input_data(RuntimeState* state) const {
     auto& local_state = get_local_state(state);
-    const BlockBudget budget(state->block_max_rows(), state->block_max_bytes());
+    const BlockBudget budget(state->batch_size(), state->preferred_block_size_bytes());
     const bool need_batch = local_state._stop_emplace_flag
                                     ? local_state._aggregated_block->empty()
                                     : budget.within_budget(local_state._aggregated_block->rows(),

be/src/exec/operator/exchange_source_operator.cpp

@@ -155,7 +155,7 @@ Status ExchangeSourceOperatorX::get_block(RuntimeState* state, Block* block, boo
         SCOPED_TIMER(local_state.create_merger_timer);
         RETURN_IF_ERROR(local_state.stream_recvr->create_merger(
                 local_state.vsort_exec_exprs.ordering_expr_ctxs(), _is_asc_order, _nulls_first,
-                state->block_max_rows(), _limit, _offset, state->block_max_bytes()));
+                state->batch_size(), _limit, _offset, state->preferred_block_size_bytes()));
         local_state.is_ready = true;
         return Status::OK();
     }

be/src/exec/operator/hashjoin_build_sink.cpp

@@ -139,15 +139,15 @@ size_t HashJoinBuildSinkLocalState::get_reserve_mem_size(RuntimeState* state, bo
         const auto bytes = _build_side_mutable_block.bytes();
         const auto allocated_bytes = _build_side_mutable_block.allocated_bytes();
         const auto bytes_per_row = bytes / build_block_rows;
-        const auto estimated_size_of_next_block = bytes_per_row * state->block_max_rows();
+        const auto estimated_size_of_next_block = bytes_per_row * state->batch_size();
         // If the new size is greater than 85% of allocalted bytes, it maybe need to realloc.
         if (((estimated_size_of_next_block + bytes) * 100 / allocated_bytes) >= 85) {
             size_to_reserve += static_cast<size_t>(static_cast<double>(allocated_bytes) * 1.15);
         }
     }
 
     if (eos) {
-        const size_t rows = build_block_rows + state->block_max_rows();
+        const size_t rows = build_block_rows + state->batch_size();
         const auto bucket_size = hash_join_table_calc_bucket_size(rows);
 
         size_to_reserve += bucket_size * sizeof(uint32_t); // JoinHashTable::first
@@ -606,7 +606,7 @@ Status HashJoinBuildSinkLocalState::process_build_block(RuntimeState* state, Blo
                           using HashTableCtxType = std::decay_t<decltype(arg)>;
                           using JoinOpType = std::decay_t<decltype(join_op)>;
                           ProcessHashTableBuild<HashTableCtxType> hash_table_build_process(
-                                  rows, raw_ptrs, this, state->block_max_rows(), state);
+                                  rows, raw_ptrs, this, state->batch_size(), state);
                           auto st = hash_table_build_process.template run<JoinOpType::value>(
                                   arg, null_map_val ? &null_map_val->get_data() : nullptr,
                                   &_shared_state->_has_null_in_build_side,