cdc_test

Author: ubyte

ydb/core/tx/schemeshard/ut_cdc_stream/ut_cdc_stream.cpp

@@ -1134,6 +1134,7 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions()
             .EnableProtoSourceIdInfo(true)
+            .EnableChangefeedInitialScan(true)
             .EnablePqBilling(serverless));
         ui64 txId = 100;
 
@@ -1202,25 +1203,44 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
             NLs::PathExist,
             NLs::ExtractTenantSchemeshard(&schemeShard)
         });
-
         UNIT_ASSERT(schemeShard != 0 && schemeShard != TTestTxConfig::SchemeShard);
 
         TestCreateTable(runtime, schemeShard, ++txId, dbName, R"(
             Name: "Table"
             Columns { Name: "key" Type: "Uint64" }
-            Columns { Name: "value" Type: "Uint64" }
+            Columns { Name: "value" Type: "Utf8" }
             KeyColumnNames: ["key"]
             UniformPartitionsCount: 2
         )");
         env.TestWaitNotification(runtime, txId, schemeShard);
 
         runtime.SetLogPriority(NKikimrServices::PERSQUEUE, NLog::PRI_NOTICE);
+        {
+            // should write at least 16MiB to exceed the topics non-metered limit
+            const TString value = TString(500_KB, 'x');
+            for (size_t i = 0; i < 100; ++i) {
+                const unsigned key = 3000 + i;
+                const TString writeQuery = Sprintf(R"(
+                (
+                    (let key   '( '('key   (Uint64 '%u ) ) ) )
+                    (let row   '( '('value (Utf8 '%s) ) ) )
+                    (return (AsList (UpdateRow '__user__%s key row) ))
+                )
+            )", key, value.c_str(), "Table");
+                NKikimrMiniKQL::TResult result;
+                TString err;
+                NKikimrProto::EReplyStatus status = LocalMiniKQL(runtime, TTestTxConfig::FakeHiveTablets + 6, writeQuery, result, err);
+                UNIT_ASSERT_VALUES_EQUAL(err, "");
+                UNIT_ASSERT_VALUES_EQUAL(status, NKikimrProto::EReplyStatus::OK);
+            }
+        }
+
         TVector<TString> meteringRecords;
         runtime.SetObserverFunc([&meteringRecords](TAutoPtr<IEventHandle>& ev) {
             if (ev->GetTypeRewrite() != NMetering::TEvMetering::EvWriteMeteringJson) {
                 return TTestActorRuntime::EEventAction::PROCESS;
             }
-
+            Cerr << "GOT METERING RECORD: " << ev->Get<NMetering::TEvMetering::TEvWriteMeteringJson>()->MeteringJson << Endl;
             meteringRecords.push_back(ev->Get<NMetering::TEvMetering::TEvWriteMeteringJson>()->MeteringJson);
             return TTestActorRuntime::EEventAction::PROCESS;
         });
@@ -1229,8 +1249,9 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
             TableName: "Table"
             StreamDescription {
               Name: "Stream"
-              Mode: ECdcStreamModeKeysOnly
+              Mode: ECdcStreamModeNewImage
               Format: ECdcStreamFormatProto
+              State: ECdcStreamStateScan
             }
         )");
         env.TestWaitNotification(runtime, txId, schemeShard);
@@ -1239,26 +1260,22 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
             env.SimulateSleep(runtime, TDuration::Seconds(10));
         }
 
-        for (const auto& rec : meteringRecords) {
-            Cerr << "GOT METERING: " << rec << "\n";
-        }
-
         UNIT_ASSERT_VALUES_EQUAL(meteringRecords.size(), (serverless ? 3 : 0));
 
         if (!meteringRecords) {
             return;
         }
 
         NJson::TJsonValue json;
-        NJson::ReadJsonTree(meteringRecords[0], &json, true);
+        NJson::ReadJsonTree(meteringRecords.back(), &json, true);
         auto& map = json.GetMap();
         UNIT_ASSERT(map.contains("schema"));
         UNIT_ASSERT(map.contains("resource_id"));
         UNIT_ASSERT(map.contains("tags"));
         UNIT_ASSERT(map.find("tags")->second.GetMap().contains("ydb_size"));
         UNIT_ASSERT_VALUES_EQUAL(map.find("schema")->second.GetString(), "ydb.serverless.v1");
         UNIT_ASSERT_VALUES_EQUAL(map.find("resource_id")->second.GetString(), Sprintf("%s/Table/Stream/streamImpl", dbName.c_str()));
-        UNIT_ASSERT_VALUES_EQUAL(map.find("tags")->second.GetMap().find("ydb_size")->second.GetInteger(), 0);
+        UNIT_ASSERT_GT(map.find("tags")->second.GetMap().find("ydb_size")->second.GetInteger(), 0);
     }
 
     Y_UNIT_TEST(MeteringServerless) {