mysql8 C++源码中创建表函数，表字段最大数量限制，表行最大存储限制

在 MySQL 8 的 C++ 源码中，表的最大字段数量限制体现在 MAX_FIELDS 宏定义中。这个宏定义了表中可以拥有的最大字段数量。

代码中的体现

在 mysql_prepare_create_table 函数中，有以下代码段检查表的字段数量是否超过最大限制：

cpp

if (alter_info->create_list.elements > MAX_FIELDS) {my_error(ER_TOO_MANY_FIELDS, MYF(0));return true;
}

这里，alter_info->create_list.elements 表示表中字段的数量，MAX_FIELDS 是最大允许的字段数量。

MAX_FIELDS 的定义

MAX_FIELDS 的定义通常在 MySQL 的头文件中，例如 include/mysql_com.h 或其他相关头文件。在 MySQL 8 中，MAX_FIELDS 的值为 4096。

源码位置

在 MySQL 8 的源码中，mysql_prepare_create_table 函数位于 sql/sql_table.cc 文件中。这个文件包含了与表创建和修改相关的许多函数。

总结

• 最大字段数量：MySQL 8 中表的最大字段数量为 4096。

• 代码中的体现：在 mysql_prepare_create_table 函数中通过 MAX_FIELDS 宏检查表的字段数量。

MySQL 中，每行记录的最大长度限制为 65,535 字节。这个限制是由 MySQL 的内部实现决定的，而不是由某个特定的源码文件或函数直接定义的。这个限制适用于所有存储引擎，包括 InnoDB 和 MyISAM。

限制的体现

在 MySQL 8 的源码中，这个限制通常在多个地方体现，特别是在处理表的创建和数据插入时。以下是一些关键点：

1. 表创建时的验证： 在 mysql_prepare_create_table 函数中，虽然没有直接的代码检查每行记录的长度，但这个限制会在表创建时通过存储引擎的接口进行验证。

2. 存储引擎的限制： 每个存储引擎都有自己的限制。例如，InnoDB 存储引擎在处理表创建时会检查每行记录的长度是否超过最大限制。

3. 数据插入时的验证： 在插入数据时，MySQL 会检查每行记录的长度是否超过 65,535 字节。如果超过，会报错。

示例代码

以下是一个简化的示例，展示如何在插入数据时检查每行记录的长度：

cpp

if (reclength > file->max_record_length()) {my_error(ER_TOO_BIG_ROWSIZE, MYF(0),static_cast<long>(file->max_record_length()));return true;
}

在这个代码片段中，reclength 是计算的每行记录的长度，file->max_record_length() 是存储引擎允许的最大记录长度。如果 reclength 超过 file->max_record_length()，则报错。

具体实现

在 MySQL 源码中，file->max_record_length() 通常由存储引擎的实现提供。例如，在 InnoDB 存储引擎中，这个值被定义为 65,535 字节。

C++源码

