Datashard added OwnedCellVecBatch

Datashard added OwnedCellVecBatch

Pull Request resolved: #304

1 files changed, 48 insertions, 0 deletions

diff --git a/ydb/core/scheme/scheme_tablecell.cpp b/ydb/core/scheme/scheme_tablecell.cpp
index 04c2aaf4a6..3d30ca1d80 100644
--- a/ydb/core/scheme/scheme_tablecell.cpp
+++ b/ydb/core/scheme/scheme_tablecell.cpp
@@ -197,6 +197,54 @@ bool TSerializedCellVec::DoTryParse(const TString& data) {
     return TryDeserializeCellVec(data, Buf, Cells);
 }
 
+TOwnedCellVecBatch::TOwnedCellVecBatch()
+    : Pool(std::make_unique<TMemoryPool>(InitialPoolSize)) {
+}
+
+size_t TOwnedCellVecBatch::Append(TConstArrayRef<TCell> cells) {
+    size_t cellsSize = cells.size();
+    if (cellsSize == 0) {
+        CellVectors.emplace_back();
+        return 0;
+    }
+
+    size_t size = sizeof(TCell) * cellsSize;
+    for (auto& cell : cells) {
+        if (!cell.IsNull() && !cell.IsInline()) {
+            const size_t cellSize = cell.Size();
+            size += AlignUp(cellSize);
+        }
+    }
+
+    char * allocatedBuffer = reinterpret_cast<char *>(Pool->Allocate(size));
+
+    TCell* ptrCell = reinterpret_cast<TCell*>(allocatedBuffer);
+    char* ptrData = reinterpret_cast<char*>(ptrCell + cellsSize);
+
+    TConstArrayRef<TCell> cellvec(ptrCell, ptrCell + cellsSize);
+
+    for (auto& cell : cells) {
+        if (cell.IsNull()) {
+            new (ptrCell) TCell();
+        } else if (cell.IsInline()) {
+            new (ptrCell) TCell(cell);
+        } else {
+            const size_t cellSize = cell.Size();
+            if (Y_LIKELY(cellSize > 0)) {
+                ::memcpy(ptrData, cell.Data(), cellSize);
+            }
+            new (ptrCell) TCell(ptrData, cellSize);
+            ptrData += AlignUp(cellSize);
+        }
+
+        ++ptrCell;
+    }
+
+    CellVectors.push_back(cellvec);
+    return size;
+}
+
+
 TString DbgPrintCell(const TCell& r, NScheme::TTypeInfo typeInfo, const NScheme::TTypeRegistry &reg) {
     auto typeId = typeInfo.GetTypeId();
     TString res;