KIKIMR-20316: fix incorrect loader usage for column restore

ydb/core/formats/arrow/common/CMakeLists.darwin-arm64.txt

@@ -12,6 +12,7 @@ target_link_libraries(formats-arrow-common PUBLIC
   contrib-libs-cxxsupp
   yutil
   libs-apache-arrow
+  cpp-actors-core
 )
 target_sources(formats-arrow-common PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/core/formats/arrow/common/validation.cpp

ydb/core/formats/arrow/common/CMakeLists.darwin-x86_64.txt

@@ -12,6 +12,7 @@ target_link_libraries(formats-arrow-common PUBLIC
   contrib-libs-cxxsupp
   yutil
   libs-apache-arrow
+  cpp-actors-core
 )
 target_sources(formats-arrow-common PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/core/formats/arrow/common/validation.cpp

ydb/core/formats/arrow/common/CMakeLists.linux-aarch64.txt

@@ -13,6 +13,7 @@ target_link_libraries(formats-arrow-common PUBLIC
   contrib-libs-cxxsupp
   yutil
   libs-apache-arrow
+  cpp-actors-core
 )
 target_sources(formats-arrow-common PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/core/formats/arrow/common/validation.cpp

ydb/core/formats/arrow/common/CMakeLists.linux-x86_64.txt

@@ -13,6 +13,7 @@ target_link_libraries(formats-arrow-common PUBLIC
   contrib-libs-cxxsupp
   yutil
   libs-apache-arrow
+  cpp-actors-core
 )
 target_sources(formats-arrow-common PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/core/formats/arrow/common/validation.cpp

ydb/core/formats/arrow/common/CMakeLists.windows-x86_64.txt

@@ -12,6 +12,7 @@ target_link_libraries(formats-arrow-common PUBLIC
   contrib-libs-cxxsupp
   yutil
   libs-apache-arrow
+  cpp-actors-core
 )
 target_sources(formats-arrow-common PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/core/formats/arrow/common/validation.cpp

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

@@ -1,5 +1,10 @@
 #include "validation.h"
+#include <library/cpp/actors/core/log.h>
 
 namespace NKikimr::NArrow {
 
+void TStatusValidator::Validate(const arrow::Status& status) {
+    AFL_VERIFY(status.ok())("problem", status.ToString().c_str());
+}
+
 }

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

@@ -8,19 +8,17 @@ namespace NKikimr::NArrow {
 
 class TStatusValidator {
 public:
-    static void Validate(const arrow::Status& status) {
-        Y_ABORT_UNLESS(status.ok(), "%s", status.ToString().c_str());
-    }
+    static void Validate(const arrow::Status& status);
 
     template <class T>
     static T GetValid(const arrow::Result<T>& result) {
-        Y_ABORT_UNLESS(result.ok(), "%s", result.status().ToString().c_str());
+        Validate(result.status());
         return *result;
     }
 
     template <class T>
     static T GetValid(arrow::Result<T>&& result) {
-        Y_ABORT_UNLESS(result.ok(), "%s", result.status().ToString().c_str());
+        Validate(result.status());
         return std::move(*result);
     }
 };

ydb/core/formats/arrow/common/ya.make

@@ -2,6 +2,7 @@ LIBRARY()
 
 PEERDIR(
     contrib/libs/apache/arrow
+    library/cpp/actors/core
 )
 
 SRCS(

ydb/core/tx/columnshard/engines/changes/compaction/column_cursor.cpp

@@ -8,7 +8,7 @@ bool TPortionColumnCursor::Fetch(TMergedColumn& column) {
     Y_ABORT_UNLESS(RecordIndexStart);
     ui32 currentStartPortionIdx = *RecordIndexStart;
     ui32 currentFinishPortionIdx = RecordIndexFinish;
-
+//    NActors::TLogContextGuard lg(NActors::TLogContextBuilder::Build()("portion_id", PortionId));
     while (currentStartPortionIdx - ChunkRecordIndexStartPosition >= CurrentColumnChunk->GetMeta().GetNumRowsVerified()) {
         if (!NextChunk()) {
             return false;
@@ -18,11 +18,11 @@ bool TPortionColumnCursor::Fetch(TMergedColumn& column) {
     ui32 currentStart = currentStartPortionIdx - ChunkRecordIndexStartPosition;
     while (currentFinishPortionIdx - ChunkRecordIndexStartPosition >= CurrentColumnChunk->GetMeta().GetNumRowsVerified()) {
         const ui32 currentFinish = CurrentColumnChunk->GetMeta().GetNumRowsVerified();
-        if (currentStart == 0) {
-            column.AppendBlob(CurrentBlobChunk->GetData(), *CurrentColumnChunk);
-        } else {
+//        if (currentStart == 0) {
+//            column.AppendBlob(CurrentBlobChunk->GetData(), *CurrentColumnChunk);
+//        } else {
             column.AppendSlice(GetCurrentArray(), currentStart, currentFinish - currentStart);
-        }
+//        }
         currentStart = 0;
         if (!NextChunk()) {
             return false;

ydb/core/tx/columnshard/engines/changes/compaction/column_cursor.h

@@ -20,6 +20,7 @@ private:
     const TColumnRecord* CurrentColumnChunk = nullptr;
     std::shared_ptr<arrow::Array> CurrentArray;
     std::shared_ptr<TColumnLoader> ColumnLoader;
+    const ui64 PortionId;
 
     const std::shared_ptr<arrow::Array>& GetCurrentArray();
 
@@ -36,11 +37,13 @@ public:
 
     bool Fetch(TMergedColumn& column);
 
-    TPortionColumnCursor(const std::vector<IPortionColumnChunk::TPtr>& columnChunks, const std::vector<const TColumnRecord*>& records, const std::shared_ptr<TColumnLoader> loader)
+    TPortionColumnCursor(const std::vector<IPortionColumnChunk::TPtr>& columnChunks, const std::vector<const TColumnRecord*>& records, const std::shared_ptr<TColumnLoader>& loader, const ui64 portionId)
         : BlobChunks(columnChunks)
         , ColumnChunks(records)
         , ColumnLoader(loader)
+        , PortionId(portionId)
     {
+        Y_UNUSED(PortionId);
         Y_ABORT_UNLESS(BlobChunks.size());
         Y_ABORT_UNLESS(ColumnChunks.size() == BlobChunks.size());
         CurrentBlobChunk = BlobChunks.front();