tmp changes

Author: kardymonds

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

@@ -225,7 +225,9 @@ class TJsonFilter::TImpl {
         const TString& whereFilter,
         TCallback callback)
         : Sql(GenerateSql(columns, types, whereFilter)) {
-        auto factory = NYql::NPureCalc::MakeProgramFactory(NYql::NPureCalc::TProgramFactoryOptions());
+
+
+        auto factory = NYql::NPureCalc::MakeProgramFactory(NYql::NPureCalc::TProgramFactoryOptions().SetUDFsDir("/home/kardymon-d/ydb3/ydb/ydb/library/yql/udfs/common"));
 
         LOG_ROW_DISPATCHER_DEBUG("Creating program...");
         Program = factory->MakePushStreamProgram(

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

@@ -232,7 +232,7 @@ class TJsonParser::TImpl {
         const TVector<TString>& columns,
         TCallback callback)
         : Sql(GenerateSql(columns)) {
-        auto options = NYql::NPureCalc::TProgramFactoryOptions();
+        auto options = NYql::NPureCalc::TProgramFactoryOptions().SetUDFsDir("/home/kardymon-d/ydb3/ydb/ydb/library/yql/udfs/common");
         auto factory = NYql::NPureCalc::MakeProgramFactory(options);
 
         LOG_ROW_DISPATCHER_DEBUG("Creating program...");

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

@@ -549,6 +549,9 @@ void TTopicSession::SendToParsing(ui64 offset, const TString& message) {
         }
     }
 
+    if (ClientsWithoutPredicate.size() == Clients.size())
+        return;
+
     try {
         Parser->Push(offset, message);
     } catch (const std::exception& e) {
@@ -617,7 +620,7 @@ void TTopicSession::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
             std::forward_as_tuple(ev->Sender), 
             std::forward_as_tuple(ev)).first->second;
 
-        TString predicate = clientInfo.Settings.GetSource().GetPredicate();
+        TString predicate;// = clientInfo.Settings.GetSource().GetPredicate();
         if (!predicate.empty()) {
             clientInfo.Filter = NewJsonFilter(
                 columns,
@@ -630,6 +633,8 @@ void TTopicSession::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
             ClientsWithoutPredicate.insert(ev->Sender);
         }
 
+        Cerr << "New client" << Endl;
+
         LOG_ROW_DISPATCHER_INFO("New client: offset " << clientInfo.NextMessageOffset << ", predicate: " << clientInfo.Settings.GetSource().GetPredicate());
 
         if (ReadSession) {

ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h

@@ -1386,6 +1386,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
                         .OutputIndex = outputIndex,
                         .StatsLevel = collectStatsLevel,
                         .TxId = TxId,
+                        .TaskId = Task.GetId(),
                         .Callback = static_cast<TSinkCallbacks*>(this),
                         .SecureParams = secureParams,
                         .TaskParams = taskParams,

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

@@ -152,7 +152,7 @@ class TDqPqWriteActor : public NActors::TActor<TDqPqWriteActor>, public IDqCompu
         , LogPrefix(TStringBuilder() << "SelfId: " << this->SelfId() << ", TxId: " << TxId << ", TaskId: " << taskId << ", PQ sink. ")
         , FreeSpace(freeSpace)
         , TopicClient(Driver, GetTopicClientSettings())
-        , File("data/result/" + std::get<TString>(TxId), EOpenModeFlag::CreateAlways |  EOpenModeFlag::WrOnly)
+        , File("data/result/" + std::get<TString>(TxId) + "_" + ToString(taskId), EOpenModeFlag::CreateAlways |  EOpenModeFlag::WrOnly)
     { 
         EgressStats.Level = statsLevel;
     }
@@ -317,10 +317,10 @@ class TDqPqWriteActor : public NActors::TActor<TDqPqWriteActor>, public IDqCompu
     }
 
     void CreateSessionIfNotExists() {
-        if (!WriteSession) {
-            WriteSession = TopicClient.CreateWriteSession(GetWriteSessionSettings());
-            SubscribeOnNextEvent();
-        }
+        // if (!WriteSession) {
+        //     WriteSession = TopicClient.CreateWriteSession(GetWriteSessionSettings());
+        //     SubscribeOnNextEvent();
+        // }
     }
 
     void SubscribeOnNextEvent() {