YQ-4221 fixed s3 write with large path (#16902)

Author: GrigoriyPA

ydb/core/kqp/proxy_service/kqp_script_executions.cpp

@@ -70,6 +70,30 @@ NYql::TIssues DeserializeIssues(const TString& issuesSerialized) {
     return issues;
 }
 
+template <typename TProto>
+void SerializeBinaryProto(const TProto& proto, NJson::TJsonValue& value) {
+    value.SetType(NJson::EJsonValueType::JSON_MAP);
+
+    const auto config = NProtobufJson::TProto2JsonConfig()
+        .AddStringTransform(MakeIntrusive<NProtobufJson::TBase64EncodeBytesTransform>());
+
+    NProtobufJson::Proto2Json(proto, value["encoded_proto"], config);
+}
+
+template <typename TProto>
+void DeserializeBinaryProto(const NJson::TJsonValue& value, TProto& proto) {
+    const auto& valueMap = value.GetMap();
+    const auto encodedProto = valueMap.find("encoded_proto");
+    if (encodedProto == valueMap.end()) {
+        return NProtobufJson::Json2Proto(value, proto, NProtobufJson::TJson2ProtoConfig());
+    }
+
+    const auto config = NProtobufJson::TJson2ProtoConfig()
+        .AddStringTransform(MakeIntrusive<NProtobufJson::TBase64DecodeBytesTransform>());
+
+    NProtobufJson::Json2Proto(encodedProto->second, proto, config);
+}
+
 
 class TQueryBase : public NKikimr::TQueryBase {
 public:
@@ -2272,7 +2296,7 @@ class TSaveScriptExternalEffectActor : public TQueryBase {
         NJson::TJsonValue::TArray& jsonArray = value.GetArraySafe();
         jsonArray.resize(sinks.size());
         for (size_t i = 0; i < sinks.size(); ++i) {
-            NProtobufJson::Proto2Json(sinks[i], jsonArray[i], NProtobufJson::TProto2JsonConfig());
+            SerializeBinaryProto(sinks[i], jsonArray[i]);
         }
 
         NJsonWriter::TBuf serializedSinks;
@@ -2416,7 +2440,7 @@ class TSaveScriptFinalStatusActor : public TQueryBase {
                 value.GetValuePointer(i, &serializedSink);
 
                 NKqpProto::TKqpExternalSink sink;
-                NProtobufJson::Json2Proto(*serializedSink, sink);
+                DeserializeBinaryProto(*serializedSink, sink);
                 Response->Sinks.push_back(sink);
             }
         }

ydb/core/kqp/ut/federated_query/s3/kqp_federated_query_ut.cpp

@@ -785,7 +785,10 @@ Y_UNIT_TEST_SUITE(KqpFederatedQuery) {
     Y_UNIT_TEST(InsertIntoBucketWithSelect) {
         const TString writeDataSourceName = "/Root/write_data_source";
         const TString writeBucket = "test_bucket_write_with_select";
-        const TString writeObject = "test_object_write/";
+
+        // Also tests large object path with size >= 128 
+        // for atomic upload commit case
+        const TString writeObject = TStringBuilder() << "test_object_write/" << TString(512, 'x') << "/";
 
         {
             Aws::S3::S3Client s3Client = MakeS3Client();
@@ -811,6 +814,8 @@ Y_UNIT_TEST_SUITE(KqpFederatedQuery) {
         UNIT_ASSERT_C(result.GetStatus() == NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
 
         const TString sql = fmt::format(R"(
+                PRAGMA s3.AtomicUploadCommit = "true";
+
                 INSERT INTO `{write_source}`.`{write_object}` WITH (FORMAT = "csv_with_names")
                 SELECT * FROM AS_TABLE([<|id: 0, payload: "#######"|>]);