[FLASH-402] Skip unneeded columns when decode row

dbms/src/Storages/Transaction/Codec.cpp

@@ -65,6 +65,24 @@ String DecodeBytes(size_t & cursor, const String & raw_value)
  return ss.str();
 }
 
+void SkipBytes(size_t & cursor, const String & raw_value)
+{
+ while (true)
+ {
+ size_t next_cursor = cursor + 9;
+ if (next_cursor > raw_value.size())
+ throw Exception("Wrong format, cursor over buffer size. (DecodeBytes)", ErrorCodes::LOGICAL_ERROR);
+ UInt8 marker = (UInt8)raw_value[cursor + 8];
+ UInt8 pad_size = ENC_MARKER - marker;
+
+ if (pad_size > 8)
+ throw Exception("Wrong format, too many padding bytes. (DecodeBytes)", ErrorCodes::LOGICAL_ERROR);
+ cursor = next_cursor;
+ if (pad_size != 0)
+ break;
+ }
+}
+
 String DecodeCompactBytes(size_t & cursor, const String & raw_value)
 {
  size_t size = DecodeVarInt(cursor, raw_value);
@@ -73,13 +91,24 @@ String DecodeCompactBytes(size_t & cursor, const String & raw_value)
  return res;
 }
 
+void SkipCompactBytes(size_t & cursor, const String & raw_value)
+{
+ size_t size = DecodeVarInt(cursor, raw_value);
+ cursor += size;
+}
+
 Int64 DecodeVarInt(size_t & cursor, const String & raw_value)
 {
  UInt64 v = DecodeVarUInt(cursor, raw_value);
  Int64 vx = v >> 1;
  return (v & 1) ? ~vx : vx;
 }
 
+void SkipVarInt(size_t & cursor, const String & raw_value)
+{
+ SkipVarUInt(cursor, raw_value);
+}
+
 UInt64 DecodeVarUInt(size_t & cursor, const String & raw_value)
 {
  UInt64 res = 0;
@@ -99,6 +128,11 @@ UInt64 DecodeVarUInt(size_t & cursor, const String & raw_value)
  throw Exception("Wrong format. (DecodeVarUInt)", ErrorCodes::LOGICAL_ERROR);
 }
 
+void SkipVarUInt(size_t & cursor, const String & raw_value)
+{
+ std::ignore = DecodeVarUInt(cursor, raw_value);
+}
+
 inline Int8 getWords(PrecType prec, ScaleType scale)
 {
  Int8 scale_word = scale / 9 + (scale % 9 > 0);
@@ -204,6 +238,15 @@ Decimal DecodeDecimal(size_t & cursor, const String & raw_value)
  return Decimal(value, prec, frac);
 }
 
+void SkipDecimal(size_t & cursor, const String & raw_value)
+{
+ PrecType prec = raw_value[cursor++];
+ ScaleType frac = raw_value[cursor++];
+
+ int binSize = getBytes(prec, frac);
+ cursor += binSize;
+}
+
 Field DecodeDatum(size_t & cursor, const String & raw_value)
 {
  switch (raw_value[cursor++])
@@ -233,6 +276,43 @@ Field DecodeDatum(size_t & cursor, const String & raw_value)
  }
 }
 
