YT-25894: Add '$remaining_row_bytes' skiff column

```
* Changelog entry
Type: feature
Component: proxy

Introduce a special skiff-column '$remaining_row_bytes'.
```
commit_hash:b3f817454e13a852fd35e17dc59f237258b8a0b4

1 files changed, 48 insertions, 7 deletions

diff --git a/library/cpp/skiff/skiff.cpp b/library/cpp/skiff/skiff.cpp
index 9b42628cc62..a24d2ec4a52 100644
--- a/library/cpp/skiff/skiff.cpp
+++ b/library/cpp/skiff/skiff.cpp
@@ -377,12 +377,14 @@ ui64 TCheckedSkiffParser::GetReadBytesCount() const
 ////////////////////////////////////////////////////////////////////////////////
 
 TUncheckedSkiffWriter::TUncheckedSkiffWriter(IZeroCopyOutput* underlying)
-    : Underlying_(underlying)
+    : UnderlyingOutputWriter_(underlying)
+    , CurrentOutputWriter_(&UnderlyingOutputWriter_)
 { }
 
 TUncheckedSkiffWriter::TUncheckedSkiffWriter(IOutputStream* underlying)
     : BufferedOutput_(MakeHolder<TBufferedOutput>(underlying))
-    , Underlying_(BufferedOutput_.Get())
+    , UnderlyingOutputWriter_(BufferedOutput_.Get())
+    , CurrentOutputWriter_(&UnderlyingOutputWriter_)
 { }
 
 TUncheckedSkiffWriter::TUncheckedSkiffWriter(const std::shared_ptr<TSkiffSchema>& /*schema*/, IZeroCopyOutput* underlying)
@@ -480,13 +482,13 @@ void TUncheckedSkiffWriter::WriteBoolean(bool value)
 void TUncheckedSkiffWriter::WriteString32(TStringBuf value)
 {
     WriteSimple<ui32>(value.size());
-    Underlying_.Write(value.data(), value.size());
+    CurrentOutputWriter_->Write(value.data(), value.size());
 }
 
 void TUncheckedSkiffWriter::WriteYson32(TStringBuf value)
 {
     WriteSimple<ui32>(value.size());
-    Underlying_.Write(value.data(), value.size());
+    CurrentOutputWriter_->Write(value.data(), value.size());
 }
 
 void TUncheckedSkiffWriter::WriteVariant8Tag(ui8 tag)
@@ -499,9 +501,37 @@ void TUncheckedSkiffWriter::WriteVariant16Tag(ui16 tag)
     WriteSimple<ui16>(tag);
 }
 
+void TUncheckedSkiffWriter::StartBlob()
+{
+    if (BlobOutputWriter_) {
+        throw TSkiffException() << "Blob start called before previous blob was finished";
+    }
+    BlobOutput_.emplace(Blob_);
+    BlobOutputWriter_.emplace(&*BlobOutput_);
+
+    CurrentOutputWriter_ = &*BlobOutputWriter_;
+}
+
+void TUncheckedSkiffWriter::FinishBlob()
+{
+    if (!BlobOutput_) {
+        throw TSkiffException() << "Blob finish called before blob was started";
+    }
+
+    BlobOutputWriter_->UndoRemaining();
+
+    BlobOutput_.reset();
+    BlobOutputWriter_.reset();
+
+    CurrentOutputWriter_ = &UnderlyingOutputWriter_;
+
+    WriteString32(Blob_);
+    Blob_.clear();
+}
+
 void TUncheckedSkiffWriter::Flush()
 {
-    Underlying_.UndoRemaining();
+    UnderlyingOutputWriter_.UndoRemaining();
     if (BufferedOutput_) {
         BufferedOutput_->Flush();
     }
@@ -512,9 +542,9 @@ Y_FORCE_INLINE void TUncheckedSkiffWriter::WriteSimple(T value)
 {
     if constexpr (std::is_integral_v<T>) {
         value = HostToLittle(value);
-        Underlying_.Write(&value, sizeof(T));
+        CurrentOutputWriter_->Write(&value, sizeof(T));
     } else {
-        Underlying_.Write(&value, sizeof(T));
+        CurrentOutputWriter_->Write(&value, sizeof(T));
     }
 }
 
@@ -645,6 +675,17 @@ void TCheckedSkiffWriter::WriteVariant16Tag(ui16 tag)
     Writer_.WriteVariant16Tag(tag);
 }
 
+void TCheckedSkiffWriter::StartBlob()
+{
+    Validator_->OnSimpleType(EWireType::Int32);
+    Writer_.StartBlob();
+}
+
+void TCheckedSkiffWriter::FinishBlob()
+{
+    Writer_.FinishBlob();
+}
+
 void TCheckedSkiffWriter::Flush()
 {
     Writer_.Flush();