fix segfault and memory leaks in CS (#7288)

Co-authored-by: ivanmorozov333 <[email protected]>

Author: zverevgeny

ydb/core/formats/arrow/common/container.cpp

@@ -117,6 +117,11 @@ TGeneralContainer::TGeneralContainer(const std::shared_ptr<arrow::RecordBatch>&
     Initialize();
 }
 
+TGeneralContainer::TGeneralContainer(const ui32 recordsCount)
+    : RecordsCount(recordsCount)
+    , Schema(std::make_shared<NModifier::TSchema>()) {
+}
+
 std::shared_ptr<NKikimr::NArrow::NAccessor::IChunkedArray> TGeneralContainer::GetAccessorByNameVerified(const std::string& fieldId) const {
     auto result = GetAccessorByNameOptional(fieldId);
     AFL_VERIFY(result)("event", "cannot_find_accessor_in_general_container")("field_id", fieldId)("schema", Schema->ToString());

ydb/core/formats/arrow/common/container.h

@@ -29,6 +29,13 @@ class TGeneralContainer {
     std::vector<std::shared_ptr<NAccessor::IChunkedArray>> Columns;
     void Initialize();
 public:
+    TGeneralContainer(const ui32 recordsCount);
+
+    ui32 GetRecordsCountVerified() const {
+        AFL_VERIFY(RecordsCount);
+        return *RecordsCount;
+    }
+
     TString DebugString() const;
 
     [[nodiscard]] TConclusionStatus SyncSchemaTo(const std::shared_ptr<arrow::Schema>& schema,
@@ -67,7 +74,6 @@ class TGeneralContainer {
 
     [[nodiscard]] TConclusionStatus AddField(const std::shared_ptr<arrow::Field>& f, const std::shared_ptr<arrow::ChunkedArray>& data);
 
-    TGeneralContainer() = default;
     TGeneralContainer(const std::shared_ptr<arrow::Table>& table);
     TGeneralContainer(const std::shared_ptr<arrow::RecordBatch>& table);
     TGeneralContainer(const std::shared_ptr<arrow::Schema>& schema, std::vector<std::shared_ptr<NAccessor::IChunkedArray>>&& columns);

ydb/core/tx/columnshard/engines/scheme/versions/abstract_scheme.cpp

@@ -35,7 +35,7 @@ TConclusion<std::shared_ptr<NArrow::TGeneralContainer>> ISnapshotSchema::Normali
     AFL_VERIFY(dataSchema.GetSnapshot() < GetSnapshot());
     const std::shared_ptr<arrow::Schema>& resultArrowSchema = GetSchema();
 
-    std::shared_ptr<NArrow::TGeneralContainer> result = std::make_shared<NArrow::TGeneralContainer>();
+    std::shared_ptr<NArrow::TGeneralContainer> result = std::make_shared<NArrow::TGeneralContainer>(batch->GetRecordsCountVerified());
     for (size_t i = 0; i < resultArrowSchema->fields().size(); ++i) {
         auto& resultField = resultArrowSchema->fields()[i];
         auto columnId = GetIndexInfo().GetColumnId(resultField->name());

ydb/core/tx/columnshard/transactions/operators/sharing.cpp

@@ -28,9 +28,9 @@ bool TSharingTransactionOperator::DoParse(TColumnShard& owner, const TString& da
         AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "session_exists")("session_id", SharingTask->GetSessionId())("info", SharingTask->DebugString());
     } else {
         SharingTask->Confirm();
+        TxPropose = SharingSessionsManager->ProposeDestSession(&owner, SharingTask);
     }
 
-    TxPropose = SharingSessionsManager->ProposeDestSession(&owner, SharingTask);
 
     return true;
 }
@@ -47,8 +47,8 @@ void TSharingTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner
     if (!SessionExistsFlag) {
         AFL_VERIFY(!!TxPropose);
         TxPropose->Complete(ctx);
+        TxPropose.reset();
     }
-    TxPropose.release();
 }
 
 bool TSharingTransactionOperator::ProgressOnExecute(