Fix bulk CS stats (#20253)

Vladilen · web-flow · commit b8c0fee34210 · 2025-07-01T10:26:51.000+02:00
diff --git a/ydb/core/protos/counters_columnshard.proto b/ydb/core/protos/counters_columnshard.proto
@@ -139,6 +139,8 @@ enum ECumulativeCounters {
     COUNTER_IMMEDIATE_TX_COMPLETED = 83                     [(CounterOpts) = {Name: "ImmediateTxCompleted"}];
     COUNTER_ROWS_ERASED = 84                                [(CounterOpts) = {Name: "RowsErased"}];
     COUNTER_OPERATIONS_ROWS_WRITTEN = 85                    [(CounterOpts) = {Name: "OperationsRowsWritten"}];
+    COUNTER_OPERATIONS_BULK_ROWS_WRITTEN = 86               [(CounterOpts) = {Name: "OperationsBulkRowsWritten"}];
+    COUNTER_OPERATIONS_BULK_BYTES_WRITTEN = 87              [(CounterOpts) = {Name: "OperationsBulkBytesWritten"}];
 }
 
 enum EPercentileCounters {
diff --git a/ydb/core/protos/data_events.proto b/ydb/core/protos/data_events.proto
@@ -104,6 +104,7 @@ message TEvWrite {
         optional EDataFormat PayloadFormat = 5;
         optional string PayloadSchema = 6;
         optional uint32 DefaultFilledColumnCount = 7;
+        optional bool IsBulk = 8;
     }
 
     // Transaction operations
diff --git a/ydb/core/protos/tx_columnshard.proto b/ydb/core/protos/tx_columnshard.proto
@@ -354,4 +354,5 @@ message TInternalOperationData {
     optional uint32 ModificationType = 2;
     optional uint64 PathId = 3;
     optional bool WritePortions = 4;
+    optional bool IsBulk = 5;
 }
diff --git a/ydb/core/tablet/tablet_counters_aggregator.cpp b/ydb/core/tablet/tablet_counters_aggregator.cpp
@@ -817,6 +817,8 @@ class TTabletMon {
         TCounterPtr ColumnShardScannedRows_;
         TCounterPtr ColumnShardOperationsRowsWritten_;
         TCounterPtr ColumnShardOperationsBytesWritten_;
+        TCounterPtr ColumnShardOperationsBulkRowsWritten_;
+        TCounterPtr ColumnShardOperationsBulkBytesWritten_;
         TCounterPtr ColumnShardErasedBytes_;
         TCounterPtr ColumnShardErasedRows_;
         THistogramPtr ColumnShardConsumedCpuHistogram;
@@ -969,6 +971,8 @@ class TTabletMon {
                 ColumnShardScannedRows_ = appGroup->GetCounter("ColumnShard/ScannedRows");
                 ColumnShardOperationsRowsWritten_ = appGroup->GetCounter("ColumnShard/OperationsRowsWritten");
                 ColumnShardOperationsBytesWritten_ = appGroup->GetCounter("ColumnShard/OperationsBytesWritten");
+                ColumnShardOperationsBulkRowsWritten_ = appGroup->GetCounter("ColumnShard/OperationsBulkRowsWritten");
+                ColumnShardOperationsBulkBytesWritten_ = appGroup->GetCounter("ColumnShard/OperationsBulkBytesWritten");
                 ColumnShardErasedBytes_ = appGroup->GetCounter("ColumnShard/BytesErased");
                 ColumnShardErasedRows_ = appGroup->GetCounter("ColumnShard/RowsErased");
                 ColumnShardConsumedCpuHistogram = appGroup->FindHistogram("HIST(ConsumedCPU)");
@@ -1023,8 +1027,8 @@ class TTabletMon {
                 ColumnShardScanBytes_->Set(ColumnShardScannedBytes_->Val());
                 ColumnShardWriteRows_->Set(ColumnShardOperationsRowsWritten_->Val());
                 ColumnShardWriteBytes_->Set(ColumnShardOperationsBytesWritten_->Val());
-                ColumnShardBulkUpsertRows_->Set(ColumnShardOperationsRowsWritten_->Val());
-                ColumnShardBulkUpsertBytes_->Set(ColumnShardOperationsBytesWritten_->Val());
+                ColumnShardBulkUpsertRows_->Set(ColumnShardOperationsBulkRowsWritten_->Val());
+                ColumnShardBulkUpsertBytes_->Set(ColumnShardOperationsBulkBytesWritten_->Val());
                 ColumnShardEraseRows_->Set(ColumnShardErasedRows_->Val());
                 ColumnShardEraseBytes_->Set(ColumnShardErasedBytes_->Val());
 
diff --git a/ydb/core/tx/columnshard/blobs_action/abstract/write.h b/ydb/core/tx/columnshard/blobs_action/abstract/write.h
@@ -23,7 +23,9 @@ class IBlobsWritingAction: public ICommonBlobsAction {
     THashSet<TUnifiedBlobId> BlobsWaiting;
     bool Aborted = false;
     std::shared_ptr<NBlobOperations::TWriteCounters> Counters;
+    YDB_FLAG_ACCESSOR(Bulk, false);
     void AddDataForWrite(const TUnifiedBlobId& blobId, const TString& data);
+
 protected:
     virtual void DoOnExecuteTxBeforeWrite(NColumnShard::TColumnShard& self, TBlobManagerDb& dbBlobs) = 0;
     virtual void DoOnCompleteTxBeforeWrite(NColumnShard::TColumnShard& self) = 0;
diff --git a/ydb/core/tx/columnshard/blobs_action/bs/write.cpp b/ydb/core/tx/columnshard/blobs_action/bs/write.cpp
@@ -15,7 +15,11 @@ void TWriteAction::DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& self, co
     ui64 blobsWritten = BlobBatch.GetBlobCount();
     ui64 bytesWritten = BlobBatch.GetTotalSize();
     if (blobsWroteSuccessfully) {
-        self.Counters.GetTabletCounters()->OnWriteSuccess(blobsWritten, bytesWritten);
+        if (IsBulk()) {
+            self.Counters.GetTabletCounters()->OnBulkWriteSuccess(blobsWritten, bytesWritten);
+        } else {
+            self.Counters.GetTabletCounters()->OnWriteSuccess(blobsWritten, bytesWritten);
+        }
         Manager->SaveBlobBatchOnComplete(std::move(BlobBatch));
     } else {
         self.Counters.GetTabletCounters()->OnWriteFailure();
diff --git a/ydb/core/tx/columnshard/columnshard__write.cpp b/ydb/core/tx/columnshard/columnshard__write.cpp
@@ -105,15 +105,23 @@ void TColumnShard::Handle(NPrivateEvents::NWrite::TEvWritePortionResult::TPtr& e
             AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_WRITE)("writing_size", i.GetDataSize())("event", "data_write_finished")(
                 "writing_id", i.GetWriteMeta().GetId());
             i.MutableWriteMeta().OnStage(NEvWrite::EWriteStage::SuccessWritingToLocalDB);
-            Counters.OnWritePutBlobsSuccess(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
+            if (i.GetWriteMeta().IsBulk()) {
+                Counters.OnWritePutBulkBlobsSuccess(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
+            } else {
+                Counters.OnWritePutBlobsSuccess(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
+            }
             Counters.GetWritesMonitor()->OnFinishWrite(i.GetDataSize(), 1);
         }
         Execute(new TTxBlobsWritingFinished(this, ev->Get()->GetWriteStatus(), ev->Get()->GetWriteAction(), std::move(writtenData)), ctx);
     } else {
         const TMonotonic now = TMonotonic::Now();
         for (auto&& i : writtenData.GetWriteResults()) {
             i.MutableWriteMeta().OnStage(NEvWrite::EWriteStage::FailWritingToLocalDB);
-            Counters.OnWritePutBlobsFailed(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
+            if (i.GetWriteMeta().IsBulk()) {
+                Counters.OnWritePutBulkBlobsFailed(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
+            } else {
+                Counters.OnWritePutBlobsFailed(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
+            }
             Counters.GetCSCounters().OnWritePutBlobsFail(now - i.GetWriteMeta().GetWriteStartInstant());
             AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("writing_size", i.GetDataSize())("event", "data_write_error")(
                 "writing_id", i.GetWriteMeta().GetId())("reason", i.GetErrorMessage());
@@ -477,9 +485,11 @@ void TColumnShard::Handle(NEvents::TDataEvents::TEvWrite::TPtr& ev, const TActor
         lockId = record.GetLockTxId();
     }
 
+    const bool isBulk = operation.HasIsBulk() && operation.GetIsBulk();
+
     Counters.GetWritesMonitor()->OnStartWrite(arrowData->GetSize());
     WriteTasksQueue->Enqueue(TWriteTask(
-        arrowData, schema, source, granuleShardingVersionId, pathId, cookie, lockId, *mType, behaviour, writeTimeout, record.GetTxId()));
+        arrowData, schema, source, granuleShardingVersionId, pathId, cookie, lockId, *mType, behaviour, writeTimeout, record.GetTxId(), isBulk));
     WriteTasksQueue->Drain(false, ctx);
 }
 
diff --git a/ydb/core/tx/columnshard/counters/counters_manager.h b/ydb/core/tx/columnshard/counters/counters_manager.h
@@ -98,10 +98,20 @@ class TCountersManager {
         CSCounters.OnWritePutBlobsSuccess(d);
     }
 
+    void OnWritePutBulkBlobsSuccess(const TDuration d, const ui64 rowsWritten) const {
+        TabletCounters->OnWritePutBulkBlobsSuccess(rowsWritten);
+        CSCounters.OnWritePutBlobsSuccess(d);
+    }
+
     void OnWritePutBlobsFailed(const TDuration d, const ui64 /*rowsWritten*/) const {
         TabletCounters->OnWriteFailure();
         CSCounters.OnWritePutBlobsFail(d);
     }
+
+    void OnWritePutBulkBlobsFailed(const TDuration d, const ui64 /*rowsWritten*/) const {
+        TabletCounters->OnWriteFailure();
+        CSCounters.OnWritePutBlobsFail(d);
+    }
 };
 
 }   // namespace NKikimr::NColumnShard
diff --git a/ydb/core/tx/columnshard/counters/tablet_counters.h b/ydb/core/tx/columnshard/counters/tablet_counters.h
@@ -75,6 +75,12 @@ class TTabletCountersHandle {
         IncCounter(NColumnShard::COUNTER_WRITE_SUCCESS);
     }
 
+    void OnBulkWriteSuccess(const ui64 blobsWritten, const ui64 bytesWritten) const {
+        IncCounter(NColumnShard::COUNTER_OPERATIONS_BLOBS_WRITTEN, blobsWritten);
+        IncCounter(NColumnShard::COUNTER_OPERATIONS_BULK_BYTES_WRITTEN, bytesWritten);
+        IncCounter(NColumnShard::COUNTER_WRITE_SUCCESS);
+    }
+
     void OnWriteFailure() const {
         IncCounter(NColumnShard::COUNTER_WRITE_FAIL);
     }
@@ -96,14 +102,18 @@ class TTabletCountersHandle {
         IncCounter(NColumnShard::COUNTER_OPERATIONS_ROWS_WRITTEN, rowsWritten);
     }
 
+    void OnWritePutBulkBlobsSuccess(const ui64 rowsWritten) const {
+        IncCounter(NColumnShard::COUNTER_OPERATIONS_BULK_ROWS_WRITTEN, rowsWritten);
+    }
+
     void OnDropPortionEvent(const ui64 rawBytes, const ui64 blobBytes, const ui64 rows) const {
         IncCounter(NColumnShard::COUNTER_RAW_BYTES_ERASED, rawBytes);
         IncCounter(NColumnShard::COUNTER_BYTES_ERASED, blobBytes);
         IncCounter(NColumnShard::COUNTER_ROWS_ERASED, rows);
     }
 
     void FillStats(::NKikimrTableStats::TTableStats& output) const {
-        output.SetRowUpdates(GetValue(COUNTER_OPERATIONS_ROWS_WRITTEN));
+        output.SetRowUpdates(GetValue(COUNTER_OPERATIONS_ROWS_WRITTEN) + GetValue(COUNTER_OPERATIONS_BULK_ROWS_WRITTEN));
         output.SetRowDeletes(GetValue(COUNTER_ROWS_ERASED));
         output.SetRowReads(0);   // all reads are range reads
         output.SetRangeReadRows(GetValue(COUNTER_READ_INDEX_ROWS));
diff --git a/ydb/core/tx/columnshard/operations/manager.cpp b/ydb/core/tx/columnshard/operations/manager.cpp
@@ -29,7 +29,7 @@ bool TOperationsManager::Load(NTabletFlatExecutor::TTransactionContext& txc) {
             Y_ABORT_UNLESS(metaProto.ParseFromString(metadata));
 
             auto operation = std::make_shared<TWriteOperation>(TUnifiedPathId{}, writeId, lockId, cookie, status, TInstant::Seconds(createdAtSec),
-                granuleShardingVersionId, NEvWrite::EModificationType::Upsert);
+                granuleShardingVersionId, NEvWrite::EModificationType::Upsert, metaProto.GetIsBulk());
             operation->FromProto(metaProto);
             AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "register_operation_on_load")("operation_id", operation->GetWriteId());
             AFL_VERIFY(operation->GetStatus() != EOperationStatus::Draft);
@@ -203,10 +203,10 @@ void TOperationsManager::LinkTransactionOnComplete(const ui64 /*lockId*/, const
 }
 
 TWriteOperation::TPtr TOperationsManager::CreateWriteOperation(const TUnifiedPathId& pathId, const ui64 lockId, const ui64 cookie,
-    const std::optional<ui32> granuleShardingVersionId, const NEvWrite::EModificationType mType) {
+    const std::optional<ui32> granuleShardingVersionId, const NEvWrite::EModificationType mType, const bool isBulk) {
     auto writeId = BuildNextOperationWriteId();
     auto operation = std::make_shared<TWriteOperation>(pathId, writeId, lockId, cookie, EOperationStatus::Draft, AppData()->TimeProvider->Now(),
-        granuleShardingVersionId, mType);
+        granuleShardingVersionId, mType, isBulk);
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_WRITE)("event", "register_operation")("operation_id", operation->GetWriteId())(
         "last", LastWriteId);
     AFL_VERIFY(Operations.emplace(operation->GetWriteId(), operation).second);
diff --git a/ydb/core/tx/columnshard/operations/manager.h b/ydb/core/tx/columnshard/operations/manager.h
@@ -208,7 +208,7 @@ class TOperationsManager {
     }
 
     TWriteOperation::TPtr CreateWriteOperation(const TUnifiedPathId& pathId, const ui64 lockId, const ui64 cookie, const std::optional<ui32> granuleShardingVersionId,
-        const NEvWrite::EModificationType mType);
+        const NEvWrite::EModificationType mType, const bool isBulk);
     bool RegisterLock(const ui64 lockId, const ui64 generationId) {
         if (LockFeatures.contains(lockId)) {
             return false;
diff --git a/ydb/core/tx/columnshard/operations/write.cpp b/ydb/core/tx/columnshard/operations/write.cpp
@@ -16,15 +16,17 @@ namespace NKikimr::NColumnShard {
 
 TWriteOperation::TWriteOperation(const TUnifiedPathId& pathId, const TOperationWriteId writeId, const ui64 lockId, const ui64 cookie,
     const EOperationStatus& status, const TInstant createdAt, const std::optional<ui32> granuleShardingVersionId,
-    const NEvWrite::EModificationType mType)
+    const NEvWrite::EModificationType mType, const bool isBulk)
     : PathId(pathId)
     , Status(status)
     , CreatedAt(createdAt)
     , WriteId(writeId)
     , LockId(lockId)
     , Cookie(cookie)
     , GranuleShardingVersionId(granuleShardingVersionId)
-    , ModificationType(mType) {
+    , ModificationType(mType)
+    , BulkFlag(isBulk)
+{
 }
 
 void TWriteOperation::Start(
@@ -36,8 +38,10 @@ void TWriteOperation::Start(
         context.GetWritingCounters()->GetWriteFlowCounters());
     writeMeta->SetLockId(LockId);
     writeMeta->SetModificationType(ModificationType);
-    NEvWrite::TWriteData writeData(writeMeta, data, owner.TablesManager.GetPrimaryIndex()->GetReplaceKey(),
-        owner.StoragesManager->GetInsertOperator()->StartWritingAction(NOlap::NBlobOperations::EConsumer::WRITING_OPERATOR));
+    writeMeta->SetBulk(IsBulk());
+    auto writingAction = owner.StoragesManager->GetInsertOperator()->StartWritingAction(NOlap::NBlobOperations::EConsumer::WRITING_OPERATOR);
+    writingAction->SetBulk(IsBulk());
+    NEvWrite::TWriteData writeData(writeMeta, data, owner.TablesManager.GetPrimaryIndex()->GetReplaceKey(), std::move(writingAction));
     std::shared_ptr<NConveyor::ITask> task = std::make_shared<NOlap::TBuildBatchesTask>(std::move(writeData), context);
     NConveyorComposite::TInsertServiceOperator::SendTaskToExecute(task);
 
@@ -95,6 +99,7 @@ void TWriteOperation::ToProto(NKikimrTxColumnShard::TInternalOperationData& prot
     }
     proto.SetModificationType((ui32)ModificationType);
     proto.SetWritePortions(true);
+    proto.SetIsBulk(IsBulk());
     PathId.InternalPathId.ToProto(proto);
 }
 
@@ -109,6 +114,7 @@ void TWriteOperation::FromProto(const NKikimrTxColumnShard::TInternalOperationDa
     } else {
         ModificationType = NEvWrite::EModificationType::Replace;
     }
+    BulkFlag = proto.GetIsBulk();
 }
 
 void TWriteOperation::AbortOnExecute(TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc) const {
diff --git a/ydb/core/tx/columnshard/operations/write.h b/ydb/core/tx/columnshard/operations/write.h
@@ -60,6 +60,7 @@ class TWriteOperation: public TMonitoringObjectsCounter<TWriteOperation> {
     YDB_ACCESSOR(EOperationBehaviour, Behaviour, EOperationBehaviour::Undefined);
     YDB_READONLY_DEF(std::optional<ui32>, GranuleShardingVersionId);
     YDB_READONLY(NEvWrite::EModificationType, ModificationType, NEvWrite::EModificationType::Upsert);
+    YDB_READONLY_FLAG(Bulk, false);
     const std::shared_ptr<TAtomicCounter> Activity = std::make_shared<TAtomicCounter>(1);
 
 public:
@@ -70,7 +71,7 @@ class TWriteOperation: public TMonitoringObjectsCounter<TWriteOperation> {
     }
 
     TWriteOperation(const TUnifiedPathId& pathId, const TOperationWriteId writeId, const ui64 lockId, const ui64 cookie, const EOperationStatus& status,
-        const TInstant createdAt, const std::optional<ui32> granuleShardingVersionId, const NEvWrite::EModificationType mType);
+        const TInstant createdAt, const std::optional<ui32> granuleShardingVersionId, const NEvWrite::EModificationType mType, const bool isBulk);
 
     void Start(
         TColumnShard& owner, const NEvWrite::IDataContainer::TPtr& data, const NActors::TActorId& source, const NOlap::TWritingContext& context);
diff --git a/ydb/core/tx/columnshard/tablet/write_queue.cpp b/ydb/core/tx/columnshard/tablet/write_queue.cpp
@@ -9,7 +9,7 @@ namespace NKikimr::NColumnShard {
 bool TWriteTask::Execute(TColumnShard* owner, const TActorContext& /* ctx */) const {
     owner->Counters.GetCSCounters().WritingCounters->OnWritingTaskDequeue(TMonotonic::Now() - Created);
     owner->OperationsManager->RegisterLock(LockId, owner->Generation());
-    auto writeOperation = owner->OperationsManager->CreateWriteOperation(PathId, LockId, Cookie, GranuleShardingVersionId, ModificationType);
+    auto writeOperation = owner->OperationsManager->CreateWriteOperation(PathId, LockId, Cookie, GranuleShardingVersionId, ModificationType, IsBulk);
 
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_WRITE)("writing_size", ArrowData->GetSize())("operation_id", writeOperation->GetIdentifier())(
         "in_flight", owner->Counters.GetWritesMonitor()->GetWritesInFlight())(
diff --git a/ydb/core/tx/columnshard/tablet/write_queue.h b/ydb/core/tx/columnshard/tablet/write_queue.h
@@ -23,6 +23,7 @@ class TWriteTask: public TMoveOnly {
     const TMonotonic Created = TMonotonic::Now();
     const std::optional<TDuration> Timeout;
     const ui64 TxId;
+    const bool IsBulk;
 
 public:
     bool operator<(const TWriteTask& item) const {
@@ -35,7 +36,7 @@ class TWriteTask: public TMoveOnly {
 
     TWriteTask(const std::shared_ptr<TArrowData>& arrowData, const NOlap::ISnapshotSchema::TPtr& schema, const NActors::TActorId sourceId,
         const std::optional<ui32>& granuleShardingVersionId, const TUnifiedPathId pathId, const ui64 cookie, const ui64 lockId,
-        const NEvWrite::EModificationType modificationType, const EOperationBehaviour behaviour, const std::optional<TDuration> timeout, const ui64 txId)
+        const NEvWrite::EModificationType modificationType, const EOperationBehaviour behaviour, const std::optional<TDuration> timeout, const ui64 txId, const bool isBulk)
         : ArrowData(arrowData)
         , Schema(schema)
         , SourceId(sourceId)
@@ -47,6 +48,7 @@ class TWriteTask: public TMoveOnly {
         , Behaviour(behaviour)
         , Timeout(timeout)
         , TxId(txId)
+        , IsBulk(isBulk)
     {
     }
 
diff --git a/ydb/core/tx/data_events/columnshard_splitter.h b/ydb/core/tx/data_events/columnshard_splitter.h
@@ -50,6 +50,7 @@ class TColumnShardShardsSplitter: public IShardsSplitter {
             operation->SetPayloadIndex(0);
             operation->MutableTableId()->SetTableId(tableId);
             operation->MutableTableId()->SetSchemaVersion(schemaVersion);
+            operation->SetIsBulk(true);
         }
     };
 
diff --git a/ydb/core/tx/data_events/write_data.h b/ydb/core/tx/data_events/write_data.h
@@ -53,6 +53,8 @@ class TWriteMeta: public NColumnShard::TMonitoringObjectsCounter<TWriteMeta>, TN
     const std::shared_ptr<TWriteFlowCounters> Counters;
     mutable NOlap::NCounters::TStateSignalsOperator<NEvWrite::EWriteStage>::TGuard StateGuard;
 
+    YDB_FLAG_ACCESSOR(Bulk, false);
+
 public:
     void OnStage(const EWriteStage stage) const;
 

Original file line number	Diff line number	Diff line change
`@@ -139,6 +139,8 @@ enum ECumulativeCounters {`
`139`	`139`	`COUNTER_IMMEDIATE_TX_COMPLETED = 83 [(CounterOpts) = {Name: "ImmediateTxCompleted"}];`
`140`	`140`	`COUNTER_ROWS_ERASED = 84 [(CounterOpts) = {Name: "RowsErased"}];`
`141`	`141`	`COUNTER_OPERATIONS_ROWS_WRITTEN = 85 [(CounterOpts) = {Name: "OperationsRowsWritten"}];`
	`142`	`+ COUNTER_OPERATIONS_BULK_ROWS_WRITTEN = 86 [(CounterOpts) = {Name: "OperationsBulkRowsWritten"}];`
	`143`	`+ COUNTER_OPERATIONS_BULK_BYTES_WRITTEN = 87 [(CounterOpts) = {Name: "OperationsBulkBytesWritten"}];`
`142`	`144`	`}`
`143`	`145`
`144`	`146`	`enum EPercentileCounters {`
Original file line number	Diff line number	Diff line change
`@@ -104,6 +104,7 @@ message TEvWrite {`
`104`	`104`	`optional EDataFormat PayloadFormat = 5;`
`105`	`105`	`optional string PayloadSchema = 6;`
`106`	`106`	`optional uint32 DefaultFilledColumnCount = 7;`
	`107`	`+ optional bool IsBulk = 8;`
`107`	`108`	`}`
`108`	`109`
`109`	`110`	`// Transaction operations`
Original file line number	Diff line number	Diff line change
`@@ -354,4 +354,5 @@ message TInternalOperationData {`
`354`	`354`	`optional uint32 ModificationType = 2;`
`355`	`355`	`optional uint64 PathId = 3;`
`356`	`356`	`optional bool WritePortions = 4;`
	`357`	`+ optional bool IsBulk = 5;`
`357`	`358`	`}`