update.

Signed-off-by: ywqzzy <592838129@qq.com> update. Signed-off-by: ywqzzy <592838129@qq.com> update. Signed-off-by: ywqzzy <592838129@qq.com> Split write and ingest throughput (pingcap#4245) close pingcap#4243 Signed-off-by: ywqzzy <592838129@qq.com> fix the problem that expired data was not recycled timely due to slow gc speed (pingcap#4224) close pingcap#4146 Signed-off-by: ywqzzy <592838129@qq.com> PageStorage: Fix some bugs (pingcap#4212) ref pingcap#3594 Signed-off-by: ywqzzy <592838129@qq.com> Make test of CreateTables using affected opts (pingcap#4239) close pingcap#4235 Signed-off-by: ywqzzy <592838129@qq.com> BlobStore: remove the old_ids, not need recycle the blob id. (pingcap#4247) ref pingcap#3594 Signed-off-by: ywqzzy <592838129@qq.com> support `PartitionTableScan` in TiFlash (pingcap#3876) close pingcap#3873 Signed-off-by: ywqzzy <592838129@qq.com> update. Signed-off-by: ywqzzy <592838129@qq.com> update. Signed-off-by: ywqzzy <592838129@qq.com> support last_day and dayofmonth pushdown to tiflash (pingcap#4183) close pingcap#4149 Signed-off-by: ywqzzy <592838129@qq.com>
ywqzzy · Mar 14, 2022 · b54dece · b54dece
1 parent 93060e8
commit b54dece
Show file tree

Hide file tree

Showing 75 changed files with 2,326 additions and 769 deletions.
diff --git a/contrib/kvproto b/contrib/kvproto
diff --git a/contrib/tipb b/contrib/tipb
diff --git a/dbms/CMakeLists.txt b/dbms/CMakeLists.txt
@@ -177,6 +177,7 @@ target_link_libraries (clickhouse_common_io
  cpptoml
 )
 target_include_directories (clickhouse_common_io BEFORE PRIVATE ${kvClient_SOURCE_DIR}/include)
+target_include_directories (clickhouse_common_io BEFORE PUBLIC ${kvproto_SOURCE_DIR} ${tipb_SOURCE_DIR} ${Protobuf_INCLUDE_DIR} ${gRPC_INCLUDE_DIRS})
 
 target_link_libraries (dbms
  clickhouse_parsers

diff --git a/dbms/src/Common/Exception.h b/dbms/src/Common/Exception.h
@@ -36,6 +36,12 @@ class Exception : public Poco::Exception
  : Poco::Exception(exc.displayText())
  {}
 
+ Exception(const Poco::Exception & exc, const std::string & arg)
+ : Poco::Exception(exc.displayText())
+ {
+ extendedMessage(arg);
+ }
+
  const char * name() const throw() override { return "DB::Exception"; }
  const char * className() const throw() override { return "DB::Exception"; }
  DB::Exception * clone() const override { return new DB::Exception(*this); }

diff --git a/dbms/src/Common/MyTime.cpp b/dbms/src/Common/MyTime.cpp
@@ -523,15 +523,7 @@ bool checkTimeValid(Int32 year, Int32 month, Int32 day, Int32 hour, Int32 minute
  {
  return false;
  }
- static int days_of_month_table[] = {0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
- if (month != 2)
- return day <= days_of_month_table[month];
- bool is_leap_year = false;
- if ((year & 0b0011) == 0)
- {
- is_leap_year = year % 100 != 0 || year % 400 == 0;
- }
- return day <= (is_leap_year ? 29 : 28);
+ return day <= getLastDay(year, month);
 }
 
 std::pair<Field, bool> parseMyDateTimeAndJudgeIsDate(const String & str, int8_t fsp, bool needCheckTimeValid)

diff --git a/dbms/src/Common/MyTime.h b/dbms/src/Common/MyTime.h
@@ -192,4 +192,21 @@ bool isValidSeperator(char c, int previous_parts);
 // Note that this function will not check if the input is logically a valid datetime value.
 bool toCoreTimeChecked(const UInt64 & year, const UInt64 & month, const UInt64 & day, const UInt64 & hour, const UInt64 & minute, const UInt64 & second, const UInt64 & microsecond, MyDateTime & result);
 
+inline bool isLeapYear(UInt16 year)
+{
+ return ((year % 4 == 0) && (year % 100 != 0)) || (year % 400 == 0);
+}
+
+// Get last day of a month. Return 0 if month if invalid.
+inline UInt8 getLastDay(UInt16 year, UInt8 month)
+{
+ static constexpr UInt8 days_of_month_table[] = {0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
+ UInt8 last_day = 0;
+ if (month > 0 && month <= 12)
+ last_day = days_of_month_table[month];
+ if (month == 2 && isLeapYear(year))
+ last_day = 29;
+ return last_day;
+}
+
 } // namespace DB
diff --git a/dbms/src/Common/TiFlashMetrics.h b/dbms/src/Common/TiFlashMetrics.h
@@ -40,7 +40,8 @@ namespace DB
  M(tiflash_coprocessor_executor_count, "Total number of each executor", Counter, F(type_ts, {"type", "table_scan"}), \
  F(type_sel, {"type", "selection"}), F(type_agg, {"type", "aggregation"}), F(type_topn, {"type", "top_n"}), \
  F(type_limit, {"type", "limit"}), F(type_join, {"type", "join"}), F(type_exchange_sender, {"type", "exchange_sender"}), \
- F(type_exchange_receiver, {"type", "exchange_receiver"}), F(type_projection, {"type", "projection"})) \
+ F(type_exchange_receiver, {"type", "exchange_receiver"}), F(type_projection, {"type", "projection"}), \
+ F(type_partition_ts, {"type", "partition_table_scan"})) \
  M(tiflash_coprocessor_request_duration_seconds, "Bucketed histogram of request duration", Histogram, \
  F(type_batch, {{"type", "batch"}}, ExpBuckets{0.0005, 2, 30}), F(type_cop, {{"type", "cop"}}, ExpBuckets{0.0005, 2, 30}), \
  F(type_super_batch, {{"type", "super_batch"}}, ExpBuckets{0.0005, 2, 30}), \

diff --git a/dbms/src/Core/Types.h b/dbms/src/Core/Types.h
@@ -133,7 +133,6 @@ enum class TypeIndex
  Tuple,
  Set,
  Interval,
- Nullable,
  Function,
  AggregateFunction,
  LowCardinality,

diff --git a/dbms/src/DataTypes/DataTypeNullable.cpp b/dbms/src/DataTypes/DataTypeNullable.cpp
@@ -58,7 +58,7 @@ void DataTypeNullable::serializeBinaryBulkWithMultipleStreams(
 
  /// First serialize null map.
  path.push_back(Substream::NullMap);
- if (auto stream = getter(path))
+ if (auto * stream = getter(path))
  DataTypeUInt8().serializeBinaryBulk(col.getNullMapColumn(), *stream, offset, limit);
 
  /// Then serialize contents of arrays.
@@ -78,7 +78,7 @@ void DataTypeNullable::deserializeBinaryBulkWithMultipleStreams(
  ColumnNullable & col = static_cast<ColumnNullable &>(column);
 
  path.push_back(Substream::NullMap);
- if (auto stream = getter(path))
+ if (auto * stream = getter(path))
  DataTypeUInt8().deserializeBinaryBulk(col.getNullMapColumn(), *stream, limit, 0);
 
  path.back() = Substream::NullableElements;
@@ -99,7 +99,7 @@ void DataTypeNullable::serializeWidenBinaryBulkWithMultipleStreams(
 
  /// First serialize null map.
  path.push_back(Substream::NullMap);
- if (auto stream = getter(path))
+ if (auto * stream = getter(path))
  DataTypeUInt8().serializeBinaryBulk(col.getNullMapColumn(), *stream, offset, limit);
 
  /// Then serialize contents of arrays.
@@ -119,7 +119,7 @@ void DataTypeNullable::deserializeWidenBinaryBulkWithMultipleStreams(
  ColumnNullable & col = static_cast<ColumnNullable &>(column);
 
  path.push_back(Substream::NullMap);
- if (auto stream = getter(path))
+ if (auto * stream = getter(path))
  DataTypeUInt8().deserializeBinaryBulk(col.getNullMapColumn(), *stream, limit, 0);
 
  path.back() = Substream::NullableElements;
@@ -172,7 +172,7 @@ void DataTypeNullable::deserializeBinary(IColumn & column, ReadBuffer & istr) co
 {
  safeDeserialize(
  column,
- [&istr] { bool is_null = 0; readBinary(is_null, istr); return is_null; },
+ [&istr] { bool is_null = false; readBinary(is_null, istr); return is_null; },
  [this, &istr](IColumn & nested) { nested_data_type->deserializeBinary(nested, istr); });
 }
 
@@ -337,7 +337,6 @@ bool DataTypeNullable::equals(const IDataType & rhs) const
  return rhs.isNullable() && nested_data_type->equals(*static_cast<const DataTypeNullable &>(rhs).nested_data_type);
 }
 
-
 static DataTypePtr create(const ASTPtr & arguments)
 {
  if (!arguments || arguments->children.size() != 1)

diff --git a/dbms/src/DataTypes/DataTypeNullable.h b/dbms/src/DataTypes/DataTypeNullable.h
@@ -12,12 +12,11 @@ class DataTypeNullable final : public IDataType
 public:
  static constexpr bool is_parametric = true;
 
- DataTypeNullable(const DataTypePtr & nested_data_type_);
+ explicit DataTypeNullable(const DataTypePtr & nested_data_type_);
  std::string getName() const override { return "Nullable(" + nested_data_type->getName() + ")"; }
  const char * getFamilyName() const override { return "Nullable"; }
 
- TypeIndex getTypeId() const override { return TypeIndex::Nullable; }
-
+ TypeIndex getTypeId() const override { return nested_data_type->getTypeId(); }
  void enumerateStreams(const StreamCallback & callback, SubstreamPath & path) const override;
 
  void serializeBinaryBulkWithMultipleStreams(
@@ -70,7 +69,7 @@ class DataTypeNullable final : public IDataType
  * Now we support only first.
  * In CSV, non-NULL string value, starting with \N characters, must be placed in quotes, to avoid ambiguity.
  */
- void deserializeTextCSV(IColumn & column, ReadBuffer & istr, const char delimiter) const override;
+ void deserializeTextCSV(IColumn & column, ReadBuffer & istr, char delimiter) const override;
 
  void serializeTextJSON(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettingsJSON &) const override;
  void deserializeTextJSON(IColumn & column, ReadBuffer & istr) const override;
@@ -92,11 +91,27 @@ class DataTypeNullable final : public IDataType
  bool canBeComparedWithCollation() const override { return nested_data_type->canBeComparedWithCollation(); }
  bool canBeUsedAsVersion() const override { return false; }
  bool isSummable() const override { return nested_data_type->isSummable(); }
+ bool canBeUsedInBitOperations() const override { return nested_data_type->canBeUsedInBitOperations(); };
  bool canBeUsedInBooleanContext() const override { return nested_data_type->canBeUsedInBooleanContext(); }
+ bool isNumber() const override { return nested_data_type->isNumber(); }
+ bool isInteger() const override { return nested_data_type->isInteger(); }
+ bool isUnsignedInteger() const override { return nested_data_type->isUnsignedInteger(); }
+ bool isFloatingPoint() const override { return nested_data_type->isFloatingPoint(); }
+ bool isDateOrDateTime() const override { return nested_data_type->isDateOrDateTime(); } // used in delta merge
+ bool isMyDateOrMyDateTime() const override { return nested_data_type->isMyDateOrMyDateTime(); }
+ bool isMyTime() const override { return nested_data_type->isMyTime(); }
+ bool isDecimal() const override { return nested_data_type->isDecimal(); }
+ bool isValueRepresentedByNumber() const override { return nested_data_type->isValueRepresentedByNumber(); }
+ bool isValueRepresentedByInteger() const override { return nested_data_type->isValueRepresentedByInteger(); }
+ bool isValueUnambiguouslyRepresentedInContiguousMemoryRegion() const override { return nested_data_type->isValueUnambiguouslyRepresentedInContiguousMemoryRegion(); }
+ bool isValueUnambiguouslyRepresentedInFixedSizeContiguousMemoryRegion() const override { return nested_data_type->isValueUnambiguouslyRepresentedInFixedSizeContiguousMemoryRegion(); }
+ bool isString() const override { return nested_data_type->isString(); }
  bool haveMaximumSizeOfValue() const override { return nested_data_type->haveMaximumSizeOfValue(); }
  size_t getMaximumSizeOfValueInMemory() const override { return 1 + nested_data_type->getMaximumSizeOfValueInMemory(); }
  bool isNullable() const override { return true; }
  size_t getSizeOfValueInMemory() const override;
+ bool isCategorial() const override { return nested_data_type->isCategorial(); }
+ bool isEnum() const override { return nested_data_type->isEnum(); }
  bool onlyNull() const override;
 
  const DataTypePtr & getNestedType() const { return nested_data_type; }

diff --git a/dbms/src/Debug/MockTiDB.cpp b/dbms/src/Debug/MockTiDB.cpp
@@ -263,10 +263,45 @@ int MockTiDB::newTables(
  Timestamp tso,
  const String & engine_type)
 {
+ std::lock_guard lock(tables_mutex);
+ if (databases.find(database_name) == databases.end())
+ {
+ throw Exception("MockTiDB not found db: " + database_name, ErrorCodes::LOGICAL_ERROR);
+ }
+
+ version++;
+ SchemaDiff diff;
+ diff.type = SchemaActionType::CreateTables;
  for (const auto & [table_name, columns, handle_pk_name] : tables)
  {
- newTable(database_name, table_name, columns, tso, handle_pk_name, engine_type);
+ String qualified_name = database_name + "." + table_name;
+ if (tables_by_name.find(qualified_name) != tables_by_name.end())
+ {
+ throw Exception("Mock TiDB table " + qualified_name + " already exists", ErrorCodes::TABLE_ALREADY_EXISTS);
+ }
+
+ auto table_info = *parseColumns(table_name, columns, handle_pk_name, engine_type);
+ table_info.id = table_id_allocator++;
+ table_info.update_timestamp = tso;
+
+ auto table = std::make_shared<Table>(database_name, databases[database_name], table_info.name, std::move(table_info));
+ tables_by_id.emplace(table->table_info.id, table);
+ tables_by_name.emplace(qualified_name, table);
+
+ AffectedOption opt;
+ opt.schema_id = table->database_id;
+ opt.table_id = table->id();
+ opt.old_schema_id = table->database_id;
+ opt.old_table_id = table->id();
+ diff.affected_opts.push_back(std::move(opt));
  }
+
+ if (diff.affected_opts.empty())
+ throw Exception("MockTiDB CreateTables should have at lease 1 table", ErrorCodes::LOGICAL_ERROR);
+
+ diff.schema_id = diff.affected_opts[0].schema_id;
+ diff.version = version;
+ version_diff[version] = diff;
  return 0;
 }