+void SkipDatum(size_t & cursor, const String & raw_value)
+{
+ switch (raw_value[cursor++])
+ {
+ case TiDB::CodecFlagNil:
+ return;
+ case TiDB::CodecFlagInt:
+ cursor += sizeof(Int64);
+ return;
+ case TiDB::CodecFlagUInt:
+ cursor += sizeof(UInt64);
+ return;
+ case TiDB::CodecFlagBytes:
+ SkipBytes(cursor, raw_value);
+ return;
+ case TiDB::CodecFlagCompactBytes:
+ SkipCompactBytes(cursor, raw_value);
+ return;
+ case TiDB::CodecFlagFloat:
+ cursor += sizeof(UInt64);
+ return;
+ case TiDB::CodecFlagVarUInt:
+ SkipVarUInt(cursor, raw_value);
+ return;
+ case TiDB::CodecFlagVarInt:
+ SkipVarInt(cursor, raw_value);
+ return;
+ case TiDB::CodecFlagDuration:
+ throw Exception("Not implented yet. DecodeDatum: CodecFlagDuration", ErrorCodes::LOGICAL_ERROR);
+ case TiDB::CodecFlagDecimal:
+ SkipDecimal(cursor, raw_value);
+ return;
+ default:
+ throw Exception("Unknown Type:" + std::to_string(raw_value[cursor - 1]), ErrorCodes::LOGICAL_ERROR);
+ }
+}
+
 void EncodeFloat64(Float64 num, std::stringstream & ss)
 {
  UInt64 u = enforce_cast<UInt64>(num);

dbms/src/Storages/Transaction/Codec.h

@@ -41,6 +41,18 @@ Decimal DecodeDecimal(size_t & cursor, const String & raw_value);
 
 Field DecodeDatum(size_t & cursor, const String & raw_value);
 
+void SkipBytes(size_t & cursor, const String & raw_value);
+
+void SkipCompactBytes(size_t & cursor, const String & raw_value);
+
+void SkipVarInt(size_t & cursor, const String & raw_value);
+
+void SkipVarUInt(size_t & cursor, const String & raw_value);
+
+void SkipDecimal(size_t & cursor, const String & raw_value);
+
+void SkipDatum(size_t & cursor, const String & raw_value);
+
 template <typename T>
 inline std::enable_if_t<std::is_unsigned_v<T>, void> EncodeUInt(T u, std::stringstream & ss)
 {

dbms/src/Storages/Transaction/RegionBlockReader.cpp

@@ -125,16 +125,32 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
  ColumnID handle_col_id = InvalidColumnID;
 
  std::unordered_map<ColumnID, std::pair<MutableColumnPtr, NameAndTypePair>> column_map;
- for (const auto & column_info : table_info.columns)
+ std::unordered_map<ColumnID, size_t> column_id_to_info_index_map;
+ std::unordered_set<ColumnID> column_ids_to_read;
+ std::unordered_set<ColumnID> schema_all_column_ids;
+ for (size_t i = 0; i < table_info.columns.size(); i++)
  {
+ auto & column_info = table_info.columns[i];
  ColumnID col_id = column_info.id;
  String col_name = column_info.name;
+ schema_all_column_ids.insert(col_id);
+ if (std::find(column_names_to_read.begin(), column_names_to_read.end(), col_name) == column_names_to_read.end())
+ {
+ continue;
+ }
  auto ch_col = columns.getPhysical(col_name);
  column_map[col_id] = std::make_pair(ch_col.type->createColumn(), ch_col);
  column_map[col_id].first->reserve(data_list.size());
  if (table_info.pk_is_handle && column_info.hasPriKeyFlag())
  handle_col_id = col_id;
+ else
+ {
+ column_ids_to_read.emplace(col_id);
+ column_id_to_info_index_map.emplace(std::make_pair(col_id, i));
+ }
  }
+ if (column_names_to_read.size() - 3 != column_ids_to_read.size())
+ throw Exception("schema doesn't contain needed columns.", ErrorCodes::LOGICAL_ERROR);
 
  if (!table_info.pk_is_handle)
  {
@@ -168,15 +184,21 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
  const auto & date_lut = DateLUT::instance();
 
- std::unordered_set<ColumnID> col_id_included;
-
- const size_t target_col_size = (!table_info.pk_is_handle ? table_info.columns.size() : table_info.columns.size() - 1) * 2;
+ const size_t target_col_size = column_names_to_read.size() - 3;
 
  Block block;
 
  // optimize for only need handle, tso, delmark.
  if (column_names_to_read.size() > 3)
  {
+ std::unordered_set<ColumnID> col_id_included;
+
+ // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
+ std::vector<ColumnID> col_ids;
+ std::vector<Field> fields;
+ col_ids.reserve(target_col_size);
+ fields.reserve(target_col_size);
+
  for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
  {
  std::ignore = handle;
@@ -185,94 +207,84 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
  if (commit_ts > start_ts)
  continue;
 
- // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
-
- std::vector<Field> row;
-
+ col_ids.clear();
+ fields.clear();
  if (write_type == Region::DelFlag)
  {
- row.reserve(table_info.columns.size() * 2);
- for (const TiDB::ColumnInfo & column : table_info.columns)
+ col_ids.reserve(target_col_size);
+ fields.reserve(target_col_size);
+ for (auto col_id : column_ids_to_read)
  {
- if (handle_col_id == column.id)
- continue;
+ const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
 
- row.push_back(Field(column.id));
- row.push_back(GenDecodeRow(column.getCodecFlag()));
+ col_ids.push_back(column.id);
+ fields.emplace_back(GenDecodeRow(column.getCodecFlag()));
  }
  }
  else
- row = RecordKVFormat::DecodeRow(*value_ptr);
-
- if (row.size() == 1 && row[0].isNull())
  {
- // all field is null
- row.clear();
+ bool schema_not_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
+ if (schema_not_match && !force_decode)
+ {
+ return std::make_tuple(block, false);
+ }
+ if (col_ids.empty() && fields.size() == 1 && fields[0].isNull())
+ {
+ // all field is null
+ fields.clear();
+ }
  }
 
- if (row.size() & 1)
+ if (col_ids.size() != fields.size())
  throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
  /// Modify `row` by adding missing column values or removing useless column values.
-
- col_id_included.clear();
- for (size_t i = 0; i < row.size(); i += 2)
- col_id_included.emplace(row[i].get<ColumnID>());
-
- // Fill in missing column values.
- for (const TiDB::ColumnInfo & column : table_info.columns)
+ if (unlikely(col_ids.size() > column_ids_to_read.size()))
  {
- if (handle_col_id == column.id)
- continue;
- if (col_id_included.count(column.id))
- continue;
-
- if (!force_decode)
- return std::make_tuple(block, false);
-
- row.emplace_back(Field(column.id));
- if (column.hasNoDefaultValueFlag())
- // Fill `zero` value if NOT NULL specified or else NULL.
- row.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
- else
- // Fill default value.
- row.push_back(column.defaultValueToField());
+ throw Exception("read unexpected columns.", ErrorCodes::LOGICAL_ERROR);
  }
-
- // Remove values of non-existing columns, which could be data inserted (but not flushed) before DDLs that drop some columns.
- // TODO: May need to log this.
- for (int i = int(row.size()) - 2; i >= 0; i -= 2)
+ if (col_ids.size() < column_ids_to_read.size())
  {
- Field & col_id = row[i];
- if (column_map.find(col_id.get<ColumnID>()) == column_map.end())
+ col_id_included.clear();
+ for (size_t i = 0; i < col_ids.size(); i++)
+ col_id_included.emplace(col_ids[i]);
+
+ // Fill in missing column values.
+ for (auto col_id : column_ids_to_read)
  {
- if (!force_decode)
- return std::make_tuple(block, false);
+ if (col_id_included.count(col_id))
+ continue;
 
- row.erase(row.begin() + i, row.begin() + i + 2);
+ const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
+ col_ids.push_back(column.id);
+ if (column.hasNoDefaultValueFlag())
+ // Fill `zero` value if NOT NULL specified or else NULL.
+ fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
+ else
+ // Fill default value.
+ fields.push_back(column.defaultValueToField());
  }
  }
 
- if (row.size() != target_col_size)
+ if (col_ids.size() != target_col_size || fields.size() != target_col_size)
  throw Exception("decode row error.", ErrorCodes::LOGICAL_ERROR);
 
  /// Transform `row` to columnar format.
-
- for (size_t i = 0; i < row.size(); i += 2)
+ for (size_t i = 0; i < col_ids.size(); i++)
  {
- Field & col_id = row[i];
- auto it = column_map.find(col_id.get<ColumnID>());
+ ColumnID col_id = col_ids[i];
+ auto it = column_map.find(col_id);
  if (it == column_map.end())
  throw Exception("col_id not found in column_map", ErrorCodes::LOGICAL_ERROR);
 
  const auto & tp = it->second.second.type;
  if (tp->isDateOrDateTime()
  || (tp->isNullable() && dynamic_cast<const DataTypeNullable *>(tp.get())->getNestedType()->isDateOrDateTime()))
  {
- Field & field = row[i + 1];
+ Field & field = fields[i];
  if (field.isNull())
  {
- it->second.first->insert(row[i + 1]);
+ it->second.first->insert(fields[i]);
  continue;
  }
  UInt64 packed = field.get<UInt64>();
@@ -317,7 +329,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
  }
  else
  {
- it->second.first->insert(row[i + 1]);
+ it->second.first->insert(fields[i]);
 
  // Check overflow for potential un-synced data type widen,
  // i.e. schema is old and narrow, meanwhile data is new and wide.
@@ -335,14 +347,14 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
  {
  // Unsigned checking by bitwise compare.
  UInt64 inserted = nested_column.get64(inserted_index);
- UInt64 orig = row[i + 1].get<UInt64>();
+ UInt64 orig = fields[i].get<UInt64>();
  overflow = inserted != orig;
  }
  else
  {
  // Singed checking by arithmetical cast.
  Int64 inserted = nested_column.getInt(inserted_index);
- Int64 orig = row[i + 1].get<Int64>();
+ Int64 orig = fields[i].get<Int64>();
  overflow = inserted != orig;
  }
  if (overflow)
@@ -352,7 +364,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
  // Otherwise return false to outer, outer should sync schema and try again.
  if (force_decode)
  throw Exception(
- "Detected overflow for data " + std::to_string(row[i + 1].get<UInt64>()) + " of type " + tp->getName(),
+ "Detected overflow for data " + std::to_string(fields[i].get<UInt64>()) + " of type " + tp->getName(),
  ErrorCodes::LOGICAL_ERROR);
 
  return std::make_tuple(block, false);
@@ -364,6 +376,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
  }
  }
 
+
  for (const auto & name : column_names_to_read)
  {
  if (name == MutableSupport::delmark_column_name)