1 /*
2 Copyright (c) 2018, 2020, Oracle and/or its affiliates. All rights reserved.
3
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU General Public License, version 2.0,
6 as published by the Free Software Foundation.
7
8 This program is also distributed with certain software (including
9 but not limited to OpenSSL) that is licensed under separate terms,
10 as designated in a particular file or component or in included license
11 documentation. The authors of MySQL hereby grant you an additional
12 permission to link the program and your derivative works with the
13 separately licensed software that they have included with MySQL.
14
15 This program is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License, version 2.0, for more details.
19
20 You should have received a copy of the GNU General Public License
21 along with this program; if not, write to the Free Software
22 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
23 */
24
25 // Implements the interface defined in
26 #include "storage/ndb/plugin/ndb_util_table.h"
27
28 #include <cstring>
29 #include <memory>
30 #include <utility>
31
32 #include "my_base.h"
33 #include "my_byteorder.h" // uint2korr
34 #include "mysql_version.h"
35 #include "ndbapi/NdbRecAttr.hpp" // NdbRecAttr
36 #include "sql/sql_class.h" // THD
37 #include "storage/ndb/plugin/ha_ndbcluster_binlog.h"
38 #include "storage/ndb/plugin/ndb_dd_client.h"
39 #include "storage/ndb/plugin/ndb_dd_table.h"
40 #include "storage/ndb/plugin/ndb_local_connection.h"
41 #include "storage/ndb/plugin/ndb_log.h"
42 #include "storage/ndb/plugin/ndb_ndbapi_util.h"
43 #include "storage/ndb/plugin/ndb_tdc.h"
44 #include "storage/ndb/plugin/ndb_thd_ndb.h"
45
46 class Db_name_guard {
47 Ndb *const m_ndb;
48 const std::string m_save_old_dbname;
49 Db_name_guard() = delete;
50 Db_name_guard(const Db_name_guard &) = delete;
51
52 public:
Db_name_guard(Ndb * ndb,const std::string dbname)53 Db_name_guard(Ndb *ndb, const std::string dbname)
54 : m_ndb(ndb), m_save_old_dbname(ndb->getDatabaseName()) {
55 m_ndb->setDatabaseName(dbname.c_str());
56 }
57
~Db_name_guard()58 ~Db_name_guard() {
59 // Restore old dbname
60 m_ndb->setDatabaseName(m_save_old_dbname.c_str());
61 }
62 };
63
64 class Util_table_creator {
65 THD *const m_thd;
66 Thd_ndb *const m_thd_ndb;
67 Ndb_util_table &m_util_table;
68 std::string m_name;
69
db_name() const70 const char *db_name() const { return m_util_table.db_name(); }
table_name() const71 const char *table_name() const { return m_util_table.table_name(); }
72
73 bool create_or_upgrade_in_NDB(bool upgrade_allowed, bool &reinstall) const;
74
75 bool install_in_DD(bool reinstall);
76
77 bool setup_table_for_binlog() const;
78
79 public:
80 Util_table_creator(THD *, Thd_ndb *, Ndb_util_table &);
81 Util_table_creator() = delete;
82 Util_table_creator(const Util_table_creator &) = delete;
83
84 bool create_or_upgrade(bool upgrade_allowed, bool create_events);
85 };
86
Ndb_util_table(Thd_ndb * thd_ndb,std::string db_name,std::string table_name,bool hidden,bool events)87 Ndb_util_table::Ndb_util_table(Thd_ndb *thd_ndb, std::string db_name,
88 std::string table_name, bool hidden, bool events)
89 : m_thd_ndb(thd_ndb),
90 m_table_guard(thd_ndb->ndb->getDictionary()),
91 m_db_name(std::move(db_name)),
92 m_table_name(std::move(table_name)),
93 m_hidden(hidden),
94 m_create_events(events) {}
95
~Ndb_util_table()96 Ndb_util_table::~Ndb_util_table() {}
97
get_thd() const98 const THD *Ndb_util_table::get_thd() const { return m_thd_ndb->get_thd(); }
99
get_ndb() const100 Ndb *Ndb_util_table::get_ndb() const { return m_thd_ndb->ndb; }
101
create_or_upgrade(THD * thd,bool upgrade_flag)102 bool Ndb_util_table::create_or_upgrade(THD *thd, bool upgrade_flag) {
103 Util_table_creator creator(thd, m_thd_ndb, *this);
104 return creator.create_or_upgrade(upgrade_flag, m_create_events);
105 }
106
push_warning(const char * fmt,...) const107 void Ndb_util_table::push_warning(const char *fmt, ...) const {
108 // Assemble the message
109 char message[512];
110 va_list args;
111 va_start(args, fmt);
112 vsnprintf(message, sizeof(message), fmt, args);
113 va_end(args);
114
115 m_thd_ndb->push_warning("[%s.%s] %s", m_db_name.c_str(), m_table_name.c_str(),
116 message);
117 }
118
push_ndb_error_warning(const NdbError & ndb_err) const119 void Ndb_util_table::push_ndb_error_warning(const NdbError &ndb_err) const {
120 push_warning("NDB error: %d %s", ndb_err.code, ndb_err.message);
121 }
122
exists() const123 bool Ndb_util_table::exists() const {
124 Ndb *ndb = m_thd_ndb->ndb;
125
126 // Set correct database name on the Ndb object
127 Db_name_guard db_guard(ndb, m_db_name.c_str());
128
129 // Load up the table definition from NDB dictionary
130 Ndb_table_guard ndb_tab(ndb->getDictionary(), m_table_name.c_str());
131
132 if (ndb_tab.get_table() == nullptr) {
133 // Table does not exist in NDB
134 return false;
135 }
136
137 // Table exists in NDB
138 return true;
139 }
140
open(bool reload_table)141 bool Ndb_util_table::open(bool reload_table) {
142 Ndb *ndb = m_thd_ndb->ndb;
143
144 // Set correct database name on the Ndb object
145 Db_name_guard db_guard(ndb, m_db_name.c_str());
146
147 if (unlikely(reload_table)) {
148 DBUG_ASSERT(m_table_guard.get_table() != nullptr);
149 // Reload the table definition from NDB dictionary
150 m_table_guard.invalidate();
151 m_table_guard.reinit();
152 } else {
153 // Load up the table definition from NDB dictionary
154 m_table_guard.init(m_table_name.c_str());
155 }
156
157 const NdbDictionary::Table *tab = m_table_guard.get_table();
158 if (!tab) {
159 push_warning("Failed to open table %s.%s from NDB", m_db_name.c_str(),
160 m_table_name.c_str());
161 return false;
162 }
163
164 return true;
165 }
166
get_table() const167 const NdbDictionary::Table *Ndb_util_table::get_table() const {
168 return m_table_guard.get_table();
169 }
170
get_column(const char * name) const171 const NdbDictionary::Column *Ndb_util_table::get_column(
172 const char *name) const {
173 return get_table()->getColumn(name);
174 }
175
check_column_exist(const char * name) const176 bool Ndb_util_table::check_column_exist(const char *name) const {
177 if (get_column(name) == nullptr) {
178 push_warning("Could not find expected column '%s'", name);
179 return false;
180 }
181 return true;
182 }
183
check_primary_key(const std::vector<const char * > columns) const184 bool Ndb_util_table::check_primary_key(
185 const std::vector<const char *> columns) const {
186 // Check that the primary key of the table matches the given columns
187 int keys = 0;
188 for (const char *name : columns) {
189 if (!get_column(name)->getPrimaryKey()) {
190 push_warning("Column '%s' is not part of primary key", name);
191 return false;
192 }
193 keys++;
194 }
195 if (keys != get_table()->getNoOfPrimaryKeys()) {
196 push_warning("Invalid primary key");
197 return false;
198 }
199 return true;
200 }
201
get_column_max_length(const char * name) const202 int Ndb_util_table::get_column_max_length(const char *name) const {
203 return get_column(name)->getLength();
204 }
205
check_column_type(const NdbDictionary::Column * col,NdbDictionary::Column::Type type,const char * type_name) const206 bool Ndb_util_table::check_column_type(const NdbDictionary::Column *col,
207 NdbDictionary::Column::Type type,
208 const char *type_name) const {
209 if (col->getType() != type) {
210 push_warning("Column '%s' must be defined as '%s'", col->getName(),
211 type_name);
212 return false;
213 }
214 return true;
215 }
216
check_column_minlength(const char * name,int min_length) const217 bool Ndb_util_table::check_column_minlength(const char *name,
218 int min_length) const {
219 if (get_column(name)->getLength() < min_length) {
220 push_warning("Column '%s' is too short, need at least %d bytes", name,
221 min_length);
222 return false;
223 }
224 return true;
225 }
226
check_column_varbinary(const char * name) const227 bool Ndb_util_table::check_column_varbinary(const char *name) const {
228 return check_column_type(get_column(name), NdbDictionary::Column::Varbinary,
229 "VARBINARY");
230 }
231
check_column_binary(const char * name) const232 bool Ndb_util_table::check_column_binary(const char *name) const {
233 return check_column_type(get_column(name), NdbDictionary::Column::Binary,
234 "BINARY");
235 }
236
check_column_unsigned(const char * name) const237 bool Ndb_util_table::check_column_unsigned(const char *name) const {
238 return check_column_type(get_column(name), NdbDictionary::Column::Unsigned,
239 "INT UNSIGNED ");
240 }
241
check_column_bigunsigned(const char * name) const242 bool Ndb_util_table::check_column_bigunsigned(const char *name) const {
243 return check_column_type(get_column(name), NdbDictionary::Column::Bigunsigned,
244 "BIGINT UNSIGNED");
245 }
246
check_column_blob(const char * name) const247 bool Ndb_util_table::check_column_blob(const char *name) const {
248 return check_column_type(get_column(name), NdbDictionary::Column::Blob,
249 "BLOB");
250 }
251
check_column_nullable(const char * name,bool nullable) const252 bool Ndb_util_table::check_column_nullable(const char *name,
253 bool nullable) const {
254 if (get_column(name)->getNullable() != nullable) {
255 push_warning("Column '%s' must be defined to %sallow NULL values", name,
256 nullable ? "" : "not ");
257 return false;
258 }
259 return true;
260 }
261
define_table_add_column(NdbDictionary::Table & new_table,const NdbDictionary::Column & new_column) const262 bool Ndb_util_table::define_table_add_column(
263 NdbDictionary::Table &new_table,
264 const NdbDictionary::Column &new_column) const {
265 if (new_table.addColumn(new_column) != 0) {
266 push_warning("Failed to add column '%s'", new_column.getName());
267 return false;
268 }
269 return true;
270 }
271
define_indexes(unsigned int) const272 bool Ndb_util_table::define_indexes(unsigned int) const {
273 // Base class implementation. Override in derived classes to define indexes.
274 return true;
275 }
276
create_index(const NdbDictionary::Index & idx) const277 bool Ndb_util_table::create_index(const NdbDictionary::Index &idx) const {
278 Db_name_guard db_guard(m_thd_ndb->ndb, m_db_name.c_str());
279
280 NdbDictionary::Dictionary *dict = m_thd_ndb->ndb->getDictionary();
281 const NdbDictionary::Table *table = get_table();
282 DBUG_ASSERT(table != nullptr);
283 if (dict->createIndex(idx, *table) != 0) {
284 push_ndb_error_warning(dict->getNdbError());
285 push_warning("Failed to create index '%s'", idx.getName());
286 return false;
287 }
288 return true;
289 }
290
create_primary_ordered_index() const291 bool Ndb_util_table::create_primary_ordered_index() const {
292 NdbDictionary::Index index("PRIMARY");
293
294 index.setType(NdbDictionary::Index::OrderedIndex);
295 index.setLogging(false);
296
297 const NdbDictionary::Table *table = get_table();
298 DBUG_ASSERT(table != nullptr);
299
300 for (int i = 0; i < table->getNoOfPrimaryKeys(); i++) {
301 index.addColumnName(table->getPrimaryKey(i));
302 }
303 return create_index(index);
304 }
305
create_table_in_NDB(const NdbDictionary::Table & new_table) const306 bool Ndb_util_table::create_table_in_NDB(
307 const NdbDictionary::Table &new_table) const {
308 // Set correct database name on the Ndb object
309 Db_name_guard db_guard(m_thd_ndb->ndb, m_db_name.c_str());
310
311 NdbDictionary::Dictionary *dict = m_thd_ndb->ndb->getDictionary();
312 if (dict->createTable(new_table) != 0) {
313 push_ndb_error_warning(dict->getNdbError());
314 push_warning("Failed to create table '%s'", new_table.getName());
315 return false;
316 }
317 return true;
318 }
319
drop_table_in_NDB(const NdbDictionary::Table & old_table) const320 bool Ndb_util_table::drop_table_in_NDB(
321 const NdbDictionary::Table &old_table) const {
322 // Set correct database name on the Ndb object
323 Db_name_guard db_guard(m_thd_ndb->ndb, m_db_name.c_str());
324 NdbDictionary::Dictionary *dict = m_thd_ndb->ndb->getDictionary();
325
326 if (!drop_events_in_NDB()) {
327 push_warning("Failed to drop events for table '%s'", m_table_name.c_str());
328 return false;
329 }
330
331 if (dict->dropTableGlobal(old_table) != 0) {
332 push_ndb_error_warning(dict->getNdbError());
333 push_warning("Failed to drop table '%s'", old_table.getName());
334 return false;
335 }
336
337 return true;
338 }
339
drop_event_in_NDB(const char * event_name) const340 bool Ndb_util_table::drop_event_in_NDB(const char *event_name) const {
341 NdbDictionary::Dictionary *dict = m_thd_ndb->ndb->getDictionary();
342 if (dict->dropEvent(event_name) != 0) {
343 if (dict->getNdbError().code == 4710 || dict->getNdbError().code == 1419) {
344 // Failed to drop event but return code says it was
345 // because the event didn't exist -> all ok
346 return true;
347 }
348 push_ndb_error_warning(dict->getNdbError());
349 push_warning("Failed to drop event '%s'", event_name);
350 return false;
351 }
352 return true;
353 }
354
create(bool is_upgrade)355 bool Ndb_util_table::create(bool is_upgrade) {
356 NdbDictionary::Table new_table(m_table_name.c_str());
357
358 unsigned mysql_version = MYSQL_VERSION_ID;
359 #ifndef DBUG_OFF
360 if (m_table_name == "ndb_schema" &&
361 DBUG_EVALUATE_IF("ndb_schema_skip_create_schema_op_id", true, false)) {
362 push_warning("Creating table definition without schema_op_id column");
363 mysql_version = 50725;
364 }
365 #endif
366 if (!define_table_ndb(new_table, mysql_version)) return false;
367
368 if (!create_table_in_NDB(new_table)) return false;
369
370 // Load the new table definition into the Ndb_util_table object.
371 if (!open(is_upgrade)) return false;
372
373 if (!define_indexes(mysql_version)) return false;
374
375 if (!post_install()) return false;
376
377 return true;
378 }
379
380 // Upgrade table
upgrade()381 bool Ndb_util_table::upgrade() {
382 const NdbDictionary::Table *old_table = get_table();
383
384 // Any additional metadata could be saved before upgrade
385 // and then restored later after install
386 if (!pre_upgrade()) return false;
387
388 // Drop the old table
389 if (!drop_table_in_NDB(*old_table)) return false;
390
391 // Create the new table
392 if (!create(true)) return false;
393
394 return true;
395 }
396
unpack_varbinary(NdbRecAttr * ndbRecAttr)397 std::string Ndb_util_table::unpack_varbinary(NdbRecAttr *ndbRecAttr) {
398 DBUG_TRACE;
399 // Function should be called only on a varbinary column
400 DBUG_ASSERT(ndbRecAttr->getType() == NdbDictionary::Column::Varbinary ||
401 ndbRecAttr->getType() == NdbDictionary::Column::Longvarbinary);
402
403 const char *value_start;
404 size_t value_length;
405 ndb_unpack_varchar(ndbRecAttr->getColumn(), 0, &value_start, &value_length,
406 ndbRecAttr->aRef());
407
408 return std::string(value_start, value_length);
409 }
410
pack_varbinary(const char * column_name,const char * src,char * dst) const411 void Ndb_util_table::pack_varbinary(const char *column_name, const char *src,
412 char *dst) const {
413 // The table has to be loaded before this function is called
414 DBUG_ASSERT(get_table() != nullptr);
415 // The type of column should be VARBINARY
416 DBUG_ASSERT(check_column_varbinary(column_name));
417 ndb_pack_varchar(get_column(column_name), 0, src, std::strlen(src), dst);
418 }
419
unpack_varbinary(const char * column_name,const char * packed_str) const420 std::string Ndb_util_table::unpack_varbinary(const char *column_name,
421 const char *packed_str) const {
422 // The table has to be loaded before this function is called
423 DBUG_ASSERT(get_table() != nullptr);
424 // The type of column should be VARBINARY
425 DBUG_ASSERT(check_column_varbinary(column_name));
426 const char *unpacked_str;
427 size_t unpacked_str_length;
428 ndb_unpack_varchar(get_column(column_name), 0, &unpacked_str,
429 &unpacked_str_length, packed_str);
430
431 return std::string(unpacked_str, unpacked_str_length);
432 }
433
unpack_blob_not_null(NdbBlob * ndb_blob_handle,std::string * blob_value)434 bool Ndb_util_table::unpack_blob_not_null(NdbBlob *ndb_blob_handle,
435 std::string *blob_value) {
436 // Read length of blob
437 Uint64 blob_len;
438 if (ndb_blob_handle->getLength(blob_len) != 0) {
439 return false;
440 }
441 if (blob_len == 0) {
442 // The blob column didn't contain anything, return empty string
443 return true;
444 }
445
446 // Read the blob content
447 Uint32 read_len = static_cast<Uint32>(blob_len);
448 std::unique_ptr<char[]> read_buf(new char[read_len]);
449 if (ndb_blob_handle->readData(read_buf.get(), read_len) != 0) {
450 return false;
451 }
452 DBUG_ASSERT(blob_len == read_len); // Assert that all has been read
453 blob_value->assign(read_buf.get(), read_len);
454
455 DBUG_PRINT("unpack_blob", ("str: '%s'", blob_value->c_str()));
456 return true;
457 }
458
459 //
460 // Util_table_creator
461 //
462
Util_table_creator(THD * thd,Thd_ndb * thd_ndb,Ndb_util_table & util_table)463 Util_table_creator::Util_table_creator(THD *thd, Thd_ndb *thd_ndb,
464 Ndb_util_table &util_table)
465 : m_thd(thd), m_thd_ndb(thd_ndb), m_util_table(util_table) {
466 m_name.append(db_name()).append(".").append(table_name());
467 }
468
create_or_upgrade_in_NDB(bool upgrade_allowed,bool & reinstall) const469 bool Util_table_creator::create_or_upgrade_in_NDB(bool upgrade_allowed,
470 bool &reinstall) const {
471 ndb_log_verbose(50, "Checking '%s' table", m_name.c_str());
472
473 if (m_util_table.exists()) {
474 // Table exists already. Upgrade it if required.
475 if (!m_util_table.open()) {
476 ndb_log_error("Failed to open '%s' table", m_name.c_str());
477 return false;
478 }
479
480 if (m_util_table.need_upgrade()) {
481 ndb_log_info("The '%s' table needs to be upgraded", m_name.c_str());
482
483 if (!upgrade_allowed) {
484 ndb_log_info("Upgrade of '%s' table not allowed!", m_name.c_str());
485 // Skip upgrading the table and continue with
486 // limited functionality
487 return true;
488 }
489
490 ndb_log_info("Upgrade of '%s' table...", m_name.c_str());
491 if (!m_util_table.upgrade()) {
492 ndb_log_error("Upgrade of '%s' table failed!", m_name.c_str());
493 return false;
494 }
495 reinstall = true;
496 ndb_log_info("Upgrade of '%s' table completed", m_name.c_str());
497 }
498
499 } else {
500 // Table did not exist. Create it.
501 ndb_log_verbose(50, "The '%s' table does not exist, creating..",
502 m_name.c_str());
503
504 // Create the table using NdbApi
505 if (!m_util_table.create()) {
506 ndb_log_error("Failed to create '%s' table", m_name.c_str());
507 return false;
508 }
509 reinstall = true;
510
511 ndb_log_info("Created '%s' table", m_name.c_str());
512 }
513
514 ndb_log_verbose(50, "The '%s' table is ok", m_name.c_str());
515 return true;
516 }
517
install_in_DD(bool reinstall)518 bool Util_table_creator::install_in_DD(bool reinstall) {
519 Ndb_dd_client dd_client(m_thd);
520
521 if (!dd_client.mdl_locks_acquire_exclusive(db_name(), table_name())) {
522 ndb_log_error("Failed to MDL lock '%s' table", m_name.c_str());
523 return false;
524 }
525
526 const dd::Table *existing;
527 if (!dd_client.get_table(db_name(), table_name(), &existing)) {
528 ndb_log_error("Failed to get '%s' table from DD", m_name.c_str());
529 return false;
530 }
531
532 // Table definition exists
533 if (existing) {
534 int table_id, table_version;
535 if (!ndb_dd_table_get_object_id_and_version(existing, table_id,
536 table_version)) {
537 ndb_log_error("Failed to extract id and version from '%s' table",
538 m_name.c_str());
539 DBUG_ASSERT(false);
540 // Continue and force removal of table definition
541 reinstall = true;
542 }
543
544 // Check if table definition in DD is outdated
545 const NdbDictionary::Table *ndbtab = m_util_table.get_table();
546 if (!reinstall && (ndbtab->getObjectId() == table_id &&
547 ndbtab->getObjectVersion() == table_version)) {
548 // Existed, didn't need reinstall and version matched
549 return true;
550 }
551
552 ndb_log_verbose(1, "Removing '%s' from DD", m_name.c_str());
553 if (!dd_client.remove_table(db_name(), table_name())) {
554 ndb_log_info("Failed to remove '%s' from DD", m_name.c_str());
555 return false;
556 }
557
558 dd_client.commit();
559
560 /*
561 The table existed in and was deleted from DD. It's possible
562 that someone has tried to use it and thus it might have been
563 inserted in the table definition cache. Close the table
564 in the table definition cace(tdc).
565 */
566 ndb_log_verbose(1, "Removing '%s' from table definition cache",
567 m_name.c_str());
568 ndb_tdc_close_cached_table(m_thd, db_name(), table_name());
569 }
570
571 // Create DD table definition
572 Thd_ndb::Options_guard thd_ndb_options(m_thd_ndb);
573 // Allow creating DD table definition although table already exist in NDB
574 thd_ndb_options.set(Thd_ndb::CREATE_UTIL_TABLE);
575 // Mark table definition as hidden in DD
576 if (m_util_table.is_hidden())
577 thd_ndb_options.set(Thd_ndb::CREATE_UTIL_TABLE_HIDDEN);
578
579 Ndb_local_connection mysqld(m_thd);
580 if (mysqld.create_util_table(m_util_table.define_table_dd())) {
581 ndb_log_error("Failed to create table definition for '%s' in DD",
582 m_name.c_str());
583 return false;
584 }
585
586 return true;
587 }
588
setup_table_for_binlog() const589 bool Util_table_creator::setup_table_for_binlog() const {
590 // Acquire exclusive MDL lock on schema and table
591 Ndb_dd_client dd_client(m_thd);
592 if (!dd_client.mdl_locks_acquire_exclusive(db_name(), table_name())) {
593 ndb_log_error("Failed to acquire MDL lock for '%s' table", m_name.c_str());
594 m_thd->clear_error();
595 return false;
596 }
597
598 const dd::Table *table_def;
599 if (!dd_client.get_table(db_name(), table_name(), &table_def)) {
600 ndb_log_error("Failed to open table definition for '%s' table",
601 m_name.c_str());
602 return false;
603 }
604
605 // Setup events for this table
606 if (ndbcluster_binlog_setup_table(m_thd, m_thd_ndb->ndb, db_name(),
607 table_name(), table_def)) {
608 ndb_log_error("Failed to setup events for '%s' table", m_name.c_str());
609 return false;
610 }
611
612 return true;
613 }
614
create_or_upgrade(bool upgrade_allowed,bool create_events)615 bool Util_table_creator::create_or_upgrade(bool upgrade_allowed,
616 bool create_events) {
617 bool reinstall = false;
618 if (!create_or_upgrade_in_NDB(upgrade_allowed, reinstall)) {
619 return false;
620 }
621
622 if (!install_in_DD(reinstall)) {
623 return false;
624 }
625
626 if (create_events) {
627 if (!setup_table_for_binlog()) {
628 return false;
629 }
630 }
631 return true;
632 }
633