[#24545] docdb: fix segfault caused by re-packing when packed row is …

…disabled

Summary:
If ysql_enable_packed_row is turned off, and there are packed rows existing in DB, compaction may hit segfault with the following stack:
```
[m-1] *** SIGSEGV (@0x0) received by PID 120292 (TID 0x7f9c4a653700) from PID 0; stack trace: ***
[m-1]     @     0x7f9c527ae697 google::(anonymous namespace)::FailureSignalHandler()
[m-1]     @     0x7f9c509e1d10 (unknown)
[m-1]     @     0x7f9c534c7082 yb::dockv::RowPackerBase::NextColumnId()
[m-1]     @     0x7f9c5639dd46 yb::docdb::(anonymous namespace)::PackedRowData::ProcessColumn()
[m-1]     @     0x7f9c563a3dfa yb::docdb::(anonymous namespace)::DocDBCompactionFeed::Feed()
[m-1]     @     0x7f9c56014067 rocksdb::CompactionJob::ProcessKeyValueCompaction()
[m-1]     @     0x7f9c56015182 rocksdb::CompactionJob::Run()
[m-1]     @     0x7f9c5603ff63 rocksdb::DBImpl::BackgroundCompaction()
[m-1]     @     0x7f9c5604b6f0 rocksdb::DBImpl::BackgroundCallCompaction()
[m-1]     @     0x7f9c56069cca rocksdb::DBImpl::CompactionTask::DoRun()
[m-1]     @     0x7f9c5605c4ff rocksdb::DBImpl::ThreadPoolTask::Run()
[m-1]     @     0x7f9c52ad13db yb::(anonymous namespace)::PriorityThreadPoolWorker::Run()
[m-1]     @     0x7f9c52acda1d std::_Function_handler<>::_M_invoke()
[m-1]     @     0x7f9c52b1f845 yb::Thread::SuperviseThread()
[m-1]     @     0x7f9c509d71ca start_thread
[m-1]     @     0x7f9c506328d3 __GI___clone
```
The root cause is that the packer is not initialized when packed row is disabled. But if there exists packed rows, it will still re-pack the rows with the new columns or new schema, which can cause segfault.

This diff is to ensure packer is created when re-packing the old rows, the packed_row_version of the old row will be used to initialize the new row packer.
Jira: DB-13579

Test Plan: PackingVersion/PgPackedRowTest.DisableRepack*/*

Reviewers: sergei

Reviewed By: sergei

Subscribers: yql, ybase, qhu

Differential Revision: https://phorge.dev.yugabyte.com/D39264

src/yb/docdb/docdb_compaction_context.cc

@@ -186,7 +186,7 @@ class PackedRowData {
     return Status::OK();
   }
 
-  void StartPacking(
+  Status StartPacking(
       const Slice& internal_key, size_t doc_key_size,
       const EncodedDocHybridTime& encoded_doc_ht,
       size_t new_doc_key_serial) {
@@ -196,7 +196,7 @@ class PackedRowData {
     control_fields_size_ = 0;
     encoded_doc_ht_ = encoded_doc_ht;
     old_value_slice_ = Slice();
-    InitPacker();
+    return InitPacker();
   }
 
   void InitKey(
@@ -288,7 +288,10 @@ class PackedRowData {
       // CotablePacking returns desired packed row type, so keep type extracted from actual row.
       old_packing_.packed_row_version = packed_row_version;
     }
-    InitPacker();
+    if (!new_packing_.packed_row_version.has_value()) {
+      new_packing_.packed_row_version = old_packing_.packed_row_version;
+    }
+    RETURN_NOT_OK(InitPacker());
     switch (*old_packing_.packed_row_version) {
       case dockv::PackedRowVersion::kV1:
         CreateOldRowDecoderHelper<dockv::PackedRowDecoderV1>();
@@ -305,7 +308,10 @@ class PackedRowData {
     old_row_decoder_.emplace<Decoder>(*old_packing_.schema_packing, old_value_slice_.data());
   }
 
-  void InitPacker() {
+  Status InitPacker() {
+    if (!new_packing_.packed_row_version.has_value()) {
+      return STATUS(IllegalState, "Packed row version is not set");
+    }
     packing_started_ = true;
     if (!packer_) {
       auto packed_row_version = *new_packing_.packed_row_version;
@@ -317,14 +323,16 @@ class PackedRowData {
       switch (packed_row_version) {
         case dockv::PackedRowVersion::kV1:
           InitPackerHelper<dockv::RowPackerV1>();
-          break;
+          return Status::OK();
         case dockv::PackedRowVersion::kV2:
           InitPackerHelper<dockv::RowPackerV2>();
-          break;
+          return Status::OK();
       }
+      FATAL_INVALID_ENUM_VALUE(dockv::PackedRowVersion, packed_row_version);
     } else {
       CHECK(false);
     }
+    return Status::OK();
   }
 
   template <class Packer>
@@ -977,10 +985,11 @@ Status DocDBCompactionFeed::Feed(const Slice& internal_key, const Slice& value)
           << ", can have other data before: " << CanHaveOtherDataBefore(encoded_doc_ht)
           << ", start packing: " << start_packing;
       if (start_packing) {
-        packed_row_.StartPacking(internal_key, doc_key_size, encoded_doc_ht, doc_key_serial_);
+        RETURN_NOT_OK(packed_row_.StartPacking(
+            internal_key, doc_key_size, encoded_doc_ht, doc_key_serial_));
         AssignPrevSubDocKey(key.cdata(), same_bytes);
       }
-      if (packed_row_.active()) {
+      if (packed_row_.can_start_packing() && packed_row_.active()) {
         if (key_type == dockv::KeyEntryType::kSystemColumnId &&
             column_id == dockv::KeyEntryValue::kLivenessColumn.GetColumnId()) {
           return Status::OK();

src/yb/yql/pgwrapper/pg_packed_row-test.cc

@@ -1054,6 +1054,31 @@ TEST_P(PgPackedRowTest, RestorePITRSnapshotAfterOldSchemaGC) {
   ASSERT_OK(snapshot_util_->RestoreSnapshot(snapshot_id, hybrid_time));
 }
 
+TEST_P(PgPackedRowTest, DisableRepackUpdatedColumns) {
+  auto conn = ASSERT_RESULT(Connect());
+
+  ASSERT_OK(conn.Execute("CREATE TABLE test(v1 INT, v2 INT) SPLIT INTO 1 TABLETS"));
+
+  ASSERT_OK(conn.Execute("INSERT INTO test VALUES (1, 1)"));
+  ASSERT_OK(conn.Execute("UPDATE test SET v2 = 3"));
+
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_ysql_enable_packed_row) = false;
+  ASSERT_OK(cluster_->CompactTablets());
+}
+
+TEST_P(PgPackedRowTest, DisableRepackWithNewSchema) {
+  auto conn = ASSERT_RESULT(Connect());
+
+  ASSERT_OK(conn.Execute("CREATE TABLE test(v1 INT, v2 INT) SPLIT INTO 1 TABLETS"));
+
+  ASSERT_OK(conn.Execute("INSERT INTO test VALUES (1, 1)"));
+  ASSERT_OK(conn.Execute(
+      "ALTER TABLE test ADD COLUMN v3 TIMESTAMP DEFAULT CURRENT_TIMESTAMP NULL"));
+
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_ysql_enable_packed_row) = false;
+  ASSERT_OK(cluster_->CompactTablets());
+}
+
 INSTANTIATE_TEST_SUITE_P(
     PackingVersion, PgPackedRowTest, ::testing::ValuesIn(dockv::kPackedRowVersionArray),
     PackedRowVersionToString);