Skip to content

Commit ca6ae0f

Browse files
ivanmorozov333ivanmorozov333
authored
Merge 612baed into 5f125c0 
2 parents 5f125c0 + 612baed commit ca6ae0f

File tree

20 files changed

+105
-83
lines changed

20 files changed

+105
-83
lines changed

‎ydb/core/formats/arrow/serializer/abstract.cpp

+5-1
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,12 @@ NKikimr::TConclusionStatus TSerializerContainer::DeserializeFromRequest(NYql::TF
2121
return TBase::GetObjectPtr()->DeserializeFromRequest(features);
2222
}
2323

24-
std::shared_ptr<NKikimr::NArrow::NSerialization::ISerializer> TSerializerContainer::GetDefaultSerializer() {
24+
std::shared_ptr<ISerializer> TSerializerContainer::GetDefaultSerializer() {
2525
return std::make_shared<TNativeSerializer>();
2626
}
27+
std::shared_ptr<ISerializer> TSerializerContainer::GetFastestSerializer() {
28+
return std::make_shared<TNativeSerializer>(arrow::Compression::UNCOMPRESSED);
29+
}
30+
2731

2832
}

‎ydb/core/formats/arrow/serializer/abstract.h

+1
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,7 @@ class TSerializerContainer: public NBackgroundTasks::TInterfaceProtoContainer<IS
146146
using TBase::DeserializeFromProto;
147147

148148
static std::shared_ptr<ISerializer> GetDefaultSerializer();
149+
static std::shared_ptr<ISerializer> GetFastestSerializer();
149150

150151
TConclusionStatus DeserializeFromProto(const NKikimrSchemeOp::TCompressionOptions& proto);
151152

‎ydb/core/formats/arrow/size_calcer.cpp

+8-5
Original file line numberDiff line numberDiff line change
@@ -242,12 +242,15 @@ ui64 GetArrayDataSize(const std::shared_ptr<arrow::Array>& column) {
242242
}
243243

244244
NKikimr::NArrow::TSerializedBatch TSerializedBatch::Build(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context) {
245-
std::optional<TString> specialKeys;
245+
std::optional<TString> specialKeysPayload;
246+
std::optional<TString> specialKeysFull;
246247
if (context.GetFieldsForSpecialKeys().size()) {
247-
specialKeys = TFirstLastSpecialKeys(batch, context.GetFieldsForSpecialKeys()).SerializeToString();
248+
TFirstLastSpecialKeys specialKeys(batch, context.GetFieldsForSpecialKeys());
249+
specialKeysPayload = specialKeys.SerializePayloadToString();
250+
specialKeysFull = specialKeys.SerializeFullToString();
248251
}
249-
return TSerializedBatch(NArrow::SerializeSchema(*batch->schema()), NArrow::SerializeBatchNoCompression(batch), batch->num_rows(),
250-
NArrow::GetBatchDataSize(batch), specialKeys);
252+
return TSerializedBatch(NArrow::SerializeBatchNoCompression(batch), batch->num_rows(),
253+
NArrow::GetBatchDataSize(batch), specialKeysPayload, specialKeysFull);
251254
}
252255

253256
TConclusionStatus TSerializedBatch::BuildWithLimit(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context, std::optional<TSerializedBatch>& sbL, std::optional<TSerializedBatch>& sbR) {
@@ -291,7 +294,7 @@ TConclusion<std::vector<TSerializedBatch>> TSerializedBatch::BuildWithLimit(std:
291294
}
292295

293296
TString TSerializedBatch::DebugString() const {
294-
return TStringBuilder() << "(data_size=" << Data.size() << ";schema_data_size=" << SchemaData.size() << ";rows_count=" << RowsCount << ";raw_bytes=" << RawBytes << ";)";
297+
return TStringBuilder() << "(data_size=" << Data.size() << ";rows_count=" << RowsCount << ";raw_bytes=" << RawBytes << ";)";
295298
}
296299

297300
}

‎ydb/core/formats/arrow/size_calcer.h

