From 2089710881ff00250ae463c103612128b311e8b4 Mon Sep 17 00:00:00 2001 From: obdev Date: Wed, 7 Feb 2024 00:43:32 +0000 Subject: [PATCH] [FEAT MERGE] New minimal mode for tenant. --- src/share/schema/ob_table_schema.cpp | 120 +++++++++++++ src/share/schema/ob_table_schema.h | 4 + src/sql/resolver/dml/ob_del_upd_resolver.cpp | 172 ++++++++++++++++++- src/sql/resolver/dml/ob_del_upd_resolver.h | 15 +- src/sql/resolver/dml/ob_delete_resolver.cpp | 6 +- src/sql/resolver/dml/ob_update_resolver.cpp | 92 +++++++++- src/sql/resolver/dml/ob_update_resolver.h | 5 + src/storage/ls/ob_ls_tablet_service.cpp | 50 +++--- tools/upgrade/upgrade_checker.py | 21 +++ tools/upgrade/upgrade_post.py | 21 +++ tools/upgrade/upgrade_pre.py | 21 +++ 11 files changed, 484 insertions(+), 43 deletions(-) diff --git a/src/share/schema/ob_table_schema.cpp b/src/share/schema/ob_table_schema.cpp index 83407fe920..5ef4b57018 100644 --- a/src/share/schema/ob_table_schema.cpp +++ b/src/share/schema/ob_table_schema.cpp @@ -3724,6 +3724,20 @@ int ObTableSchema::is_unique_key_column(ObSchemaGetterGuard &schema_guard, uint64_t column_id, bool &is_uni) const { + // This interface is compatible with MySQL. + // For table t1(c1 int, c2 int, UNIQUE u1(c1, c2)), the + // show columns query result is as following. + // > show columns from t1; + // +-------+---------+------+-----+---------+-------+ + // | Field | Type | Null | Key | Default | Extra | + // +-------+---------+------+-----+---------+-------+ + // | c1 | int(11) | YES | MUL | NULL | | + // | c2 | int(11) | YES | | NULL | | + // +-------+---------+------+-----+---------+-------+ + // Only column within a single column unique index is considered + // as UNI key. + // So we cannot use this interface to judge whether one column + // is really an unique index column. int ret = OB_SUCCESS; is_uni = false; ObSEArray simple_index_infos; @@ -5855,6 +5869,112 @@ int ObTableSchema::get_generated_column_ids(ObIArray &column_ids) cons } return ret; } + +int ObTableSchema::is_real_unique_index_column(ObSchemaGetterGuard &schema_guard, + uint64_t column_id, + bool &is_uni) const +{ + // Whether the argument column with column_id is an unique index column. + // This interface is different with is_unique_index_column(), the latter one + // is compatible with MySQL, for multiple columns unique index, the first column + // is considered as MUL key but not unique key. + int ret = OB_SUCCESS; + is_uni = false; + ObSEArray simple_index_infos; + if (OB_FAIL(get_simple_index_infos(simple_index_infos))) { + LOG_WARN("get simple_index_infos failed", K(ret)); + } else { + for (int64_t i = 0; !is_uni && OB_SUCC(ret) && i < simple_index_infos.count(); ++i) { + const ObSimpleTableSchemaV2 *simple_index_schema = NULL; + const ObTableSchema *index_schema = NULL; + if (OB_FAIL(schema_guard.get_simple_table_schema(get_tenant_id(), + simple_index_infos.at(i).table_id_, + simple_index_schema))) { + LOG_WARN("fail to get simple table schema", K(ret), "table_id", simple_index_infos.at(i).table_id_); + } else if (OB_UNLIKELY(NULL == simple_index_schema)) { + ret = OB_TABLE_NOT_EXIST; + LOG_WARN("simple index schema from schema guard is NULL", K(ret), K(simple_index_schema)); + } else if (!simple_index_schema->is_unique_index()) { + // This is not an unique index, skip. + } else if (OB_FAIL(schema_guard.get_table_schema(get_tenant_id(), + simple_index_infos.at(i).table_id_, + index_schema))) { + LOG_WARN("fail to get table schema", K(ret), "table_id", simple_index_infos.at(i).table_id_); + } else if (OB_UNLIKELY(NULL == index_schema)) { + ret = OB_TABLE_NOT_EXIST; + LOG_WARN("index schema from schema guard is NULL", K(ret), K(index_schema)); + } else { + // check whether the columns of unique index are not nullable + ObTableSchema::const_column_iterator iter = index_schema->column_begin(); + for ( ; OB_SUCC(ret) && !is_uni && iter != index_schema->column_end(); iter++) { + const ObColumnSchemaV2 *column = *iter; + if (OB_ISNULL(column)) { + ret = OB_ERR_UNDEFINED; + LOG_WARN("unexpected err", K(ret), KPC(column)); + } else if (!column->is_index_column()) { + // this column is not index column, skip + } else if (column_id == column->get_column_id()) { + is_uni = true; + } else { /*do nothing*/ } + } + } + } // for + } + return ret; +} + +int ObTableSchema::has_not_null_unique_key(ObSchemaGetterGuard &schema_guard, bool &bool_result) const +{ + int ret = OB_SUCCESS; + bool_result = false; + ObSEArray simple_index_infos; + if (!is_valid()) { + ret = OB_SCHEMA_ERROR; + LOG_WARN("The ObTableSchema is invalid", K(ret)); + } else if (OB_FAIL(get_simple_index_infos(simple_index_infos))) { + LOG_WARN("get simple_index_infos failed", K(ret)); + } else { + const uint64_t tenant_id = get_tenant_id(); + for (int64_t i = 0; !bool_result && OB_SUCC(ret) && i < simple_index_infos.count(); ++i) { + const ObTableSchema *index_table_schema = NULL; + const ObSimpleTableSchemaV2 *simple_index_schema = NULL; + if (OB_FAIL(schema_guard.get_simple_table_schema(tenant_id, + simple_index_infos.at(i).table_id_, simple_index_schema))) { + LOG_WARN("fail to get simple table schema", K(ret), "table_id", simple_index_infos.at(i).table_id_); + } else if (OB_UNLIKELY(NULL == simple_index_schema)) { + ret = OB_TABLE_NOT_EXIST; + LOG_WARN("simple index schema from schema guard is NULL", K(ret), K(simple_index_schema)); + } else if (!simple_index_schema->is_unique_index()) { + // This is not an unique index, skip. + } else if (OB_FAIL(schema_guard.get_table_schema(tenant_id, + simple_index_infos.at(i).table_id_, index_table_schema))) { + LOG_WARN("fail to get table schema", K(tenant_id), + K(simple_index_infos.at(i).table_id_), K(ret)); + } else if (OB_ISNULL(index_table_schema)) { + ret = OB_TABLE_NOT_EXIST; + LOG_WARN("index table schema must not be NULL", K(ret), "table_id", simple_index_infos.at(i).table_id_); + } else { + // check whether the columns of unique index are not nullable + ObTableSchema::const_column_iterator iter = index_table_schema->column_begin(); + for ( ; OB_SUCC(ret) && !bool_result && iter != index_table_schema->column_end(); iter++) { + const ObColumnSchemaV2 *column = *iter; + if (OB_ISNULL(column)) { + ret = OB_ERR_UNDEFINED; + LOG_WARN("unexpected err", K(ret), KPC(column)); + } else if (!column->is_index_column()) { + // this column is not index column, skip + } else if (false == column->is_nullable() || // mysql mode + true == column->has_not_null_constraint()) { // oracle mode + // find not null column, end loop + bool_result = true; + } else { /*do nothing*/ } + } + } + } + } + return ret; +} + bool ObTableSchema::has_generated_and_partkey_column() const { bool result = false; diff --git a/src/share/schema/ob_table_schema.h b/src/share/schema/ob_table_schema.h index f07e5a3275..d9e3fb651d 100644 --- a/src/share/schema/ob_table_schema.h +++ b/src/share/schema/ob_table_schema.h @@ -1282,6 +1282,7 @@ public: int get_column_ids_without_rowkey(common::ObIArray &column_ids, const bool no_virtual = false) const; int get_generated_column_ids(common::ObIArray &column_ids) const; inline bool has_generated_column() const { return generated_columns_.num_members() > 0; } + int has_not_null_unique_key(ObSchemaGetterGuard &schema_guard, bool &bool_result) const; // The table has a generated column that is a partition key. bool has_generated_and_partkey_column() const; int check_is_stored_generated_column_base_column(uint64_t column_id, bool &is_stored_base_col) const; @@ -1555,6 +1556,9 @@ public: int sort_column_array_by_column_id(); int check_column_array_sorted_by_column_id(const bool skip_rowkey) const; int check_has_local_index(ObSchemaGetterGuard &schema_guard, bool &has_local_index) const; + int is_real_unique_index_column(ObSchemaGetterGuard &schema_guard, + uint64_t column_id, + bool &is_uni) const; int is_unique_key_column(ObSchemaGetterGuard &schema_guard, uint64_t column_id, bool &is_uni) const; diff --git a/src/sql/resolver/dml/ob_del_upd_resolver.cpp b/src/sql/resolver/dml/ob_del_upd_resolver.cpp index 187f0da8f4..a2e22be2fe 100644 --- a/src/sql/resolver/dml/ob_del_upd_resolver.cpp +++ b/src/sql/resolver/dml/ob_del_upd_resolver.cpp @@ -1767,14 +1767,36 @@ int ObDelUpdResolver::build_returning_lob_expr(ObColumnRefRawExpr *ref_expr, ObS return ret; } -bool ObDelUpdResolver::need_all_columns(const ObTableSchema &table_schema, int64_t binlog_row_image) +int ObDelUpdResolver::need_all_columns(const ObTableSchema &table_schema, + const int64_t binlog_row_image, + const int64_t need_check_uk, + bool &need_all_columns) { - return (table_schema.is_heap_table() || - table_schema.get_foreign_key_infos().count() > 0 || - table_schema.get_trigger_list().count() > 0 || - table_schema.has_check_constraint() || - table_schema.has_generated_and_partkey_column() || - binlog_row_image == ObBinlogRowImage::FULL); + int ret = OB_SUCCESS; + const bool is_binlog_full_mode = (ObBinlogRowImage::FULL == binlog_row_image) ? true : false; + bool has_not_null_uk = false; + if (!is_binlog_full_mode && need_check_uk && table_schema.is_heap_table()) { + // Need check UK for heap table only when binlog_row_image is not FULL. + ObSchemaGetterGuard *schema_guard = NULL; + CK (OB_NOT_NULL(schema_checker_)); + CK (OB_NOT_NULL(schema_guard = schema_checker_->get_schema_guard())); + if (OB_SUCC(ret) && OB_FAIL(table_schema.has_not_null_unique_key(*schema_guard, has_not_null_uk))) { + LOG_WARN("has_not_null_unique_key failed", K(ret), K(table_schema)); + } + } + // For heap table: + // 1) If it does not have [NOT NULL UK], need record all columns. + // 2) If it has [NOT NULL UK], only need record all columns when UK is one of the updated columns, + // or just need record UK in OldRow. + if (OB_SUCC(ret)) { + need_all_columns = (is_binlog_full_mode || + (table_schema.is_heap_table() && !has_not_null_uk) || + table_schema.get_foreign_key_infos().count() > 0 || + table_schema.get_trigger_list().count() > 0 || + table_schema.has_check_constraint() || + table_schema.has_generated_and_partkey_column()); + } + return ret; } int ObDelUpdResolver::add_all_columns_to_stmt(const TableItem &table_item, @@ -1809,7 +1831,6 @@ int ObDelUpdResolver::add_all_columns_to_stmt(const TableItem &table_item, } return ret; } - int ObDelUpdResolver::add_all_lob_columns_to_stmt(const TableItem &table_item, ObIArray &column_exprs) { @@ -1908,6 +1929,72 @@ int ObDelUpdResolver::add_all_rowkey_columns_to_stmt(const TableItem &table_item return ret; } +int ObDelUpdResolver::add_all_unique_key_columns_to_stmt(const TableItem &table_item, + ObIArray &column_exprs) +{ + int ret = OB_SUCCESS; + const ObTableSchema *table_schema = NULL; + const TableItem &base_table_item = table_item.get_base_table_item(); + ObDelUpdStmt *stmt = get_del_upd_stmt(); + + ObSchemaGetterGuard *schema_guard = NULL; + uint64_t view_id = OB_INVALID_ID; + CK (OB_NOT_NULL(schema_checker_)); + CK (OB_NOT_NULL(schema_guard = schema_checker_->get_schema_guard())); + + if (OB_ISNULL(stmt)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("get unexpected null", K(stmt)); + } else if (stmt->has_instead_of_trigger()) { + // do nothing, instead of trigger doesn't have rowkey + } else if (OB_ISNULL(params_.session_info_)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("params_.session_info_ is null", K(ret)); + } else if (OB_FAIL(schema_checker_->get_table_schema(params_.session_info_->get_effective_tenant_id(), + base_table_item.ref_id_, + table_schema, + base_table_item.is_link_table()))) { + LOG_WARN("table schema not found", K(base_table_item)); + } else if (NULL == table_schema) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("get invalid table schema", K(table_item)); + } else { + // extract all column_ids for unique key, and add them into stmt + ObSEArray simple_index_infos; + if (OB_FAIL(table_schema->get_simple_index_infos(simple_index_infos))) { + LOG_WARN("get simple_index_infos failed", K(ret)); + } else { + const uint64_t tenant_id = table_schema->get_tenant_id(); + for (int64_t i = 0; OB_SUCCESS == ret && i < simple_index_infos.count(); ++i) { + const ObTableSchema *index_table_schema = NULL; + if (OB_FAIL(schema_guard->get_table_schema(tenant_id, + simple_index_infos.at(i).table_id_, index_table_schema))) { + LOG_WARN("fail to get table schema", K(tenant_id), + K(simple_index_infos.at(i).table_id_), K(ret)); + } else if (OB_ISNULL(index_table_schema)) { + ret = OB_TABLE_NOT_EXIST; + LOG_WARN("index table schema must not be NULL", K(ret)); + } else if (!index_table_schema->is_unique_index()) { + // not unique index, skip + } else { + ObTableSchema::const_column_iterator iter = index_table_schema->column_begin(); + for ( ; iter != index_table_schema->column_end(); iter++) { + const ObColumnSchemaV2 *column_schema = *iter; + if (OB_ISNULL(column_schema)) { + LOG_WARN("unexpected err", KPC(column_schema)); + } else if (!column_schema->is_index_column()) { + // skip non index column + } else if (OB_FAIL(add_column_to_stmt(table_item, *column_schema, column_exprs))) { + LOG_WARN("add column to stmt failed", K(ret), K(table_item)); + } else { /*do nothing*/ } + } + } + } + } + } + return ret; +} + //for ObDelUpdStmt // add column's related columns in index to stmt // if column_id is OB_INVALID_ID, all indexes' columns would be added to stmt @@ -2012,6 +2099,47 @@ int ObDelUpdResolver::add_all_index_rowkey_to_stmt(const TableItem &table_item, return ret; } +int ObDelUpdResolver::add_necessary_columns_for_minimal_mode(const TableItem &table_item, + ObIArray &column_exprs) +{ + // The timestamp and datetime columns need to be recorded unconditionly for MINIMAL mode. + int ret = OB_SUCCESS; + const ObTableSchema *table_schema = NULL; + const TableItem& base_table_item = table_item.get_base_table_item(); + if (OB_ISNULL(params_.session_info_)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("params_.session_info_ is null", K(ret)); + } else if (OB_FAIL(schema_checker_->get_table_schema(params_.session_info_->get_effective_tenant_id(), + base_table_item.ref_id_, + table_schema, + base_table_item.is_link_table()))) { + LOG_WARN("not find table schema", K(ret), K(base_table_item)); + } else if (OB_ISNULL(table_schema)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("get unexpected null", K(table_schema), K(ret)); + } else { + ObTableSchema::const_column_iterator iter = table_schema->column_begin(); + ObTableSchema::const_column_iterator end = table_schema->column_end(); + for (; OB_SUCC(ret) && iter != end; ++iter) { + const ObColumnSchemaV2 *column = *iter; + if (OB_ISNULL(column)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("invalid column schema", K(column)); + } else if (column->get_meta_type().is_datetime() || + column->get_meta_type().is_timestamp() || + column->get_meta_type().is_time() || + column->get_meta_type().is_date() || + column->get_meta_type().is_otimestamp_type()) { + // date/datatime and time/timestamp column need to be added to old_row + if (OB_FAIL(add_column_to_stmt(table_item, *column, column_exprs))) { + LOG_WARN("add column item to stmt failed", K(ret)); + } + } else { /* do nothing */ } + } + } + return ret; +} + int ObDelUpdResolver::add_all_index_rowkey_to_stmt(const TableItem &table_item, const ObTableSchema *index_schema, ObIArray &column_items) @@ -2097,6 +2225,34 @@ int ObDelUpdResolver::add_all_partition_key_columns_to_stmt(const TableItem &tab return ret; } +int ObDelUpdResolver::add_udt_hidden_columns_for_minimal_mode(const TableItem &table_item, + const ObTableSchema *table_schema, + ObColumnRefRawExpr *col_expr, + ObIArray &column_items) +{ + int ret = OB_SUCCESS; + if (OB_ISNULL(table_schema) || OB_ISNULL(col_expr)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("invalid arguments", K(ret)); + } else if (!col_expr->get_result_type().is_user_defined_sql_type()) { + // not udt column ref, do nothing + } else { + ObSEArray hidden_cols; + if (OB_FAIL(table_schema->get_column_schema_in_same_col_group(col_expr->get_column_id(), + col_expr->get_udt_set_id(), + hidden_cols))) { + LOG_WARN("failed to get column schema", K(ret)); + } else { + for (int i = 0; i < hidden_cols.count() && OB_SUCC(ret); i++) { + if (OB_FAIL(add_column_to_stmt(table_item, *(hidden_cols[i]), column_items))) { + LOG_WARN("add column to stmt failed", K(ret), K(table_item), K(i), K(*(hidden_cols[i]))); + } + } + } + } + return ret; +} + int ObDelUpdResolver::uv_check_key_preserved(const TableItem &table_item, bool &key_preserved) { int ret = OB_SUCCESS; diff --git a/src/sql/resolver/dml/ob_del_upd_resolver.h b/src/sql/resolver/dml/ob_del_upd_resolver.h index cf525d2d03..c2d0f456ec 100644 --- a/src/sql/resolver/dml/ob_del_upd_resolver.h +++ b/src/sql/resolver/dml/ob_del_upd_resolver.h @@ -125,15 +125,19 @@ protected: virtual int process_values_function(ObRawExpr *&expr); virtual int recursive_values_expr(ObRawExpr *&expr); - bool need_all_columns(const share::schema::ObTableSchema &table_schema, int64_t binlog_row_image); + int need_all_columns(const share::schema::ObTableSchema &table_schema, + const int64_t binlog_row_image, + const int64_t need_check_uk, + bool &need_all_columns); int add_all_columns_to_stmt(const TableItem &table_item, common::ObIArray &column_exprs); int add_all_columns_to_stmt_for_trigger(const TableItem &table_item, common::ObIArray &column_exprs); int add_all_rowkey_columns_to_stmt(const TableItem &table_item, - common::ObIArray &column_exprs); - + common::ObIArray &column_exprs); + int add_all_unique_key_columns_to_stmt(const TableItem &table_item, + common::ObIArray &column_exprs); int add_index_related_columns_to_stmt(const TableItem &table_item, const uint64_t column_id, common::ObIArray &column_exprs); @@ -224,6 +228,11 @@ protected: int add_select_items(ObSelectStmt &select_stmt, const ObIArray& select_items); int add_select_list_for_set_stmt(ObSelectStmt &select_stmt); int add_all_lob_columns_to_stmt(const TableItem &table_item, ObIArray &column_exprs); + int add_necessary_columns_for_minimal_mode(const TableItem &table_item, ObIArray &column_exprs); + int add_udt_hidden_columns_for_minimal_mode(const TableItem &table_item, + const ObTableSchema *table_schema, + ObColumnRefRawExpr *col_expr, + ObIArray &column_items); protected: int generate_insert_table_info(const TableItem &table_item, ObInsertTableInfo &table_info, diff --git a/src/sql/resolver/dml/ob_delete_resolver.cpp b/src/sql/resolver/dml/ob_delete_resolver.cpp index 646d7dd3b9..80d3a72cb2 100644 --- a/src/sql/resolver/dml/ob_delete_resolver.cpp +++ b/src/sql/resolver/dml/ob_delete_resolver.cpp @@ -387,6 +387,7 @@ int ObDeleteResolver::generate_delete_table_info(const TableItem &table_item) uint64_t index_tid[OB_MAX_INDEX_PER_TABLE]; int64_t gindex_cnt = OB_MAX_INDEX_PER_TABLE; int64_t binlog_row_image = ObBinlogRowImage::FULL; + bool is_need_all_columns = true; if (OB_ISNULL(schema_checker_) || OB_ISNULL(params_.session_info_) || OB_ISNULL(allocator_) || OB_ISNULL(delete_stmt)) { ret = OB_ERR_UNEXPECTED; @@ -411,13 +412,16 @@ int ObDeleteResolver::generate_delete_table_info(const TableItem &table_item) LOG_WARN("failed to allocate table info", K(ret)); } else { table_info = new(ptr) ObDeleteTableInfo(); + const bool need_check_uk = true; if (OB_FAIL(table_info->part_ids_.assign(base_table_item.part_ids_))) { LOG_WARN("failed to assign part ids", K(ret)); } else if (!delete_stmt->has_instead_of_trigger()) { // todo @zimiao error logging also need all columns ? if (OB_FAIL(add_all_rowkey_columns_to_stmt(table_item, table_info->column_exprs_))) { LOG_WARN("add all rowkey columns to stmt failed", K(ret)); - } else if (need_all_columns(*table_schema, binlog_row_image)) { + } else if (OB_FAIL(need_all_columns(*table_schema, binlog_row_image, need_check_uk, is_need_all_columns))) { + LOG_WARN("call need_all_columns failed", K(ret), K(binlog_row_image)); + } else if (is_need_all_columns) { if (OB_FAIL(add_all_columns_to_stmt(table_item, table_info->column_exprs_))) { LOG_WARN("fail to add all column to stmt", K(ret), K(table_item)); } diff --git a/src/sql/resolver/dml/ob_update_resolver.cpp b/src/sql/resolver/dml/ob_update_resolver.cpp index 5c0acdc9f4..d0a7dd7074 100644 --- a/src/sql/resolver/dml/ob_update_resolver.cpp +++ b/src/sql/resolver/dml/ob_update_resolver.cpp @@ -483,6 +483,68 @@ int ObUpdateResolver::resolve_table_list(const ParseNode &parse_tree) return ret; } +int ObUpdateResolver::is_table_has_unique_key(const ObTableSchema *table_schema, + bool &is_has_uk) const +{ + int ret = OB_SUCCESS; + is_has_uk = false; + ObSchemaGetterGuard *schema_guard = NULL; + CK (OB_NOT_NULL(schema_checker_)); + CK (OB_NOT_NULL(schema_guard = schema_checker_->get_schema_guard())); + ObSEArray simple_index_infos; + if (NULL == table_schema) { + ret = OB_INVALID_ARGUMENT; + LOG_WARN("invalid table schema", K(table_schema)); + } else if (OB_FAIL(table_schema->get_simple_index_infos(simple_index_infos))) { + LOG_WARN("get simple_index_infos failed", K(ret)); + } else { + const uint64_t tenant_id = table_schema->get_tenant_id(); + for (int64_t i = 0; OB_SUCC(ret) && !is_has_uk && i < simple_index_infos.count(); ++i) { + const ObTableSchema *index_table_schema = NULL; + if (OB_FAIL(schema_guard->get_table_schema(tenant_id, + simple_index_infos.at(i).table_id_, index_table_schema))) { + LOG_WARN("fail to get table schema", K(tenant_id), + K(simple_index_infos.at(i).table_id_), K(ret)); + } else if (OB_ISNULL(index_table_schema)) { + ret = OB_TABLE_NOT_EXIST; + LOG_WARN("index table schema must not be NULL", K(ret)); + } else if (index_table_schema->is_unique_index()) { + is_has_uk = true; + } else { + // not unique index, skip + } + } + } + return ret; +} + +int ObUpdateResolver::check_unique_key_is_updated(const ObTableSchema *table_schema, + const common::ObIArray &assigns, + bool &is_updated) const +{ + int ret = OB_SUCCESS; + is_updated = false; + ObSchemaGetterGuard *schema_guard = NULL; + if (OB_ISNULL(schema_guard = schema_checker_->get_schema_guard())) { + } else { + for (int64_t i = 0; OB_SUCC(ret) && !is_updated && i < assigns.count(); i++) { + // We cannot use col_expr->is_unique_key_column_ here, which may be false for a unique index column. + ObColumnRefRawExpr *col_expr = assigns.at(i).column_expr_; + bool is_unique_col = false; + if (OB_ISNULL(col_expr)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("get unexpected null", K(ret)); + } else if (OB_FAIL(table_schema->is_real_unique_index_column(*schema_guard, col_expr->get_column_id(), is_unique_col))) { + LOG_WARN("is_unique_key_column failed", K(ret)); + } else if (is_unique_col) { + is_updated = true; + break; + } else { /*do nothing*/ } + } + } + return ret; +} + int ObUpdateResolver::generate_update_table_info(ObTableAssignment &table_assign) { int ret = OB_SUCCESS; @@ -494,6 +556,7 @@ int ObUpdateResolver::generate_update_table_info(ObTableAssignment &table_assign uint64_t index_tid[OB_MAX_INDEX_PER_TABLE]; int64_t gindex_cnt = OB_MAX_INDEX_PER_TABLE; int64_t binlog_row_image = ObBinlogRowImage::FULL; + bool is_need_all_columns = false; if (OB_ISNULL(schema_checker_) || OB_ISNULL(params_.session_info_) || OB_ISNULL(allocator_) || OB_ISNULL(update_stmt)) { ret = OB_ERR_UNEXPECTED; @@ -526,13 +589,35 @@ int ObUpdateResolver::generate_update_table_info(ObTableAssignment &table_assign } else if (OB_FAIL(table_info->part_ids_.assign(table_item->get_base_table_item().part_ids_))) { LOG_WARN("failed to assign part ids", K(ret)); } else if (!update_stmt->has_instead_of_trigger()) { + const bool is_binlog_full_mode = (ObBinlogRowImage::FULL == binlog_row_image) ? true : false; + is_need_all_columns = is_binlog_full_mode; + bool is_has_uk = false; + bool is_uk_updated = false; if (OB_FAIL(add_all_rowkey_columns_to_stmt(*table_item, table_info->column_exprs_))) { LOG_WARN("add all rowkey columns to stmt failed", K(ret)); - } else if (need_all_columns(*table_schema, binlog_row_image) || - update_stmt->is_error_logging()) { + } else if (!is_need_all_columns && + OB_FAIL(is_table_has_unique_key(table_schema, is_has_uk))) { + LOG_WARN("is_table_has_unique_key failed", K(ret)); + } else if (!is_need_all_columns && + is_has_uk && + OB_FAIL(check_unique_key_is_updated(table_schema, table_assign.assignments_, is_uk_updated))) { + // Check whether UK is updated only when binlog_row_image is not FULL. + LOG_WARN("check_unique_key_is_updated failed", K(ret)); + } else if (!is_need_all_columns && + OB_FAIL(need_all_columns(*table_schema, binlog_row_image, is_has_uk, is_need_all_columns))) { + LOG_WARN("call need_all_columns failed", K(ret), K(binlog_row_image)); + } else if (is_need_all_columns || + update_stmt->is_error_logging() || + is_uk_updated) { if (OB_FAIL(add_all_columns_to_stmt(*table_item, table_info->column_exprs_))) { LOG_WARN("fail to add all column to stmt", K(ret), K(*table_item)); } + } else if (is_has_uk && + OB_FAIL(add_all_unique_key_columns_to_stmt(*table_item, table_info->column_exprs_))) { + LOG_WARN("add all unique columns to stmt failed", K(ret)); + } else if (!is_binlog_full_mode && + OB_FAIL(add_necessary_columns_for_minimal_mode(*table_item, table_info->column_exprs_))) { + LOG_WARN("fail to add necessary columns for minimal mode", K(ret), K(*table_item)); } else { for (int64_t i = 0; OB_SUCC(ret) && i < table_assign.assignments_.count(); ++i) { ObAssignment &assign = table_assign.assignments_.at(i); @@ -541,6 +626,9 @@ int ObUpdateResolver::generate_update_table_info(ObTableAssignment &table_assign } else if (OB_FAIL(add_index_related_columns_to_stmt(*table_item, assign.column_expr_->get_column_id(), table_info->column_exprs_))) { LOG_WARN("failed to add index columns", K(ret)); + } else if (OB_FAIL(add_udt_hidden_columns_for_minimal_mode(*table_item, + table_schema, assign.column_expr_, table_info->column_exprs_))) { + LOG_WARN("failed to add udt hidden columns", K(ret)); } else { /*do nothing*/ } } } diff --git a/src/sql/resolver/dml/ob_update_resolver.h b/src/sql/resolver/dml/ob_update_resolver.h index a5d8070b6a..7232252acc 100644 --- a/src/sql/resolver/dml/ob_update_resolver.h +++ b/src/sql/resolver/dml/ob_update_resolver.h @@ -58,6 +58,11 @@ private: int check_safe_update_mode(ObUpdateStmt *update_stmt); int resolve_update_constraints(); int generate_batched_stmt_info(); + int is_table_has_unique_key(const ObTableSchema *table_schema, + bool &is_has_uk) const; + int check_unique_key_is_updated(const ObTableSchema *table_schema, + const common::ObIArray &assigns, + bool &is_updated) const; }; } // namespace sql diff --git a/src/storage/ls/ob_ls_tablet_service.cpp b/src/storage/ls/ob_ls_tablet_service.cpp index 6573a2c6c2..d683797859 100644 --- a/src/storage/ls/ob_ls_tablet_service.cpp +++ b/src/storage/ls/ob_ls_tablet_service.cpp @@ -4584,25 +4584,15 @@ int ObLSTabletService::process_old_row( if (OB_FAIL(ret)) { } else if (data_tbl_rowkey_change) { ObStoreRow del_row(tbl_row); - del_row.flag_.set_flag(ObDmlFlag::DF_DELETE); - if (!is_delete_total_quantity_log) { - if (OB_FAIL(tablet_handle.get_obj()->insert_row_without_rowkey_check(relative_table, - run_ctx.store_ctx_, col_descs, del_row, run_ctx.dml_param_.encrypt_meta_))) { - if (OB_TRY_LOCK_ROW_CONFLICT != ret && OB_TRANSACTION_SET_VIOLATION != ret) { - LOG_WARN("failed to write data tablet row", K(ret), K(del_row)); - } - } - } else { - ObStoreRow new_tbl_row; - new_tbl_row.flag_.set_flag(ObDmlFlag::DF_DELETE); - new_tbl_row.row_val_ = tbl_row.row_val_; - del_row.flag_.set_flag(ObDmlFlag::DF_UPDATE); - ObSEArray update_idx; - if (OB_FAIL(tablet_handle.get_obj()->update_row(relative_table, - run_ctx.store_ctx_, col_descs, update_idx, del_row, new_tbl_row, run_ctx.dml_param_.encrypt_meta_))) { - if (OB_TRY_LOCK_ROW_CONFLICT != ret && OB_TRANSACTION_SET_VIOLATION != ret) { - LOG_WARN("failed to write data tablet row", K(ret), K(del_row), K(new_tbl_row)); - } + ObStoreRow new_tbl_row; + new_tbl_row.flag_.set_flag(ObDmlFlag::DF_DELETE); + new_tbl_row.row_val_ = tbl_row.row_val_; + del_row.flag_.set_flag(ObDmlFlag::DF_UPDATE); + ObSEArray update_idx; + if (OB_FAIL(tablet_handle.get_obj()->update_row(relative_table, + run_ctx.store_ctx_, col_descs, update_idx, del_row, new_tbl_row, run_ctx.dml_param_.encrypt_meta_))) { + if (OB_TRY_LOCK_ROW_CONFLICT != ret && OB_TRANSACTION_SET_VIOLATION != ret) { + LOG_WARN("failed to write data tablet row", K(ret), K(del_row), K(new_tbl_row)); } } } else if (lob_update) { @@ -4665,7 +4655,7 @@ int ObLSTabletService::process_data_table_row( int ret = OB_SUCCESS; ObStoreCtx &ctx = run_ctx.store_ctx_; ObRelativeTable &relative_table = run_ctx.relative_table_; - bool is_update_total_quantity_log = run_ctx.dml_param_.is_total_quantity_log_; + const bool is_update_total_quantity_log = run_ctx.dml_param_.is_total_quantity_log_; const common::ObTimeZoneInfo *tz_info = run_ctx.dml_param_.tz_info_; if (OB_UNLIKELY(!ctx.is_valid() || !relative_table.is_valid() @@ -4707,10 +4697,18 @@ int ObLSTabletService::process_data_table_row( ObStoreRow new_row; new_row.flag_.set_flag(rowkey_change ? ObDmlFlag::DF_INSERT : ObDmlFlag::DF_UPDATE); new_row.row_val_ = new_tbl_row.row_val_; - if (is_update_total_quantity_log && !rowkey_change) { + if (!rowkey_change) { ObStoreRow old_row; old_row.flag_.set_flag(ObDmlFlag::DF_UPDATE); old_row.row_val_ = old_tbl_row.row_val_; + if (!is_update_total_quantity_log) { + // For minimal mode, set pk columns of old_row to nop value, because + // they are already stored in new_row. + const int64_t rowkey_col_cnt = relative_table.get_rowkey_column_num(); + for (int64_t i = 0; i < rowkey_col_cnt; ++i) { + (old_row.row_val_.cells_[i]).set_nop_value(); + } + } if (OB_FAIL(data_tablet.get_obj()->update_row(relative_table, ctx, col_descs, update_idx, old_row, new_row, run_ctx.dml_param_.encrypt_meta_))) { if (OB_TRY_LOCK_ROW_CONFLICT != ret && OB_TRANSACTION_SET_VIOLATION != ret) { @@ -5161,7 +5159,8 @@ int ObLSTabletService::process_old_row_lob_col( const ObColDesc &column = run_ctx.col_descs_->at(i); if (is_lob_storage(column.col_type_.get_type())) { has_lob_col = true; - need_reread = need_reread || !(tbl_row.row_val_.cells_[i].has_lob_header()); + ObObj &obj = tbl_row.row_val_.cells_[i]; + need_reread = need_reread || (!obj.is_null() && !obj.is_nop_value() && !obj.has_lob_header()); break; } } @@ -5297,13 +5296,6 @@ int ObLSTabletService::delete_row_in_tablet( LOG_WARN("failed to process old row lob col", K(ret), K(tbl_row)); } else if (OB_FAIL(delete_lob_tablet_rows(run_ctx, tablet_handle, tbl_row, row))) { LOG_WARN("failed to delete lob rows.", K(ret), K(tbl_row), K(row)); - } else if (!dml_param.is_total_quantity_log_) { - if (OB_FAIL(tablet_handle.get_obj()->insert_row_without_rowkey_check(relative_table, - ctx, *run_ctx.col_descs_, tbl_row, dml_param.encrypt_meta_))) { - if (OB_TRY_LOCK_ROW_CONFLICT != ret && OB_TRANSACTION_SET_VIOLATION != ret) { - LOG_WARN("failed to set row", K(ret), K(*run_ctx.col_descs_), K(tbl_row)); - } - } } else { update_idx.reset(); // update_idx is a dummy param here new_tbl_row.reset(); diff --git a/tools/upgrade/upgrade_checker.py b/tools/upgrade/upgrade_checker.py index 03e93bf6a0..4d1be06b51 100755 --- a/tools/upgrade/upgrade_checker.py +++ b/tools/upgrade/upgrade_checker.py @@ -590,6 +590,26 @@ def check_not_supported_tenant_name(query_cur): break logging.info('check special tenant name success') +# 17. 检查是否有租户在升到4.3.0版本之前已将binlog_row_image设为MINIMAL +def check_variable_binlog_row_image(query_cur): +# 4.3.0.0之前的版本,MINIMAL模式生成的日志CDC无法正常消费(DELETE日志). +# 4.3.0版本开始,MINIMAL模式做了改进,支持CDC消费,需要在升级到4.3.0.0之后再打开. + min_cluster_version = 0 + sql = """select distinct value from GV$OB_PARAMETERS where name='min_observer_version'""" + (desc, results) = query_cur.exec_query(sql) + if len(results) != 1: + fail_list.append('min_observer_version is not sync') + elif len(results[0]) != 1: + fail_list.append('column cnt not match') + else: + min_cluster_version = get_version(results[0][0]) + # check cluster version + if min_cluster_version < get_version("4.3.0.0"): + (desc, results) = query_cur.exec_query("""select count(*) from CDB_OB_SYS_VARIABLES where NAME='binlog_row_image' and VALUE = '0'""") + if results[0][0] > 0 : + fail_list.append('Sys Variable binlog_row_image is set to MINIMAL, please check'.format(results[0][0])) + logging.info('check variable binlog_row_image success') + # last check of do_check, make sure no function execute after check_fail_list def check_fail_list(): if len(fail_list) != 0 : @@ -632,6 +652,7 @@ def do_check(my_host, my_port, my_user, my_passwd, timeout, upgrade_params): check_schema_status(query_cur) check_server_version(query_cur) check_not_supported_tenant_name(query_cur) + check_variable_binlog_row_image(query_cur) # all check func should execute before check_fail_list check_fail_list() modify_server_permanent_offline_time(cur) diff --git a/tools/upgrade/upgrade_post.py b/tools/upgrade/upgrade_post.py index ea3f2cf42c..75ec211068 100755 --- a/tools/upgrade/upgrade_post.py +++ b/tools/upgrade/upgrade_post.py @@ -2160,6 +2160,26 @@ # break # logging.info('check special tenant name success') # +## 17. 检查是否有租户在升到4.3.0版本之前已将binlog_row_image设为MINIMAL +#def check_variable_binlog_row_image(query_cur): +## 4.3.0.0之前的版本,MINIMAL模式生成的日志CDC无法正常消费(DELETE日志). +## 4.3.0版本开始,MINIMAL模式做了改进,支持CDC消费,需要在升级到4.3.0.0之后再打开. +# min_cluster_version = 0 +# sql = """select distinct value from GV$OB_PARAMETERS where name='min_observer_version'""" +# (desc, results) = query_cur.exec_query(sql) +# if len(results) != 1: +# fail_list.append('min_observer_version is not sync') +# elif len(results[0]) != 1: +# fail_list.append('column cnt not match') +# else: +# min_cluster_version = get_version(results[0][0]) +# # check cluster version +# if min_cluster_version < get_version("4.3.0.0"): +# (desc, results) = query_cur.exec_query("""select count(*) from CDB_OB_SYS_VARIABLES where NAME='binlog_row_image' and VALUE = '0'""") +# if results[0][0] > 0 : +# fail_list.append('Sys Variable binlog_row_image is set to MINIMAL, please check'.format(results[0][0])) +# logging.info('check variable binlog_row_image success') +# ## last check of do_check, make sure no function execute after check_fail_list #def check_fail_list(): # if len(fail_list) != 0 : @@ -2202,6 +2222,7 @@ # check_schema_status(query_cur) # check_server_version(query_cur) # check_not_supported_tenant_name(query_cur) +# check_variable_binlog_row_image(query_cur) # # all check func should execute before check_fail_list # check_fail_list() # modify_server_permanent_offline_time(cur) diff --git a/tools/upgrade/upgrade_pre.py b/tools/upgrade/upgrade_pre.py index 4ac72b8915..180769a5c8 100755 --- a/tools/upgrade/upgrade_pre.py +++ b/tools/upgrade/upgrade_pre.py @@ -2160,6 +2160,26 @@ # break # logging.info('check special tenant name success') # +## 17. 检查是否有租户在升到4.3.0版本之前已将binlog_row_image设为MINIMAL +#def check_variable_binlog_row_image(query_cur): +## 4.3.0.0之前的版本,MINIMAL模式生成的日志CDC无法正常消费(DELETE日志). +## 4.3.0版本开始,MINIMAL模式做了改进,支持CDC消费,需要在升级到4.3.0.0之后再打开. +# min_cluster_version = 0 +# sql = """select distinct value from GV$OB_PARAMETERS where name='min_observer_version'""" +# (desc, results) = query_cur.exec_query(sql) +# if len(results) != 1: +# fail_list.append('min_observer_version is not sync') +# elif len(results[0]) != 1: +# fail_list.append('column cnt not match') +# else: +# min_cluster_version = get_version(results[0][0]) +# # check cluster version +# if min_cluster_version < get_version("4.3.0.0"): +# (desc, results) = query_cur.exec_query("""select count(*) from CDB_OB_SYS_VARIABLES where NAME='binlog_row_image' and VALUE = '0'""") +# if results[0][0] > 0 : +# fail_list.append('Sys Variable binlog_row_image is set to MINIMAL, please check'.format(results[0][0])) +# logging.info('check variable binlog_row_image success') +# ## last check of do_check, make sure no function execute after check_fail_list #def check_fail_list(): # if len(fail_list) != 0 : @@ -2202,6 +2222,7 @@ # check_schema_status(query_cur) # check_server_version(query_cur) # check_not_supported_tenant_name(query_cur) +# check_variable_binlog_row_image(query_cur) # # all check func should execute before check_fail_list # check_fail_list() # modify_server_permanent_offline_time(cur)