Refactored StaticBufferSize -> BufferCellCount

GrigoriyPA · GrigoriyPA · commit d05bdbde7803 · 2024-11-17T15:47:05.000Z
diff --git a/ydb/core/fq/libs/config/protos/row_dispatcher.proto b/ydb/core/fq/libs/config/protos/row_dispatcher.proto
@@ -17,7 +17,7 @@ message TRowDispatcherCoordinatorConfig {
 message TJsonParserConfig {
     uint64 BatchSizeBytes = 1;  // default 1 MiB
     uint64 BatchCreationTimeoutMs = 2;
-    uint64 StaticBufferSize = 3;  // (number rows) * (number columns) limit, default 10^6
+    uint64 BufferCellCount = 3;  // (number rows) * (number columns) limit, default 10^6
 }
 
 message TRowDispatcherConfig {
diff --git a/ydb/core/fq/libs/row_dispatcher/json_parser.cpp b/ydb/core/fq/libs/row_dispatcher/json_parser.cpp
@@ -20,7 +20,7 @@ namespace {
 TString LogPrefix = "JsonParser: ";
 
 constexpr ui64 DEFAULT_BATCH_SIZE = 1_MB;
-constexpr ui64 DEFAULT_STATIC_BUFFER_SIZE = 1000000;
+constexpr ui64 DEFAULT_BUFFER_CELL_COUNT = 1000000;
 
 struct TJsonParserBuffer {
     size_t NumberValues = 0;
@@ -269,11 +269,11 @@ namespace NFq {
 
 class TJsonParser::TImpl {
 public:
-    TImpl(const TVector<TString>& columns, const TVector<TString>& types, TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 staticBufferSize)
+    TImpl(const TVector<TString>& columns, const TVector<TString>& types, TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 bufferCellCount)
         : Alloc(__LOCATION__, NKikimr::TAlignedPagePoolCounters(), true, false)
         , TypeEnv(std::make_unique<NKikimr::NMiniKQL::TTypeEnvironment>(Alloc))
         , BatchSize(batchSize ? batchSize : DEFAULT_BATCH_SIZE)
-        , MaxNumberRows(((staticBufferSize ? staticBufferSize : DEFAULT_STATIC_BUFFER_SIZE) - 1) / columns.size() + 1)
+        , MaxNumberRows(((bufferCellCount ? bufferCellCount : DEFAULT_BUFFER_CELL_COUNT) - 1) / columns.size() + 1)
         , BatchCreationTimeout(batchCreationTimeout)
         , ParseCallback(parseCallback)
         , ParsedValues(columns.size())
@@ -444,8 +444,8 @@ class TJsonParser::TImpl {
     TVector<TVector<NYql::NUdf::TUnboxedValue>> ParsedValues;
 };
 
-TJsonParser::TJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 staticBufferSize)
-    : Impl(std::make_unique<TJsonParser::TImpl>(columns, types, parseCallback, batchSize, batchCreationTimeout, staticBufferSize))
+TJsonParser::TJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 bufferCellCount)
+    : Impl(std::make_unique<TJsonParser::TImpl>(columns, types, parseCallback, batchSize, batchCreationTimeout, bufferCellCount))
 {}
 
 TJsonParser::~TJsonParser() {
@@ -479,8 +479,8 @@ TString TJsonParser::GetDescription() const {
     return Impl->GetDescription();
 }
 
-std::unique_ptr<TJsonParser> NewJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TJsonParser::TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 staticBufferSize) {
-    return std::unique_ptr<TJsonParser>(new TJsonParser(columns, types, parseCallback, batchSize, batchCreationTimeout, staticBufferSize));
+std::unique_ptr<TJsonParser> NewJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TJsonParser::TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 bufferCellCount) {
+    return std::unique_ptr<TJsonParser>(new TJsonParser(columns, types, parseCallback, batchSize, batchCreationTimeout, bufferCellCount));
 }
 
 } // namespace NFq
diff --git a/ydb/core/fq/libs/row_dispatcher/json_parser.h b/ydb/core/fq/libs/row_dispatcher/json_parser.h
@@ -11,7 +11,7 @@ class TJsonParser {
     using TCallback = std::function<void(ui64 rowsOffset, ui64 numberRows, const TVector<TVector<NYql::NUdf::TUnboxedValue>>& parsedValues)>;
 
 public:
-    TJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 staticBufferSize);
+    TJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 bufferCellCount);
     ~TJsonParser();
 
     bool IsReady() const;
@@ -29,6 +29,6 @@ class TJsonParser {
     const std::unique_ptr<TImpl> Impl;
 };
 
-std::unique_ptr<TJsonParser> NewJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TJsonParser::TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 staticBufferSize);
+std::unique_ptr<TJsonParser> NewJsonParser(const TVector<TString>& columns, const TVector<TString>& types, TJsonParser::TCallback parseCallback, ui64 batchSize, TDuration batchCreationTimeout, ui64 bufferCellCount);
 
 } // namespace NFq
