[Backport 2.15.0] [#13037] DocDB: Fix repacking rows after alter table

Summary:
It could happen that a packed row is already near the size limit, and then user adds new columns to the table.
Since each column uses 4 bytes in a packed row, such row could grow over the limit after repacking.
Previously we assumed that repacking row could not make it larger that the limit, and there is a check for that. But clearly it is not so in the scenario above.

Changed the code to force-repack a row even if the repacked row overflows the specified limit, so we could have rows larger than the limit without crashing. It should not be an issue, since we expect it to happen quite rarely in an actual DB. And large packed rows are just less effective, but they still work fine.

Original diff: b608dda/D17904

Test Plan: PgPackedRowTest.PackOverflow

Reviewers: mbautin, rthallam

Reviewed By: mbautin, rthallam

Subscribers: ybase, bogdan

Differential Revision: https://phabricator.dev.yugabyte.com/D18012

src/yb/docdb/docdb_compaction_context.cc

@@ -261,7 +261,9 @@ class PackedRowData {
       column_value = Slice();
     }
     VLOG(4) << "Keep value for column " << column_id << ": " << column_value->ToDebugHexString();
-    auto result = VERIFY_RESULT(packer_->AddValue(column_id, *column_value, /* tail_size= */ 0));
+    // Use min ssize_t value to be sure that packing always succeed.
+    constexpr auto kUnlimitedTail = std::numeric_limits<ssize_t>::min();
+    auto result = VERIFY_RESULT(packer_->AddValue(column_id, *column_value, kUnlimitedTail));
     RSTATUS_DCHECK(result, Corruption, "Unable to pack old value for $0", column_id);
     return Status::OK();
   }

src/yb/docdb/packed_row-test.cc

@@ -44,7 +44,7 @@ void TestRowPacking(const Schema& schema, const std::vector<QLValuePB>& values)
   ASSERT_EQ(schema.num_columns() - schema.num_key_columns(), values.size());
   constexpr int kVersion = 1;
   SchemaPacking schema_packing(schema);
-  RowPacker packer(kVersion, schema_packing, std::numeric_limits<uint64_t>::max());
+  RowPacker packer(kVersion, schema_packing, std::numeric_limits<int64_t>::max());
   size_t idx = schema.num_key_columns();
   for (const auto& value : values) {
     auto column_id = schema.column_id(idx);

src/yb/docdb/packed_row.cc

@@ -276,12 +276,12 @@ Result<bool> RowPacker::AddValue(ColumnId column_id, const QLValuePB& value) {
   return DoAddValue(column_id, value, 0);
 }
 
-Result<bool> RowPacker::AddValue(ColumnId column_id, const Slice& value, size_t tail_size) {
+Result<bool> RowPacker::AddValue(ColumnId column_id, const Slice& value, ssize_t tail_size) {
   return DoAddValue(column_id, value, tail_size);
 }
 
 template <class Value>
-Result<bool> RowPacker::DoAddValue(ColumnId column_id, const Value& value, size_t tail_size) {
+Result<bool> RowPacker::DoAddValue(ColumnId column_id, const Value& value, ssize_t tail_size) {
   RSTATUS_DCHECK(
       idx_ < packing_.columns(),
       InvalidArgument, "Add extra column $0, while already have $1 of $2 columns",
@@ -302,7 +302,7 @@ Result<bool> RowPacker::DoAddValue(ColumnId column_id, const Value& value, size_
           "Missing value for non nullable column $0, while adding $1", column_data.id, column_id);
     } else if (!column_data.nullable || !IsNull(value)) {
       if (column_data.varlen() &&
-          prev_size + PackedValueSize(value) + tail_size > packed_size_limit_) {
+          make_signed(prev_size + PackedValueSize(value)) + tail_size > packed_size_limit_) {
         result = false;
       } else {
         PackValue(value, &result_);

src/yb/docdb/packed_row.h

@@ -159,7 +159,7 @@ class RowPacker {
   RowPacker(SchemaVersion version, std::reference_wrapper<const SchemaPacking> packing,
             size_t packed_size_limit);
 
-  RowPacker(const std::pair<SchemaVersion, const SchemaPacking&>& pair, size_t packed_size_limit)
+  RowPacker(const std::pair<SchemaVersion, const SchemaPacking&>& pair, ssize_t packed_size_limit)
       : RowPacker(pair.first, pair.second, packed_size_limit) {
   }
 
@@ -177,17 +177,19 @@ class RowPacker {
   Result<const ColumnPackingData&> NextColumnData() const;
 
   // Returns false when unable to add value due to packed size limit.
+  // tail_size is added to proposed encoded size, to make decision whether encoded value fits
+  // into bounds or not.
+  Result<bool> AddValue(ColumnId column_id, const Slice& value, ssize_t tail_size);
   Result<bool> AddValue(ColumnId column_id, const QLValuePB& value);
-  Result<bool> AddValue(ColumnId column_id, const Slice& value, size_t tail_size);
 
   Result<Slice> Complete();
 
  private:
   template <class Value>
-  Result<bool> DoAddValue(ColumnId column_id, const Value& value, size_t tail_size);
+  Result<bool> DoAddValue(ColumnId column_id, const Value& value, ssize_t tail_size);
 
   const SchemaPacking& packing_;
-  const size_t packed_size_limit_;
+  const ssize_t packed_size_limit_;
   size_t idx_ = 0;
   size_t prefix_end_;
   ValueBuffer result_;

src/yb/docdb/primitive_value.cc

@@ -1087,10 +1087,8 @@ Status KeyEntryValue::DecodeKey(Slice* slice, KeyEntryValue* out) {
 }
 
 Status PrimitiveValue::DecodeFromValue(const Slice& rocksdb_slice) {
-  if (rocksdb_slice.empty()) {
-    return STATUS(Corruption, "Cannot decode a value from an empty slice");
-  }
-  rocksdb::Slice slice(rocksdb_slice);
+  RSTATUS_DCHECK(!rocksdb_slice.empty(), Corruption, "Cannot decode a value from an empty slice");
+  Slice slice(rocksdb_slice);
   this->~PrimitiveValue();
   // Ensure we are not leaving the object in an invalid state in case e.g. an exception is thrown
   // due to inability to allocate memory.

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

@@ -429,5 +429,25 @@ TEST_F(PgPackedRowTest, YB_DISABLE_TEST_IN_TSAN(AddColumn)) {
   ASSERT_OK(conn2.Execute("INSERT INTO t (key, ival) VALUES (2, 2)"));
 }
 
+// Checks repacking of columns then would not fit into limit with new schema due to added columns.
+TEST_F(PgPackedRowTest, YB_DISABLE_TEST_IN_TSAN(PackOverflow)) {
+  constexpr int kRange = 32;
+
+  FLAGS_ysql_packed_row_size_limit = 128;
+  auto conn = ASSERT_RESULT(Connect());
+  ASSERT_OK(conn.Execute(
+      "CREATE TABLE t (key INT PRIMARY KEY, v1 TEXT) SPLIT INTO 1 TABLETS"));
+
+  for (auto key : Range(0, kRange + 1)) {
+    auto len = FLAGS_ysql_packed_row_size_limit - kRange / 2 + key;
+    ASSERT_OK(conn.ExecuteFormat(
+        "INSERT INTO t VALUES ($0, '$1')", key, RandomHumanReadableString(len)));
+  }
+
+  ASSERT_OK(conn.Execute("ALTER TABLE t ADD COLUMN v2 TEXT"));
+
+  ASSERT_OK(cluster_->CompactTablets());
+}
+
 } // namespace pgwrapper
 } // namespace yb