YQ-3736 Shared reading: fix unsupported predicates (#10256)

Author: kardymonds

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

@@ -168,7 +168,6 @@ void TActorCoordinator::AddRowDispatcher(NActors::TActorId actorId, bool isLocal
 void TActorCoordinator::Handle(NActors::TEvents::TEvPing::TPtr& ev) {
     LOG_ROW_DISPATCHER_TRACE("TEvPing received, " << ev->Sender);
     AddRowDispatcher(ev->Sender, false);
-    PrintInternalState();
     LOG_ROW_DISPATCHER_TRACE("Send TEvPong to " << ev->Sender);
     Send(ev->Sender, new NActors::TEvents::TEvPong(), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
 }

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

@@ -299,8 +299,8 @@ void TRowDispatcher::Handle(TEvPrivate::TEvCoordinatorPing::TPtr&) {
     if (!CoordinatorActorId) {
         return;
     }
-    LOG_ROW_DISPATCHER_DEBUG("Send ping to " << *CoordinatorActorId);
-    Send(*CoordinatorActorId, new NActors::TEvents::TEvPing(), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
+    LOG_ROW_DISPATCHER_TRACE("Send ping to " << *CoordinatorActorId);
+    Send(*CoordinatorActorId, new NActors::TEvents::TEvPing());
 }
 
 void TRowDispatcher::Handle(NActors::TEvents::TEvPong::TPtr&) {

ydb/library/yql/providers/pq/async_io/dq_pq_rd_read_actor.cpp

@@ -598,7 +598,7 @@ void TDqPqRdReadActor::Handle(NFq::TEvRowDispatcher::TEvMessageBatch::TPtr& ev)
         return;
     }
 
-    Metrics.InFlyGetNextBatch->Dec();
+    Metrics.InFlyGetNextBatch->Set(0);
     auto& sessionInfo = it->second;
     if (!sessionInfo.EventsQueue.OnEventReceived(ev)) {
         SRC_LOG_W("Wrong seq num ignore message, seqNo " << meta.GetSeqNo());

ydb/library/yql/providers/pq/provider/yql_pq_dq_integration.cpp

@@ -263,7 +263,8 @@ class TPqDqIntegration: public TDqIntegrationBase {
                 if (auto predicate = topicSource.FilterPredicate(); !NYql::IsEmptyFilterPredicate(predicate)) {
                     TStringBuilder err;
                     if (!NYql::SerializeFilterPredicate(predicate, &predicateProto, err)) {
-                        ythrow yexception() << "Failed to serialize filter predicate for source: " << err;
+                        ctx.AddWarning(TIssue(ctx.GetPosition(node.Pos()), "Failed to serialize filter predicate for source: " + err));
+                        predicateProto.Clear();
                     }
                 }
 

ydb/library/yql/providers/pq/provider/yql_pq_logical_opt.cpp

@@ -257,7 +257,6 @@ class TPqLogicalOptProposalTransformer : public TOptimizeTransformerBase {
 
         auto newFilterLambda = MakePushdownPredicate(flatmap.Lambda(), ctx, node.Pos(), TPushdownSettings());
         if (!newFilterLambda) {
-            ctx.AddWarning(TIssue(ctx.GetPosition(node.Pos()), "No predicate to pushdown"));
             return node;
         }
         YQL_CLOG(INFO, ProviderPq) << "Build new TCoFlatMap with predicate";

ydb/tests/fq/yds/test_row_dispatcher.py

@@ -243,6 +243,31 @@ def test_filter(self, kikimr, client):
         issues = str(client.describe_query(query_id).result.query.transient_issue)
         assert "Row dispatcher will use the predicate: WHERE (`time` > 101" in issues, "Incorrect Issues: " + issues
 
+    @yq_v1
+    def test_filter_use_unsupported_predicate(self, kikimr, client):
+        client.create_yds_connection(
+            YDS_CONNECTION, os.getenv("YDB_DATABASE"), os.getenv("YDB_ENDPOINT"), shared_reading=True
+        )
+        self.init_topics("test_filter_use_unsupported_predicate")
+
+        sql = Rf'''
+            INSERT INTO {YDS_CONNECTION}.`{self.output_topic}`
+            SELECT Cast(time as String) FROM {YDS_CONNECTION}.`{self.input_topic}`
+                WITH (format=json_each_row, SCHEMA (time UInt64 NOT NULL, data String NOT NULL, event String NOT NULL))
+                WHERE  event LIKE 'event2%';'''
+
+        query_id = start_yds_query(kikimr, client, sql)
+        wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_SESSION", 1)
+
+        data = [
+            '{"time": 102, "data": "hello2", "event": "event2"}',
+        ]
+
+        self.write_stream(data)
+        assert self.read_stream(1, topic_path=self.output_topic) == ['102']
+        wait_actor_count(kikimr, "DQ_PQ_READ_ACTOR", 1)
+        stop_yds_query(client, query_id)
+
     @yq_v1
     def test_filter_with_mr(self, kikimr, client):
         client.create_yds_connection(