YQ-3738 Fix metafields in read_actor / to stable (#12725)

Author: kardymonds

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

@@ -797,7 +797,7 @@ void TDqPqRdReadActor::AddMessageBatch(TRope&& messageBatch, NKikimr::NMiniKQL::
                 *(itemPtr++) = parsedRow[*index];
             } else {
                 // TODO: support metadata fields here
-                *(itemPtr++) = NUdf::TUnboxedValue();
+                *(itemPtr++) = NUdf::TUnboxedValuePod::Zero();
             }
         }
 

ydb/tests/fq/yds/test_row_dispatcher.py

@@ -146,7 +146,6 @@ def test_simple_not_null(self, kikimr, client):
 
         query_id = start_yds_query(kikimr, client, sql)
         wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_SESSION", 1)
-        time.sleep(10)
 
         data = [
             '{"time": 101, "data": "hello1", "event": "event1"}',
@@ -167,6 +166,35 @@ def test_simple_not_null(self, kikimr, client):
         assert len(read_rules) == 0, read_rules
         wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_SESSION", 0)
 
+    @yq_v1
+    def test_metadatafields(self, kikimr, client):
+        client.create_yds_connection(
+            YDS_CONNECTION, os.getenv("YDB_DATABASE"), os.getenv("YDB_ENDPOINT"), shared_reading=True
+        )
+        self.init_topics("test_metadatafields")
+
+        # Its not completely clear why metadatafields appear in this request(
+        sql = Rf'''
+            PRAGMA FeatureR010="prototype";
+            PRAGMA config.flags("TimeOrderRecoverDelay", "-10");
+            PRAGMA config.flags("TimeOrderRecoverAhead", "10");
+            INSERT INTO {YDS_CONNECTION}.`{self.output_topic}`
+            SELECT ToBytes(Unwrap(Json::SerializeJson(Yson::From(TableRow())))) FROM {YDS_CONNECTION}.`{self.input_topic}`
+                WITH (format=json_each_row, SCHEMA (time Int32 NOT NULL))
+                MATCH_RECOGNIZE(
+                    ORDER BY CAST(time as Timestamp)
+                    MEASURES LAST(A.time) as b_key
+                    PATTERN (A )
+                    DEFINE A as A.time > 4
+                );'''
+
+        query_id = start_yds_query(kikimr, client, sql)
+        wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_SESSION", 1)
+
+        self.write_stream(['{"time": 100}', '{"time": 120}'])
+        assert len(self.read_stream(1, topic_path=self.output_topic)) == 1
+        stop_yds_query(client, query_id)
+
     @yq_v1
     def test_simple_optional(self, kikimr, client):
         client.create_yds_connection(