[yugabyte#27202] YSQL: Fix analyze for large rows

Summary:
If multiple batched request are sent to multiple nodes in parallel, their results are combined into single response. This single response can exceed size limits.

There are two ways to fix it.
1. set the fetch size limit to ~ `rpc_max_message_size * max_buffer_size_to_rpc_limit_ratio / send_count`
2. modify the batching behaviour

#1 is much simpler, so we go with that approach.

If the request has no size limit (yb_fetch_size_limit = 0) or a size limit greater than the max allowed, set the size limit to `rpc_max_message_size * max_buffer_size_to_rpc_limit_ratio / send_count`.
Jira: DB-16688

Test Plan:
```lang=sql
CREATE TABLE tbl1(col0 TEXT) SPLIT INTO 10 TABLETS;
INSERT INTO tbl1 SELECT repeat('a', 30000) FROM generate_series(0, 30000);
ANALYZE tbl1; -- fails before this diff
```

```lang=sh
./yb_build.sh --cxx-test pg_tablet_split-test --gtest_filter PgManyTabletsSelect.AnalyzeTableWithLargeRows -n 20
```
Note that this test uses a separate test class because RPC_max_message_size is a non-runtime flag, and setting it at runtime can result in flaky failures as some sections of the code are aware of the new limit and other's aren't.

Reviewers: amartsinchyk

Reviewed By: amartsinchyk

Subscribers: yql

Differential Revision: https://phorge.dev.yugabyte.com/D44455

Author: timothy-e

src/yb/yql/pggate/pg_doc_op.cc

@@ -39,6 +39,9 @@
 #include "yb/yql/pggate/util/pg_doc_data.h"
 #include "yb/yql/pggate/util/ybc_util.h"
 
+DECLARE_uint64(rpc_max_message_size);
+DECLARE_double(max_buffer_size_to_rpc_limit_ratio);
+
 namespace yb::pggate {
 namespace {
 
@@ -375,6 +378,26 @@ Status PgDocOp::SendRequestImpl(ForceNonBufferable force_non_bufferable) {
   if (active_op_count_ > 0) {
     // Send at most "parallelism_level_" number of requests at one time.
     size_t send_count = std::min(parallelism_level_, active_op_count_);
+
+    uint64_t max_size = FLAGS_rpc_max_message_size * FLAGS_max_buffer_size_to_rpc_limit_ratio
+                        / send_count;
+
+    for (const auto& op : pgsql_ops_) {
+      if (op->is_active() && op->is_read()) {
+        auto& read_op = down_cast<PgsqlReadOp&>(*op);
+        auto req_size_limit = read_op.read_request().size_limit();
+        if (req_size_limit > 0) {
+          VLOG(2) << "Capping read op at size limit: " << max_size
+                  << " (from " << req_size_limit << ")";
+        }
+
+        // Cap the size limit if the size limit is unset or exceeds the maximum size.
+        if (req_size_limit > max_size || req_size_limit == 0) {
+              read_op.read_request().set_size_limit(max_size);
+        }
+      }
+    }
+
     VLOG(1) << "Number of operations to send: " << send_count;
     response_ = VERIFY_RESULT(sender_(
         pg_session_.get(), pgsql_ops_.data(), send_count, *table_,

src/yb/yql/pgwrapper/pg_tablet_split-test.cc

@@ -73,6 +73,8 @@ DECLARE_int32(ysql_client_read_write_timeout_ms);
 DECLARE_int64(db_block_size_bytes);
 DECLARE_uint64(post_split_compaction_input_size_threshold_bytes);
 DECLARE_string(ysql_pg_conf_csv);
+DECLARE_int32(ysql_select_parallelism);
+DECLARE_uint64(rpc_max_message_size);
 
 DECLARE_bool(TEST_asyncrpc_common_response_check_fail_once);
 DECLARE_bool(TEST_pause_before_full_compaction);
@@ -776,6 +778,43 @@ TEST_F(PgTabletSplitTest, TestMetaCacheLookupsPostSplit) {
   ASSERT_NE(remote_child->tablet_id(), parent_tablet_id);
 }
 
+class PgManyTabletsSelect : public PgTabletSplitTest {
+ protected:
+ protected:
+  void SetUp() override {
+    ANNOTATE_UNPROTECTED_WRITE(FLAGS_ysql_select_parallelism) = 20;
+    ANNOTATE_UNPROTECTED_WRITE(FLAGS_rpc_max_message_size) = 1_MB;
+
+    PgTabletSplitTest::SetUp();
+  }
+};
+
+TEST_F(PgManyTabletsSelect, AnalyzeTableWithLargeRows) {
+  const auto table_name = "big_table";
+  const auto rows = 1000;
+  auto conn = ASSERT_RESULT(Connect());
+
+  ASSERT_OK(conn.ExecuteFormat(
+      "CREATE TABLE $0(wide TEXT) SPLIT INTO $1 TABLETS",
+      table_name, FLAGS_ysql_select_parallelism));
+  ASSERT_OK(conn.ExecuteFormat(
+      "INSERT INTO $0 SELECT repeat('a', 1100) FROM generate_series(1, $1)", table_name, rows));
+  // 1100 bytes per row * 1000 rows ~= 1.1 MB total size.
+
+  ASSERT_OK(conn.ExecuteFormat("SET yb_fetch_row_limit = 0"));
+
+  const auto explain_str = ASSERT_RESULT(conn.FetchAllAsString(Format(
+      "EXPLAIN (ANALYZE, DIST, FORMAT JSON) SELECT * FROM $0 WHERE wide = wide", table_name)));
+  LOG(INFO) << "Explain output: " << explain_str;
+
+  rapidjson::Document explain_json;
+  explain_json.Parse(explain_str.c_str());
+
+  // The response is too large to fit in one request (RPC_MAX_SIZE_LIMIT), so it is split into two
+  ASSERT_EQ(explain_json[0]["Storage Read Requests"].GetInt(), 2);
+  ASSERT_EQ(explain_json[0]["Plan"]["Actual Rows"].GetInt(), rows);
+}
+
 class PgPartitioningVersionTest :
     public PgTabletSplitTest,
     public testing::WithParamInterface<uint32_t> {