oceanbase

1
/**
2
 * Copyright (c) 2021 OceanBase
3
 * OceanBase CE is licensed under Mulan PubL v2.
4
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
5
 * You may obtain a copy of Mulan PubL v2 at:
6
 *          http://license.coscl.org.cn/MulanPubL-2.0
7
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
8
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
9
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10
 * See the Mulan PubL v2 for more details.
11
 */
12

13
#define USING_LOG_PREFIX RS
14
#include "rootserver/parallel_ddl/ob_create_table_helper.h"
15
#include "rootserver/parallel_ddl/ob_index_name_checker.h"
16
#include "rootserver/ob_index_builder.h"
17
#include "rootserver/ob_lob_meta_builder.h"
18
#include "rootserver/ob_lob_piece_builder.h"
19
#include "rootserver/ob_table_creator.h"
20
#include "rootserver/ob_balance_group_ls_stat_operator.h"
21
#include "rootserver/freeze/ob_major_freeze_helper.h"
22
#include "share/ob_rpc_struct.h"
23
#include "share/ob_index_builder_util.h"
24
#include "share/ob_debug_sync_point.h"
25
#include "share/sequence/ob_sequence_option_builder.h" // ObSequenceOptionBuilder
26
#include "share/schema/ob_table_sql_service.h"
27
#include "share/schema/ob_security_audit_sql_service.h"
28
#include "share/schema/ob_sequence_sql_service.h"
29
#include "share/schema/ob_multi_version_schema_service.h"
30
#include "sql/resolver/ob_resolver_utils.h"
31

32
using namespace oceanbase::lib;
33
using namespace oceanbase::common;
34
using namespace oceanbase::share;
35
using namespace oceanbase::share::schema;
36
using namespace oceanbase::rootserver;
37

38
uint64_t ObCreateTableHelper::MockFKParentTableNameWrapper::hash() const
39
{
40
  uint64_t hash_ret = 0;
41
  common::ObCollationType cs_type = ObSchema::get_cs_type_with_cmp_mode(OB_ORIGIN_AND_INSENSITIVE);
42
  hash_ret = common::ObCharset::hash(cs_type, parent_database_, hash_ret);
43
  hash_ret = common::ObCharset::hash(cs_type, parent_table_, hash_ret);
44
  return hash_ret;
45
}
46

47
int ObCreateTableHelper::MockFKParentTableNameWrapper::hash(uint64_t &hash_val) const
48
{
49
  hash_val = hash();
50
  return OB_SUCCESS;
51
}
52

53
bool ObCreateTableHelper::MockFKParentTableNameWrapper::operator==(const MockFKParentTableNameWrapper &rv) const
54
{
55
  return 0 == parent_database_.case_compare(rv.parent_database_)
56
         && 0 == parent_table_.case_compare(rv.parent_table_);
57
}
58

59
ObCreateTableHelper::ObCreateTableHelper(
60
    share::schema::ObMultiVersionSchemaService *schema_service,
61
    const uint64_t tenant_id,
62
    const obrpc::ObCreateTableArg &arg,
63
    obrpc::ObCreateTableRes &res)
64
  : ObDDLHelper(schema_service, tenant_id),
65
    arg_(arg),
66
    res_(res),
67
    replace_mock_fk_parent_table_id_(common::OB_INVALID_ID),
68
    new_tables_(),
69
    new_mock_fk_parent_tables_(),
70
    new_mock_fk_parent_table_map_(),
71
    new_audits_(),
72
    new_sequences_(),
73
    has_index_(false)
74
{}
75

76
ObCreateTableHelper::~ObCreateTableHelper()
77
{
78
}
79

80
int ObCreateTableHelper::init_()
81
{
82
  int ret = OB_SUCCESS;
83
  const int64_t BUCKET_NUM = 100;
84
  if (OB_FAIL(new_mock_fk_parent_table_map_.create(BUCKET_NUM, "MockFkPMap", "MockFkPMap"))) {
85
    LOG_WARN("fail to init mock fk parent table map", KR(ret));
86
  }
87
  return ret;
88
}
89

90
int ObCreateTableHelper::execute()
91
{
92
  RS_TRACE(create_table_begin);
93
  int ret = OB_SUCCESS;
94
  if (OB_FAIL(check_inner_stat_())) {
95
    LOG_WARN("fail to check inner stat", KR(ret));
96
  } else if (OB_FAIL(init_())) {
97
    LOG_WARN("fail to init struct", KR(ret));
98
  } else if (OB_FAIL(start_ddl_trans_())) {
99
    LOG_WARN("fail to start ddl trans", KR(ret));
100
  } else if (OB_FAIL(lock_objects_())) {
101
    LOG_WARN("fail to lock objects", KR(ret));
102
  } else if (OB_FAIL(generate_schemas_())) {
103
    LOG_WARN("fail to generate schemas", KR(ret));
104
  } else if (OB_FAIL(calc_schema_version_cnt_())) {
105
    LOG_WARN("fail to calc schema version cnt", KR(ret));
106
  } else if (OB_FAIL(gen_task_id_and_schema_versions_())) {
107
    LOG_WARN("fail to gen task id and schema versions", KR(ret));
108
  } else if (OB_FAIL(create_schemas_())) {
109
    LOG_WARN("fail create schemas", KR(ret));
110
  } else if (OB_FAIL(create_tablets_())) {
111
    LOG_WARN("fail create schemas", KR(ret));
112
  } else if (OB_FAIL(serialize_inc_schema_dict_())) {
113
    LOG_WARN("fail to serialize inc schema dict", KR(ret));
114
  } else if (OB_FAIL(wait_ddl_trans_())) {
115
    LOG_WARN("fail to wait ddl trans", KR(ret));
116
  } else if (OB_FAIL(add_index_name_to_cache_())) {
117
    LOG_WARN("fail to add index name to cache", KR(ret));
118
  }
119

120
  const bool commit = OB_SUCC(ret);
121
  if (OB_FAIL(end_ddl_trans_(ret))) { // won't overwrite ret
122
    LOG_WARN("fail to end ddl trans", KR(ret));
123
    if (commit && has_index_) {
124
      // Because index name is added to cache before trans commit,
125
      // it will remain garbage in cache when trans commit failed and false alarm will occur.
126
      //
127
      // To solve this problem:
128
      // 1. check_index_name_exist() will double check by inner_sql and erase garbage if index name conflicts.
129
      // 2. (Fully unnecessary) clean up index name cache when trans commit failed.
130
      int tmp_ret = OB_SUCCESS;
131
      if (OB_ISNULL(ddl_service_)) {
132
        tmp_ret = OB_ERR_UNEXPECTED;
133
        LOG_WARN("ddl_service_ is null", KR(tmp_ret));
134
      } else if (OB_TMP_FAIL(ddl_service_->get_index_name_checker().reset_cache(tenant_id_))) {
135
        LOG_ERROR("fail to reset cache", K(ret), KR(tmp_ret), K_(tenant_id));
136
      }
137
    }
138
  }
139

140
  if (OB_SUCC(ret)) {
141
    auto *tsi_generator = GET_TSI(TSISchemaVersionGenerator);
142
    int64_t last_schema_version = OB_INVALID_VERSION;
143
    int64_t end_schema_version = OB_INVALID_VERSION;
144
    if (OB_UNLIKELY(new_tables_.count() <= 0)) {
145
      ret = OB_ERR_UNEXPECTED;
146
      LOG_WARN("table cnt is invalid", KR(ret));
147
    } else if (OB_ISNULL(tsi_generator)) {
148
      ret = OB_ERR_UNEXPECTED;
149
      LOG_WARN("tsi schema version generator is null", KR(ret));
150
    } else if (OB_FAIL(tsi_generator->get_current_version(last_schema_version))) {
151
      LOG_WARN("fail to get end version", KR(ret), K_(tenant_id), K_(arg));
152
    } else if (OB_FAIL(tsi_generator->get_end_version(end_schema_version))) {
153
      LOG_WARN("fail to get end version", KR(ret), K_(tenant_id), K_(arg));
154
    } else if (OB_UNLIKELY(last_schema_version != end_schema_version)) {
155
      ret = OB_ERR_UNEXPECTED;
156
      LOG_WARN("too much schema versions may be allocated", KR(ret), KPC(tsi_generator));
157
    } else {
158
      res_.table_id_ = new_tables_.at(0).get_table_id();
159
      res_.schema_version_ = last_schema_version;
160
    }
161
  }
162

163
  if (OB_ERR_TABLE_EXIST == ret) {
164
    const ObTableSchema &table = arg_.schema_;
165
    //create table xx if not exist (...)
166
    if (arg_.if_not_exist_) {
167
      ret = OB_SUCCESS;
168
      LOG_INFO("table is exist, no need to create again",
169
               "tenant_id", table.get_tenant_id(),
170
               "database_id", table.get_database_id(),
171
               "table_name", table.get_table_name());
172
    } else {
173
      LOG_WARN("table is exist, cannot create it twice", KR(ret),
174
               "tenant_id", table.get_tenant_id(),
175
               "database_id", table.get_database_id(),
176
               "table_name", table.get_table_name());
177
      LOG_USER_ERROR(OB_ERR_TABLE_EXIST,
178
                     table.get_table_name_str().length(),
179
                     table.get_table_name_str().ptr());
180
    }
181
  }
182

183
  RS_TRACE(create_table_end);
184
  FORCE_PRINT_TRACE(THE_RS_TRACE, "[parallel create table]");
185
  return ret;
186
}
187

188
int ObCreateTableHelper::lock_objects_()
189
{
190
  int ret = OB_SUCCESS;
191
  if (OB_FAIL(check_inner_stat_())) {
192
    LOG_WARN("fail to check inner stat", KR(ret));
193
  }
194
  // 1. lock database name first
195
  if (FAILEDx(lock_database_by_obj_name_())) {
196
    LOG_WARN("fail to lock databases by obj name", KR(ret), K_(tenant_id));
197
  }
198
  // 2. lock objects by name
199
  if (FAILEDx(lock_objects_by_name_())) {
200
    LOG_WARN("fail to lock objects by name", KR(ret), K_(tenant_id));
201
  }
202
  DEBUG_SYNC(AFTER_PARALLEL_DDL_LOCK_OBJ_BY_NAME);
203
  // 3. prefetch schemas
204
  if (FAILEDx(prefetch_schemas_())) {
205
    LOG_WARN("fail to prefech schemas", KR(ret), K_(tenant_id));
206
  }
207
  // 4. lock objects by id
208
  if (FAILEDx(lock_objects_by_id_())) {
209
    LOG_WARN("fail to lock objects by name", KR(ret), K_(tenant_id));
210
  }
211
  // 5. lock objects by id after related objects are locked.
212
  if (FAILEDx(post_lock_objects_by_id_())) {
213
    LOG_WARN("fail to lock objects by id in post", KR(ret));
214
  }
215
  // 6. check ddl conflict
216
  if (FAILEDx(check_ddl_conflict_())) {
217
    LOG_WARN("fail to check ddl confict", KR(ret));
218
  }
219
  RS_TRACE(lock_objects);
220
  return ret;
221
}
222

223
// Lock ddl related database by name
224
// 1. database        (S)
225
// - to create table
226
// 2. parent database (S)
227
// - in foreign key
228
int ObCreateTableHelper::lock_database_by_obj_name_()
229
{
230
  int ret = OB_SUCCESS;
231
  const int64_t start_ts =  ObTimeUtility::current_time();
232
  if (OB_FAIL(check_inner_stat_())) {
233
    LOG_WARN("fail to check inner stat", KR(ret));
234
  } else {
235
    const ObString &database_name = arg_.db_name_;
236
    if (OB_FAIL(add_lock_object_by_database_name_(database_name, transaction::tablelock::SHARE))) {
237
      LOG_WARN("fail to lock database by name", KR(ret), K_(tenant_id), K(database_name));
238
    }
239

240
    for (int64_t i = 0; OB_SUCC(ret) && i < arg_.foreign_key_arg_list_.count(); i++) {
241
      const obrpc::ObCreateForeignKeyArg &foreign_key_arg = arg_.foreign_key_arg_list_.at(i);
242
      const ObString &parent_database_name = foreign_key_arg.parent_database_;
243
       if (OB_FAIL(add_lock_object_by_database_name_(parent_database_name, transaction::tablelock::SHARE))) {
244
         LOG_WARN("fail to lock database by name", KR(ret), K_(tenant_id), K(parent_database_name));
245
       }
246
    } // end for
247

248
    if (FAILEDx(lock_databases_by_name_())) {
249
      LOG_WARN("fail to lock databases by name", KR(ret), K_(tenant_id));
250
    }
251
  }
252
  const int64_t cost_ts = ObTimeUtility::current_time() - start_ts;
253
  LOG_INFO("lock databases", KR(ret), K_(tenant_id), K(cost_ts));
254
  return ret;
255
}
256

