@@ -40,6 +40,8 @@ import third_party;
4040import base_table_ref;
4141import index_secondary;
4242import data_block;
43+ import bg_task;
44+ import logger;
4345
4446using namespace infinity ;
4547
@@ -413,3 +415,183 @@ TEST_F(CatalogDeltaReplayTest, replay_append) {
413415 infinity::InfinityContext::instance ().UnInit ();
414416 }
415417}
418+
419+ TEST_F (CatalogDeltaReplayTest, replay_with_full_checkpoint) {
420+ auto config_path = std::make_shared<std::string>(std::string (test_data_path ()) + " /config/test_catalog_delta.toml"
421+
422+ auto db_name = std::make_shared<std::string>(" default"
423+
424+ auto column_def1 =
425+ std::make_shared<ColumnDef>(0 , std::make_shared<DataType>(LogicalType::kInteger ), " col1"
426+ auto column_def2 =
427+ std::make_shared<ColumnDef>(0 , std::make_shared<DataType>(LogicalType::kVarchar ), " col2"
428+ auto table_name = std::make_shared<std::string>(" tb1"
429+ auto table_name_committed = std::make_shared<std::string>(" tb_committed"
430+ auto table_name_uncommitted = std::make_shared<std::string>(" tb_uncommitted"
431+ auto table_def = TableDef::Make (db_name, table_name, {column_def1, column_def2});
432+ auto table_def_committed = TableDef::Make (db_name, table_name_committed, {column_def1, column_def2});
433+ auto table_def_uncommitted = TableDef::Make (db_name, table_name_uncommitted, {column_def1, column_def2});
434+
435+ {
436+ InfinityContext::instance ().Init (config_path);
437+ Storage *storage = InfinityContext::instance ().storage ();
438+
439+ TxnManager *txn_mgr = storage->txn_manager ();
440+ TxnTimeStamp last_commit_ts = 0 ;
441+ // create table and insert two records
442+ {
443+ auto *txn = txn_mgr->CreateTxn ();
444+ txn->Begin ();
445+
446+ txn->CreateTable (*db_name, table_def, ConflictType::kError );
447+
448+ auto [table_entry, status] = txn->GetTableByName (*db_name, *table_name);
449+ EXPECT_TRUE (status.ok ());
450+
451+ last_commit_ts = txn_mgr->CommitTxn (txn);
452+ }
453+ {
454+ auto *txn = txn_mgr->CreateTxn ();
455+ txn->Begin ();
456+
457+ Vector<SharedPtr<ColumnVector>> column_vectors;
458+ for (SizeT i = 0 ; i < table_def->columns ().size (); ++i) {
459+ SharedPtr<DataType> data_type = table_def->columns ()[i]->type ();
460+ column_vectors.push_back (MakeShared<ColumnVector>(data_type));
461+ column_vectors.back ()->Initialize ();
462+ }
463+ {
464+ int v1 = 1 ;
465+ column_vectors[0 ]->AppendByPtr (reinterpret_cast <const_ptr_t >(&v1));
466+ }
467+ {
468+ std::string v2 = " v2v2v2v2v2v2v2v2v2v2v2v2v2v2v2v2v2v2v2v2"
469+ column_vectors[1 ]->AppendByStringView (v2, ' ,'
470+ }
471+ auto data_block = DataBlock::Make ();
472+ data_block->Init (column_vectors);
473+
474+ auto status = txn->Append (*db_name, *table_name, data_block);
475+ ASSERT_TRUE (status.ok ());
476+ last_commit_ts = txn_mgr->CommitTxn (txn);
477+ }
478+ {
479+ auto *txn = txn_mgr->CreateTxn ();
480+ txn->Begin ();
481+
482+ Vector<SharedPtr<ColumnVector>> column_vectors;
483+ for (SizeT i = 0 ; i < table_def->columns ().size (); ++i) {
484+ SharedPtr<DataType> data_type = table_def->columns ()[i]->type ();
485+ column_vectors.push_back (MakeShared<ColumnVector>(data_type));
486+ column_vectors.back ()->Initialize ();
487+ }
488+ {
489+ int v1 = 2 ;
490+ column_vectors[0 ]->AppendByPtr (reinterpret_cast <const_ptr_t >(&v1));
491+ }
492+ {
493+ std::string v2 = " askljhfasfhuiwqeriqkldfnramgeasfklhllfjas"
494+ column_vectors[1 ]->AppendByStringView (v2, ' ,'
495+ }
496+ auto data_block = DataBlock::Make ();
497+ data_block->Init (column_vectors);
498+
499+ auto status = txn->Append (*db_name, *table_name, data_block);
500+ ASSERT_TRUE (status.ok ());
501+ last_commit_ts = txn_mgr->CommitTxn (txn);
502+ }
503+
504+ // 1. remain some uncommitted txn before force full checkpoint
505+ // 2. wait delta checkpoint
506+ {
507+ auto *txn_uncommitted = txn_mgr->CreateTxn ();
508+ txn_uncommitted->Begin ();
509+ txn_uncommitted->CreateTable (*db_name, table_def_uncommitted, ConflictType::kError );
510+
511+ LOG_INFO (" BEFORE FULL CKP"
512+ auto *txn_force_ckp = txn_mgr->CreateTxn ();
513+ txn_force_ckp->Begin ();
514+ auto force_ckp_task = MakeShared<ForceCheckpointTask>(txn_force_ckp, true );
515+ storage->bg_processor ()->Submit (force_ckp_task);
516+ force_ckp_task->Wait ();
517+ LOG_INFO (" AFTER FULL CKP"
518+ txn_mgr->CommitTxn (txn_force_ckp);
519+
520+ txn_mgr->CommitTxn (txn_uncommitted);
521+
522+ // some txn in delta checkpoint
523+ auto *txn_committed = txn_mgr->CreateTxn ();
524+ txn_committed->Begin ();
525+ txn_committed->CreateTable (*db_name, table_def_committed, ConflictType::kError );
526+ txn_mgr->CommitTxn (txn_committed);
527+
528+ auto *txn_record3 = txn_mgr->CreateTxn ();
529+ txn_record3->Begin ();
530+ Vector<SharedPtr<ColumnVector>> column_vectors;
531+ for (SizeT i = 0 ; i < table_def_committed->columns ().size (); ++i) {
532+ SharedPtr<DataType> data_type = table_def_committed->columns ()[i]->type ();
533+ column_vectors.push_back (MakeShared<ColumnVector>(data_type));
534+ column_vectors.back ()->Initialize ();
535+ }
536+ {
537+ int v1 = 3 ;
538+ column_vectors[0 ]->AppendByPtr (reinterpret_cast <const_ptr_t >(&v1));
539+ }
540+ {
541+ std::string v2 = " this is a test for replay with full checkpoint"
542+ column_vectors[1 ]->AppendByStringView (v2, ' ,'
543+ }
544+ auto data_block = DataBlock::Make ();
545+ data_block->Init (column_vectors);
546+
547+ auto status = txn_record3->Append (*db_name, *table_name, data_block);
548+ ASSERT_TRUE (status.ok ());
549+
550+ last_commit_ts = txn_mgr->CommitTxn (txn_record3);
551+ WaitFlushDeltaOp (txn_mgr, last_commit_ts);
552+
553+ infinity::InfinityContext::instance ().UnInit ();
554+ }
555+ }
556+
557+ // now restart and the table `tb_uncommitted` should exist
558+ {
559+ InfinityContext::instance ().Init (config_path);
560+ Storage *storage = InfinityContext::instance ().storage ();
561+
562+ TxnManager *txn_mgr = storage->txn_manager ();
563+
564+ {
565+ auto *txn = txn_mgr->CreateTxn ();
566+ txn->Begin ();
567+ {
568+ auto [table_uncommitted, uncommitted_status] = txn->GetTableByName (*db_name, *table_name_uncommitted);
569+ EXPECT_TRUE (uncommitted_status.ok ());
570+
571+ auto [table_committed, committed_status] = txn->GetTableByName (*db_name, *table_name_committed);
572+ EXPECT_TRUE (committed_status.ok ());
573+
574+ auto [table_entry, table_status] = txn->GetTableByName (*db_name, *table_name);
575+ EXPECT_TRUE (table_status.ok ());
576+
577+ EXPECT_EQ (table_entry->row_count (), 3ul );
578+ ASSERT_EQ (table_entry->segment_map ().size (), 1ul );
579+ {
580+ auto &segment_entry = table_entry->segment_map ().begin ()->second ;
581+ EXPECT_EQ (segment_entry->row_count (), 3ul );
582+ ASSERT_EQ (segment_entry->block_entries ().size (), 1ul );
583+ {
584+ BlockEntry *block_entry = segment_entry->block_entries ()[0 ].get ();
585+ EXPECT_EQ (block_entry->row_count (), 3ul );
586+ ASSERT_EQ (block_entry->columns ().size (), 2ul );
587+ {
588+ auto &col2 = block_entry->columns ()[1 ];
589+ EXPECT_EQ (col2->OutlineBufferCount (), 3ul );
590+ }
591+ }
592+ }
593+ }
594+ }
595+ infinity::InfinityContext::instance ().UnInit ();
596+ }
597+ }
0 commit comments