diff --git a/ydb/core/fq/libs/row_dispatcher/topic_session.cpp b/ydb/core/fq/libs/row_dispatcher/topic_session.cpp
@@ -870,7 +870,7 @@ void TTopicSession::UpdateParser() {
         const auto& parserConfig = Config.GetJsonParser();
         Parser = NewJsonParser(names, types, [this](ui64 rowsOffset, ui64 numberRows, const TVector<TVector<NYql::NUdf::TUnboxedValue>>& parsedValues) {
             DoFiltering(rowsOffset, numberRows, parsedValues);
-        }, parserConfig.GetBatchSizeBytes(), TDuration::MilliSeconds(parserConfig.GetBatchCreationTimeoutMs()), parserConfig.GetStaticBufferSize());
+        }, parserConfig.GetBatchSizeBytes(), TDuration::MilliSeconds(parserConfig.GetBatchCreationTimeoutMs()), parserConfig.GetBufferCellCount());
     } catch (const NYql::NPureCalc::TCompileError& e) {
         FatalError(e.GetIssues(), nullptr, true);
     }
diff --git a/ydb/core/fq/libs/row_dispatcher/ut/json_parser_ut.cpp b/ydb/core/fq/libs/row_dispatcher/ut/json_parser_ut.cpp
@@ -46,8 +46,8 @@ class TFixture : public NUnitTest::TBaseFixture {
         }
     }
 
-    void MakeParser(TVector<TString> columns, TVector<TString> types, TJsonParser::TCallback callback, ui64 batchSize = 1_MB, ui64 staticBufferSize = 1000) {
-        Parser = NFq::NewJsonParser(columns, types, callback, batchSize, TDuration::Hours(1), staticBufferSize);
+    void MakeParser(TVector<TString> columns, TVector<TString> types, TJsonParser::TCallback callback, ui64 batchSize = 1_MB, ui64 bufferCellCount = 1000) {
+        Parser = NFq::NewJsonParser(columns, types, callback, batchSize, TDuration::Hours(1), bufferCellCount);
     }
 
     void MakeParser(TVector<TString> columns, TJsonParser::TCallback callback) {

Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@ message TRowDispatcherCoordinatorConfig {`
`17`	`17`	`message TJsonParserConfig {`
`18`	`18`	`uint64 BatchSizeBytes = 1; // default 1 MiB`
`19`	`19`	`uint64 BatchCreationTimeoutMs = 2;`
`20`		`- uint64 StaticBufferSize = 3; // (number rows) * (number columns) limit, default 10^6`
	`20`	`+ uint64 BufferCellCount = 3; // (number rows) * (number columns) limit, default 10^6`
`21`	`21`	`}`
`22`	`22`
`23`	`23`	`message TRowDispatcherConfig {`
Original file line number	Diff line number	Diff line change
`@@ -870,7 +870,7 @@ void TTopicSession::UpdateParser() {`
`870`	`870`	`const auto& parserConfig = Config.GetJsonParser();`
`871`	`871`	`Parser = NewJsonParser(names, types, [this](ui64 rowsOffset, ui64 numberRows, const TVector<TVector<NYql::NUdf::TUnboxedValue>>& parsedValues) {`
`872`	`872`	`DoFiltering(rowsOffset, numberRows, parsedValues);`
`873`		`- }, parserConfig.GetBatchSizeBytes(), TDuration::MilliSeconds(parserConfig.GetBatchCreationTimeoutMs()), parserConfig.GetStaticBufferSize());`
	`873`	`+ }, parserConfig.GetBatchSizeBytes(), TDuration::MilliSeconds(parserConfig.GetBatchCreationTimeoutMs()), parserConfig.GetBufferCellCount());`
`874`	`874`	`} catch (const NYql::NPureCalc::TCompileError& e) {`
`875`	`875`	`FatalError(e.GetIssues(), nullptr, true);`
`876`	`876`	`}`
Original file line number	Diff line number	Diff line change
`@@ -46,8 +46,8 @@ class TFixture : public NUnitTest::TBaseFixture {`
`46`	`46`	`}`
`47`	`47`	`}`
`48`	`48`
`49`		`- void MakeParser(TVector<TString> columns, TVector<TString> types, TJsonParser::TCallback callback, ui64 batchSize = 1_MB, ui64 staticBufferSize = 1000) {`
`50`		`- Parser = NFq::NewJsonParser(columns, types, callback, batchSize, TDuration::Hours(1), staticBufferSize);`
	`49`	`+ void MakeParser(TVector<TString> columns, TVector<TString> types, TJsonParser::TCallback callback, ui64 batchSize = 1_MB, ui64 bufferCellCount = 1000) {`
	`50`	`+ Parser = NFq::NewJsonParser(columns, types, callback, batchSize, TDuration::Hours(1), bufferCellCount);`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`void MakeParser(TVector<TString> columns, TJsonParser::TCallback callback) {`