Allow sending/reading values larger than 16MB to RPC proxy via wire protocol to/from methods dealing with static tables

Historically, it data written through RPC proxies had pass dynamic table row validation standards, like max 16MB value size. This is inconvenient for static tables.
SPYT needs to write large values – ytsaurus/ytsaurus-spyt#43.
Disable wire input Max*ValueLength validation on RPC proxies.
The methods affected by this change are WriteTable and ReadShuffleData in rpc proxy server and TTableReader in rpc proxy client.

* Changelog entry
Type: feature
Component: proxy

Allow sending/reading values larger than 16MB to RPC proxy via wire protocol to/from methods dealing with static tables

---

Pull Request resolved: ytsaurus/ytsaurus#1019
commit_hash:264beb456994ba459ef06d8f5c25bf6d52abc08b

Author: faucct

yt/yt/client/api/rpc_proxy/row_batch_reader.cpp

@@ -18,7 +18,7 @@ TRowBatchReader::TRowBatchReader(
     IAsyncZeroCopyInputStreamPtr underlying,
     bool isStreamWithStatistics)
     : Underlying_(std::move(underlying))
-    , Decoder_(CreateWireRowStreamDecoder(NameTable_))
+    , Decoder_(CreateWireRowStreamDecoder(NameTable_, CreateUnlimitedWireProtocolOptions()))
     , IsStreamWithStatistics_(isStreamWithStatistics)
 {
     YT_VERIFY(Underlying_);

yt/yt/client/api/rpc_proxy/wire_row_stream.cpp

@@ -74,8 +74,11 @@ class TWireRowStreamDecoder
     : public IRowStreamDecoder
 {
 public:
-    explicit TWireRowStreamDecoder(TNameTablePtr nameTable)
+    explicit TWireRowStreamDecoder(
+        TNameTablePtr nameTable,
+        TWireProtocolOptions wireProtocolOptions = {})
         : NameTable_(std::move(nameTable))
+        , WireProtocolOptions_(std::move(wireProtocolOptions))
     {
         Descriptor_.set_wire_format_version(NApi::NRpcProxy::CurrentWireFormatVersion);
         Descriptor_.set_rowset_kind(NApi::NRpcProxy::NProto::RK_UNVERSIONED);
@@ -86,7 +89,7 @@ class TWireRowStreamDecoder
         const NProto::TRowsetDescriptor& descriptorDelta) override
     {
         struct TWireRowStreamDecoderTag { };
-        auto reader = CreateWireProtocolReader(payloadRef, New<TRowBuffer>(TWireRowStreamDecoderTag()));
+        auto reader = CreateWireProtocolReader(payloadRef, New<TRowBuffer>(TWireRowStreamDecoderTag()), WireProtocolOptions_);
         auto rows = reader->ReadUnversionedRowset(true);
 
         auto oldNameTableSize = Descriptor_.name_table_entries_size();
@@ -125,18 +128,20 @@ class TWireRowStreamDecoder
 
 private:
     const TNameTablePtr NameTable_;
+    const TWireProtocolOptions WireProtocolOptions_;
 
     NApi::NRpcProxy::NProto::TRowsetDescriptor Descriptor_;
     TNameTableToSchemaIdMapping IdMapping_;
     bool HasNontrivialIdMapping_ = false;
 };
 
-IRowStreamDecoderPtr CreateWireRowStreamDecoder(TNameTablePtr nameTable)
+IRowStreamDecoderPtr CreateWireRowStreamDecoder(
+    TNameTablePtr nameTable,
+    TWireProtocolOptions wireProtocolOptions)
 {
-    return New<TWireRowStreamDecoder>(std::move(nameTable));
+    return New<TWireRowStreamDecoder>(std::move(nameTable), std::move(wireProtocolOptions));
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 
 } // namespace NYT::NApi::NRpcProxy
-

yt/yt/client/api/rpc_proxy/wire_row_stream.h

@@ -3,13 +3,16 @@
 #include "public.h"
 
 #include <yt/yt/client/table_client/public.h>
+#include <yt/yt/client/table_client/wire_protocol.h>
 
 namespace NYT::NApi::NRpcProxy {
 
 ////////////////////////////////////////////////////////////////////////////////
 
 IRowStreamEncoderPtr CreateWireRowStreamEncoder(NTableClient::TNameTablePtr nameTable);
-IRowStreamDecoderPtr CreateWireRowStreamDecoder(NTableClient::TNameTablePtr nameTable);
+IRowStreamDecoderPtr CreateWireRowStreamDecoder(
+    NTableClient::TNameTablePtr nameTable,
+    NTableClient::TWireProtocolOptions wireProtocolOptions = {});
 
 ////////////////////////////////////////////////////////////////////////////////
 

yt/yt/client/table_client/wire_protocol.cpp

@@ -1055,6 +1055,17 @@ auto IWireProtocolReader::GetSchemaData(const TTableSchema& schema) -> TSchemaDa
 
 ////////////////////////////////////////////////////////////////////////////////
 
+TWireProtocolOptions CreateUnlimitedWireProtocolOptions()
+{
+    return {
+        .MaxStringValueLength = std::numeric_limits<i64>::max(),
+        .MaxAnyValueLength = std::numeric_limits<i64>::max(),
+        .MaxCompositeValueLength = std::numeric_limits<i64>::max(),
+    };
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 std::unique_ptr<IWireProtocolReader> CreateWireProtocolReader(TSharedRef data, TRowBufferPtr rowBuffer, TWireProtocolOptions options)
 {
     return std::make_unique<TWireProtocolReader>(std::move(data), std::move(rowBuffer), std::move(options));

yt/yt/client/table_client/wire_protocol.h

@@ -288,6 +288,8 @@ struct TWireProtocolOptions
     i64 MaxVersionedRowDataWeight = NTableClient::MaxServerVersionedRowDataWeight;
 };
 
+TWireProtocolOptions CreateUnlimitedWireProtocolOptions();
+
 ////////////////////////////////////////////////////////////////////////////////
 
 //! Creates wire protocol reader.