Fixed issues 1

Author: GrigoriyPA

ydb/core/fq/libs/row_dispatcher/format_handler/filters/filters_set.cpp

@@ -185,7 +185,7 @@ class TTopicFilters : public ITopicFilters {
 
         IPurecalcFilter::TPtr purecalcFilter;
         if (const auto& predicate = filter->GetWhereFilter()) {
-            LOG_ROW_DISPATCHER_TRACE("Create purecalc filter for predicate '" << predicate << "' (client id: " << filter->GetFilterId() << ")");
+            LOG_ROW_DISPATCHER_TRACE("Create purecalc filter for predicate '" << predicate << "' (filter id: " << filter->GetFilterId() << ")");
 
             auto filterStatus = CreatePurecalcFilter(filter);
             if (filterStatus.IsFail()) {
@@ -237,13 +237,11 @@ class TTopicFilters : public ITopicFilters {
         }
 
         if (const auto filter = filterHandler.GetPurecalcFilter()) {
-            LOG_ROW_DISPATCHER_TRACE("Pass " << numberRows << " rows to purecalc filter (client id: " << consumer->GetFilterId() << ")");
+            LOG_ROW_DISPATCHER_TRACE("Pass " << numberRows << " rows to purecalc filter (filter id: " << consumer->GetFilterId() << ")");
             filter->FilterData(result, numberRows);
-        } else {
+        } else if (numberRows) {
             LOG_ROW_DISPATCHER_TRACE("Add " << numberRows << " rows to client " << consumer->GetFilterId() << " without filtering");
-            for (ui64 rowId = 0; rowId < numberRows; ++rowId) {
-                consumer->OnFilteredData(rowId);
-            }
+            consumer->OnFilteredBatch(0, numberRows - 1);
         }
     }
 

ydb/core/fq/libs/row_dispatcher/format_handler/filters/filters_set.h

@@ -17,6 +17,8 @@ class IFilteredDataConsumer : public IPurecalcFilterConsumer {
     virtual const TVector<ui64>& GetColumnIds() const = 0;
     virtual TMaybe<ui64> GetNextMessageOffset() const = 0;
 
+    virtual void OnFilteredBatch(ui64 firstRow, ui64 lastRow) = 0;  // inclusive interval [firstRow, lastRow]
+
     virtual void OnFilterStarted() = 0;
     virtual void OnFilteringError(TStatus status) = 0;
 };

ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.cpp

@@ -207,6 +207,13 @@ class TTopicFormatHandler : public NActors::TActor<TTopicFormatHandler>, public
             Client->StartClientSession();
         }
 
+        void OnFilteredBatch(ui64 firstRow, ui64 lastRow) override {
+            LOG_ROW_DISPATCHER_TRACE("OnFilteredBatch, rows [" << firstRow << ", " << lastRow << "]");
+            for (ui64 rowId = firstRow; rowId <= lastRow; ++rowId) {
+                OnFilteredData(rowId);
+            }
+        }
+
         void OnFilteredData(ui64 rowId) override {
             const ui64 offset = Self.Offsets->at(rowId);
             if (const auto nextOffset = Client->GetNextMessageOffset(); nextOffset && offset < *nextOffset) {

ydb/core/fq/libs/row_dispatcher/format_handler/parsers/json_parser.cpp

@@ -414,23 +414,23 @@ class TJsonParser : public TTopicParserBase {
 
         simdjson::ondemand::document_stream documents;
         CHECK_JSON_ERROR(Parser.iterate_many(values, size, simdjson::ondemand::DEFAULT_BATCH_SIZE).get(documents)) {
-            return TStatus::Fail(EStatusId::BAD_REQUEST, TStringBuilder() << "Failed to parse message batch from offset " << Buffer.Offsets.front() << ", json documents was corrupted: " << simdjson::error_message(error) << " Current data batch:\n" << TruncateString(std::string_view(values, size)));
+            return TStatus::Fail(EStatusId::BAD_REQUEST, TStringBuilder() << "Failed to parse message batch from offset " << Buffer.Offsets.front() << ", json documents was corrupted: " << simdjson::error_message(error) << " Current data batch: " << TruncateString(std::string_view(values, size)));
         }
 
         size_t rowId = 0;
         for (auto document : documents) {
             if (Y_UNLIKELY(rowId >= Buffer.NumberValues)) {
-                return TStatus::Fail(EStatusId::INTERNAL_ERROR, TStringBuilder() << "Failed to parse json messages, expected " << Buffer.NumberValues << " json rows from offset " << Buffer.Offsets.front() << " but got " << rowId + 1 << " (expected one json row for each offset from topic API in json each row format, maybe initial data was corrupted or messages is not in json format), current data batch:\n" << TruncateString(std::string_view(values, size)));
+                return TStatus::Fail(EStatusId::INTERNAL_ERROR, TStringBuilder() << "Failed to parse json messages, expected " << Buffer.NumberValues << " json rows from offset " << Buffer.Offsets.front() << " but got " << rowId + 1 << " (expected one json row for each offset from topic API in json each row format, maybe initial data was corrupted or messages is not in json format), current data batch: " << TruncateString(std::string_view(values, size)));
             }
 
             const ui64 offset = Buffer.Offsets[rowId];
             CHECK_JSON_ERROR(document.error()) {
-                return TStatus::Fail(EStatusId::BAD_REQUEST, TStringBuilder() << "Failed to parse json message for offset " << offset << ", json document was corrupted: " << simdjson::error_message(error) << " Current data batch:\n" << TruncateString(std::string_view(values, size)));
+                return TStatus::Fail(EStatusId::BAD_REQUEST, TStringBuilder() << "Failed to parse json message for offset " << offset << ", json document was corrupted: " << simdjson::error_message(error) << " Current data batch: " << TruncateString(std::string_view(values, size)));
             }
 
             for (auto item : document.get_object()) {
                 CHECK_JSON_ERROR(item.error()) {
-                    return TStatus::Fail(EStatusId::BAD_REQUEST, TStringBuilder() << "Failed to parse json message for offset " << offset << ", json item was corrupted: " << simdjson::error_message(error) << " Current data batch:\n" << TruncateString(std::string_view(values, size)));
+                    return TStatus::Fail(EStatusId::BAD_REQUEST, TStringBuilder() << "Failed to parse json message for offset " << offset << ", json item was corrupted: " << simdjson::error_message(error) << " Current data batch: " << TruncateString(std::string_view(values, size)));
                 }
 
                 const auto it = ColumnsIndex.find(item.escaped_key().value());
@@ -445,7 +445,7 @@ class TJsonParser : public TTopicParserBase {
         }
 
         if (Y_UNLIKELY(rowId != Buffer.NumberValues)) {
-            return TStatus::Fail(EStatusId::INTERNAL_ERROR, TStringBuilder() << "Failed to parse json messages, expected " << Buffer.NumberValues << " json rows from offset " << Buffer.Offsets.front() << " but got " << rowId << " (expected one json row for each offset from topic API in json each row format, maybe initial data was corrupted or messages is not in json format), current data batch:\n" << TruncateString(std::string_view(values, size)));
+            return TStatus::Fail(EStatusId::INTERNAL_ERROR, TStringBuilder() << "Failed to parse json messages, expected " << Buffer.NumberValues << " json rows from offset " << Buffer.Offsets.front() << " but got " << rowId << " (expected one json row for each offset from topic API in json each row format, maybe initial data was corrupted or messages is not in json format), current data batch: " << TruncateString(std::string_view(values, size)));
         }
 
         const ui64 firstOffset = Buffer.Offsets.front();

ydb/core/fq/libs/row_dispatcher/format_handler/ut/topic_filter_ut.cpp

@@ -64,6 +64,13 @@ class TFiterFixture : public TBaseFixture {
             }
         }
 
+        void OnFilteredBatch(ui64 firstRow, ui64 lastRow) override {
+            UNIT_ASSERT_C(Started, "Unexpected data for not started filter");
+            for (ui64 rowId = firstRow; rowId <= lastRow; ++rowId) {
+                Callback(rowId);
+            }
+        }
+
         void OnFilteredData(ui64 rowId) override {
             UNIT_ASSERT_C(Started, "Unexpected data for not started filter");
             Callback(rowId);