// Prepares the table and key structures for table creation.
bool mysql_prepare_create_table(THD *thd, const char *error_schema_name, const char *error_table_name,HA_CREATE_INFO *create_info, Alter_info *alter_info, handler *file,bool is_partitioned, KEY **key_info_buffer, uint *key_count,FOREIGN_KEY **fk_key_info_buffer, uint *fk_key_count,FOREIGN_KEY *existing_fks, uint existing_fks_count,const dd::Table *existing_fks_table, uint fk_max_generated_name_number,int select_field_count, bool find_parent_keys) {DBUG_TRACE;/*Validation of table properties.*/LEX_STRING *connect_string = &create_info->connect_string;if (connect_string->length != 0 &&connect_string->length > CONNECT_STRING_MAXLEN &&(system_charset_info->cset->charpos(system_charset_info, connect_string->str,(connect_string->str + connect_string->length),CONNECT_STRING_MAXLEN) < connect_string->length)) {my_error(ER_WRONG_STRING_LENGTH, MYF(0), connect_string->str, "CONNECTION",CONNECT_STRING_MAXLEN);return true;}LEX_STRING *compress = &create_info->compress;if (compress->length != 0 && compress->length > TABLE_COMMENT_MAXLEN &&system_charset_info->cset->charpos(system_charset_info, compress->str, compress->str + compress->length,TABLE_COMMENT_MAXLEN) < compress->length) {my_error(ER_WRONG_STRING_LENGTH, MYF(0), compress->str, "COMPRESSION",TABLE_COMMENT_MAXLEN);return true;}LEX_STRING *encrypt_type = &create_info->encrypt_type;if (encrypt_type->length != 0 &&encrypt_type->length > TABLE_COMMENT_MAXLEN &&system_charset_info->cset->charpos(system_charset_info, encrypt_type->str,encrypt_type->str + encrypt_type->length,TABLE_COMMENT_MAXLEN) < encrypt_type->length) {my_error(ER_WRONG_STRING_LENGTH, MYF(0), encrypt_type->str, "ENCRYPTION",TABLE_COMMENT_MAXLEN);return true;}// Validate table comment stringstd::string invalid_sub_str;if (is_invalid_string({create_info->comment.str, create_info->comment.length},system_charset_info, invalid_sub_str)) {my_error(ER_COMMENT_CONTAINS_INVALID_STRING, MYF(0), "table",(std::string(error_schema_name) + "." + std::string(error_table_name)).c_str(),system_charset_info->csname, invalid_sub_str.c_str());return true;}if (validate_comment_length(thd, create_info->comment.str, &create_info->comment.length,TABLE_COMMENT_MAXLEN, ER_TOO_LONG_TABLE_COMMENT, error_table_name)) {return true;}if (alter_info->create_list.elements > MAX_FIELDS) {my_error(ER_TOO_MANY_FIELDS, MYF(0));return true;}/*Checks which previously were done during .FRM creation.TODO: Check if the old .FRM limitations still make sensewith the new DD.*//* Fix this when we have new .frm files;  Current limit is 4G rows (QQ) */constexpr ulonglong u32max = UINT_MAX32;if (create_info->max_rows > UINT_MAX32) {// Values larger than uint32_max are capped to uint32_max.// Emit a warning about this.push_warning_printf(thd, Sql_condition::SL_WARNING, ER_VALUE_OUT_OF_RANGE,ER_THD(thd, ER_VALUE_OUT_OF_RANGE), "max_rows",create_info->max_rows, 0ULL, u32max, u32max);create_info->max_rows = UINT_MAX32;}if (create_info->min_rows > UINT_MAX32) {// Values larger than uint32_max are capped to uint32_max.// Emit a warning about this.push_warning_printf(thd, Sql_condition::SL_WARNING, ER_VALUE_OUT_OF_RANGE,ER_THD(thd, ER_VALUE_OUT_OF_RANGE), "min_rows",create_info->min_rows, 0ULL, u32max, u32max);create_info->min_rows = UINT_MAX32;}if (create_info->row_type == ROW_TYPE_DYNAMIC)create_info->table_options |= HA_OPTION_PACK_RECORD;/*Prepare fields, which must be done before callingadd_functional_index_to_create_list(). The reason is thatprepare_create_field() sets several properties of all Create_fields, such ascharacter set. We need the character set in order to get the correctdisplay width for each Create_field, which is in turn needed to resolve thecorrect data type/length for each hidden generated column added byadd_functional_index_to_create_list().*/int select_field_pos = alter_info->create_list.elements - select_field_count;create_info->null_bits = 0;int field_no = 0;Create_field *sql_field;List_iterator<Create_field> it(alter_info->create_list);for (; (sql_field = it++); field_no++) {if (prepare_create_field(thd, error_schema_name, error_table_name,create_info, &alter_info->create_list,&select_field_pos, file, sql_field, field_no))return true;}// Go through all functional key parts. For each functional key part, resolve// the expression and add a hidden generated column to the create list.for (Key_spec *key : alter_info->key_list) {if (key->type == KEYTYPE_FOREIGN) continue;for (size_t j = 0; j < key->columns.size(); ++j) {Key_part_spec *key_part_spec = key->columns[j];// In the case of procedures, the Key_part_spec may both have an// expression and a field name assigned to it. But the hidden generated// will not exist in the create list, so we will have to add it.if (!key_part_spec->has_expression() ||(key_part_spec->get_field_name() != nullptr &&column_exists_in_create_list(key_part_spec->get_field_name(),alter_info->create_list))) {continue;}Create_field *new_create_field = add_functional_index_to_create_list(thd, key, alter_info, key_part_spec, j, create_info);if (new_create_field == nullptr) {return true;}// Call prepare_create_field on the Create_field that was added by// add_functional_index_to_create_list().assert(is_field_for_functional_index(new_create_field));if (prepare_create_field(thd, error_schema_name, error_table_name,create_info, &alter_info->create_list,&select_field_pos, file, new_create_field,++field_no)) {return true;}}}// Now that we have all the Create_fields available, calculate the offsets// for each column.calculate_field_offsets(&alter_info->create_list);/*Auto increment and blob checks.*/int auto_increment = 0;int blob_columns = 0;it.rewind();while ((sql_field = it++)) {if (sql_field->auto_flags & Field::NEXT_NUMBER) auto_increment++;switch (sql_field->sql_type) {case MYSQL_TYPE_GEOMETRY:case MYSQL_TYPE_BLOB:case MYSQL_TYPE_MEDIUM_BLOB:case MYSQL_TYPE_TINY_BLOB:case MYSQL_TYPE_LONG_BLOB:case MYSQL_TYPE_JSON:blob_columns++;break;default:if (sql_field->is_array) blob_columns++;break;}}if (auto_increment > 1) {my_error(ER_WRONG_AUTO_KEY, MYF(0));return true;}if (auto_increment && (file->ha_table_flags() & HA_NO_AUTO_INCREMENT)) {my_error(ER_TABLE_CANT_HANDLE_AUTO_INCREMENT, MYF(0));return true;}if (blob_columns && (file->ha_table_flags() & HA_NO_BLOBS)) {my_error(ER_TABLE_CANT_HANDLE_BLOB, MYF(0));return true;}/*CREATE TABLE[with auto_increment column] SELECT is unsafe as the rowsinserted in the created table depends on the order of the rows fetchedfrom the select tables. This order may differ on master and slave. Wetherefore mark it as unsafe.*/if (select_field_count > 0 && auto_increment)thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_CREATE_SELECT_AUTOINC);/*Count keys and key segments.Also mark redundant keys to be ignored.*/uint key_parts;Mem_root_array<bool> redundant_keys(thd->mem_root,alter_info->key_list.size(), false);if (count_keys(alter_info->key_list, key_count, &key_parts, fk_key_count,&redundant_keys, file->ha_table_flags()))return true;if (*key_count > file->max_keys()) {my_error(ER_TOO_MANY_KEYS, MYF(0), file->max_keys());return true;}/*Make KEY objects for the keys in the new table.*/KEY *key_info;(*key_info_buffer) = key_info = (KEY *)sql_calloc(sizeof(KEY) * (*key_count));KEY_PART_INFO *key_part_info =(KEY_PART_INFO *)sql_calloc(sizeof(KEY_PART_INFO) * key_parts);if (!*key_info_buffer || !key_part_info) return true;  // Out of memoryMem_root_array<const KEY *> keys_to_check(thd->mem_root);if (keys_to_check.reserve(*key_count)) return true;  // Out of memoryuint key_number = 0;bool primary_key = false;// First prepare non-foreign keys so that they are ready when// we prepare foreign keys.for (size_t i = 0; i < alter_info->key_list.size(); i++) {if (redundant_keys[i]) continue;  // Skip redundant keysconst Key_spec *key = alter_info->key_list[i];if (key->type == KEYTYPE_PRIMARY) {if (primary_key) {my_error(ER_MULTIPLE_PRI_KEY, MYF(0));return true;}primary_key = true;}if (key->type != KEYTYPE_FOREIGN) {if (prepare_key(thd, error_schema_name, error_table_name, create_info,&alter_info->create_list, key, key_info_buffer, key_info,&key_part_info, keys_to_check, key_number, file,&auto_increment))return true;key_info++;key_number++;}}// If the table is created without PK, we must check if this has// been disabled and return error. Limit the effect of sql_require_primary_key// to only those SEs that can participate in replication.if (!primary_key && !thd->is_dd_system_thread() &&!thd->is_initialize_system_thread() &&(file->ha_table_flags() &(HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE)) != 0 &&thd->variables.sql_require_primary_key) {my_error(ER_TABLE_WITHOUT_PK, MYF(0));return true;}/*At this point all KEY objects are for indexes are fully constructed.So we can check for duplicate indexes for keys for which it was requested.*/const KEY **dup_check_key;for (dup_check_key = keys_to_check.begin();dup_check_key != keys_to_check.end(); dup_check_key++) {if (check_duplicate_key(thd, error_schema_name, error_table_name,*dup_check_key, *key_info_buffer, *key_count,alter_info))return true;}if (!primary_key && check_promoted_index(file, *key_info_buffer, *key_count))return true;/*Any auto increment columns not found during prepare_key?*/if (auto_increment > 0) {my_error(ER_WRONG_AUTO_KEY, MYF(0));return true;}/* Sort keys in optimized order */std::sort(*key_info_buffer, *key_info_buffer + *key_count, sort_keys());/*Normal keys are done, now prepare foreign keys.We do this after sorting normal keys to get predictable behaviorwhen searching for parent keys for self-referencing foreign keys.*/bool se_supports_fks =(create_info->db_type->flags & HTON_SUPPORTS_FOREIGN_KEYS);assert(se_supports_fks || existing_fks_count == 0);(*fk_key_count) += existing_fks_count;FOREIGN_KEY *fk_key_info;(*fk_key_info_buffer) = fk_key_info =(FOREIGN_KEY *)sql_calloc(sizeof(FOREIGN_KEY) * (*fk_key_count));if (!fk_key_info) return true;  // Out of memory// Copy pre-existing foreign keys.if (existing_fks_count > 0)memcpy(*fk_key_info_buffer, existing_fks,existing_fks_count * sizeof(FOREIGN_KEY));uint fk_number = existing_fks_count;fk_key_info += existing_fks_count;/*Check if we are trying to add partitioning to the table with existingforeign keys and table's storage engine doesn't support foreign keysover partitioned tables.*/if (is_partitioned && existing_fks_count > 0 &&(!create_info->db_type->partition_flags ||create_info->db_type->partition_flags() & HA_CANNOT_PARTITION_FK)) {my_error(ER_FOREIGN_KEY_ON_PARTITIONED, MYF(0));return true;}/*Check that definitions of existing foreign keys are not broken by thisALTER TABLE. Update FOREIGN_KEY::unique_constraint_name if necessary.*/for (FOREIGN_KEY *fk = *fk_key_info_buffer;fk < (*fk_key_info_buffer) + existing_fks_count; fk++) {if (prepare_preexisting_foreign_key(thd, create_info, alter_info, error_schema_name, error_table_name,*key_info_buffer, *key_count, existing_fks_table, fk))return true;}// Prepare new foreign keys.for (size_t i = 0; i < alter_info->key_list.size(); i++) {if (redundant_keys[i]) continue;  // Skip redundant keysKey_spec *key = alter_info->key_list[i];if (key->type == KEYTYPE_FOREIGN) {if (prepare_foreign_key(thd, create_info, alter_info, error_schema_name,error_table_name, is_partitioned,*key_info_buffer, *key_count, *fk_key_info_buffer,fk_number, se_supports_fks, find_parent_keys,down_cast<Foreign_key_spec *>(key),&fk_max_generated_name_number, fk_key_info))return true;if (se_supports_fks) {fk_key_info++;fk_number++;}}}/*Check if  STRICT SQL mode is active and server is not started with--explicit-defaults-for-timestamp. Below check was added to prevent implicitdefault 0 value of timestamp. When explicit-defaults-for-timestamp serveroption is removed, whole set of check can be removed.Note that this check must be after KEYs have been created as thiscan cause the NOT_NULL_FLAG to be set.*/if (thd->variables.sql_mode & MODE_NO_ZERO_DATE &&!thd->variables.explicit_defaults_for_timestamp) {it.rewind();while ((sql_field = it++)) {if (!sql_field->constant_default && !sql_field->gcol_info &&is_timestamp_type(sql_field->sql_type) &&(sql_field->flags & NOT_NULL_FLAG) &&!(sql_field->auto_flags & Field::DEFAULT_NOW)) {/*An error should be reported if:- there is no explicit DEFAULT clause (default column value);- this is a TIMESTAMP column;- the column is not NULL;- this is not the DEFAULT CURRENT_TIMESTAMP column.And from checks before while loop,- STRICT SQL mode is active;- server is not started with --explicit-defaults-for-timestampIn other words, an error should be reported if- STRICT SQL mode is active;- the column definition is equivalent to'column_name TIMESTAMP DEFAULT 0'.*/my_error(ER_INVALID_DEFAULT, MYF(0), sql_field->field_name);return true;}}}/* If fixed row records, we need one bit to check for deleted rows */if (!(create_info->table_options & HA_OPTION_PACK_RECORD))create_info->null_bits++;const ulong data_offset = (create_info->null_bits + 7) / 8;size_t reclength = data_offset;it.rewind();while ((sql_field = it++)) {const size_t length = sql_field->pack_length();if (sql_field->offset + data_offset + length > reclength)reclength = sql_field->offset + data_offset + length;}if (reclength > file->max_record_length()) {my_error(ER_TOO_BIG_ROWSIZE, MYF(0),static_cast<long>(file->max_record_length()));return true;}return false;
}