257
// Lock related objects' name for create table (`X` for EXCLUSIVE, `S` for SHARE):
258
// 1. table               (X)
259
// 2. index               (X)
260
// - in oracle mode
261
// 3. constraint          (X)
262
// 4. foreign key         (X)
263
// 5. tablegroup          (S)
264
// 6. sequence            (X)
265
// - Operation to lock sequence name will be delayed to generate_schemas_() stage.
266
// 7. parent table        (X)
267
// 8. mock fk parent table(X)
268
int ObCreateTableHelper::lock_objects_by_name_()
269
{
270
  int ret = OB_SUCCESS;
271
  bool is_oracle_mode = false;
272
  const int64_t start_ts =  ObTimeUtility::current_time();
273
  if (OB_FAIL(check_inner_stat_())) {
274
    LOG_WARN("fail to check inner stat", KR(ret));
275
  } else if (OB_FAIL(ObCompatModeGetter::check_is_oracle_mode_with_tenant_id(
276
             tenant_id_, is_oracle_mode))) {
277
    LOG_WARN("fail to check is oracle mode", KR(ret));
278
  } else {
279
    const ObString &database_name = arg_.db_name_;
280
    const ObTableSchema &table = arg_.schema_;
281

282
    // 1. table
283
    const ObString &table_name = table.get_table_name();
284
    if (OB_FAIL(add_lock_object_by_name_(database_name, table_name,
285
        share::schema::TABLE_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
286
      LOG_WARN("fail to lock object by table name", KR(ret), K_(tenant_id), K(database_name), K(table_name));
287
    }
288

289
    // 2. index (oracle) :
290
    // 1) oracle mode: index name is unique in user(database)
291
    // 2) mysql mode: index name is unique in table
292
    if (OB_SUCC(ret) && is_oracle_mode) {
293
      for (int64_t i = 0; OB_SUCC(ret) && i < arg_.index_arg_list_.size(); i++) {
294
        const ObString &index_name = arg_.index_arg_list_.at(i).index_name_; // original index name
295
        if (OB_FAIL(add_lock_object_by_name_(database_name, index_name,
296
            share::schema::TABLE_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
297
          LOG_WARN("fail to lock object by index name", KR(ret), K_(tenant_id), K(database_name), K(index_name));
298
        }
299
      } // end for
300
    }
301

302
    // 3. constraint
303
    for (int64_t i = 0; OB_SUCC(ret) && i < arg_.constraint_list_.count(); i++) {
304
      const ObString &cst_name = arg_.constraint_list_.at(i).get_constraint_name_str();
305
      if (OB_FAIL(add_lock_object_by_name_(database_name, cst_name,
306
          share::schema::CONSTRAINT_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
307
        LOG_WARN("fail to lock object by constraint name", KR(ret), K_(tenant_id), K(database_name), K(cst_name));
308
      }
309
    } // end for
310

311
    // 4. foreign key
312
    for (int64_t i = 0; OB_SUCC(ret) && i < arg_.foreign_key_arg_list_.count(); i++) {
313
      const ObString &fk_name = arg_.foreign_key_arg_list_.at(i).foreign_key_name_;
314
      if (OB_FAIL(add_lock_object_by_name_(database_name, fk_name,
315
          share::schema::FOREIGN_KEY_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
316
        LOG_WARN("fail to lock object by foreign key name", KR(ret), K_(tenant_id), K(database_name), K(fk_name));
317
      }
318
    } // end for
319

320
    // 5. tablegroup
321
    const ObString &tablegroup_name = arg_.schema_.get_tablegroup_name();
322
    if (OB_SUCC(ret) && !tablegroup_name.empty()) {
323
      ObString mock_database_name(OB_SYS_DATABASE_NAME);  // consider that tablegroup may across databases.
324
      if (OB_FAIL(add_lock_object_by_name_(mock_database_name, tablegroup_name,
325
          share::schema::TABLEGROUP_SCHEMA, transaction::tablelock::SHARE))) {
326
        LOG_WARN("fail to lock object by tablegroup name", KR(ret), K_(tenant_id), K(database_name), K(tablegroup_name));
327
      }
328
    }
329

330
    // 6. sequence
331
    // - will be delayed to generate_schemas_() stage.
332

333
    // 7. parent table/mock fk parent table
334
    // - here we don't distinguish between table and mocked table.
335
    for (int64_t i = 0; OB_SUCC(ret) && i < arg_.foreign_key_arg_list_.count(); i++) {
336
      const obrpc::ObCreateForeignKeyArg &foreign_key_arg = arg_.foreign_key_arg_list_.at(i);
337
      const ObString &parent_database_name = foreign_key_arg.parent_database_;
338
      const ObString &parent_table_name = foreign_key_arg.parent_table_;
339
      if (OB_FAIL(add_lock_object_by_name_(parent_database_name, parent_table_name,
340
          share::schema::TABLE_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
341
        LOG_WARN("fail to lock object by parent table", KR(ret),
342
                 K_(tenant_id), K(parent_database_name), K(parent_table_name));
343
      }
344
    } // end for
345

346
    if (FAILEDx(lock_existed_objects_by_name_())) {
347
      LOG_WARN("fail to lock objects by name", KR(ret), K_(tenant_id));
348
    }
349
  }
350
  const int64_t cost_ts = ObTimeUtility::current_time() - start_ts;
351
  LOG_INFO("lock objects by name", KR(ret), K_(tenant_id), K(cost_ts));
352
  return ret;
353
}
354

355
// lock related objects' id for create table (`X` for EXCLUSIVE, `S` for SHARE):
356
// 1. tablegroup          (S)
357
// 2. audit               (S)
358
// - add share lock for OB_AUDIT_MOCK_USER_ID
359
// 3. tablespace          (S)
360
// 4. parent table        (X)
361
// 5. mock fk parent table(X)
362
// 6. udt                 (S)
363

364
// Specially, when table name is duplicated with existed mock fk parent table name,
365
// we may replace mock fk parent table with new table and modified related foreign key/child table.
366
// So after lock table/mock fk parent table by name, we should lock all foreign keys/child tables
367
// from mock fk parent table by id.
368
//
369
// 6. mock fk parent table for replacement (X)
370
int ObCreateTableHelper::lock_objects_by_id_()
371
{
372
  int ret = OB_SUCCESS;
373
  const int64_t start_ts =  ObTimeUtility::current_time();
374
  const ObTableSchema &table = arg_.schema_;
375
  if (OB_FAIL(check_inner_stat_())) {
376
    LOG_WARN("fail to check inner stat", KR(ret));
377
  }
378
  // 1. tablegroup
379
  const uint64_t tablegroup_id = table.get_tablegroup_id();
380
  if (OB_SUCC(ret) && OB_INVALID_ID != tablegroup_id) {
381
    if (OB_FAIL(add_lock_object_by_id_(tablegroup_id, // filled in set_tablegroup_id_()
382
        share::schema::TABLEGROUP_SCHEMA, transaction::tablelock::SHARE))) {
383
      LOG_WARN("fail to lock tablegroup_id", KR(ret), K_(tenant_id), K(tablegroup_id));
384
    }
385
  }
386
  // 2. audit (lock by user_id)
387
  if (OB_SUCC(ret)) {
388
    if (OB_FAIL(add_lock_object_by_id_(OB_AUDIT_MOCK_USER_ID,
389
        share::schema::USER_SCHEMA, transaction::tablelock::SHARE))) {
390
      LOG_WARN("fail to lock user_id", KR(ret), K_(tenant_id), "user_id", OB_AUDIT_MOCK_USER_ID);
391
    }
392
  }
393
  // 3. tablespace
394
  const uint64_t tablespace_id = arg_.schema_.get_tablespace_id();
395
  if (OB_SUCC(ret) && OB_INVALID_ID != tablespace_id) {
396
    if (OB_FAIL(add_lock_object_by_id_(tablespace_id,
397
        share::schema::TABLESPACE_SCHEMA, transaction::tablelock::SHARE))) {
398
      LOG_WARN("fail to lock tablespace_id", KR(ret), K_(tenant_id), K(tablespace_id));
399
    }
400
  }
401
  // 4. parent table/mock fk parent table
402
  for (int64_t i = 0; OB_SUCC(ret) && i < arg_.foreign_key_arg_list_.count(); i++) {
403
    const obrpc::ObCreateForeignKeyArg &foreign_key_arg = arg_.foreign_key_arg_list_.at(i);
404
    if (OB_INVALID_ID != foreign_key_arg.parent_table_id_) { // filled in check_and_set_parent_table_id_()
405
      if (OB_FAIL(add_lock_object_by_id_(foreign_key_arg.parent_table_id_,
406
          share::schema::TABLE_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
407
        LOG_WARN("fail to lock parent table id",
408
                 KR(ret), K_(tenant_id), "parent_table_id", foreign_key_arg.parent_table_id_);
409
      }
410
    }
411
  } // end for
412
  // 5. lock mock fk parent table for replacement
413
  if (OB_SUCC(ret)) {
414
    const ObString &table_name = table.get_table_name();
415
    const uint64_t database_id = table.get_database_id();
416
    //TODO(yanmu.ztl): this interface has poor performance.
417
    if (OB_FAIL(latest_schema_guard_.get_mock_fk_parent_table_id(
418
        database_id, table_name, replace_mock_fk_parent_table_id_))) {
419
      LOG_WARN("fail to ge mock fk parent table id",
420
               KR(ret), K_(tenant_id), K(database_id), K(table_name));
421
    } else if (OB_INVALID_ID != replace_mock_fk_parent_table_id_) {
422
      // has existed mock fk parent table
423
      if (OB_FAIL(add_lock_object_by_id_(replace_mock_fk_parent_table_id_,
424
          share::schema::TABLE_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
425
        LOG_WARN("fail to lock mock fk parent table id",
426
                 KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
427
      }
428
    }
429
  }
430
  // 6. udt
431
  if (OB_SUCC(ret)) {
432
    ObTableSchema::const_column_iterator begin = table.column_begin();
433
    ObTableSchema::const_column_iterator end = table.column_end();
434
    ObSchemaGetterGuard guard;
435
    if (OB_FAIL(schema_service_->get_tenant_schema_guard(OB_SYS_TENANT_ID, guard))) {
436
      LOG_WARN("fail to get schema guard", KR(ret));
437
    }
438
    for (; OB_SUCC(ret) && begin != end; begin++) {
439
      ObColumnSchemaV2 *col = (*begin);
440
      if (OB_ISNULL(col)) {
441
        ret = OB_ERR_UNEXPECTED;
442
        LOG_WARN("get column schema failed", KR(ret));
443
      } else if (col->is_extend()) {
444
        const uint64_t udt_id = col->get_sub_data_type();
445
        if (is_inner_object_id(udt_id) && !is_sys_tenant(tenant_id_)) {
446
          // can't add object lock across tenant, assumed that sys inner udt won't be changed.
447
          const ObUDTTypeInfo *udt_info = NULL;
448
          if (OB_FAIL(guard.get_udt_info(OB_SYS_TENANT_ID, udt_id, udt_info))) {
449
            LOG_WARN("fail to get udt info", KR(ret), K(udt_id));
450
          } else if (OB_ISNULL(udt_info)) {
451
            ret = OB_ERR_PARALLEL_DDL_CONFLICT;
452
            LOG_WARN("inner udt not found", KR(ret), K(udt_id));
453
          }
454
        } else if (OB_FAIL(add_lock_object_by_id_(udt_id,
455
                   share::schema::UDT_SCHEMA, transaction::tablelock::SHARE))) {
456
          LOG_WARN("fail to lock udt id", KR(ret), K_(tenant_id), K(udt_id));
457
        }
458
      }
459
    } // end for
460
  }
461

462
  if (FAILEDx(lock_existed_objects_by_id_())) {
463
    LOG_WARN("fail to lock objects by id", KR(ret));
464
  }
465
  const int64_t cost_ts = ObTimeUtility::current_time() - start_ts;
466
  LOG_INFO("lock objects by id", KR(ret), K_(tenant_id), K(cost_ts));
467
  return ret;
468
}
469

470
// tablegroup/mock fk parent table for replacement are locked by id in lock_objects_by_id_() first.
471
// 1. foreign key         (X)
472
// - from mock fk parent table
473
// 2. child table         (X)
474
// - from mock fk parent table
475
// 3. primary table       (S)
476
// - in tablegroup
477
//
478
// TODO:(yanmu.ztl)
479
// small timeout should be used here to avoid deadlock problem
480
// since we have already locked some objects by id.
481
int ObCreateTableHelper::post_lock_objects_by_id_()
482
{
483
  int ret = OB_SUCCESS;
484
  const int64_t start_ts =  ObTimeUtility::current_time();
485
  if (OB_FAIL(check_inner_stat_())) {
486
    LOG_WARN("fail to check inner stat", KR(ret));
487
  }
488
  // lock foreign key/child table if need replace existed mock fk parent table.
489
  if (OB_SUCC(ret) && OB_INVALID_ID != replace_mock_fk_parent_table_id_) {
490
    const ObMockFKParentTableSchema *mock_fk_parent_table = NULL;
491
    if (OB_FAIL(latest_schema_guard_.get_mock_fk_parent_table_schema(
492
        replace_mock_fk_parent_table_id_, mock_fk_parent_table))) {
493
      LOG_WARN("fail to get mock fk parent table schema",
494
               KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
495
    } else if (OB_ISNULL(mock_fk_parent_table)) {
496
      ret = OB_ERR_PARALLEL_DDL_CONFLICT;
497
      LOG_WARN("mock fk parent table not exist, ddl need retry",
498
               KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
499
    } else {
500
      for (int64_t i = 0; OB_SUCC(ret) && i < mock_fk_parent_table->get_foreign_key_infos().count(); i++) {
501
        const ObForeignKeyInfo &foreign_key = mock_fk_parent_table->get_foreign_key_infos().at(i);
502
        const uint64_t foreign_key_id = foreign_key.foreign_key_id_;
503
        const uint64_t child_table_id = foreign_key.child_table_id_;
504
        if (OB_FAIL(add_lock_object_by_id_(child_table_id,
505
            share::schema::TABLE_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
506
          LOG_WARN("fail to lock child table",
507
                   KR(ret), K_(tenant_id), K(child_table_id));
508
        } else if (OB_FAIL(add_lock_object_by_id_(foreign_key_id,
509
            share::schema::FOREIGN_KEY_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
510
          LOG_WARN("fail to lock foreign key",
511
                   KR(ret), K_(tenant_id), K(foreign_key_id));
512
        }
513
      } // end for
514
    }
515
  }
516

517
  // TODO:(yanmu.ztl) lock primary table in tablegroup
518
  if (OB_SUCC(ret) && OB_INVALID_ID != arg_.schema_.get_tablegroup_id()) {
519
  }
520

521
  if (FAILEDx(lock_existed_objects_by_id_())) {
522
    LOG_WARN("fail to lock objects by id", KR(ret));
523
  }
524
  const int64_t cost_ts = ObTimeUtility::current_time() - start_ts;
525
  LOG_INFO("lock objects by id in post", KR(ret), K_(tenant_id), K(cost_ts));
526
  return ret;
527
}
528

529
int ObCreateTableHelper::check_ddl_conflict_()
530
{
531
  int ret = OB_SUCCESS;
532
  const int64_t start_ts =  ObTimeUtility::current_time();
533
  if (OB_FAIL(check_inner_stat_())) {
534
    LOG_WARN("fail to check inner stat", KR(ret));
535
  } else if (!arg_.is_need_check_based_schema_objects()) {
536
    // skip
537
  } else {
538
    ObArray<uint64_t> parent_table_ids;
539
    ObArray<uint64_t> mock_fk_parent_table_ids;
540
    // check schema object infos are all existed.
541
    for (int64_t i = 0; OB_SUCC(ret) && (i < arg_.based_schema_object_infos_.count()); ++i) {
542
      const ObBasedSchemaObjectInfo &info = arg_.based_schema_object_infos_.at(i);
543
      if (MOCK_FK_PARENT_TABLE_SCHEMA == info.schema_type_
544
          || TABLE_SCHEMA == info.schema_type_) {
545
        bool find = false;
546
        for (int64_t j = 0; OB_SUCC(ret) && !find && j < arg_.foreign_key_arg_list_.count(); j++) {
547
          const obrpc::ObCreateForeignKeyArg &foreign_key_arg = arg_.foreign_key_arg_list_.at(j);
548
          if (MOCK_FK_PARENT_TABLE_SCHEMA == info.schema_type_
549
              && foreign_key_arg.is_parent_table_mock_
550
              && info.schema_id_ == foreign_key_arg.parent_table_id_) {
551
            find = true;
552
          } else if (TABLE_SCHEMA == info.schema_type_
553
                     && !foreign_key_arg.is_parent_table_mock_
554
                     && info.schema_id_ == foreign_key_arg.parent_table_id_) {
555
            find = true;
556
          }
557
        } // end for
558
        if (OB_SUCC(ret) && !find) {
559
          ret = OB_ERR_PARALLEL_DDL_CONFLICT;
560
          LOG_WARN("parent table may change, ddl need retry",
561
                   KR(ret), K_(tenant_id), K(info));
562
        }
563
        if (OB_FAIL(ret)) {
564
        } else if (MOCK_FK_PARENT_TABLE_SCHEMA == info.schema_type_) {
565
          if (!has_exist_in_array(mock_fk_parent_table_ids, info.schema_id_)
566
              && OB_FAIL(mock_fk_parent_table_ids.push_back(info.schema_id_))) {
567
            LOG_WARN("fail to push back mock fk parent table id", KR(ret), K(info));
568
          }
569
        } else if (TABLE_SCHEMA == info.schema_type_) {
570
          if (!has_exist_in_array(parent_table_ids, info.schema_id_)
571
              && OB_FAIL(parent_table_ids.push_back(info.schema_id_))) {
572
            LOG_WARN("fail to push back parent table id", KR(ret), K(info));
573
          }
574
        }
575
      }
576
    } // end for
577

578
    ObArray<ObSchemaIdVersion> parent_table_versions;
579
    if (OB_SUCC(ret) && parent_table_ids.count() > 0) {
580
      if (OB_FAIL(parent_table_versions.reserve(parent_table_ids.count()))) {
581
        LOG_WARN("fail to reserve array", KR(ret));
582
      } else if (OB_FAIL(latest_schema_guard_.get_table_schema_versions(
583
                 parent_table_ids, parent_table_versions))) {
584
        LOG_WARN("fail to get table schema versions", KR(ret));
585
      } else if (parent_table_ids.count() != parent_table_versions.count()) {
586
        ret = OB_ERR_PARALLEL_DDL_CONFLICT;
587
        LOG_WARN("parent table may be deleted, ddl need retry",
588
                 KR(ret), K_(tenant_id), "base_objs_cnt", parent_table_ids.count(),
589
                 "fetch_cnt", parent_table_versions.count());
590
      }
591
    }
592

593
    ObArray<ObSchemaIdVersion> mock_fk_parent_table_versions;
594
    if (OB_SUCC(ret) && mock_fk_parent_table_ids.count() > 0) {
595
      if (OB_FAIL(mock_fk_parent_table_versions.reserve(mock_fk_parent_table_ids.count()))) {
596
        LOG_WARN("fail to reserve array", KR(ret));
597
      } else if (OB_FAIL(latest_schema_guard_.get_mock_fk_parent_table_schema_versions(
598
                 mock_fk_parent_table_ids, mock_fk_parent_table_versions))) {
599
        LOG_WARN("fail to get table schema versions", KR(ret));
600
      } else if (mock_fk_parent_table_ids.count() != mock_fk_parent_table_versions.count()) {
601
        ret = OB_ERR_PARALLEL_DDL_CONFLICT;
602
        LOG_WARN("mock fk parent table may be deleted, ddl need retry",
603
                 KR(ret), K_(tenant_id), "base_objs_cnt", mock_fk_parent_table_ids.count(),
604
                 "fetch_cnt", mock_fk_parent_table_versions.count());
605
      }
606
    }
607

608
    if (OB_SUCC(ret)) {
609
      for (int64_t i = 0; OB_SUCC(ret) && (i < arg_.based_schema_object_infos_.count()); ++i) {
610
        const ObBasedSchemaObjectInfo &info = arg_.based_schema_object_infos_.at(i);
611
        if (MOCK_FK_PARENT_TABLE_SCHEMA == info.schema_type_
612
            || TABLE_SCHEMA == info.schema_type_) {
613
          bool find = false;
614
          for (int64_t j = 0; OB_SUCC(ret) && !find && j < parent_table_versions.count(); j++) {
615
            const ObSchemaIdVersion &version = parent_table_versions.at(j);
616
            if (version.get_schema_id() == info.schema_id_) {
617
              find = true;
618
              if (version.get_schema_version() != info.schema_version_) {
619
                ret = OB_ERR_PARALLEL_DDL_CONFLICT;
620
                LOG_WARN("parent table may be changed, ddl need retry",
621
                         KR(ret), K_(tenant_id), K(info), K(version));
622
              }
623
            }
624
          } // end for
625
          for (int64_t j = 0; OB_SUCC(ret) && !find && j < mock_fk_parent_table_versions.count(); j++) {
626
            const ObSchemaIdVersion &version = mock_fk_parent_table_versions.at(j);
627
            if (version.get_schema_id() == info.schema_id_) {
628
              find = true;
629
              if (version.get_schema_version() != info.schema_version_) {
630
                ret = OB_ERR_PARALLEL_DDL_CONFLICT;
631
                LOG_WARN("mock fk parent table may be changed, ddl need retry",
632
                         KR(ret), K_(tenant_id), K(info), K(version));
633
              }
634
            }
635
          } // end for
636
          if (OB_SUCC(ret) && !find) {
637
            ret = OB_ERR_PARALLEL_DDL_CONFLICT;
638
            LOG_WARN("parent table may be deleted, ddl need retry",
639
                     KR(ret), K_(tenant_id), K(info));
640
          }
641
        }
642
      } // end for
643
    }
644

645
    // for replace mock fk parent table:
646
    // 1. check child tables still exist.
647
    // 2. cache child tables' schema before gen_task_id_and_schema_versions_() to increase throughput.
648
    if (OB_SUCC(ret) && OB_INVALID_ID != replace_mock_fk_parent_table_id_) {
649
      const ObMockFKParentTableSchema *mock_fk_parent_table = NULL;
650
      if (OB_FAIL(latest_schema_guard_.get_mock_fk_parent_table_schema(
651
          replace_mock_fk_parent_table_id_, mock_fk_parent_table))) {
652
        LOG_WARN("fail to get mock fk parent table schema",
653
                 KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
654
      } else if (OB_ISNULL(mock_fk_parent_table)) {
655
        ret = OB_ERR_UNEXPECTED;
656
        LOG_WARN("mock fk parent table not exist after lock obj",
657
                 KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
658
      }
659
      for (int64_t i = 0; OB_SUCC(ret) && i < mock_fk_parent_table->get_foreign_key_infos().count(); i++) {
660
        const ObForeignKeyInfo &foreign_key = mock_fk_parent_table->get_foreign_key_infos().at(i);
661
        const uint64_t child_table_id = foreign_key.child_table_id_;
662
        const ObTableSchema *child_table = NULL;
663
        if (OB_FAIL(latest_schema_guard_.get_table_schema(child_table_id, child_table))) {
664
          LOG_WARN("fail to get table schema", KR(ret), K_(tenant_id), K(child_table_id));
665
        } else if (OB_ISNULL(child_table)) {
666
          ret = OB_ERR_PARALLEL_DDL_CONFLICT;
667
          LOG_WARN("child table is not exist", KR(ret), K_(tenant_id), K(child_table_id));
668
        }
669
      } // end for
670
    }
671

672
    // check udt exist & not changed
673
    for (int64_t i = 0; OB_SUCC(ret) && (i < arg_.based_schema_object_infos_.count()); ++i) {
674
      const ObBasedSchemaObjectInfo &info = arg_.based_schema_object_infos_.at(i);
675
      if (UDT_SCHEMA == info.schema_type_) {
676
        const uint64_t udt_id = info.schema_id_;
677
        const ObUDTTypeInfo *udt_info = NULL;
678
        if (is_inner_object_id(udt_id) && !is_sys_tenant(tenant_id_)) {
679
          // can't add object lock across tenant, assumed that sys inner udt won't be changed.
680
        } else if (OB_FAIL(latest_schema_guard_.get_udt_info(udt_id, udt_info))) {
681
          LOG_WARN("fail to get udt info", KR(ret), K_(tenant_id), K(udt_id), K(info));
682
        } else if (OB_ISNULL(udt_info)) {
683
          ret = OB_ERR_PARALLEL_DDL_CONFLICT;
684
          LOG_WARN("udt doesn't exist", KR(ret), K_(tenant_id), K(udt_id));
685
        } else if (udt_info->get_schema_version() != info.schema_version_) {
686
          ret = OB_ERR_PARALLEL_DDL_CONFLICT;
687
          LOG_WARN("udt changed", KR(ret), K(info), KPC(udt_info));
688
        }
689
      }
690
    } // end for
691
  }
692
  const int64_t cost_ts = ObTimeUtility::current_time() - start_ts;
693
  LOG_INFO("check ddl confict", KR(ret), K_(tenant_id), K(cost_ts));
694
  return ret;
695
}
696

697
int ObCreateTableHelper::prefetch_schemas_()
698
{
699
  int ret = OB_SUCCESS;
700
  const int64_t start_ts =  ObTimeUtility::current_time();
701
  if (OB_FAIL(check_inner_stat_())) {
702
    LOG_WARN("fail to check inner stat", KR(ret));
703
  } else if (OB_FAIL(check_and_set_database_id_())) {
704
    LOG_WARN("fail to check and set database id", KR(ret));
705
  } else if (OB_FAIL(check_table_name_())) {
706
    LOG_WARN("fail to check table name", KR(ret));
707
  } else if (OB_FAIL(set_tablegroup_id_())) {
708
    LOG_WARN("fail to set tablegroup id", KR(ret));
709
  } else if (OB_FAIL(check_and_set_parent_table_id_())) {
710
    LOG_WARN("fail to check and set parent table id", KR(ret));
711
  }
712
  const int64_t cost_ts = ObTimeUtility::current_time() - start_ts;
713
  LOG_INFO("prefetch schemas", KR(ret), K_(tenant_id), K(cost_ts));
714
  return ret;
715
}
716

717
int ObCreateTableHelper::check_and_set_database_id_()
718
{
719
  int ret = OB_SUCCESS;
720
  const ObString &database_name = arg_.db_name_;
721
  uint64_t database_id = OB_INVALID_ID;
722
  const ObDatabaseSchema *database_schema = NULL;
723
  if (OB_FAIL(check_inner_stat_())) {
724
    LOG_WARN("fail to check inner stat", KR(ret));
725
  } else if (OB_FAIL(latest_schema_guard_.get_database_id(database_name, database_id))) {
726
    LOG_WARN("fail to get database id", KR(ret), K_(tenant_id), K(database_name));
727
  } else if (OB_UNLIKELY(OB_INVALID_ID == database_id)) {
728
    ret = OB_ERR_BAD_DATABASE;
729
    LOG_WARN("database not exist",  KR(ret), K_(tenant_id), K(database_name));
730
    LOG_USER_ERROR(OB_ERR_BAD_DATABASE, database_name.length(), database_name.ptr());
731
  } else if (OB_FAIL(latest_schema_guard_.get_database_schema(database_id, database_schema))) {
732
    LOG_WARN("fail to get database schema", KR(ret), K_(tenant_id), K(database_id), K(database_name));
733
  } else if (OB_ISNULL(database_schema)) {
734
    ret = OB_ERR_BAD_DATABASE;
735
    LOG_WARN("database not exist", KR(ret), K_(tenant_id), K(database_id), K(database_name));
736
    LOG_USER_ERROR(OB_ERR_BAD_DATABASE, database_name.length(), database_name.ptr());
737
  } else if (!arg_.is_inner_ && database_schema->is_in_recyclebin()) {
738
    ret = OB_ERR_OPERATION_ON_RECYCLE_OBJECT;
739
    LOG_WARN("Can't not create table in database which is in recyclebin",
740
             KR(ret), K_(tenant_id), K(database_id), K(database_name));
741
  } else {
742
    (void) const_cast<ObTableSchema&>(arg_.schema_).set_database_id(database_id);
743
  }
744
  return ret;
745
}
746

747
int ObCreateTableHelper::check_table_name_()
748
{
749
  int ret = OB_SUCCESS;
750
  bool is_oracle_mode = false;
751
  const ObTableSchema &table = arg_.schema_;
752
  const uint64_t database_id = table.get_database_id();
753
  const ObString &table_name = table.get_table_name();
754
  // session_id is > 0 when table is mysql tmp table or creating ctas table.
755
  const uint64_t session_id = table.get_session_id();
756
  bool if_not_exist = arg_.if_not_exist_;
757
  if (OB_FAIL(check_inner_stat_())) {
758
    LOG_WARN("fail to check inner stat", KR(ret));
759
  } else if (OB_FAIL(ObCompatModeGetter::check_is_oracle_mode_with_tenant_id(tenant_id_, is_oracle_mode))) {
760
    LOG_WARN("fail to check is oracle mode", KR(ret));
761
  } else if (is_oracle_mode) {
762
    if (OB_FAIL(latest_schema_guard_.check_oracle_object_exist(
763
        database_id, session_id, table_name, TABLE_SCHEMA,
764
        INVALID_ROUTINE_TYPE, if_not_exist))) {
765
      LOG_WARN("fail to check oracle object exist", KR(ret),
766
               K(database_id), K(session_id), K(table_name), K(if_not_exist));
767
    }
768
  } else {
769
    uint64_t synonym_id = OB_INVALID_ID;
770
    uint64_t table_id = OB_INVALID_ID;
771
    ObTableType table_type = MAX_TABLE_TYPE;
772
    int64_t schema_version = OB_INVALID_VERSION;
773
    if (OB_FAIL(latest_schema_guard_.get_synonym_id(database_id, table_name, synonym_id))) {
774
      LOG_WARN("fail to get synonymn_id", KR(ret), K_(tenant_id), K(database_id), K(table_name));
775
    } else if (OB_UNLIKELY(OB_INVALID_ID != synonym_id)) {
776
      ret = OB_ERR_EXIST_OBJECT;
777
      LOG_WARN("Name is already used by an existing object",
778
               KR(ret), K_(tenant_id), K(database_id), K(table_name), K(synonym_id));
779
    } else if (OB_FAIL(latest_schema_guard_.get_table_id(
780
               database_id, session_id, table_name, table_id, table_type, schema_version))) {
781
      LOG_WARN("fail to get table_id", KR(ret), K_(tenant_id), K(database_id), K(session_id), K(table_name));
782
    } else if (OB_UNLIKELY(OB_INVALID_ID != table_id)) {
783
      if (table.is_mysql_tmp_table()
784
          && !(is_inner_table(table_id) || is_mysql_tmp_table(table_type))) {
785
        // mysql tmp table name is only duplicated in the same session.
786
      } else {
787
        // Raise error here to skip the following steps,
788
        // ret will be overwrite if if_not_exist_ is true before rpc returns.
789
        ret = OB_ERR_TABLE_EXIST;
790
        res_.table_id_ = table_id;
791
        res_.schema_version_ = schema_version;
792
        LOG_WARN("table exist", KR(ret), K_(tenant_id), K(database_id),
793
                 K(session_id), K(table_name), K(table_id), K(schema_version),
794
                 K(arg_.if_not_exist_));
795
      }
796
    }
797
  }
798
  return ret;
799
}
800

801
int ObCreateTableHelper::set_tablegroup_id_()
802
{
803
  int ret = OB_SUCCESS;
804
  const ObTableSchema &table = arg_.schema_;
805
  const ObString &tablegroup_name = table.get_tablegroup_name();
806
  uint64_t tablegroup_id = OB_INVALID_ID;
807
  if (OB_FAIL(check_inner_stat_())) {
808
    LOG_WARN("fail to check inner stat", KR(ret));
809
  } else if (tablegroup_name.empty()) {
810
    if (table.has_partition()) {
811
      // try use default tablegroup id
812
      const uint64_t database_id = table.get_database_id();
813
      const ObDatabaseSchema *database_schema = NULL;
814
      if (OB_FAIL(latest_schema_guard_.get_database_schema(database_id, database_schema))) {
815
        LOG_WARN("fail to get database schema", KR(ret), K_(tenant_id), K(database_id));
816
      } else if (OB_ISNULL(database_schema)) {
817
        ret = OB_ERR_BAD_DATABASE;
818
        LOG_WARN("database not exist", KR(ret), K_(tenant_id), K(database_id));
819
      } else {
820
        tablegroup_id = database_schema->get_default_tablegroup_id();
821
      }
822

823
      if (OB_SUCC(ret) && OB_INVALID_ID == tablegroup_id) {
824
        const ObTenantSchema *tenant_schema = NULL;
825
        if (OB_FAIL(latest_schema_guard_.get_tenant_schema(tenant_id_, tenant_schema))) {
826
          LOG_WARN("fail to get tenant schema", KR(ret), K_(tenant_id));
827
        } else if (OB_ISNULL(tenant_schema)) {
828
          ret = OB_TENANT_NOT_EXIST;
829
          LOG_WARN("tenant not exist", KR(ret), K_(tenant_id));
830
        } else {
831
          tablegroup_id = tenant_schema->get_default_tablegroup_id();
832
        }
833
      }
834
    }
835
  } else {
836
    if (!table.has_partition()) {
837
      ret = OB_NOT_SUPPORTED;
838
      LOG_WARN("table which has no partitions with tablegroup", KR(ret), K(table));
839
    } else if (OB_FAIL(latest_schema_guard_.get_tablegroup_id(
840
               tablegroup_name, tablegroup_id))) {
841
      LOG_WARN("fail to get tablegroup id", KR(ret), K_(tenant_id), K(tablegroup_name));
842
    } else if (OB_UNLIKELY(OB_INVALID_ID == tablegroup_id)) {
843
      ret = OB_TABLEGROUP_NOT_EXIST;
844
      LOG_WARN("tabelgroup not exist ", KR(ret), K_(tenant_id), K(tablegroup_name));
845
    } else {}
846
  }
847

848
  if (OB_SUCC(ret) && OB_INVALID_ID != tablegroup_id) {
849
    // TODO:(yanmu.ztl) after 4.2, we can use ObSimpleTableSchema instead of ObTablegroupSchema
850
    const ObTablegroupSchema *tablegroup_schema = NULL;
851
    if (OB_FAIL(latest_schema_guard_.get_tablegroup_schema(tablegroup_id, tablegroup_schema))) {
852
      LOG_WARN("fail to get tablegroup schema", KR(ret), K_(tenant_id), K(tablegroup_id));
853
    } else if (OB_ISNULL(tablegroup_schema)) {
854
      ret = OB_TABLEGROUP_NOT_EXIST;
855
      LOG_WARN("tabelgroup not exist ", KR(ret), K_(tenant_id), K(tablegroup_id));
856
    } else if (OB_UNLIKELY(ObDuplicateScope::DUPLICATE_SCOPE_NONE != table.get_duplicate_scope()
857
               && OB_INVALID_ID != tablegroup_id)) {
858
      ret = OB_NOT_SUPPORTED;
859
      LOG_WARN("duplicated table in tablegroup is not supported", K(ret),
860
               "table_id", table.get_table_id(),
861
               "tablegroup_id", table.get_tablegroup_id());
862
      LOG_USER_ERROR(OB_NOT_SUPPORTED, "duplicated table in tablegroup");
863
    }
864
  }
865

866
  if (OB_SUCC(ret)) {
867
    (void) const_cast<ObTableSchema&>(table).set_tablegroup_id(tablegroup_id);
868
  }
869
  return ret;
870
}
871

872
// parent table id won't change after parent table name is locked.
873
int ObCreateTableHelper::check_and_set_parent_table_id_()
874
{
875
  int ret = OB_SUCCESS;
876
  const ObTableSchema &table = arg_.schema_;
877
  const ObString &database_name = arg_.db_name_;
878
  const ObString &table_name = table.get_table_name();
879
  const uint64_t session_id = table.get_session_id();
880
  if (OB_FAIL(check_inner_stat_())) {
881
    LOG_WARN("fail to check inner stat", KR(ret));
882
  } else {
883
    for (int64_t i = 0; OB_SUCC(ret) && i < arg_.foreign_key_arg_list_.count(); i++) {
884
      const obrpc::ObCreateForeignKeyArg &foreign_key_arg = arg_.foreign_key_arg_list_.at(i);
885
      const ObString &parent_database_name = foreign_key_arg.parent_database_;
886
      const ObString &parent_table_name = foreign_key_arg.parent_table_;
887
      if (0 == parent_database_name.case_compare(database_name)
888
          && 0 == parent_table_name.case_compare(table_name)) {
889
        // self reference
890
      } else {
891
        uint64_t parent_database_id = OB_INVALID_ID;
892
        const ObDatabaseSchema *parent_database = NULL;
893
        if (0 == parent_database_name.case_compare(database_name)) {
894
          parent_database_id = table.get_database_id();
895
        } else if (OB_FAIL(latest_schema_guard_.get_database_id(parent_database_name, parent_database_id))) {
896
          LOG_WARN("fail to get database id", KR(ret), K_(tenant_id), K(parent_database_name));
897
        } else if (OB_UNLIKELY(OB_INVALID_ID == parent_database_id)) {
898
          ret = OB_ERR_BAD_DATABASE;
899
          LOG_WARN("parent database not exist", KR(ret), K_(tenant_id), K(parent_database_name));
900
          LOG_USER_ERROR(OB_ERR_BAD_DATABASE, parent_database_name.length(), parent_database_name.ptr());
901
        } else if (OB_FAIL(latest_schema_guard_.get_database_schema(parent_database_id, parent_database))) {
902
          LOG_WARN("fail to get database schema", KR(ret), K_(tenant_id), K(parent_database_id));
903
        } else if (OB_ISNULL(parent_database)) {
904
          ret = OB_ERR_UNEXPECTED;
905
          LOG_WARN("database schema is null", KR(ret), K_(tenant_id), K(parent_database_id));
906
        } else if (parent_database->is_in_recyclebin()) {
907
          ret = OB_ERR_OPERATION_ON_RECYCLE_OBJECT;
908
          LOG_WARN("Can't not do ddl on db in recyclebin", KR(ret), K_(tenant_id), K(parent_database_id));
909
        }
910

911
        uint64_t parent_table_id = OB_INVALID_ID;
912
        ObTableType parent_table_type = ObTableType::MAX_TABLE_TYPE;
913
        int64_t parent_table_schema_version = OB_INVALID_VERSION; // not used
914
        if (FAILEDx(latest_schema_guard_.get_table_id(
915
            parent_database_id, session_id, parent_table_name,
916
            parent_table_id, parent_table_type, parent_table_schema_version))) {
917
          LOG_WARN("fail to get parent table id", KR(ret), K_(tenant_id),
918
                   K(session_id), K(parent_database_id), K(parent_table_name));
919
        } else if (OB_UNLIKELY(OB_INVALID_ID == parent_table_id)) {
920
          if (!foreign_key_arg.is_parent_table_mock_) {
921
            ret = OB_TABLE_NOT_EXIST;
922
            LOG_WARN("parent table not exist", KR(ret), K_(tenant_id),
923
                     K(session_id), K(parent_database_id), K(parent_table_name));
924
            LOG_USER_ERROR(OB_TABLE_NOT_EXIST,
925
                           to_cstring(parent_database_name),
926
                           to_cstring(parent_table_name));
927
          } else {
928
            //TODO(yanmu.ztl): this interface has poor performance.
929
            if (OB_FAIL(latest_schema_guard_.get_mock_fk_parent_table_id(
930
                parent_database_id, parent_table_name, parent_table_id))) {
931
              LOG_WARN("fail to get mock fk parent table id", KR(ret),
932
                       K_(tenant_id), K(parent_database_id), K(parent_table_name));
933
            } else if (OB_UNLIKELY(OB_INVALID_ID == parent_table_id)) {
934
              LOG_INFO("mock fk parent table not exist", KR(ret), K_(tenant_id),
935
                       K(parent_database_id), K(parent_table_name));
936
            }
937
          }
938
        } else {
939
          if (foreign_key_arg.is_parent_table_mock_) {
940
            ret = OB_ERR_PARALLEL_DDL_CONFLICT;
941
            LOG_WARN("parenet table already exist, should retry",
942
                     KR(ret), K_(tenant_id), K(parent_table_id), K(foreign_key_arg));
943
          }
944
        }
945
        // parent_table_id will be OB_INVALID_ID in the following cases:
946
        // 1. foreign key is self reference.
947
        // 2. mock fk parent table doesn't exist.
948
        if (OB_SUCC(ret)) {
949
          const_cast<obrpc::ObCreateForeignKeyArg&>(foreign_key_arg).parent_database_id_ = parent_database_id;
950
          const_cast<obrpc::ObCreateForeignKeyArg&>(foreign_key_arg).parent_table_id_ = parent_table_id;
951
        }
952
      }
953
    } // end for
954
  }
955
  return ret;
956
}
957

958
int ObCreateTableHelper::generate_schemas_()
959
{
960
  int ret = OB_SUCCESS;
961
  if (OB_FAIL(check_inner_stat_())) {
962
    LOG_WARN("fail to check inner stat", KR(ret));
963
  } else if (OB_FAIL(generate_table_schema_())) {
964
    LOG_WARN("fail to generate table schema", KR(ret));
965
  } else if (OB_FAIL(generate_aux_table_schemas_())) {
966
    LOG_WARN("fail to generate aux table schemas", KR(ret));
967
  } else if (OB_FAIL(gen_partition_object_and_tablet_ids_(new_tables_))) {
968
    LOG_WARN("fail to gen partition object/tablet ids", KR(ret));
969
  } else if (OB_FAIL(generate_foreign_keys_())) {
970
    LOG_WARN("fail to generate foreign keys", KR(ret));
971
  } else if (OB_FAIL(generate_sequence_object_())) {
972
    LOG_WARN("fail to generate sequence object", KR(ret));
973
  } else if (OB_FAIL(generate_audit_schema_())) {
974
    LOG_WARN("fail to generate audit schema", KR(ret));
975
  }
976
  RS_TRACE(generate_schemas);
977
  return ret;
978
}
979

980
int ObCreateTableHelper::generate_table_schema_()
981
{
982
  int ret = OB_SUCCESS;
983
  HEAP_VAR(ObTableSchema, new_table) {
984

985
  // to make try_format_partition_schema() passed
986
  const uint64_t mock_table_id = OB_MIN_USER_OBJECT_ID + 1;
987
  bool is_oracle_mode = false;
988
  if (OB_FAIL(check_inner_stat_())) {
989
    LOG_WARN("fail to check inner stat", KR(ret));
990
  } else if (OB_UNLIKELY(OB_INVALID_ID != arg_.schema_.get_table_id())) {
991
    ret = OB_NOT_SUPPORTED;
992
    LOG_WARN("create table with table_id in 4.x is not supported",
993
             KR(ret), K_(tenant_id), "table_id", arg_.schema_.get_table_id());
994
    LOG_USER_ERROR(OB_NOT_SUPPORTED, "create table with id is");
995
  } else if (OB_FAIL(new_table.assign(arg_.schema_))) {
996
    LOG_WARN("fail to assign table schema", KR(ret), K_(tenant_id));
997
  } else if (FALSE_IT(new_table.set_table_id(mock_table_id))) {
998
  } else if (OB_FAIL(ddl_service_->try_format_partition_schema(new_table))) {
999
    LOG_WARN("fail to format partition schema", KR(ret), K_(tenant_id));
1000
  } else if (OB_FAIL(new_table.check_if_oracle_compat_mode(is_oracle_mode))) {
1001
    LOG_WARN("failed to get compat mode", KR(ret), K_(tenant_id));
1002
  }
1003

1004
  const uint64_t tablespace_id = new_table.get_tablespace_id();
1005
  if (OB_SUCC(ret) && OB_INVALID_ID != tablespace_id) {
1006
    const ObTablespaceSchema *tablespace = NULL;
1007
    if (OB_FAIL(latest_schema_guard_.get_tablespace_schema(
1008
        tablespace_id, tablespace))) {
1009
      LOG_WARN("fail to get tablespace schema", KR(ret), K_(tenant_id), K(tablespace_id));
1010
    } else if (OB_ISNULL(tablespace)) {
1011
      ret = OB_ERR_PARALLEL_DDL_CONFLICT;
1012
      LOG_WARN("tablespace not exist, need retry", KR(ret), K_(tenant_id), K(tablespace_id));
1013
    } else if (OB_FAIL(new_table.set_encrypt_key(tablespace->get_encrypt_key()))) {
1014
      LOG_WARN("fail to set encrypt key", KR(ret), K_(tenant_id), KPC(tablespace));
1015
    } else {
1016
      new_table.set_master_key_id(tablespace->get_master_key_id());
1017
    }
1018
  }
1019

1020
  const uint64_t tablegroup_id = new_table.get_tablegroup_id();
1021
  if (OB_SUCC(ret) && OB_INVALID_ID != tablegroup_id) {
1022
    //TODO:(yanmu.ztl) local schema maybe too old for concurrent create table
1023
    // to check partition options with the primary table in tablegroup.
1024
    ObSchemaGetterGuard guard;
1025
    if (OB_FAIL(schema_service_->get_tenant_schema_guard(tenant_id_, guard))) {
1026
      LOG_WARN("fail to get tenant schema guard", KR(ret), K_(tenant_id));
1027
    } else if (OB_FAIL(ddl_service_->try_check_and_set_table_schema_in_tablegroup(guard, new_table))) {
1028
      LOG_WARN("fail to check table in tablegorup", KR(ret), K(new_table));
1029
    }
1030
  }
1031

1032
  if (OB_SUCC(ret)) {
1033
    ObTableSchema::const_column_iterator begin = new_table.column_begin();
1034
    ObTableSchema::const_column_iterator end = new_table.column_end();
1035
    ObSchemaGetterGuard guard;
1036
    if (OB_FAIL(schema_service_->get_tenant_schema_guard(OB_SYS_TENANT_ID, guard))) {
1037
      LOG_WARN("fail to get schema guard", KR(ret));
1038
    }
1039
    for (; OB_SUCC(ret) && begin != end; begin++) {
1040
      ObColumnSchemaV2 *col = (*begin);
1041
      if (OB_ISNULL(col)) {
1042
        ret = OB_ERR_UNEXPECTED;
1043
        LOG_WARN("get column schema failed", KR(ret));
1044
      } else if (col->is_extend()) {
1045
        const uint64_t udt_id = col->get_sub_data_type();
1046
        const ObUDTTypeInfo *udt_info = NULL;
1047
        if (is_inner_object_id(udt_id) && !is_sys_tenant(tenant_id_)) {
1048
          // can't add object lock across tenant, assumed that sys inner udt won't be changed.
1049
          if (OB_FAIL(guard.get_udt_info(OB_SYS_TENANT_ID, udt_id, udt_info))) {
1050
            LOG_WARN("fail to get udt info", KR(ret), K(udt_id));
1051
          } else if (OB_ISNULL(udt_info)) {
1052
            ret = OB_ERR_UNEXPECTED;
1053
            LOG_WARN("inner udt not found", KR(ret), K(udt_id));
1054
          }
1055
        } else if (OB_FAIL(latest_schema_guard_.get_udt_info(udt_id, udt_info))) {
1056
          LOG_WARN("fail to get udt info", KR(ret), K_(tenant_id), K(udt_id));
1057
        } else if (OB_ISNULL(udt_info)) {
1058
          ret = OB_ERR_UNEXPECTED;
1059
          LOG_WARN("udt doesn't exist", KR(ret), K_(tenant_id), K(udt_id));
1060
        }
1061
      }
1062
    } // end for
1063
  }
1064

1065
  // check if constraint name duplicated
1066
  const uint64_t database_id = new_table.get_database_id();
1067
  bool cst_exist = false;
1068
  const ObIArray<ObConstraint> &constraints = arg_.constraint_list_;
1069
  const int64_t cst_cnt = constraints.count();
1070
  for (int64_t i = 0; OB_SUCC(ret) && i < cst_cnt; i++) {
1071
    const ObConstraint &cst = constraints.at(i);
1072
    const ObString &cst_name = cst.get_constraint_name_str();
1073
    if (OB_UNLIKELY(cst.get_constraint_name_str().empty())) {
1074
      ret = OB_ERR_UNEXPECTED;
1075
      LOG_WARN("cst name is empty", KR(ret), K_(tenant_id), K(database_id), K(cst_name));
1076
    } else if (OB_FAIL(check_constraint_name_exist_(new_table, cst_name, false /*is_foreign_key*/, cst_exist))) {
1077
      LOG_WARN("fail to check constraint name exist", KR(ret), K_(tenant_id), K(database_id), K(cst_name));
1078
    } else if (cst_exist) {
1079
      ret = OB_ERR_CONSTRAINT_NAME_DUPLICATE;
1080
      if (!is_oracle_mode) {
1081
        LOG_USER_ERROR(OB_ERR_CONSTRAINT_NAME_DUPLICATE, cst_name.length(), cst_name.ptr());
1082
      }
1083
      LOG_WARN("cst name is duplicate", KR(ret), K_(tenant_id), K(database_id), K(cst_name));
1084
    }
1085
  } // end for
1086

1087
  // fetch object_ids (data table + constraints)
1088
  ObIDGenerator id_generator;
1089
  const uint64_t object_cnt = cst_cnt + 1;
1090
  uint64_t object_id = OB_INVALID_ID;
1091
  if (FAILEDx(gen_object_ids_(object_cnt, id_generator))) {
1092
    LOG_WARN("fail to gen object ids", KR(ret), K_(tenant_id), K(object_cnt));
1093
  } else if (OB_FAIL(id_generator.next(object_id))) {
1094
    LOG_WARN("fail to get next object_id", KR(ret));
1095
  } else {
1096
    (void) new_table.set_table_id(object_id);
1097
  }
1098

1099
  // generate constraints
1100
  for (int64_t i = 0; OB_SUCC(ret) && i < cst_cnt; i++) {
1101
    ObConstraint &cst = const_cast<ObConstraint &>(constraints.at(i));
1102
    cst.set_tenant_id(tenant_id_);
1103
    cst.set_table_id(new_table.get_table_id());
1104
    if (OB_FAIL(id_generator.next(object_id))) {
1105
      LOG_WARN("fail to get next object_id", KR(ret));
1106
    } else if (FALSE_IT(cst.set_constraint_id(object_id))) {
1107
    } else if (OB_FAIL(new_table.add_constraint(cst))) {
1108
      LOG_WARN("fail to add constraint", KR(ret), K(cst));
1109
    }
1110
  } // end for
1111

1112
  // fill table schema for interval part
1113
  if (OB_SUCC(ret)
1114
      && new_table.has_partition()
1115
      && new_table.is_interval_part()) {
1116
    int64_t part_num = new_table.get_part_option().get_part_num();
1117
    ObPartition **part_array = new_table.get_part_array();
1118
    const ObRowkey *transition_point = NULL;
1119
    if (OB_UNLIKELY(PARTITION_LEVEL_TWO == new_table.get_part_level()
1120
        && !new_table.has_sub_part_template_def())) {
1121
      ret = OB_NOT_SUPPORTED;
1122
      LOG_WARN("interval part of composited-partitioned table not support", KR(ret), K(new_table));
1123
      LOG_USER_ERROR(OB_NOT_SUPPORTED, "interval part of composited-partitioned table without template");
1124
    } else if (OB_UNLIKELY(1 != new_table.get_partition_key_column_num())) {
1125
      ret = OB_OP_NOT_ALLOW;
1126
      LOG_WARN("more than one partition key not support", KR(ret), K(new_table));
1127
      LOG_USER_ERROR(OB_OP_NOT_ALLOW, "more than one partition key");
1128
    } else if (OB_ISNULL(part_array)
1129
               || OB_UNLIKELY(0 == part_num)) {
1130
      ret = OB_ERR_UNEXPECTED;
1131
      LOG_WARN("range part array is null or part_num is 0", KR(ret), K(new_table));
1132
    } else if (OB_ISNULL(transition_point = &part_array[part_num - 1]->get_high_bound_val())) {
1133
      ret = OB_ERR_UNEXPECTED;
1134
      LOG_WARN("transition_point is null", KR(ret), KP(transition_point));
1135
    } else if (OB_FAIL(ObPartitionUtils::check_interval_partition_table(
1136
                       *transition_point, new_table.get_interval_range()))) {
1137
      LOG_WARN("fail to check_interval_partition_table", KR(ret), K(new_table));
1138
    } else if (OB_FAIL(new_table.set_transition_point(*transition_point))) {
1139
      LOG_WARN("fail to set transition point", KR(ret), K(new_table));
1140
    }
1141
  }
1142

1143
  if (FAILEDx(new_tables_.push_back(new_table))) {
1144
    LOG_WARN("fail to push back table", KR(ret));
1145
  }
1146

1147
  } // end HEAP_VAR
1148
  return ret;
1149
}
1150

1151
int ObCreateTableHelper::generate_aux_table_schemas_()
1152
{
1153
  int ret = OB_SUCCESS;
1154
  HEAP_VAR(ObTableSchema, index_schema) {
1155

1156
  if (OB_FAIL(check_inner_stat_())) {
1157
    LOG_WARN("fail to check inner stat", KR(ret));
1158
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
1159
    ret = OB_ERR_UNEXPECTED;
1160
    LOG_WARN("table cnt not match", KR(ret), "table_cnt", new_tables_.count());
1161
  } else {
1162
    ObTableSchema *data_table = &(new_tables_.at(0));
1163

1164
    // 0. fetch object_ids
1165
    ObIDGenerator id_generator;
1166
    int64_t object_cnt = arg_.index_arg_list_.size();
1167
    bool has_lob_table = false;
1168
    uint64_t object_id = OB_INVALID_ID;
1169
    if (!data_table->is_external_table()) {
1170
      for (int64_t i = 0; OB_SUCC(ret) && !has_lob_table && i < data_table->get_column_count(); i++) {
1171
        const ObColumnSchemaV2 *column = data_table->get_column_schema_by_idx(i);
1172
        if (OB_ISNULL(column)) {
1173
          ret = OB_ERR_UNEXPECTED;
1174
          LOG_WARN("column is null", KR(ret), K(i), KPC(data_table));
1175
        } else if (is_lob_storage(column->get_data_type())) {
1176
          has_lob_table = true;
1177
          object_cnt += 2;
1178
        }
1179
      } // end for
1180
    }
1181
    if (FAILEDx(gen_object_ids_(object_cnt, id_generator))) {
1182
      LOG_WARN("fail to gen object ids", KR(ret), K_(tenant_id), K(object_cnt));
1183
    }
1184

1185
    // 1. build index table
1186
    ObIndexBuilder index_builder(*ddl_service_);
1187
    for (int64_t i = 0; OB_SUCC(ret) && i < arg_.index_arg_list_.size(); ++i) {
1188
      index_schema.reset();
1189
      obrpc::ObCreateIndexArg &index_arg = const_cast<obrpc::ObCreateIndexArg&>(arg_.index_arg_list_.at(i));
1190
      if (!index_arg.index_schema_.is_partitioned_table()
1191
          && !data_table->is_partitioned_table()) {
1192
        if (INDEX_TYPE_NORMAL_GLOBAL == index_arg.index_type_) {
1193
          index_arg.index_type_ = INDEX_TYPE_NORMAL_GLOBAL_LOCAL_STORAGE;
1194
        } else if (INDEX_TYPE_UNIQUE_GLOBAL == index_arg.index_type_) {
1195
          index_arg.index_type_ = INDEX_TYPE_UNIQUE_GLOBAL_LOCAL_STORAGE;
1196
        } else if (INDEX_TYPE_SPATIAL_GLOBAL == index_arg.index_type_) {
1197
          index_arg.index_type_ = INDEX_TYPE_SPATIAL_GLOBAL_LOCAL_STORAGE;
1198
        }
1199
      }
1200
      // the global index has generated column schema during resolve, RS no need to generate index schema,
1201
      // just assign column schema
1202
      if (INDEX_TYPE_NORMAL_GLOBAL == index_arg.index_type_
1203
          || INDEX_TYPE_UNIQUE_GLOBAL == index_arg.index_type_
1204
          || INDEX_TYPE_SPATIAL_GLOBAL == index_arg.index_type_) {
1205
        if (OB_FAIL(index_schema.assign(index_arg.index_schema_))) {
1206
          LOG_WARN("fail to assign schema", KR(ret), K(index_arg));
1207
        }
1208
      }
1209
      // TODO(yanmu.ztl): should check if index name is duplicated in oracle mode.
1210
      const bool global_index_without_column_info = false;
1211
      ObSEArray<ObColumnSchemaV2 *, 1> gen_columns;
1212
      ObIAllocator *allocator = index_arg.index_schema_.get_allocator();
1213
      bool index_exist = false;
1214
      if (OB_FAIL(ret)) {
1215
      } else if (OB_ISNULL(allocator)) {
1216
        ret = OB_ERR_UNEXPECTED;
1217
        LOG_WARN("invalid allocator", K(ret));
1218
      } else if (OB_FAIL(ObIndexBuilderUtil::adjust_expr_index_args(
1219
                 index_arg, *data_table, *allocator, gen_columns))) {
1220
          LOG_WARN("fail to adjust expr index args", KR(ret), K(index_arg), KPC(data_table));
1221
      } else if (OB_FAIL(index_builder.generate_schema(index_arg,
1222
                                                       *data_table,
1223
                                                       global_index_without_column_info,
1224
                                                       false, /*generate_id*/
1225
                                                       index_schema))) {
1226
        LOG_WARN("generate_schema for index failed", KR(ret), K(index_arg), KPC(data_table));
1227
      } else if (OB_FAIL(id_generator.next(object_id))) {
1228
        LOG_WARN("fail to get next object_id", KR(ret));
1229
      } else if (FALSE_IT(index_schema.set_table_id(object_id))) {
1230
      } else if (OB_FAIL(index_schema.generate_origin_index_name())) {
1231
        // For the later operation ObIndexNameChecker::add_index_name()
1232
        LOG_WARN("fail to generate origin index name", KR(ret), K(index_schema));
1233
      } else if (OB_FAIL(ddl_service_->get_index_name_checker().check_index_name_exist(
1234
                 index_schema.get_tenant_id(),
1235
                 index_schema.get_database_id(),
1236
                 index_schema.get_table_name_str(),
1237
                 index_exist))) {
1238
      } else if (index_exist) {
1239
        // actually, only index name in oracle tenant will be checked.
1240
        ret = OB_ERR_KEY_NAME_DUPLICATE;
1241
        LOG_WARN("duplicate index name", KR(ret), K_(tenant_id),
1242
                 "database_id", index_schema.get_database_id(),
1243
                 "index_name", index_schema.get_origin_index_name_str());
1244
        LOG_USER_ERROR(OB_ERR_KEY_NAME_DUPLICATE,
1245
                       index_schema.get_origin_index_name_str().length(),
1246
                       index_schema.get_origin_index_name_str().ptr());
1247
      } else if (OB_FAIL(new_tables_.push_back(index_schema))) {
1248
        LOG_WARN("push_back failed", KR(ret));
1249
      } else {
1250
        data_table = &(new_tables_.at(0)); // memory of data table may change after add table to new_tables_
1251
      }
1252
    } // end for
1253

1254
    // 2. build lob table
1255
    if (OB_SUCC(ret) && has_lob_table) {
1256
      HEAP_VARS_2((ObTableSchema, lob_meta_schema), (ObTableSchema, lob_piece_schema)) {
1257
      ObLobMetaBuilder lob_meta_builder(*ddl_service_);
1258
      ObLobPieceBuilder lob_piece_builder(*ddl_service_);
1259
      bool need_object_id = false;
1260
      if (OB_FAIL(id_generator.next(object_id))) {
1261
        LOG_WARN("fail to get next object_id", KR(ret));
1262
      } else if (OB_FAIL(lob_meta_builder.generate_aux_lob_meta_schema(
1263
        schema_service_->get_schema_service(), *data_table, object_id, lob_meta_schema, need_object_id))) {
1264
        LOG_WARN("generate lob meta table failed", KR(ret), KPC(data_table));
1265
      } else if (OB_FAIL(id_generator.next(object_id))) {
1266
        LOG_WARN("fail to get next object_id", KR(ret));
1267
      } else if (OB_FAIL(lob_piece_builder.generate_aux_lob_piece_schema(
1268
        schema_service_->get_schema_service(), *data_table, object_id, lob_piece_schema, need_object_id))) {
1269
        LOG_WARN("generate_schema for lob data table failed", KR(ret), KPC(data_table));
1270
      } else if (OB_FAIL(new_tables_.push_back(lob_meta_schema))) {
1271
        LOG_WARN("push_back lob meta table failed", KR(ret));
1272
      } else if (OB_FAIL(new_tables_.push_back(lob_piece_schema))) {
1273
        LOG_WARN("push_back lob piece table failed", KR(ret));
1274
      } else {
1275
        data_table = &(new_tables_.at(0)); // memory of data table may change after add table to new_tables_
1276
        data_table->set_aux_lob_meta_tid(lob_meta_schema.get_table_id());
1277
        data_table->set_aux_lob_piece_tid(lob_piece_schema.get_table_id());
1278
      }
1279

1280
      } // end HEAP_VARS_2
1281
    }
1282
  }
1283

1284
  } // end HEAP_VAR
1285
  return ret;
1286
}
1287

1288
int ObCreateTableHelper::generate_foreign_keys_()
1289
{
1290
  int ret = OB_SUCCESS;
1291
  if (OB_FAIL(check_inner_stat_())) {
1292
    LOG_WARN("fail to check inner stat", KR(ret));
1293
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
1294
    ret = OB_ERR_UNEXPECTED;
1295
    LOG_WARN("invalid table cnt", KR(ret), "table_cnt", new_tables_.count());
1296
  } else {
1297
    bool is_oracle_mode = false;
1298
    ObTableSchema &data_table = new_tables_.at(0);
1299
    const uint64_t session_id = data_table.get_session_id();
1300
    ObIDGenerator id_generator; // for foreign key
1301
    const uint64_t object_cnt = arg_.foreign_key_arg_list_.count();
1302
    if (OB_FAIL(gen_object_ids_(object_cnt, id_generator))) {
1303
      LOG_WARN("fail to gen object ids", KR(ret), K_(tenant_id), K(object_cnt));
1304
    } else if (OB_FAIL(data_table.check_if_oracle_compat_mode(is_oracle_mode))) {
1305
      LOG_WARN("fail to check oracle mode", KR(ret));
1306
    }
1307
    // genernate foreign keys
1308
    for (int64_t i = 0; OB_SUCC(ret) && i < arg_.foreign_key_arg_list_.count(); i++) {
1309
      const obrpc::ObCreateForeignKeyArg &foreign_key_arg = arg_.foreign_key_arg_list_.at(i);
1310
      const ObString &foreign_key_name = foreign_key_arg.foreign_key_name_;
1311
      bool fk_exist = false;
1312
      ObForeignKeyInfo foreign_key_info;
1313
      // check if foreign key name is duplicated
1314
      if (OB_UNLIKELY(foreign_key_name.empty())) {
1315
        ret = OB_ERR_UNEXPECTED;
1316
        LOG_WARN("fk name is empty", KR(ret), K(foreign_key_arg));
1317
      } else if (OB_FAIL(check_constraint_name_exist_(data_table, foreign_key_name, true /*is_foreign_key*/, fk_exist))) {
1318
        LOG_WARN("fail to check foreign key name exist", KR(ret), K_(tenant_id), K(foreign_key_name));
1319
      } else if (fk_exist) {
1320
        if (is_oracle_mode) {
1321
          ret = OB_ERR_CONSTRAINT_NAME_DUPLICATE;
1322
          LOG_WARN("fk name is duplicate", KR(ret), K(foreign_key_name));
1323
        } else { // mysql mode
1324
          ret = OB_ERR_DUP_KEY;
1325
          LOG_USER_ERROR(OB_ERR_DUP_KEY,
1326
                         data_table.get_table_name_str().length(),
1327
                         data_table.get_table_name_str().ptr());
1328
        }
1329
      } else {
1330
        const ObTableSchema *parent_table = NULL;
1331
        ObMockFKParentTableSchema *new_mock_fk_parent_table = NULL;
1332
        const ObString &parent_database_name = foreign_key_arg.parent_database_;
1333
        const ObString &parent_table_name = foreign_key_arg.parent_table_;
1334
        const bool self_reference = (0 == parent_table_name.case_compare(data_table.get_table_name_str())
1335
                                     && 0 == parent_database_name.case_compare(arg_.db_name_));
1336
        // 1. fill ref_cst_type_/ref_cst_id_
1337
        if (self_reference) {
1338
          // TODO: is it necessory to determine whether it is case sensitive by check sys variable
1339
          // check whether it belongs to self reference, if so, the parent schema is child schema.
1340
          parent_table = &data_table;
1341
          if (CONSTRAINT_TYPE_PRIMARY_KEY == foreign_key_arg.ref_cst_type_) {
1342
            if (is_oracle_mode) {
1343
              ObTableSchema::const_constraint_iterator iter = parent_table->constraint_begin();
1344
              for ( ; iter != parent_table->constraint_end(); ++iter) {
1345
                if (CONSTRAINT_TYPE_PRIMARY_KEY == (*iter)->get_constraint_type()) {
1346
                  foreign_key_info.ref_cst_type_ = CONSTRAINT_TYPE_PRIMARY_KEY;
1347
                  foreign_key_info.ref_cst_id_ = (*iter)->get_constraint_id();
1348
                  break;
1349
                }
1350
              } // end for
1351
            } else {
1352
              foreign_key_info.ref_cst_type_ = CONSTRAINT_TYPE_PRIMARY_KEY;
1353
              foreign_key_info.ref_cst_id_ = common::OB_INVALID_ID;
1354
            }
1355
          } else if (CONSTRAINT_TYPE_UNIQUE_KEY == foreign_key_arg.ref_cst_type_) {
1356
            if (OB_FAIL(ddl_service_->get_uk_cst_id_for_self_ref(new_tables_, foreign_key_arg, foreign_key_info))) {
1357
              LOG_WARN("failed to get uk cst id for self ref", KR(ret), K(foreign_key_arg));
1358
            }
1359
          } else {
1360
            ret = OB_ERR_UNEXPECTED;
1361
            LOG_WARN("invalid foreign key ref cst type", KR(ret), K(foreign_key_arg));
1362
          }
1363
        } else {
1364
          foreign_key_info.ref_cst_type_ = foreign_key_arg.ref_cst_type_;
1365
          foreign_key_info.ref_cst_id_ = foreign_key_arg.ref_cst_id_;
1366
          if (foreign_key_arg.is_parent_table_mock_) {
1367
            // skip
1368
          } else {
1369
            const uint64_t parent_table_id = foreign_key_arg.parent_table_id_;
1370
            if (OB_FAIL(latest_schema_guard_.get_table_schema(parent_table_id, parent_table))) {
1371
              LOG_WARN("fail to get table schema", KR(ret), K_(tenant_id), K(parent_table_id));
1372
            } else if (OB_ISNULL(parent_table)) {
1373
              ret = OB_TABLE_NOT_EXIST;
1374
              LOG_WARN("parent table is not exist", KR(ret), K(foreign_key_arg));
1375
            }
1376
          }
1377
        }
1378

1379
        // 2. fill parent columns
1380
        if (OB_FAIL(ret)) {
1381
        } else if (foreign_key_arg.is_parent_table_mock_) {
1382
          if (OB_FAIL(get_mock_fk_parent_table_info_(foreign_key_arg, foreign_key_info, new_mock_fk_parent_table))) {
1383
            LOG_WARN("fail to get mock fk parent table info", KR(ret), K(foreign_key_arg));
1384
          }
1385
        } else {
1386
          if (false == parent_table->is_tmp_table()
1387
              && 0 != parent_table->get_session_id()
1388
              && OB_INVALID_ID != arg_.schema_.get_session_id()) {
1389
            ret = OB_TABLE_NOT_EXIST;
1390
            LOG_USER_ERROR(OB_TABLE_NOT_EXIST, to_cstring(parent_database_name), to_cstring(parent_table_name));
1391
          } else if (!arg_.is_inner_ && parent_table->is_in_recyclebin()) {
1392
            ret = OB_ERR_OPERATION_ON_RECYCLE_OBJECT;
1393
            LOG_WARN("parent table is in recyclebin", KR(ret), K(foreign_key_arg));
1394
          }
1395
          for (int64_t j = 0; OB_SUCC(ret) && j < foreign_key_arg.parent_columns_.count(); j++) {
1396
            const ObString &column_name = foreign_key_arg.parent_columns_.at(j);
1397
            const ObColumnSchemaV2 *column_schema = parent_table ->get_column_schema(column_name);
1398
            if (OB_ISNULL(column_schema)) {
1399
              ret = OB_ERR_COLUMN_NOT_FOUND;
1400
              LOG_WARN("parent column is not exist", KR(ret), K(column_name));
1401
            } else if (OB_FAIL(foreign_key_info.parent_column_ids_.push_back(column_schema->get_column_id()))) {
1402
              LOG_WARN("failed to push parent column id", KR(ret), K(column_name));
1403
            }
1404
          } // end for
1405
        }
1406

1407
        // 3. fill child columns
1408
        if (OB_SUCC(ret)) {
1409
          foreign_key_info.child_table_id_ = data_table.get_table_id();
1410
          foreign_key_info.parent_table_id_ = foreign_key_arg.is_parent_table_mock_ ?
1411
                                              new_mock_fk_parent_table->get_mock_fk_parent_table_id() :
1412
                                              parent_table->get_table_id();
1413
          for (int64_t j = 0; OB_SUCC(ret) && j < foreign_key_arg.child_columns_.count(); j++) {
1414
            const ObString &column_name = foreign_key_arg.child_columns_.at(j);
1415
            const ObColumnSchemaV2 *column_schema = data_table.get_column_schema(column_name);
1416
            if (OB_ISNULL(column_schema)) {
1417
              ret = OB_ERR_COLUMN_NOT_FOUND;
1418
              LOG_WARN("child column is not exist", KR(ret), K(column_name));
1419
            } else if (OB_FAIL(foreign_key_info.child_column_ids_.push_back(column_schema->get_column_id()))) {
1420
              LOG_WARN("failed to push child column id", KR(ret), K(column_name));
1421
            }
1422
          } // end for
1423
        }
1424

1425
        // 4. get reference option and foreign key name.
1426
        if (OB_SUCC(ret)) {
1427
          foreign_key_info.update_action_ = foreign_key_arg.update_action_;
1428
          foreign_key_info.delete_action_ = foreign_key_arg.delete_action_;
1429
          foreign_key_info.foreign_key_name_ = foreign_key_arg.foreign_key_name_;
1430
          foreign_key_info.enable_flag_ = foreign_key_arg.enable_flag_;
1431
          foreign_key_info.validate_flag_ = foreign_key_arg.validate_flag_;
1432
          foreign_key_info.rely_flag_ = foreign_key_arg.rely_flag_;
1433
          foreign_key_info.is_parent_table_mock_ = foreign_key_arg.is_parent_table_mock_;
1434
          foreign_key_info.name_generated_type_  = foreign_key_arg.name_generated_type_;
1435
        }
1436

1437
        // 5. add foreign key info
1438
        if (OB_SUCC(ret)) {
1439
          uint64_t object_id = OB_INVALID_ID;
1440
          if (OB_FAIL(id_generator.next(object_id))) {
1441
            LOG_WARN("fail to get next object_id", KR(ret));
1442
          } else if (FALSE_IT(foreign_key_info.foreign_key_id_ = object_id)) {
1443
          } else if (OB_FAIL(data_table.add_foreign_key_info(foreign_key_info))) {
1444
            LOG_WARN("fail to add foreign key info", KR(ret), K(foreign_key_info));
1445
          } else if (foreign_key_arg.is_parent_table_mock_) {
1446
            if (OB_FAIL(new_mock_fk_parent_table->add_foreign_key_info(foreign_key_info))) {
1447
              LOG_WARN("fail to add foreign key info", KR(ret), K(foreign_key_info));
1448
            }
1449
          } else {
1450
            // this logic is duplicated because of add_foreign_key() will also update data table's schema_version.
1451
            //if (parent_table->get_table_id() != data_table.get_table_id()) {
1452
            //  // no need to update sync_versin_for_cascade_table while the refrence table is itself
1453
            //  if (OB_FAIL(data_table.add_depend_table_id(parent_table->get_table_id()))) {
1454
            //    LOG_WARN("failed to add depend table id", KR(ret), K(foreign_key_info));
1455
            //  }
1456
            //}
1457
          }
1458
        }
1459
      }
1460
    } // end for
1461

1462
    FOREACH_X(iter, new_mock_fk_parent_table_map_, OB_SUCC(ret)) {
1463
      ObMockFKParentTableSchema *&new_mock_fk_parent_table = iter->second;
1464
      if (OB_ISNULL(new_mock_fk_parent_table)) {
1465
        ret = OB_ERR_UNEXPECTED;
1466
        LOG_WARN("mock fk parent table is null", KR(ret));
1467
      } else if (OB_FAIL(new_mock_fk_parent_tables_.push_back(new_mock_fk_parent_table))) {
1468
        LOG_WARN("fail to push back mock fk parent table", KR(ret), KPC(new_mock_fk_parent_table));
1469
      }
1470
    } // end foreach
1471

1472
    // check & replace existed mock fk parent table with data table
1473
    if (OB_SUCC(ret)) {
1474
      ObMockFKParentTableSchema *new_mock_fk_parent_table = NULL;
1475
      if (OB_FAIL(try_replace_mock_fk_parent_table_(new_mock_fk_parent_table))) {
1476
        LOG_WARN("replace mock fk parent table failed", KR(ret));
1477
      } else if (OB_NOT_NULL(new_mock_fk_parent_table)
1478
                 && OB_FAIL(new_mock_fk_parent_tables_.push_back(new_mock_fk_parent_table))) {
1479
        LOG_WARN("fail to push back mock fk parent table", KR(ret), K(new_mock_fk_parent_table));
1480
      }
1481
    }
1482
  }
1483
  return ret;
1484
}
1485

1486
int ObCreateTableHelper::get_mock_fk_parent_table_info_(
1487
    const obrpc::ObCreateForeignKeyArg &foreign_key_arg,
1488
    ObForeignKeyInfo &foreign_key_info,
1489
    ObMockFKParentTableSchema *&new_mock_fk_parent_table_schema)
1490
{
1491
  int ret = OB_SUCCESS;
1492
  new_mock_fk_parent_table_schema = NULL;
1493
  if (OB_FAIL(check_inner_stat_())) {
1494
    LOG_WARN("fail to check inner stat", KR(ret));
1495
  } else if (OB_UNLIKELY(!foreign_key_arg.is_parent_table_mock_)) {
1496
    ret = OB_ERR_UNEXPECTED;
1497
    LOG_WARN("foreign_key_arg shoud be parent_table_mock", KR(ret), K(foreign_key_arg));
1498
  } else {
1499
    const ObMockFKParentTableSchema *ori_mock_fk_parent_table_schema = NULL;
1500
    if (OB_INVALID_ID != foreign_key_arg.parent_table_id_) {
1501
      // mock fk parent table already exist
1502
      if (OB_FAIL(latest_schema_guard_.get_mock_fk_parent_table_schema(
1503
          foreign_key_arg.parent_table_id_, ori_mock_fk_parent_table_schema))) {
1504
        LOG_WARN("fail to get mock fk parent table", KR(ret), K_(tenant_id), K(foreign_key_arg));
1505
      } else if (OB_ISNULL(ori_mock_fk_parent_table_schema)) {
1506
        ret = OB_ERR_PARALLEL_DDL_CONFLICT;
1507
        LOG_WARN("mock fk parent table may be dropped, ddl need retry", KR(ret), K_(tenant_id), K(foreign_key_arg));
1508
      }
1509
    }
1510

1511
    if (OB_SUCC(ret)) {
1512
      MockFKParentTableNameWrapper name_wrapper(foreign_key_arg.parent_database_,
1513
                                                foreign_key_arg.parent_table_);
1514
      int hash_ret = new_mock_fk_parent_table_map_.get_refactored(name_wrapper, new_mock_fk_parent_table_schema);
1515
      if (OB_SUCCESS == hash_ret) {
1516
        // skip
1517
      } else if (OB_HASH_NOT_EXIST != hash_ret) {
1518
        ret = hash_ret;
1519
        LOG_WARN("fail to get new mock fk parent table from map", KR(ret), K_(tenant_id), K(name_wrapper));
1520
      } else {
1521
        // 1. try init new mock fk parent table schema
1522
        if (OB_FAIL(ObSchemaUtils::alloc_schema(allocator_, new_mock_fk_parent_table_schema))) {
1523
          LOG_WARN("fail to new ObMockFKParentTableSchema", KR(ret), K_(tenant_id));
1524
        } else if (OB_INVALID_ID != foreign_key_arg.parent_table_id_) {
1525
          // mock fk parent table already exist
1526
          if (OB_ISNULL(ori_mock_fk_parent_table_schema)) {
1527
            ret = OB_ERR_UNEXPECTED;
1528
            LOG_WARN("ori mock fk parent table not exist", KR(ret), K(foreign_key_arg.parent_table_id_));
1529
          } else if (OB_FAIL(new_mock_fk_parent_table_schema->assign(*ori_mock_fk_parent_table_schema))) {
1530
            LOG_WARN("fail to assign mock fk parent table schema", KR(ret), KPC(ori_mock_fk_parent_table_schema));
1531
          } else {
1532
            (void) new_mock_fk_parent_table_schema->reset_column_array();
1533
            (void) new_mock_fk_parent_table_schema->set_operation_type(
1534
                   ObMockFKParentTableOperationType::MOCK_FK_PARENT_TABLE_OP_UPDATE_SCHEMA_VERSION);
1535
          }
1536
        } else {
1537
          // mock fk parent table not exist
1538
          ObIDGenerator id_generator;
1539
          const uint64_t object_cnt = 1;
1540
          uint64_t object_id = OB_INVALID_ID;
1541
          if (OB_FAIL(gen_object_ids_(object_cnt, id_generator))) {
1542
            LOG_WARN("fail to gen object ids", KR(ret), K_(tenant_id), K(object_cnt));
1543
          } else if (OB_FAIL(id_generator.next(object_id))) {
1544
            LOG_WARN("fail to get next object_id", KR(ret));
1545
          } else if (OB_FAIL(new_mock_fk_parent_table_schema->set_mock_fk_parent_table_name(foreign_key_arg.parent_table_))) {
1546
            LOG_WARN("fail to set mock fk parent table name", KR(ret), K(foreign_key_arg));
1547
          } else {
1548
            (void) new_mock_fk_parent_table_schema->set_tenant_id(tenant_id_);
1549
            (void) new_mock_fk_parent_table_schema->set_database_id(foreign_key_arg.parent_database_id_);
1550
            (void) new_mock_fk_parent_table_schema->set_mock_fk_parent_table_id(object_id);
1551
            (void) new_mock_fk_parent_table_schema->set_operation_type(
1552
                   ObMockFKParentTableOperationType::MOCK_FK_PARENT_TABLE_OP_CREATE_TABLE_BY_ADD_FK_IN_CHILD_TBALE);
1553
          }
1554
        }
1555

1556
        if (FAILEDx(new_mock_fk_parent_table_map_.set_refactored(name_wrapper, new_mock_fk_parent_table_schema))) {
1557
          LOG_WARN("fail to set mock fk parent table to map", KR(ret), KPC(new_mock_fk_parent_table_schema));
1558
        }
1559
      }
1560
    }
1561

1562
    // 2. try add new column to mock fk parent table
1563
    if (OB_FAIL(ret)) {
1564
    } else if (OB_ISNULL(new_mock_fk_parent_table_schema)) {
1565
      ret = OB_ERR_UNEXPECTED;
1566
      LOG_WARN("mock fk parent table should not be null", KR(ret), K(foreign_key_arg));
1567
    } else {
1568
      bool is_alter_table = ObMockFKParentTableOperationType::MOCK_FK_PARENT_TABLE_OP_CREATE_TABLE_BY_ADD_FK_IN_CHILD_TBALE
1569
                            != new_mock_fk_parent_table_schema->get_operation_type();
1570

1571
      uint64_t max_used_column_id = 0;
1572
      if (new_mock_fk_parent_table_schema->get_column_array().count() > 0) {
1573
        // 1. new mock fk parent table : update schema_version/add column
1574
        // 2. existed mock fk parent table: update schema_version/add column
1575
        max_used_column_id = new_mock_fk_parent_table_schema->get_column_array()
1576
                             .at(new_mock_fk_parent_table_schema->get_column_array().count() - 1).first;
1577
      } else if (!is_alter_table) {
1578
        // 1. new mock fk parent table: create
1579
        max_used_column_id = 0;
1580
      } else {
1581
        // 1. update schema_version of existed mock fk parent table
1582
        // 2. add new column to mock fk parent table at the first time.
1583
        if (OB_ISNULL(ori_mock_fk_parent_table_schema)) {
1584
          ret = OB_ERR_UNEXPECTED;
1585
          LOG_WARN("mock fk parent table doesn't exist", KR(ret), K_(tenant_id), K(foreign_key_arg));
1586
        } else if (OB_UNLIKELY(ori_mock_fk_parent_table_schema->get_column_array().count() <= 0)) {
1587
          ret = OB_ERR_UNEXPECTED;
1588
          LOG_WARN("column cnt unexpected", KR(ret), KPC(ori_mock_fk_parent_table_schema));
1589
        } else {
1590
          max_used_column_id = ori_mock_fk_parent_table_schema->get_column_array()
1591
                               .at(ori_mock_fk_parent_table_schema->get_column_array().count() - 1).first;
1592
        }
1593
      }
1594

1595
      for (int64_t i = 0; OB_SUCC(ret) && i < foreign_key_arg.parent_columns_.count(); ++i) {
1596
        uint64_t column_id = OB_INVALID_ID;
1597
        bool is_column_exist = false;
1598
        const ObString &column_name = foreign_key_arg.parent_columns_.at(i);
1599
        (void) new_mock_fk_parent_table_schema->get_column_id_by_column_name(column_name, column_id, is_column_exist);
1600

1601
        if (!is_column_exist && is_alter_table) {
1602
          if (OB_ISNULL(ori_mock_fk_parent_table_schema)) {
1603
            ret = OB_ERR_UNEXPECTED;
1604
            LOG_WARN("mock fk parent table doesn't exist", KR(ret), K_(tenant_id), K(foreign_key_arg));
1605
          } else {
1606
            (void) ori_mock_fk_parent_table_schema->get_column_id_by_column_name(column_name, column_id, is_column_exist);
1607
          }
1608
        }
1609

1610
        if (OB_FAIL(ret)) {
1611
        } else if (!is_column_exist) {
1612
          if (OB_FAIL(new_mock_fk_parent_table_schema->add_column_info_to_column_array(
1613
                      std::make_pair(++max_used_column_id, column_name)))) {
1614
            LOG_WARN("fail to add_column_info_to_column_array for mock_fk_parent_table_schema",
1615
                     KR(ret), K(max_used_column_id), K(column_name));
1616
          } else if (OB_FAIL(foreign_key_info.parent_column_ids_.push_back(max_used_column_id))) {
1617
            LOG_WARN("failed to push parent column id", KR(ret), K(max_used_column_id));
1618
          }
1619
        } else {
1620
          if (OB_FAIL(foreign_key_info.parent_column_ids_.push_back(column_id))) {
1621
            LOG_WARN("failed to push parent column id", KR(ret), K(column_id));
1622
          }
1623
        }
1624
      } // end for
1625
    }
1626

1627
    if (OB_SUCC(ret)
1628
        && ObMockFKParentTableOperationType::MOCK_FK_PARENT_TABLE_OP_UPDATE_SCHEMA_VERSION
1629
           == new_mock_fk_parent_table_schema->get_operation_type()
1630
        && new_mock_fk_parent_table_schema->get_column_array().count() > 0) {
1631
      (void) new_mock_fk_parent_table_schema->set_operation_type(
1632
             ObMockFKParentTableOperationType::MOCK_FK_PARENT_TABLE_OP_ADD_COLUMN);
1633
    }
1634
  }
1635
  return ret;
1636
}
1637

1638
int ObCreateTableHelper::try_replace_mock_fk_parent_table_(
1639
    ObMockFKParentTableSchema *&new_mock_fk_parent_table)
1640
{
1641
  int ret = OB_SUCCESS;
1642
  new_mock_fk_parent_table = NULL;
1643
  if (OB_FAIL(check_inner_stat_())) {
1644
    LOG_WARN("fail to check inner stat", KR(ret));
1645
  } else if (OB_INVALID_ID == replace_mock_fk_parent_table_id_) {
1646
    // do nothing
1647
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
1648
    ret = OB_ERR_UNEXPECTED;
1649
    LOG_WARN("unexpected table cnt", KR(ret), K(new_tables_.count()));
1650
  } else {
1651
    // check if data table's columns are matched with existed mock fk parent table
1652
    const ObTableSchema &data_table = new_tables_.at(0);
1653

1654
    ObArray<const share::schema::ObTableSchema*> index_schemas;
1655
    for (int64_t i = 1; OB_SUCC(ret) && i < new_tables_.count(); ++i) {
1656
      if (new_tables_.at(i).is_unique_index()
1657
          && OB_FAIL(index_schemas.push_back(&new_tables_.at(i)))) {
1658
        LOG_WARN("failed to push back to index_schemas", KR(ret));
1659
      }
1660
    } // end for
1661

1662
    const ObMockFKParentTableSchema *mock_fk_parent_table = NULL;
1663
    if (FAILEDx(latest_schema_guard_.get_mock_fk_parent_table_schema(
1664
        replace_mock_fk_parent_table_id_, mock_fk_parent_table))) {
1665
      LOG_WARN("fail to get mock fk parent table schema",
1666
               KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
1667
    } else if (OB_ISNULL(mock_fk_parent_table)) {
1668
      ret = OB_ERR_UNEXPECTED;
1669
      LOG_WARN("mock fk parent table not exist",
1670
               KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
1671
    } else if (OB_FAIL(check_fk_columns_type_for_replacing_mock_fk_parent_table_(
1672
               data_table, *mock_fk_parent_table))) {
1673
      LOG_WARN("fail to check if data table is matched with mock_fk_parent_table", KR(ret));
1674
    } else if (OB_FAIL(ObSchemaUtils::alloc_schema(
1675
               allocator_, *mock_fk_parent_table, new_mock_fk_parent_table))) {
1676
      LOG_WARN("fail to alloc schema", KR(ret), KPC(mock_fk_parent_table));
1677
    } else {
1678
      (void) new_mock_fk_parent_table->set_operation_type(
1679
             ObMockFKParentTableOperationType::MOCK_FK_PARENT_TABLE_OP_REPLACED_BY_REAL_PREANT_TABLE);
1680
      const ObIArray<ObForeignKeyInfo> &ori_mock_fk_infos_array = mock_fk_parent_table->get_foreign_key_infos();
1681
      // modify the parent column id of fk，make it fit with real parent table
1682
      // mock_column_id -> column_name -> real_column_id
1683
      for (int64_t i = 0; OB_SUCC(ret) && i < ori_mock_fk_infos_array.count(); ++i) {
1684
        const ObForeignKeyInfo &ori_foreign_key_info = mock_fk_parent_table->get_foreign_key_infos().at(i);
1685
        ObForeignKeyInfo &new_foreign_key_info = new_mock_fk_parent_table->get_foreign_key_infos().at(i);
1686
        new_foreign_key_info.parent_column_ids_.reuse();
1687
        new_foreign_key_info.ref_cst_type_ = CONSTRAINT_TYPE_INVALID;
1688
        new_foreign_key_info.is_parent_table_mock_ = false;
1689
        new_foreign_key_info.parent_table_id_ = data_table.get_table_id();
1690
        // replace parent table columns
1691
        for (int64_t j = 0;  OB_SUCC(ret) && j < ori_foreign_key_info.parent_column_ids_.count(); ++j) {
1692
          bool is_column_exist = false;
1693
          uint64_t mock_parent_table_column_id = ori_foreign_key_info.parent_column_ids_.at(j);
1694
          ObString column_name;
1695
          const ObColumnSchemaV2 *col_schema = NULL;
1696
          (void) mock_fk_parent_table->get_column_name_by_column_id(
1697
                 mock_parent_table_column_id, column_name, is_column_exist);
1698
          if (!is_column_exist) {
1699
            ret = OB_ERR_UNEXPECTED;
1700
            LOG_WARN("column is not exist", KR(ret), K(mock_parent_table_column_id), KPC(mock_fk_parent_table));
1701
          } else if (OB_ISNULL(col_schema = data_table.get_column_schema(column_name))) {
1702
            ret = OB_ERR_UNEXPECTED;
1703
            LOG_WARN("get column schema failed", KR(ret), K(column_name));
1704
          } else if (OB_FAIL(new_foreign_key_info.parent_column_ids_.push_back(col_schema->get_column_id()))) {
1705
            LOG_WARN("push_back to parent_column_ids failed", KR(ret), K(col_schema->get_column_id()));
1706
          }
1707
        } // end for
1708
        // check and mofidy ref cst type and ref cst id of fk
1709
        const ObRowkeyInfo &rowkey_info = data_table.get_rowkey_info();
1710
        common::ObArray<uint64_t> pk_column_ids;
1711
        for (int64_t j = 0; OB_SUCC(ret) && j < rowkey_info.get_size(); ++j) {
1712
          uint64_t column_id = 0;
1713
          const ObColumnSchemaV2 *col_schema = NULL;
1714
          if (OB_FAIL(rowkey_info.get_column_id(j, column_id))) {
1715
            LOG_WARN("fail to get rowkey info", KR(ret), K(j), K(rowkey_info));
1716
          } else if (OB_ISNULL(col_schema = data_table.get_column_schema(column_id))) {
1717
            ret = OB_ERR_UNEXPECTED;
1718
            LOG_WARN("get index column schema failed", K(ret));
1719
          } else if (col_schema->is_hidden() || col_schema->is_shadow_column()) {
1720
            // do nothing
1721
          } else if (OB_FAIL(pk_column_ids.push_back(col_schema->get_column_id()))) {
1722
            LOG_WARN("push back column_id failed", KR(ret), K(col_schema->get_column_id()));
1723
          }
1724
        } // end for
1725
        bool is_match = false;
1726
        if (FAILEDx(sql::ObResolverUtils::check_match_columns(
1727
            pk_column_ids, new_foreign_key_info.parent_column_ids_, is_match))) {
1728
          LOG_WARN("check_match_columns failed", KR(ret));
1729
        } else if (is_match) {
1730
          new_foreign_key_info.ref_cst_type_ = CONSTRAINT_TYPE_PRIMARY_KEY;
1731
        } else { // pk is not match, check if uk match
1732
          if (OB_FAIL(ddl_service_->get_uk_cst_id_for_replacing_mock_fk_parent_table(
1733
              index_schemas, new_foreign_key_info))) {
1734
            LOG_WARN("fail to get_uk_cst_id_for_replacing_mock_fk_parent_table", KR(ret));
1735
          } else if (CONSTRAINT_TYPE_INVALID == new_foreign_key_info.ref_cst_type_) {
1736
            ret = OB_ERR_CANNOT_ADD_FOREIGN;
1737
            LOG_WARN("ref_cst_type is invalid", KR(ret), KPC(mock_fk_parent_table));
1738
          }
1739
        }
1740
      }
1741
    }
1742
  }
1743
  return ret;
1744
}
1745

1746
int ObCreateTableHelper::check_fk_columns_type_for_replacing_mock_fk_parent_table_(
1747
    const ObTableSchema &parent_table_schema,
1748
    const ObMockFKParentTableSchema &mock_parent_table_schema)
1749
{
1750
  int ret = OB_SUCCESS;
1751
  bool is_oracle_mode = false;
1752
  if (OB_FAIL(check_inner_stat_())) {
1753
    LOG_WARN("fail to check inner stat", KR(ret));
1754
  } else if (OB_FAIL(parent_table_schema.check_if_oracle_compat_mode(is_oracle_mode))) {
1755
    LOG_WARN("check if oracle compat mode failed", K(ret));
1756
  }
1757
  for (int64_t i = 0; OB_SUCC(ret) && i < mock_parent_table_schema.get_foreign_key_infos().count(); ++i) {
1758
    const ObTableSchema *child_table_schema = NULL;
1759
    const ObForeignKeyInfo &fk_info = mock_parent_table_schema.get_foreign_key_infos().at(i);
1760
    if (OB_FAIL(latest_schema_guard_.get_table_schema(fk_info.child_table_id_, child_table_schema))) {
1761
      LOG_WARN("table is not exist", KR(ret), K(fk_info));
1762
    } else if (OB_ISNULL(child_table_schema)) {
1763
      ret = OB_ERR_PARALLEL_DDL_CONFLICT;
1764
      LOG_WARN("child table schema is null, need retry", KR(ret), K(fk_info));
1765
    } else {
1766
      // prepare params for check_foreign_key_columns_type
1767
      ObArray<ObString> child_columns;
1768
      ObArray<ObString> parent_columns;
1769
      bool is_column_exist = false;
1770
      for (int64_t j = 0; OB_SUCC(ret) && j < fk_info.child_column_ids_.count(); ++j) {
1771
        ObString child_column_name;
1772
        const ObColumnSchemaV2 *child_col = child_table_schema->get_column_schema(fk_info.child_column_ids_.at(j));
1773
        if (OB_ISNULL(child_col)) {
1774
          ret = OB_ERR_UNEXPECTED;
1775
          LOG_WARN("column is not exist", KR(ret), K(fk_info));
1776
        } else if (OB_FAIL(child_columns.push_back(child_col->get_column_name_str()))) {
1777
          LOG_WARN("fail to push_back to child_columns", KR(ret), KPC(child_col));
1778
        }
1779
      } // end for
1780
      for (int64_t j = 0; OB_SUCC(ret) && j < fk_info.parent_column_ids_.count(); ++j) {
1781
        ObString parent_column_name;
1782
        (void) mock_parent_table_schema.get_column_name_by_column_id(
1783
               fk_info.parent_column_ids_.at(j), parent_column_name, is_column_exist);
1784
        if (!is_column_exist) {
1785
          ret = OB_ERR_UNEXPECTED;
1786
          LOG_WARN("column is not exist", KR(ret), K(fk_info));
1787
        } else if (OB_FAIL(parent_columns.push_back(parent_column_name))) {
1788
          LOG_WARN("fail to push_back to real_parent_table_schema_columns", KR(ret), K(fk_info));
1789
        }
1790
      } // end for
1791
      if (FAILEDx(sql::ObResolverUtils::check_foreign_key_columns_type(
1792
          !is_oracle_mode/*is_mysql_compat_mode*/,
1793
          *child_table_schema,
1794
          parent_table_schema,
1795
          child_columns,
1796
          parent_columns,
1797
          NULL))) {
1798
        ret = OB_ERR_CANNOT_ADD_FOREIGN;
1799
        LOG_WARN("Failed to check_foreign_key_columns_type", KR(ret));
1800
      }
1801
    }
1802
  } // end for
1803
  return ret;
1804
}
1805

1806
int ObCreateTableHelper::generate_sequence_object_()
1807
{
1808
  int ret = OB_SUCCESS;
1809
  bool is_oracle_mode = false;
1810
  if (OB_FAIL(check_inner_stat_())) {
1811
    LOG_WARN("fail to check inner stat", KR(ret));
1812
  } else if (OB_FAIL(ObCompatModeGetter::check_is_oracle_mode_with_tenant_id(tenant_id_, is_oracle_mode))) {
1813
    LOG_WARN("fail to check is oracle mode", KR(ret));
1814
  } else if (!is_oracle_mode) {
1815
    // skip
1816
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
1817
    ret = OB_ERR_UNEXPECTED;
1818
    LOG_WARN("unexpected table cnt", KR(ret), K(new_tables_.count()));
1819
  } else if (!(new_tables_.at(0).is_user_table()
1820
               || new_tables_.at(0).is_oracle_tmp_table())) {
1821
    // skip
1822
  } else {
1823
    // 1. lock object name
1824
    // - Sequence object name for table is encoded with table_id which is determinated in generate_schemas_() stage,
1825
    // - so lock_objects_() stage will be delayed. And sequence_name won't be conficted for most cases since it's encoded.
1826
    const obrpc::ObSequenceDDLArg &sequence_ddl_arg = arg_.sequence_ddl_arg_;
1827
    const ObTableSchema &data_table = new_tables_.at(0);
1828
    const uint64_t tenant_id = data_table.get_tenant_id();
1829
    const uint64_t database_id = data_table.get_database_id();
1830
    const uint64_t session_id = arg_.schema_.get_session_id();
1831
    const ObString &database_name = arg_.db_name_;
1832
    ObSequenceSchema *new_sequence = NULL;
1833
    char sequence_name[OB_MAX_SEQUENCE_NAME_LENGTH + 1] = { 0 };
1834
    const bool is_or_replace = false;
1835
    for (ObTableSchema::const_column_iterator iter = data_table.column_begin();
1836
         OB_SUCC(ret) && iter != data_table.column_end(); ++iter) {
1837
      ObColumnSchemaV2 *column_schema = *iter;
1838
      if (OB_ISNULL(column_schema)) {
1839
        ret = OB_ERR_UNEXPECTED;
1840
        LOG_WARN("column is null", KR(ret));
1841
      } else if (!column_schema->is_identity_column()) {
1842
        continue;
1843
      } else if (OB_FAIL(ObSchemaUtils::alloc_schema(allocator_, new_sequence))) {
1844
        LOG_WARN("fail to alloc schema", KR(ret));
1845
      } else if (OB_FAIL(new_sequence->assign(sequence_ddl_arg.seq_schema_))) {
1846
        LOG_WARN("fail to assign sequence schema", KR(ret), K(sequence_ddl_arg));
1847
      } else {
1848
        ObSequenceOptionBuilder opt_builder;
1849
        int32_t len = snprintf(sequence_name, sizeof(sequence_name),
1850
                               "%s%lu%c%lu", IDENTITY_COLUMN_SEQUENCE_OBJECT_NAME_PREFIX,
1851
                                data_table.get_table_id(), '_', column_schema->get_column_id());
1852
        new_sequence->set_tenant_id(tenant_id);
1853
        new_sequence->set_database_id(database_id);
1854
        if (OB_UNLIKELY(len < 0)) {
1855
          ret = OB_ERR_UNEXPECTED;
1856
          LOG_WARN("create sequence name fail", KR(ret), KPC(column_schema));
1857
        } else if (OB_FAIL(new_sequence->set_sequence_name(sequence_name))) {
1858
          LOG_WARN("fail to set sequence name", KR(ret), K(sequence_name));
1859
        } else if (OB_FAIL(opt_builder.build_create_sequence_option(
1860
          sequence_ddl_arg.option_bitset_, new_sequence->get_sequence_option()))) {
1861
          LOG_WARN("fail to build sequence_option", KR(ret), K(sequence_ddl_arg));
1862
        } else if (OB_FAIL(add_lock_object_by_name_(
1863
                   database_name, new_sequence->get_sequence_name(),
1864
                   share::schema::SEQUENCE_SCHEMA, transaction::tablelock::EXCLUSIVE))) {
1865
          LOG_WARN("fail to lock object by sequence name prefix", KR(ret), K_(tenant_id),
1866
                   K(database_name), KPC(new_sequence));
1867
        } else if (OB_FAIL(latest_schema_guard_.check_oracle_object_exist(
1868
                   new_sequence->get_database_id(), session_id,
1869
                   new_sequence->get_sequence_name(), SEQUENCE_SCHEMA,
1870
                   INVALID_ROUTINE_TYPE, is_or_replace))) {
1871
          LOG_WARN("fail to check oracle object exist", KR(ret), KPC(new_sequence));
1872
        } else if (OB_FAIL(new_sequences_.push_back(new_sequence))) {
1873
          LOG_WARN("fail to push back new sequence ptr", KR(ret));
1874
        }
1875
      }
1876
    } // end for
1877

1878
    if (FAILEDx(lock_existed_objects_by_name_())) {
1879
      LOG_WARN("fail to lock objects by name", KR(ret), K_(tenant_id));
1880
    }
1881

1882
    // 2. generate object ids
1883
    ObIDGenerator id_generator;
1884
    const uint64_t object_cnt = new_sequences_.count();
1885
    if (FAILEDx(gen_object_ids_(object_cnt, id_generator))) {
1886
      LOG_WARN("fail to gen object ids", KR(ret), K_(tenant_id), K(object_cnt));
1887
    }
1888

1889
    // 3. generate schema
1890
    int64_t idx = 0;
1891
    uint64_t sequence_id = OB_INVALID_ID;
1892
    char sequence_string[OB_MAX_SEQUENCE_NAME_LENGTH + 1] = { 0 };
1893
    for (ObTableSchema::const_column_iterator iter = data_table.column_begin();
1894
         OB_SUCC(ret) && iter != data_table.column_end(); ++iter) {
1895
      ObColumnSchemaV2 *column_schema = *iter;
1896
      if (OB_ISNULL(column_schema)) {
1897
        ret = OB_ERR_UNEXPECTED;
1898
        LOG_WARN("column is null", KR(ret));
1899
      } else if (!column_schema->is_identity_column()) {
1900
        continue;
1901
      } else if (OB_UNLIKELY(idx >= new_sequences_.count())) {
1902
        ret = OB_ERR_UNEXPECTED;
1903
        LOG_WARN("invalid idx", KR(ret), K(idx), K(new_sequences_.count()));
1904
      } else if (OB_ISNULL(new_sequence = new_sequences_.at(idx++))) {
1905
        ret = OB_ERR_UNEXPECTED;
1906
        LOG_WARN("sequence is null", KR(ret));
1907
      } else if (OB_FAIL(id_generator.next(sequence_id))) {
1908
        LOG_WARN("fail to get next object_id", KR(ret));
1909
      } else {
1910
        // 3.1. generate sequence id
1911
        new_sequence->set_sequence_id(sequence_id);
1912
        // 3.2. modify column schema
1913
        column_schema->set_sequence_id(sequence_id);
1914
        int32_t len = snprintf(sequence_string, sizeof(sequence_string), "%lu", sequence_id);
1915
        if (OB_UNLIKELY(len < 0)) {
1916
          ret = OB_ERR_UNEXPECTED;
1917
          LOG_WARN("create sequence string fail", KR(ret), K(sequence_id));
1918
        } else {
1919
          ObObjParam cur_default_value;  // for desc table
1920
          ObObjParam orig_default_value; // for store pure_sequence_id
1921
          cur_default_value.set_varchar("SEQUENCE.NEXTVAL");
1922
          cur_default_value.set_collation_type(ObCharset::get_system_collation());
1923
          cur_default_value.set_collation_level(CS_LEVEL_IMPLICIT);
1924
          cur_default_value.set_param_meta();
1925
          orig_default_value.set_varchar(sequence_string);
1926
          orig_default_value.set_collation_type(ObCharset::get_system_collation());
1927
          orig_default_value.set_collation_level(CS_LEVEL_IMPLICIT);
1928
          orig_default_value.set_param_meta();
1929
          if (OB_FAIL(column_schema->set_cur_default_value(cur_default_value))) {
1930
            LOG_WARN("set current default value fail", KR(ret));
1931
          } else if (OB_FAIL(column_schema->set_orig_default_value(orig_default_value))) {
1932
            LOG_WARN("set origin default value fail", KR(ret), K(column_schema));
1933
          }
1934
        }
1935
      }
1936
    } // end for
1937
  }
1938
  return ret;
1939
}
1940

1941
// create audit schema for table/sequence
1942
int ObCreateTableHelper::generate_audit_schema_()
1943
{
1944
  int ret = OB_SUCCESS;
1945
  if (OB_FAIL(check_inner_stat_())) {
1946
    LOG_WARN("fail to check inner stat", KR(ret));
1947
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
1948
    ret = OB_ERR_UNEXPECTED;
1949
    LOG_WARN("unexpected table cnt", KR(ret), K(new_tables_.count()));
1950
  } else {
1951
    // check if data table's columns are matched with existed mock fk parent table
1952
    const ObTableSchema &data_table = new_tables_.at(0);
1953
    ObArray<ObSAuditSchema> audits;
1954
    if (OB_FAIL(latest_schema_guard_.get_default_audit_schemas(audits))) {
1955
      LOG_WARN("fail to get audits", KR(ret));
1956
    } else if (!audits.empty()) {
1957
      ObSAuditSchema *new_audit_schema = NULL;
1958
      for (int64_t i = 0; OB_SUCC(ret) && i < audits.count(); i++) {
1959
        const ObSAuditSchema &audit = audits.at(i);
1960
        if (audit.is_access_operation_for_sequence()) {
1961
          for (int64_t j = 0; OB_SUCC(ret) && j < new_sequences_.count(); j++) {
1962
            const ObSequenceSchema *new_sequence = new_sequences_.at(j);
1963
            if (OB_ISNULL(new_sequence)) {
1964
              ret = OB_ERR_UNEXPECTED;
1965
              LOG_WARN("new sequence is null", KR(ret));
1966
            } else if (OB_FAIL(ObSchemaUtils::alloc_schema(allocator_, new_audit_schema))) {
1967
              LOG_WARN("fail to alloc schema", KR(ret));
1968
            } else if (OB_FAIL(new_audit_schema->assign(audit))) {
1969
              LOG_WARN("fail to assign audit", KR(ret));
1970
            } else {
1971
              new_audit_schema->set_audit_type(AUDIT_SEQUENCE);
1972
              new_audit_schema->set_owner_id(new_sequence->get_sequence_id());
1973
              if (OB_FAIL(new_audits_.push_back(new_audit_schema))) {
1974
                LOG_WARN("fail to push back audit", KR(ret), KPC(new_audit_schema));
1975
              }
1976
            }
1977
          } // end for
1978
        }
1979
        if (OB_FAIL(ret)) {
1980
        } else if (audit.is_access_operation_for_table()) {
1981
          if (OB_FAIL(ObSchemaUtils::alloc_schema(allocator_, new_audit_schema))) {
1982
            LOG_WARN("fail to alloc schema", KR(ret));
1983
          } else if (OB_FAIL(new_audit_schema->assign(audit))) {
1984
            LOG_WARN("fail to assign audit", KR(ret));
1985
          } else {
1986
            new_audit_schema->set_audit_type(AUDIT_TABLE);
1987
            new_audit_schema->set_owner_id(data_table.get_table_id());
1988
            if (OB_FAIL(new_audits_.push_back(new_audit_schema))) {
1989
              LOG_WARN("fail to push back audit", KR(ret), KPC(new_audit_schema));
1990
            }
1991
          }
1992
        }
1993
      } // end for
1994

1995
      if (OB_SUCC(ret) && !new_audits_.empty()) {
1996
        ObIDGenerator id_generator;
1997
        const uint64_t object_cnt = new_audits_.count();
1998
        uint64_t object_id = OB_INVALID_ID;
1999
        if (OB_FAIL(gen_object_ids_(object_cnt, id_generator))) {
2000
          LOG_WARN("fail to gen object ids", KR(ret), K_(tenant_id), K(object_cnt));
2001
        }
2002
        for (int64_t i = 0; OB_SUCC(ret) && i < new_audits_.count(); i++) {
2003
          ObSAuditSchema *audit = new_audits_.at(i);
2004
          if (OB_ISNULL(audit)) {
2005
            ret = OB_ERR_UNEXPECTED;
2006
            LOG_WARN("audit is null", KR(ret));
2007
          } else if (OB_FAIL(id_generator.next(object_id))) {
2008
            LOG_WARN("fail to get next object_id", KR(ret));
2009
          } else {
2010
            audit->set_audit_id(object_id);
2011
          }
2012
        } // end for
2013
      }
2014
    }
2015
  }
2016
  return ret;
2017
}
2018

2019
int ObCreateTableHelper::calc_schema_version_cnt_()
2020
{
2021
  int ret = OB_SUCCESS;
2022
  if (OB_FAIL(check_inner_stat_())) {
2023
    LOG_WARN("fail to check inner stat", KR(ret));
2024
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
2025
    ret = OB_ERR_UNEXPECTED;
2026
    LOG_WARN("unexpected table cnt", KR(ret), K(new_tables_.count()));
2027
  } else {
2028
    const ObTableSchema &data_table = new_tables_.at(0);
2029
    // 0. data table
2030
    schema_version_cnt_ = 1; // init
2031

2032
    // 1. sequence
2033
    schema_version_cnt_ += new_sequences_.count();
2034

2035
    // 2. audit
2036
    schema_version_cnt_ += new_audits_.count();
2037

2038
    // 3. create index/lob table
2039
    if (OB_SUCC(ret) && new_tables_.count() > 1) {
2040
      schema_version_cnt_ += (new_tables_.count() - 1);
2041
      // update data table schema version
2042
      schema_version_cnt_++;
2043
    }
2044

2045
    // 4. foreign key (without mock fk parent table)
2046
    if (OB_SUCC(ret)) {
2047
    // this logic is duplicated because of add_foreign_key() will also update data table's schema_version.
2048
    //schema_version_cnt_ += data_table.get_depend_table_ids();
2049
      const ObIArray<ObForeignKeyInfo> &foreign_key_infos = data_table.get_foreign_key_infos();
2050
      for (int64_t i = 0; OB_SUCC(ret) && i < foreign_key_infos.count(); i++) {
2051
        const ObForeignKeyInfo &foreign_key_info = foreign_key_infos.at(i);
2052
        if (foreign_key_info.is_modify_fk_state_) {
2053
          continue;
2054
        } else if (!foreign_key_info.is_parent_table_mock_) {
2055
          schema_version_cnt_++;
2056
          // TODO(yanmu.ztl): can be optimized in the following cases:
2057
          // - self reference
2058
          // - foreign keys has same parent table.
2059
        }
2060
      } // end for
2061
    }
2062

2063
    // 5. mock fk parent table
2064
    if (OB_SUCC(ret)) {
2065
      // schema version for new mock fk parent tables
2066
      schema_version_cnt_ += new_mock_fk_parent_tables_.count();
2067
      for (int64_t i = 0; OB_SUCC(ret) && i < new_mock_fk_parent_tables_.count(); i++) {
2068
        const ObMockFKParentTableSchema *new_mock_fk_parent_table = new_mock_fk_parent_tables_.at(i);
2069
        if (OB_ISNULL(new_mock_fk_parent_table)) {
2070
          ret = OB_ERR_UNEXPECTED;
2071
          LOG_WARN("mock fk parent table is null", KR(ret));
2072
        } else if (MOCK_FK_PARENT_TABLE_OP_CREATE_TABLE_BY_ADD_FK_IN_CHILD_TBALE
2073
                   == new_mock_fk_parent_table->get_operation_type()) {
2074
          // skip
2075
        } else if (MOCK_FK_PARENT_TABLE_OP_UPDATE_SCHEMA_VERSION
2076
                   == new_mock_fk_parent_table->get_operation_type()
2077
                   || MOCK_FK_PARENT_TABLE_OP_ADD_COLUMN
2078
                   == new_mock_fk_parent_table->get_operation_type()) {
2079
          // update new mock fk parent table(is useless here, just to be compatible with other logic)
2080
          schema_version_cnt_++;
2081
        } else if (MOCK_FK_PARENT_TABLE_OP_REPLACED_BY_REAL_PREANT_TABLE
2082
                   == new_mock_fk_parent_table->get_operation_type()) {
2083
          // update foreign keys' schema version.
2084
          schema_version_cnt_++;
2085
          // update child tables' schema version.
2086
          // TODO(yanmu.ztl): can be optimized when child table is duplicated.
2087
          schema_version_cnt_ += (new_mock_fk_parent_table->get_foreign_key_infos().count());
2088
          // update data table's schema version at last.
2089
          schema_version_cnt_++;
2090
        } else {
2091
          ret = OB_NOT_SUPPORTED;
2092
          LOG_WARN("not supported operation type", KR(ret), KPC(new_mock_fk_parent_table));
2093
        }
2094
      } // end for
2095
    }
2096

2097
    // 6. for 1503 boundary ddl operation
2098
    schema_version_cnt_++;
2099
  }
2100
  return ret;
2101
}
2102

2103
int ObCreateTableHelper::create_schemas_()
2104
{
2105
  int ret = OB_SUCCESS;
2106
  if (OB_FAIL(check_inner_stat_())) {
2107
    LOG_WARN("fail to check inner stat", KR(ret));
2108
  } else if (OB_FAIL(create_sequence_())) {
2109
    LOG_WARN("fail to create sequences", KR(ret));
2110
  } else if (OB_FAIL(create_tables_())) {
2111
    LOG_WARN("fail to create tables", KR(ret));
2112
  } else if (OB_FAIL(create_audits_())) {
2113
    LOG_WARN("fail to create tables", KR(ret));
2114
  } else if (OB_FAIL(deal_with_mock_fk_parent_tables_())) {
2115
    LOG_WARN("fail to deal with mock fk parent tables", KR(ret));
2116
  }
2117
  RS_TRACE(create_schemas);
2118
  return ret;
2119
}
2120

2121
int ObCreateTableHelper::create_sequence_()
2122
{
2123
  int ret = OB_SUCCESS;
2124
  ObSchemaService *schema_service_impl = NULL;
2125
  if (OB_FAIL(check_inner_stat_())) {
2126
    LOG_WARN("fail to check inner stat", KR(ret));
2127
  } else if (new_sequences_.count() <= 0) {
2128
    // skip
2129
  } else if (OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
2130
    ret = OB_ERR_UNEXPECTED;
2131
    LOG_WARN("schema_service impl is null", KR(ret));
2132
  } else {
2133
    ObString *ddl_stmt_str = NULL;
2134
    uint64_t *old_sequence_id = NULL; // means don't sync sequence value
2135
    int64_t new_schema_version = OB_INVALID_VERSION;
2136
    for (int64_t i = 0; OB_SUCC(ret) && i < new_sequences_.count(); i++) {
2137
      ObSequenceSchema *new_sequence = new_sequences_.at(i);
2138
      if (OB_ISNULL(new_sequence)) {
2139
        ret = OB_ERR_UNEXPECTED;
2140
        LOG_WARN("new sequence is null", KR(ret));
2141
      } else if (OB_FAIL(schema_service_->gen_new_schema_version(tenant_id_, new_schema_version))) {
2142
        LOG_WARN("fail to gen new schema_version", KR(ret), K_(tenant_id));
2143
      } else if (FALSE_IT(new_sequence->set_schema_version(new_schema_version))) {
2144
      } else if (OB_FAIL(schema_service_impl->get_sequence_sql_service().insert_sequence(
2145
                 *new_sequence, &trans_, ddl_stmt_str, old_sequence_id))) {
2146
        LOG_WARN("fail to create sequence", KR(ret), KPC(new_sequence));
2147
      }
2148
    } // end for
2149
  }
2150
  return ret;
2151
}
2152

2153
int ObCreateTableHelper::create_audits_()
2154
{
2155
  int ret = OB_SUCCESS;
2156
  ObSchemaService *schema_service_impl = NULL;
2157
  if (OB_FAIL(check_inner_stat_())) {
2158
    LOG_WARN("fail to check inner stat", KR(ret));
2159
  } else if (OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
2160
    ret = OB_ERR_UNEXPECTED;
2161
    LOG_WARN("schema_service impl is null", KR(ret));
2162
  } else {
2163
    common::ObSqlString public_sql_string;
2164
    int64_t new_schema_version = OB_INVALID_VERSION;
2165
    for (int64_t i = 0; OB_SUCC(ret) && i < new_audits_.count(); i++) {
2166
      ObSAuditSchema *new_audit = new_audits_.at(i);
2167
      if (OB_ISNULL(new_audit)) {
2168
        ret = OB_ERR_UNEXPECTED;
2169
        LOG_WARN("audit is null", KR(ret));
2170
      } else if (OB_FAIL(schema_service_->gen_new_schema_version(tenant_id_, new_schema_version))) {
2171
        LOG_WARN("fail to gen new schema_version", KR(ret), K_(tenant_id));
2172
      } else if (FALSE_IT(new_audit->set_schema_version(new_schema_version))) {
2173
      } else if (OB_FAIL(schema_service_impl->get_audit_sql_service().handle_audit_metainfo(
2174
                 *new_audit,
2175
                 AUDIT_MT_ADD,
2176
                 false,
2177
                 new_schema_version,
2178
                 NULL,
2179
                 trans_,
2180
                 public_sql_string))) {
2181
        LOG_WARN("fail to add audit", KR(ret), K_(tenant_id), K(new_audit));
2182
      }
2183
    } // end for
2184
  }
2185
  return ret;
2186
}
2187

2188
int ObCreateTableHelper::create_tables_()
2189
{
2190
  int ret = OB_SUCCESS;
2191
  ObSchemaService *schema_service_impl = NULL;
2192
  if (OB_FAIL(check_inner_stat_())) {
2193
    LOG_WARN("fail to check inner stat", KR(ret));
2194
  } else if (OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
2195
    ret = OB_ERR_UNEXPECTED;
2196
    LOG_WARN("schema_service impl is null", KR(ret));
2197
  } else {
2198
    int64_t new_schema_version = OB_INVALID_VERSION;
2199
    for (int64_t i = 0; OB_SUCC(ret) && i < new_tables_.count(); i++) {
2200
      ObTableSchema &new_table = new_tables_.at(i);
2201
      const ObString *ddl_stmt_str = (0 == i) ? &arg_.ddl_stmt_str_ : NULL;
2202
      const bool need_sync_schema_version = (new_tables_.count() - 1 == i);
2203
      if (OB_FAIL(schema_service_->gen_new_schema_version(tenant_id_, new_schema_version))) {
2204
          LOG_WARN("fail to gen new schema_version", KR(ret), K_(tenant_id));
2205
      } else if (FALSE_IT(new_table.set_schema_version(new_schema_version))) {
2206
      } else if (OB_FAIL(schema_service_impl->get_table_sql_service().create_table(
2207
                 new_table,
2208
                 trans_,
2209
                 ddl_stmt_str,
2210
                 need_sync_schema_version,
2211
                 false/*is_truncate_table*/))) {
2212
        LOG_WARN("fail to create table", KR(ret), K(new_table));
2213
      } else if (OB_FAIL(schema_service_impl->get_table_sql_service().insert_temp_table_info(
2214
                 trans_, new_table))) {
2215
        LOG_WARN("insert_temp_table_info failed", KR(ret), K(new_table));
2216
      }
2217
    } // end for
2218
  }
2219
  return ret;
2220
}
2221

2222
int ObCreateTableHelper::deal_with_mock_fk_parent_tables_()
2223
{
2224
  int ret = OB_SUCCESS;
2225
  ObSchemaService *schema_service_impl = NULL;
2226
  if (OB_FAIL(check_inner_stat_())) {
2227
    LOG_WARN("fail to check inner stat", KR(ret));
2228
  } else if (OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
2229
    ret = OB_ERR_UNEXPECTED;
2230
    LOG_WARN("schema_service impl is null", KR(ret));
2231
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
2232
    ret = OB_ERR_UNEXPECTED;
2233
    LOG_WARN("unexpected table cnt", KR(ret), K(new_tables_.count()));
2234
  } else {
2235
    const ObTableSchema &data_table = new_tables_.at(0);
2236
    const uint64_t data_table_id = data_table.get_table_id();
2237
    int64_t new_schema_version = OB_INVALID_VERSION;
2238
    for (int64_t i = 0; OB_SUCC(ret) && i < new_mock_fk_parent_tables_.count(); i++) {
2239
      ObMockFKParentTableSchema *new_mock_fk_parent_table = new_mock_fk_parent_tables_.at(i);
2240
      if (OB_ISNULL(new_mock_fk_parent_table)) {
2241
        ret = OB_ERR_UNEXPECTED;
2242
        LOG_WARN("mock fk parent table is null", KR(ret));
2243
      } else if (OB_FAIL(schema_service_->gen_new_schema_version(tenant_id_, new_schema_version))) {
2244
          LOG_WARN("fail to gen new schema_version", KR(ret), K_(tenant_id));
2245
      } else {
2246
        new_mock_fk_parent_table->set_schema_version(new_schema_version);
2247
        ObMockFKParentTableOperationType operation_type = new_mock_fk_parent_table->get_operation_type();
2248
        if (MOCK_FK_PARENT_TABLE_OP_CREATE_TABLE_BY_ADD_FK_IN_CHILD_TBALE == operation_type) {
2249
          // 1. create table: mock fk parent table doesn't exist.
2250
          if (OB_FAIL(schema_service_impl->get_table_sql_service().add_mock_fk_parent_table(
2251
              &trans_, *new_mock_fk_parent_table, false /*need_update_foreign_key*/))) {
2252
            LOG_WARN("fail to add mock fk parent table", KR(ret), KPC(new_mock_fk_parent_table));
2253
          }
2254
        } else if (MOCK_FK_PARENT_TABLE_OP_UPDATE_SCHEMA_VERSION == operation_type
2255
                   || MOCK_FK_PARENT_TABLE_OP_ADD_COLUMN == operation_type) {
2256
          // 2. alter table: mock fk parent table has new child table.
2257
          if (OB_FAIL(schema_service_impl->get_table_sql_service().alter_mock_fk_parent_table(
2258
                      &trans_, *new_mock_fk_parent_table))) {
2259
            LOG_WARN("fail to alter mock fk parent table", KR(ret), KPC(new_mock_fk_parent_table));
2260
          }
2261
        } else if (MOCK_FK_PARENT_TABLE_OP_REPLACED_BY_REAL_PREANT_TABLE == operation_type) {
2262
          // 3. replace table: replace existed mock fk parent table with data table
2263
          const ObMockFKParentTableSchema *ori_mock_fk_parent_table = NULL;
2264
          if (OB_UNLIKELY(OB_INVALID_ID == replace_mock_fk_parent_table_id_)) {
2265
            ret = OB_ERR_UNEXPECTED;
2266
            LOG_WARN("invalid replace mock fk parent table id", KR(ret), K_(replace_mock_fk_parent_table_id));
2267
          } else if (OB_FAIL(latest_schema_guard_.get_mock_fk_parent_table_schema(
2268
            replace_mock_fk_parent_table_id_, ori_mock_fk_parent_table))) {
2269
            LOG_WARN("fail to get mock fk parent table schema",
2270
                     KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
2271
          } else if (OB_ISNULL(ori_mock_fk_parent_table)) {
2272
            ret = OB_ERR_UNEXPECTED;
2273
            LOG_WARN("mock fk parent table not exist, unexpected",
2274
                     KR(ret), K_(tenant_id), K_(replace_mock_fk_parent_table_id));
2275
          } else {
2276
            // 3.1. drop mock fk parent table.
2277
            // 3.2. update foreign keys from mock fk parent table.
2278
            if (OB_FAIL(schema_service_impl->get_table_sql_service().replace_mock_fk_parent_table(
2279
                        &trans_, *new_mock_fk_parent_table, ori_mock_fk_parent_table))) {
2280
              LOG_WARN("fail to replace mock fk parent table", KR(ret), KPC(new_mock_fk_parent_table));
2281
            }
2282

2283
            // 3.3. update child tables' schema version.
2284
            for (int64_t j = 0; OB_SUCC(ret) && j < new_mock_fk_parent_table->get_foreign_key_infos().count(); j++) {
2285
              const ObForeignKeyInfo &foreign_key = new_mock_fk_parent_table->get_foreign_key_infos().at(j);
2286
              const uint64_t child_table_id = foreign_key.child_table_id_;
2287
              const ObTableSchema *child_table = NULL;
2288
              if (OB_FAIL(latest_schema_guard_.get_table_schema(child_table_id, child_table))) {
2289
                LOG_WARN("fail to get table schema", KR(ret), K_(tenant_id), K(child_table_id));
2290
              } else if (OB_ISNULL(child_table)) {
2291
                ret = OB_ERR_UNEXPECTED;
2292
                LOG_WARN("child table is not exist", KR(ret), K_(tenant_id), K(child_table_id));
2293
              } else if (OB_FAIL(schema_service_impl->get_table_sql_service().update_data_table_schema_version(
2294
                          trans_, tenant_id_, child_table_id, child_table->get_in_offline_ddl_white_list()))) {
2295
                LOG_WARN("fail to update child table's schema version", KR(ret), K_(tenant_id), K(child_table_id));
2296
              }
2297
            } // end for
2298

2299
            // 3.4. update data table's schema version at last.
2300
            if (FAILEDx(schema_service_impl->get_table_sql_service().update_data_table_schema_version(
2301
                        trans_, tenant_id_, data_table_id, false/*in_offline_ddl_white_list*/))) {
2302
              LOG_WARN("fail to update data table's schema version", KR(ret), K_(tenant_id), K(data_table_id));
2303
            }
2304
          }
2305
        } else {
2306
          ret = OB_NOT_SUPPORTED;
2307
          LOG_WARN("mock fk parent table operation type is not supported", KR(ret), K(operation_type));
2308
        }
2309
      }
2310
    } // end for
2311
  }
2312
  return ret;
2313
}
2314

2315
int ObCreateTableHelper::create_tablets_()
2316
{
2317
  int ret = OB_SUCCESS;
2318
  SCN frozen_scn;
2319
  ObSchemaGetterGuard schema_guard;
2320
  ObSchemaService *schema_service_impl = NULL;
2321
  uint64_t tenant_data_version = 0;
2322

2323
  if (OB_FAIL(check_inner_stat_())) {
2324
    LOG_WARN("fail to check inner stat", KR(ret));
2325
  } else if (OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
2326
    ret = OB_ERR_UNEXPECTED;
2327
    LOG_WARN("schema_service impl is null", KR(ret));
2328
  } else if (OB_FAIL(ObMajorFreezeHelper::get_frozen_scn(tenant_id_, frozen_scn))) {
2329
    LOG_WARN("failed to get frozen status for create tablet", KR(ret), K_(tenant_id));
2330
  } else if (OB_FAIL(schema_service_->get_tenant_schema_guard(tenant_id_, schema_guard))) {
2331
    LOG_WARN("fail to get tenant schema guard", KR(ret), K_(tenant_id));
2332
  } else if (OB_UNLIKELY(new_tables_.count() <= 0)) {
2333
    ret = OB_ERR_UNEXPECTED;
2334
    LOG_WARN("unexpected table cnt", KR(ret), K(new_tables_.count()));
2335
  } else if (OB_FAIL(GET_MIN_DATA_VERSION(tenant_id_, tenant_data_version))) {
2336
    LOG_WARN("get min data version failed", K(ret), K_(tenant_id));
2337
  } else {
2338
    ObTableCreator table_creator(
2339
                   tenant_id_,
2340
                   frozen_scn,
2341
                   trans_);
2342
    // TODO:(yanmu.ztl)
2343
    // schema_guard is used to get primary table in tablegroup or data table for local index.
2344
    // - primary table may be incorrect when ddl execute concurrently.
2345
    ObNewTableTabletAllocator new_table_tablet_allocator(
2346
                              tenant_id_,
2347
                              schema_guard,
2348
                              sql_proxy_,
2349
                              true /*use parallel ddl*/);
2350
    int64_t last_schema_version = OB_INVALID_VERSION;
2351
    auto *tsi_generator = GET_TSI(TSISchemaVersionGenerator);
2352
    if (OB_FAIL(table_creator.init(true/*need_tablet_cnt_check*/))) {
2353
      LOG_WARN("fail to init table creator", KR(ret));
2354
    } else if (OB_FAIL(new_table_tablet_allocator.init())) {
2355
      LOG_WARN("fail to init new table tablet allocator", KR(ret));
2356
    } else if (OB_ISNULL(tsi_generator)) {
2357
      ret = OB_ERR_UNEXPECTED;
2358
      LOG_WARN("tsi schema version generator is null", KR(ret));
2359
    } else if (OB_FAIL(tsi_generator->get_current_version(last_schema_version))) {
2360
      LOG_WARN("fail to get end version", KR(ret), K_(tenant_id), K_(arg));
2361
    } else if (OB_UNLIKELY(last_schema_version <= 0)) {
2362
      ret = OB_ERR_UNEXPECTED;
2363
      LOG_WARN("last schema version is invalid", KR(ret), K_(tenant_id), K(last_schema_version));
2364
    } else {
2365
      ObArray<const ObTableSchema*> schemas;
2366
      common::ObArray<share::ObLSID> ls_id_array;
2367
      ObArray<bool> need_create_empty_majors;
2368
      for (int64_t i = 0; OB_SUCC(ret) && i < new_tables_.count(); i++) {
2369
        const ObTableSchema &new_table = new_tables_.at(i);
2370
        const uint64_t table_id = new_table.get_table_id();
2371
        if (!new_table.has_tablet()) {
2372
          // eg. external table ...
2373
        } else if (!new_table.is_global_index_table()) {
2374
          if (OB_FAIL(schemas.push_back(&new_table))) {
2375
            LOG_WARN("fail to push back new table", KR(ret));
2376
          } else if (OB_FAIL(need_create_empty_majors.push_back(true))) {
2377
            LOG_WARN("fail to push back need create empty major", KR(ret));
2378
          }
2379
        } else {
2380
          if (OB_FAIL(new_table_tablet_allocator.prepare(trans_, new_table))) {
2381
            LOG_WARN("fail to prepare ls for global index", KR(ret), K(new_table));
2382
          } else if (OB_FAIL(new_table_tablet_allocator.get_ls_id_array(ls_id_array))) {
2383
            LOG_WARN("fail to get ls id array", KR(ret));
2384
          } else if (OB_FAIL(table_creator.add_create_tablets_of_table_arg(
2385
                     new_table, ls_id_array, tenant_data_version, true/*need create major sstable*/))) {
2386
            LOG_WARN("create table partitions failed", KR(ret), K(new_table));
2387
          }
2388
        }
2389
        //TODO:(yanmu.ztl) can be optimized into one sql
2390
        if (FAILEDx(schema_service_impl->get_table_sql_service().insert_ori_schema_version(
2391
                    trans_, tenant_id_, table_id, last_schema_version))) {
2392
          LOG_WARN("fail to get insert ori schema version",
2393
                   KR(ret), K_(tenant_id), K(table_id), K(last_schema_version));
2394
        }
2395
      } // end for
2396

2397
      if (OB_FAIL(ret)) {
2398
      } else if (schemas.count() > 0) {
2399
        const ObTableSchema &data_table = new_tables_.at(0);
2400
        if (OB_FAIL(new_table_tablet_allocator.prepare(trans_, data_table))) {
2401
          LOG_WARN("fail to prepare ls for data table", KR(ret));
2402
        } else if (OB_FAIL(new_table_tablet_allocator.get_ls_id_array(ls_id_array))) {
2403
          LOG_WARN("fail to get ls id array", KR(ret));
2404
        } else if (OB_FAIL(table_creator.add_create_tablets_of_tables_arg(
2405
                   schemas, ls_id_array, tenant_data_version, need_create_empty_majors /*need create major sstable*/))) {
2406
          LOG_WARN("create table partitions failed", KR(ret), K(data_table));
2407
        } else if (OB_FAIL(table_creator.execute())) {
2408
          LOG_WARN("execute create partition failed", KR(ret));
2409
        }
2410
      }
2411
    }
2412
    // finish() will do nothing here, can be removed
2413
    int tmp_ret = OB_SUCCESS;
2414
    if (OB_SUCCESS != (tmp_ret = new_table_tablet_allocator.finish(OB_SUCCESS == ret))) {
2415
      LOG_WARN("fail to finish new table tablet allocator", KR(tmp_ret));
2416
    }
2417
  }
2418
  RS_TRACE(create_tablets);
2419
  return ret;
2420
}
2421

2422
int ObCreateTableHelper::add_index_name_to_cache_()
2423
{
2424
  int ret = OB_SUCCESS;
2425
  if (OB_FAIL(check_inner_stat_())) {
2426
    LOG_WARN("fail to check inner stat", KR(ret));
2427
  } else {
2428
    ObIndexNameChecker &checker = ddl_service_->get_index_name_checker();
2429
    for (int64_t i = 0; OB_SUCC(ret) && i < new_tables_.count(); i++) {
2430
      const ObTableSchema &table = new_tables_.at(i);
2431
      if (table.is_index_table()) {
2432
        has_index_ = true;
2433
        if (OB_FAIL(checker.add_index_name(table))) {
2434
          LOG_WARN("fail to add index name", KR(ret), K(table));
2435
        }
2436
      }
2437
    } // end for
2438
  }
2439
  return ret;
2440
}
2441