+18-12
Original file line numberDiff line numberDiff line change
@@ -70,23 +70,29 @@ class TBatchSplitttingContext {
7070

7171
class TSerializedBatch {
7272
private:
73-
YDB_READONLY_DEF(TString, SchemaData);
7473
YDB_READONLY_DEF(TString, Data);
7574
YDB_READONLY(ui32, RowsCount, 0);
7675
YDB_READONLY(ui32, RawBytes, 0);
77-
std::optional<TString> SpecialKeys;
76+
std::optional<TString> SpecialKeysFull;
77+
std::optional<TString> SpecialKeysPayload;
78+
7879
public:
7980
size_t GetSize() const {
8081
return Data.size();
8182
}
8283

83-
const TString& GetSpecialKeysSafe() const {
84-
AFL_VERIFY(SpecialKeys);
85-
return *SpecialKeys;
84+
const TString& GetSpecialKeysPayloadSafe() const {
85+
AFL_VERIFY(SpecialKeysPayload);
86+
return *SpecialKeysPayload;
87+
}
88+
89+
const TString& GetSpecialKeysFullSafe() const {
90+
AFL_VERIFY(SpecialKeysFull);
91+
return *SpecialKeysFull;
8692
}
8793

8894
bool HasSpecialKeys() const {
89-
return !!SpecialKeys;
95+
return !!SpecialKeysFull;
9096
}
9197

9298
TString DebugString() const;
@@ -95,14 +101,14 @@ class TSerializedBatch {
95101
static TConclusionStatus BuildWithLimit(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context, std::optional<TSerializedBatch>& sbL, std::optional<TSerializedBatch>& sbR);
96102
static TSerializedBatch Build(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context);
97103

98-
TSerializedBatch(TString&& schemaData, TString&& data, const ui32 rowsCount, const ui32 rawBytes, const std::optional<TString>& specialKeys)
99-
: SchemaData(schemaData)
100-
, Data(data)
104+
TSerializedBatch(TString&& data, const ui32 rowsCount, const ui32 rawBytes,
105+
const std::optional<TString>& specialKeysPayload, const std::optional<TString>& specialKeysFull)
106+
: Data(data)
101107
, RowsCount(rowsCount)
102108
, RawBytes(rawBytes)
103-
, SpecialKeys(specialKeys)
104-
{
105-
109+
, SpecialKeysFull(specialKeysFull)
110+
, SpecialKeysPayload(specialKeysPayload) {
111+
AFL_VERIFY(!!SpecialKeysPayload == !!SpecialKeysFull);
106112
}
107113
};
108114

‎ydb/core/formats/arrow/special_keys.cpp

+14-10
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,12 @@ NKikimr::NArrow::TReplaceKey TSpecialKeys::GetKeyByIndex(const ui32 position, co
2727
}
2828
}
2929

30-
TString TSpecialKeys::SerializeToString() const {
31-
return NArrow::NSerialization::TSerializerContainer::GetDefaultSerializer()->SerializeFull(Data);
30+
TString TSpecialKeys::SerializePayloadToString() const {
31+
return NArrow::NSerialization::TSerializerContainer::GetFastestSerializer()->SerializePayload(Data);
3232
}
3333

34-
TString TSpecialKeys::SerializeToStringDataOnlyNoCompression() const {
35-
return NArrow::SerializeBatchNoCompression(Data);
34+
TString TSpecialKeys::SerializeFullToString() const {
35+
return NArrow::NSerialization::TSerializerContainer::GetFastestSerializer()->SerializeFull(Data);
3636
}
3737

3838
ui64 TSpecialKeys::GetMemoryBytes() const {
@@ -50,13 +50,17 @@ TFirstLastSpecialKeys::TFirstLastSpecialKeys(const std::shared_ptr<arrow::Record
5050
if (columnNames.size()) {
5151
keyBatch = NArrow::TColumnOperator().VerifyIfAbsent().Extract(batch, columnNames);
5252
}
53-
std::vector<ui64> indexes = {0};
54-
if (batch->num_rows() > 1) {
55-
indexes.emplace_back(batch->num_rows() - 1);
56-
}
53+
if (keyBatch->num_rows() <= 2) {
54+
Data = keyBatch;
55+
} else {
56+
std::vector<ui64> indexes = { 0 };
57+
if (batch->num_rows() > 1) {
58+
indexes.emplace_back(batch->num_rows() - 1);
59+
}
5760

58-
Data = NArrow::CopyRecords(keyBatch, indexes);
59-
Y_ABORT_UNLESS(Data->num_rows() == 1 || Data->num_rows() == 2);
61+
Data = NArrow::CopyRecords(keyBatch, indexes);
62+
Y_ABORT_UNLESS(Data->num_rows() == 1 || Data->num_rows() == 2);
63+
}
6064
}
6165

6266
TMinMaxSpecialKeys::TMinMaxSpecialKeys(std::shared_ptr<arrow::RecordBatch> batch, const std::shared_ptr<arrow::Schema>& schema) {

‎ydb/core/formats/arrow/special_keys.h

+2-3
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,6 @@ class TSpecialKeys {
2222
public:
2323
ui64 GetMemoryBytes() const;
2424

25-
TString SerializeToStringDataOnlyNoCompression() const;
26-
2725
TSpecialKeys(const TString& data, const std::shared_ptr<arrow::Schema>& schema) {
2826
Data = NArrow::DeserializeBatch(data, schema);
2927
Y_ABORT_UNLESS(Data);
@@ -34,7 +32,8 @@ class TSpecialKeys {
3432
Y_ABORT_UNLESS(DeserializeFromString(data));
3533
}
3634

37-
TString SerializeToString() const;
35+
TString SerializePayloadToString() const;
36+
TString SerializeFullToString() const;
3837
ui64 GetMemorySize() const;
3938
};
4039

‎ydb/core/protos/tx_columnshard.proto

+1
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@ message TLogicalMetadata {
8989
optional string SpecialKeysRawData = 6;
9090
optional TEvWrite.EModificationType ModificationType = 7;
9191
optional NKikimrArrowSchema.TSchemaSubset SchemaSubset = 8;
92+
optional string SpecialKeysPayloadData = 9;
9293
}
9394

9495
message TEvWriteResult {

‎ydb/core/tx/columnshard/blobs_action/transaction/tx_write.cpp

+2-1
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,8 @@ bool TTxWrite::InsertOneBlob(TTransactionContext& txc, const NOlap::TWideSeriali
1010
meta.SetNumRows(batch->GetRowsCount());
1111
meta.SetRawBytes(batch->GetRawBytes());
1212
meta.SetDirtyWriteTimeSeconds(batch.GetStartInstant().Seconds());
13-
meta.SetSpecialKeysRawData(batch->GetSpecialKeysSafe());
13+
meta.SetSpecialKeysRawData(batch->GetSpecialKeysFullSafe());
14+
meta.SetSpecialKeysPayloadData(batch->GetSpecialKeysPayloadSafe());
1415

1516
const auto& blobRange = batch.GetRange();
1617
Y_ABORT_UNLESS(blobRange.GetBlobId().IsValid());

‎ydb/core/tx/columnshard/columnshard__progress_tx.cpp

+3-1
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ class TColumnShard::TTxProgressTx: public TTransactionBase<TColumnShard> {
1515
std::optional<NOlap::TSnapshot> LastCompletedTx;
1616
std::optional<TTxController::TPlanQueueItem> PlannedQueueItem;
1717
std::optional<TMonotonic> StartExecution;
18+
const TMonotonic ConstructionInstant = TMonotonic::Now();
1819

1920
public:
2021
TTxProgressTx(TColumnShard* self)
@@ -95,8 +96,9 @@ class TColumnShard::TTxProgressTx: public TTransactionBase<TColumnShard> {
9596
Self->LastCompletedTx = std::max(*LastCompletedTx, Self->LastCompletedTx);
9697
}
9798
if (StartExecution) {
98-
Self->GetProgressTxController().GetCounters().OnTxProgressDuration(TxOperator->GetOpType(), TMonotonic::Now() - *StartExecution);
99+
Self->GetProgressTxController().GetCounters().OnTxExecuteDuration(TxOperator->GetOpType(), TMonotonic::Now() - *StartExecution);
99100
}
101+
Self->GetProgressTxController().GetCounters().OnTxLiveDuration(TxOperator->GetOpType(), TMonotonic::Now() - ConstructionInstant);
100102
Self->SetupIndexation();
101103
}
102104
};

‎ydb/core/tx/columnshard/counters/tx_progress.h

+10-4
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,8 @@ class TTxProgressCounters: public TCommonCountersOwner {
2424
NMonitoring::TDynamicCounters::TCounterPtr FinishProposeOnComplete;
2525
NMonitoring::TDynamicCounters::TCounterPtr FinishPlannedTx;
2626
NMonitoring::TDynamicCounters::TCounterPtr AbortTx;
27-
NMonitoring::THistogramPtr HistogramTxProgressDuration;
27+
NMonitoring::THistogramPtr HistogramTxExecuteDuration;
28+
NMonitoring::THistogramPtr HistogramTxLiveDuration;
2829
NMonitoring::THistogramPtr HistogramTxProgressLag;
2930

3031
TProgressCounters(const TCommonCountersOwner& owner)
@@ -37,16 +38,21 @@ class TTxProgressCounters: public TCommonCountersOwner {
3738
, FinishProposeOnComplete(TBase::GetDeriviative("FinishProposeOnComplete"))
3839
, FinishPlannedTx(TBase::GetDeriviative("FinishPlannedTx"))
3940
, AbortTx(TBase::GetDeriviative("AbortTx"))
40-
, HistogramTxProgressDuration(TBase::GetHistogram("TxProgress/Execution/DurationMs", NMonitoring::ExponentialHistogram(18, 2, 5)))
41+
, HistogramTxExecuteDuration(TBase::GetHistogram("TxProgress/Execution/DurationMs", NMonitoring::ExponentialHistogram(18, 2, 5)))
42+
, HistogramTxLiveDuration(TBase::GetHistogram("TxProgress/Live/DurationMs", NMonitoring::ExponentialHistogram(18, 2, 5)))
4143
, HistogramTxProgressLag(TBase::GetHistogram("TxProgress/LagOnComplete/DurationMs", NMonitoring::ExponentialHistogram(18, 2, 5))) {
4244
}
4345
};
4446

4547
THashMap<TOpType, TProgressCounters> CountersByOpType;
4648

4749
public:
48-
void OnTxProgressDuration(const TString& opType, const TDuration d) {
49-
GetSubGroup(opType).HistogramTxProgressDuration->Collect(d.MilliSeconds());
50+
void OnTxExecuteDuration(const TString& opType, const TDuration d) {
51+
GetSubGroup(opType).HistogramTxExecuteDuration->Collect(d.MilliSeconds());
52+
}
53+
54+
void OnTxLiveDuration(const TString& opType, const TDuration d) {
55+
GetSubGroup(opType).HistogramTxLiveDuration->Collect(d.MilliSeconds());
5056
}
5157

5258
void OnTxProgressLag(const TString& opType, const TDuration d) {

‎ydb/core/tx/columnshard/engines/insert_table/meta.cpp

+16-7
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,23 @@ NKikimrTxColumnShard::TLogicalMetadata TInsertedDataMeta::SerializeToProto() con
66
return OriginalProto;
77
}
88

9-
const std::optional<NKikimr::NArrow::TFirstLastSpecialKeys>& TInsertedDataMeta::GetSpecialKeys() const {
10-
if (!KeysParsed) {
11-
if (OriginalProto.HasSpecialKeysRawData()) {
12-
SpecialKeysParsed = NArrow::TFirstLastSpecialKeys(OriginalProto.GetSpecialKeysRawData());
13-
}
14-
KeysParsed = true;
9+
std::shared_ptr<NArrow::TFirstLastSpecialKeys> TInsertedDataMeta::GetSpecialKeys(const std::shared_ptr<arrow::Schema>& schema) const {
10+
if (KeyInitialized.Val()) {
11+
return SpecialKeysParsed;
1512
}
16-
return SpecialKeysParsed;
13+
std::shared_ptr<NArrow::TFirstLastSpecialKeys> result;
14+
if (OriginalProto.HasSpecialKeysPayloadData()) {
15+
result = std::make_shared<NArrow::TFirstLastSpecialKeys>(OriginalProto.GetSpecialKeysPayloadData(), schema);
16+
} else if (OriginalProto.HasSpecialKeysRawData()) {
17+
result = std::make_shared<NArrow::TFirstLastSpecialKeys>(OriginalProto.GetSpecialKeysRawData());
18+
} else {
19+
AFL_VERIFY(false);
20+
}
21+
if (AtomicCas(&KeyInitialization, 1, 0)) {
22+
SpecialKeysParsed = result;
23+
KeyInitialized = 1;
24+
}
25+
return result;
1726
}
1827

1928
}

‎ydb/core/tx/columnshard/engines/insert_table/meta.h

+8-16
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,12 @@ class TInsertedDataMeta {
1717
YDB_READONLY(NEvWrite::EModificationType, ModificationType, NEvWrite::EModificationType::Upsert);
1818
YDB_READONLY_DEF(NArrow::TSchemaSubset, SchemaSubset);
1919

20-
mutable bool KeysParsed = false;
21-
mutable std::optional<NArrow::TFirstLastSpecialKeys> SpecialKeysParsed;
22-
20+
mutable TAtomicCounter KeyInitialized = 0;
21+
mutable TAtomic KeyInitialization = 0;
22+
mutable std::shared_ptr<NArrow::TFirstLastSpecialKeys> SpecialKeysParsed;
2323
NKikimrTxColumnShard::TLogicalMetadata OriginalProto;
24+
std::shared_ptr<NArrow::TFirstLastSpecialKeys> GetSpecialKeys(const std::shared_ptr<arrow::Schema>& schema) const;
2425

25-
const std::optional<NArrow::TFirstLastSpecialKeys>& GetSpecialKeys() const;
2626
public:
2727
ui64 GetTxVolume() const {
2828
return 2 * sizeof(ui64) + sizeof(ui32) + sizeof(OriginalProto) + (SpecialKeysParsed ? SpecialKeysParsed->GetMemoryBytes() : 0);
@@ -43,19 +43,11 @@ class TInsertedDataMeta {
4343
}
4444
}
4545

46-
std::optional<NArrow::TReplaceKey> GetFirstPK(const std::shared_ptr<arrow::Schema>& schema) const {
47-
if (GetSpecialKeys()) {
48-
return GetSpecialKeys()->GetFirst(schema);
49-
} else {
50-
return {};
51-
}
46+
NArrow::TReplaceKey GetFirstPK(const std::shared_ptr<arrow::Schema>& schema) const {
47+
return GetSpecialKeys(schema)->GetFirst();
5248
}
53-
std::optional<NArrow::TReplaceKey> GetLastPK(const std::shared_ptr<arrow::Schema>& schema) const {
54-
if (GetSpecialKeys()) {
55-
return GetSpecialKeys()->GetLast(schema);
56-
} else {
57-
return {};
58-
}
49+
NArrow::TReplaceKey GetLastPK(const std::shared_ptr<arrow::Schema>& schema) const {
50+
return GetSpecialKeys(schema)->GetLast();
5951
}
6052

6153
NKikimrTxColumnShard::TLogicalMetadata SerializeToProto() const;

‎ydb/core/tx/columnshard/engines/portions/meta.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ NKikimrTxColumnShard::TIndexPortionMeta TPortionMeta::SerializeToProto() const {
3333
break;
3434
}
3535

36-
portionMeta.SetPrimaryKeyBorders(ReplaceKeyEdges.SerializeToStringDataOnlyNoCompression());
36+
portionMeta.SetPrimaryKeyBorders(ReplaceKeyEdges.SerializePayloadToString());
3737

3838
RecordSnapshotMin.SerializeToProto(*portionMeta.MutableRecordSnapshotMin());
3939
RecordSnapshotMax.SerializeToProto(*portionMeta.MutableRecordSnapshotMax());

‎ydb/core/tx/columnshard/engines/predicate/range.cpp

+1-19
Original file line numberDiff line numberDiff line change
@@ -40,25 +40,7 @@ NKikimr::NArrow::TColumnFilter TPKRangeFilter::BuildFilter(const arrow::Datum& d
4040
}
4141

4242
bool TPKRangeFilter::IsPortionInUsage(const TPortionInfo& info) const {
43-
if (const auto& from = PredicateFrom.GetReplaceKey()) {
44-
const auto& portionEnd = info.IndexKeyEnd();
45-
const int commonSize = std::min(from->Size(), portionEnd.Size());
46-
if (std::is_gt(from->ComparePartNotNull(portionEnd, commonSize))) {
47-
return false;
48-
}
49-
}
50-
51-
if (const auto& to = PredicateTo.GetReplaceKey()) {
52-
const auto& portionStart = info.IndexKeyStart();
53-
const int commonSize = std::min(to->Size(), portionStart.Size());
54-
if (std::is_lt(to->ComparePartNotNull(portionStart, commonSize))) {
55-
return false;
56-
}
57-
}
58-
// AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("start", info.IndexKeyStart().DebugString())("end", info.IndexKeyEnd().DebugString())(
59-
// "from", PredicateFrom.DebugString())("to", PredicateTo.DebugString());
60-
61-
return true;
43+
return IsPortionInPartialUsage(info.IndexKeyStart(), info.IndexKeyEnd()) != TPKRangeFilter::EUsageClass::DontUsage;
6244
}
6345

6446
TPKRangeFilter::EUsageClass TPKRangeFilter::IsPortionInPartialUsage(const NArrow::TReplaceKey& start, const NArrow::TReplaceKey& end) const {

‎ydb/core/tx/columnshard/normalizer/tablet/broken_insertion_dedup.cpp renamed to ‎ydb/core/tx/columnshard/normalizer/insert_table/broken_dedup.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
#include "broken_insertion_dedup.h"
1+
#include "broken_dedup.h"
22

33
#include <ydb/core/tx/columnshard/columnshard_private_events.h>
44
#include <ydb/core/tx/columnshard/columnshard_schema.h>

‎ydb/core/tx/columnshard/normalizer/tablet/broken_insertion_dedup.h renamed to ‎ydb/core/tx/columnshard/normalizer/insert_table/broken_dedup.h

File renamed without changes.

‎ydb/core/tx/columnshard/normalizer/insert_table/ya.make

+11
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
LIBRARY()
2+
3+
SRCS(
4+
GLOBAL broken_dedup.cpp
5+
)
6+
7+
PEERDIR(
8+
ydb/core/tx/columnshard/normalizer/abstract
9+
)
10+
11+
END()

‎ydb/core/tx/columnshard/normalizer/tablet/ya.make

-1
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ LIBRARY()
33
SRCS(
44
GLOBAL gc_counters.cpp
55
GLOBAL broken_txs.cpp
6-
GLOBAL broken_insertion_dedup.cpp
76
)
87

98
PEERDIR(

‎ydb/core/tx/columnshard/normalizer/ya.make

+1
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ PEERDIR(
66
ydb/core/tx/columnshard/normalizer/tablet
77
ydb/core/tx/columnshard/normalizer/tables
88
ydb/core/tx/columnshard/normalizer/portion
9+
ydb/core/tx/columnshard/normalizer/insert_table
910
)
1011

1112
END()

‎ydb/core/tx/data_events/columnshard_splitter.cpp

+2-1
Original file line numberDiff line numberDiff line change
@@ -66,9 +66,10 @@ NKikimr::NEvWrite::IShardsSplitter::TYdbConclusionStatus TColumnShardShardsSplit
6666
}
6767

6868
TFullSplitData result(sharding->GetShardsCount());
69+
const TString schemaString = NArrow::SerializeSchema(*batch->schema());
6970
for (auto&& [shardId, chunks] : split.GetResult()) {
7071
for (auto&& c : chunks) {
71-
result.AddShardInfo(shardId, std::make_shared<TShardInfo>(c.GetSchemaData(), c.GetData(), c.GetRowsCount(), sharding->GetShardInfoVerified(shardId).GetShardingVersion()));
72+
result.AddShardInfo(shardId, std::make_shared<TShardInfo>(schemaString, c.GetData(), c.GetRowsCount(), sharding->GetShardInfoVerified(shardId).GetShardingVersion()));
7273
}
7374
}
7475

0 commit comments

Comments
 (0)