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 "observer/ob_inner_sql_connection.h"  //ObInnerSQLConnection
15
#include "rootserver/parallel_ddl/ob_ddl_helper.h"
16
#include "share/schema/ob_multi_version_schema_service.h"
17
#include "share/schema/ob_ddl_sql_service.h"
18
#include "share/ob_max_id_fetcher.h"
19
#include "storage/tablelock/ob_table_lock_rpc_struct.h" //ObLockObjRequest
20
#include "storage/tablelock/ob_lock_inner_connection_util.h" //ObInnerConnectionLockUtil
21

22
using namespace oceanbase::lib;
23
using namespace oceanbase::common;
24
using namespace oceanbase::share;
25
using namespace oceanbase::share::schema;
26
using namespace oceanbase::rootserver;
27
using namespace oceanbase::transaction::tablelock;
28

29
ObDDLHelper::ObLockObjPair::ObLockObjPair()
30
  : obj_id_(0),
31
    lock_mode_(transaction::tablelock::MAX_LOCK_MODE)
32
{
33
}
34

35
ObDDLHelper::ObLockObjPair::ObLockObjPair(
36
  const uint64_t obj_id,
37
  transaction::tablelock::ObTableLockMode lock_mode)
38
  : obj_id_(obj_id),
39
    lock_mode_(lock_mode)
40
{
41
}
42

43
int ObDDLHelper::ObLockObjPair::init(
44
  const uint64_t obj_id,
45
  transaction::tablelock::ObTableLockMode lock_mode)
46
{
47
  int ret = OB_SUCCESS;
48
  reset();
49
  obj_id_ = obj_id;
50
  lock_mode_ = lock_mode;
51
  return ret;
52
}
53

54
void ObDDLHelper::ObLockObjPair::reset()
55
{
56
  obj_id_ = 0;
57
  lock_mode_ = transaction::tablelock::MAX_LOCK_MODE;
58
}
59

60
bool ObDDLHelper::ObLockObjPair::less_than(
61
     const ObLockObjPair &left,
62
     const ObLockObjPair &right)
63
{
64
  bool bret = false;
65
  if (left.get_obj_id() != right.get_obj_id()) {
66
    bret = (left.get_obj_id() < right.get_obj_id());
67
  } else {
68
    bret = (left.get_lock_mode() < right.get_lock_mode());
69
  }
70
  return bret;
71
}
72

73
ObDDLHelper::ObDDLHelper(
74
  share::schema::ObMultiVersionSchemaService *schema_service,
75
  const uint64_t tenant_id)
76
  : inited_(false),
77
    schema_service_(schema_service),
78
    ddl_service_(NULL),
79
    sql_proxy_(NULL),
80
    ddl_trans_controller_(NULL),
81
    tenant_id_(tenant_id),
82
    task_id_(common::OB_INVALID_ID),
83
    schema_version_cnt_(0),
84
    object_id_cnt_(0),
85
    trans_(schema_service_,
86
           false, /*need_end_signal*/
87
           false, /*enable_query_stash*/
88
           true   /*enable_ddl_parallel*/),
89
    lock_database_name_map_(),
90
    lock_object_name_map_(),
91
    lock_object_id_map_(),
92
    latest_schema_guard_(schema_service, tenant_id)
93
{}
94

95
ObDDLHelper::~ObDDLHelper()
96
{
97
}
98

99
int ObDDLHelper::init(rootserver::ObDDLService &ddl_service)
100
{
101
  int ret = OB_SUCCESS;
102
  if (OB_UNLIKELY(inited_)) {
103
    ret = OB_INIT_TWICE;
104
    LOG_WARN("ddl_helper already inited", KR(ret));
105
  } else if (OB_ISNULL(schema_service_)) {
106
    ret = OB_INVALID_ARGUMENT;
107
    LOG_WARN("schema_service is null", KR(ret));
108
  } else if (OB_FAIL(lock_database_name_map_.create(
109
             OBJECT_BUCKET_NUM, "LockDBNameMap", "LockDBNameMap"))) {
110
    LOG_WARN("fail to create lock database name map", KR(ret));
111
  } else if (OB_FAIL(lock_object_name_map_.create(
112
             OBJECT_BUCKET_NUM, "LockObjNameMap", "LockObjNameMap"))) {
113
    LOG_WARN("fail to create lock object name map", KR(ret));
114
  } else if (OB_FAIL(lock_object_id_map_.create(
115
             OBJECT_BUCKET_NUM, "LockObjIDMap", "LockObjIDMap"))) {
116
    LOG_WARN("fail to create lock object id map", KR(ret));
117
  } else {
118
    ddl_service_ = &ddl_service;
119
    sql_proxy_ = &(ddl_service.get_sql_proxy());
120
    ddl_trans_controller_ = &(schema_service_->get_ddl_trans_controller());
121
    task_id_ = OB_INVALID_ID;
122
    schema_version_cnt_ = 0;
123
    object_id_cnt_ = 0;
124
    inited_ = true;
125
  }
126
  return ret;
127
}
128

129
int ObDDLHelper::check_inner_stat_()
130
{
131
  int ret = OB_SUCCESS;
132
  if (OB_UNLIKELY(!inited_)) {
133
    ret = OB_NOT_INIT;
134
    LOG_WARN("ddl_helper not init yet", KR(ret));
135
  } else if (OB_ISNULL(ddl_service_)
136
             || OB_ISNULL(sql_proxy_)
137
             || OB_ISNULL(schema_service_)
138
             || OB_ISNULL(ddl_trans_controller_)) {
139
    ret = OB_NOT_INIT;
140
    LOG_WARN("ptr is null", KR(ret), KP_(ddl_service), KP_(schema_service),
141
             KP_(sql_proxy), K_(ddl_trans_controller));
142
  } else if (OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id_)) {
143
    ret = OB_NOT_INIT;
144
    LOG_WARN("invalid tenant_id", KR(ret), K_(tenant_id));
145
  }
146
  return ret;
147
}
148

149
int ObDDLHelper::start_ddl_trans_()
150
{
151
  int ret = OB_SUCCESS;
152
  bool with_snapshot = false;
153
  int64_t fake_schema_version = 1000;
154
  if (OB_FAIL(check_inner_stat_())) {
155
    LOG_WARN("fail to check inner stat", KR(ret));
156
  } else if (OB_FAIL(trans_.start(sql_proxy_, tenant_id_, fake_schema_version, with_snapshot))) {
157
    LOG_WARN("fail to start trans", KR(ret), K_(tenant_id), K(fake_schema_version), K(with_snapshot));
158
  }
159
  RS_TRACE(start_ddl_trans);
160
  return ret;
161
}
162

163
int ObDDLHelper::gen_task_id_and_schema_versions_()
164
{
165
  int ret = OB_SUCCESS;
166
  // just for interface compatibility, schema version can be fetched from TSISchemaVersionGenerator
167
  ObArray<int64_t> schema_versions;
168
  int64_t version_cnt = OB_INVALID_INDEX;
169
  auto *tsi_generator = GET_TSI(TSISchemaVersionGenerator);
170
  if (OB_FAIL(check_inner_stat_())) {
171
    LOG_WARN("fail to check inner stat", KR(ret));
172
  } else if (OB_FAIL(ddl_trans_controller_->create_task_and_assign_schema_version(
173
             tenant_id_, schema_version_cnt_, task_id_, schema_versions))) {
174
    LOG_WARN("fail to gen task id and schema_versions", KR(ret), K_(tenant_id), K_(schema_version_cnt));
175
  } else if (OB_UNLIKELY(OB_INVALID_ID == task_id_
176
             || schema_version_cnt_ != schema_versions.count())) {
177
    ret = OB_INVALID_ARGUMENT;
178
    LOG_WARN("task_id or schema version cnt not match", KR(ret), K_(tenant_id), K_(task_id),
179
             K_(schema_version_cnt), "schema_versions_cnt", schema_versions.count());
180
  } else if (OB_ISNULL(tsi_generator)) {
181
    ret = OB_ERR_UNEXPECTED;
182
    LOG_WARN("tsi schema version generator is null", KR(ret));
183
  } else if (OB_FAIL(tsi_generator->get_version_cnt(version_cnt))) {
184
    LOG_WARN("fail to get id cnt", KR(ret));
185
  } else if (OB_UNLIKELY(schema_version_cnt_ != version_cnt)) {
186
    ret = OB_INVALID_ARGUMENT;
187
    LOG_WARN("schema version cnt not match", KR(ret), K_(tenant_id), K_(task_id),
188
             K_(schema_version_cnt), K(version_cnt));
189
  }
190
  RS_TRACE(gen_task_id_and_versions);
191
  return ret;
192
}
193

194
int ObDDLHelper::serialize_inc_schema_dict_()
195
{
196
  int ret = OB_SUCCESS;
197
  auto *tsi_generator = GET_TSI(TSISchemaVersionGenerator);
198
  int64_t start_schema_version = OB_INVALID_VERSION;
199
  if (OB_FAIL(check_inner_stat_())) {
200
    LOG_WARN("fail to check inner stat", KR(ret));
201
  } else if (OB_ISNULL(tsi_generator)) {
202
    ret = OB_ERR_UNEXPECTED;
203
    LOG_WARN("tsi schema version generator is null", KR(ret));
204
  } else if (OB_FAIL(tsi_generator->get_start_version(start_schema_version))) {
205
    LOG_WARN("fail to get start schema version", KR(ret));
206
  } else if (OB_FAIL(trans_.serialize_inc_schemas(start_schema_version  - 1))) {
207
    LOG_WARN("fail to serialize inc schemas", KR(ret), K_(tenant_id),
208
             "start_schema_version", start_schema_version - 1);
209
  }
210
  RS_TRACE(inc_schema_dict);
211
  return ret;
212
}
213

214
int ObDDLHelper::wait_ddl_trans_()
215
{
216
  int ret = OB_SUCCESS;
217
  ObTimeoutCtx ctx;
218
  const int64_t DEFAULT_TS = 10 * 1000 * 1000L; // 10s
219
  if (OB_FAIL(check_inner_stat_())) {
220
    LOG_WARN("fail to check inner stat", KR(ret));
221
  } else if (OB_FAIL(ObShareUtil::set_default_timeout_ctx(ctx, DEFAULT_TS))) {
222
    LOG_WARN("fail to set default ts", KR(ret));
223
  } else if (OB_FAIL(ddl_trans_controller_->wait_task_ready(tenant_id_, task_id_, ctx.get_timeout()))) {
224
    LOG_WARN("fail to wait ddl trans", KR(ret), K_(tenant_id), K_(task_id));
225
  }
226
  RS_TRACE(wait_ddl_trans);
227
  return ret;
228
}
229

230
// this function should be always called
231
int ObDDLHelper::end_ddl_trans_(const int return_ret)
232
{
233
  int ret = return_ret;
234

235
  // write 1503 ddl operation
236
  if (OB_SUCC(ret)) {
237
    auto *tsi_generator = GET_TSI(TSISchemaVersionGenerator);
238
    int64_t version_cnt = OB_INVALID_INDEX;
239
    int64_t boundary_schema_version = OB_INVALID_VERSION;
240
    share::schema::ObSchemaService *schema_service_impl = NULL;
241
    if (OB_ISNULL(tsi_generator)) {
242
      ret = OB_ERR_UNEXPECTED;
243
      LOG_WARN("tsi schema version generator is null", KR(ret));
244
    } else if (OB_FAIL(tsi_generator->get_version_cnt(version_cnt))) {
245
      LOG_WARN("fail to get version cnt", KR(ret), K(version_cnt));
246
    } else if (0 == version_cnt) {
247
      // no schema change, just skip
248
    } else if (OB_UNLIKELY(version_cnt < 2)) {
249
      ret = OB_ERR_UNEXPECTED;
250
      LOG_WARN("not enough version cnt for boudary ddl operation", KR(ret), K(version_cnt));
251
    } else if (OB_ISNULL(schema_service_)
252
               || OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
253
      ret = OB_ERR_UNEXPECTED;
254
      LOG_WARN("ptr is null", KR(ret), KP_(schema_service));
255
    } else if (OB_FAIL(schema_service_->gen_new_schema_version(tenant_id_, boundary_schema_version))) {
256
      LOG_WARN("fail to gen new schema version", KR(ret), K_(tenant_id));
257
    } else {
258
      share::schema::ObDDLSqlService ddl_sql_service(*schema_service_impl);
259
      obrpc::ObDDLNopOpreatorArg arg;
260
      arg.schema_operation_.op_type_ = OB_DDL_END_SIGN;
261
      arg.schema_operation_.tenant_id_ = tenant_id_;
262
      if (OB_FAIL(ddl_sql_service.log_nop_operation(arg.schema_operation_,
263
                                                    boundary_schema_version,
264
                                                    NULL,
265
                                                    trans_))) {
266
        LOG_WARN("fail to log ddl operation", KR(ret), K(arg));
267
      }
268
    }
269
  }
270

271
  if (trans_.is_started()) {
272
    int tmp_ret = OB_SUCCESS;
273
    bool is_commit = OB_SUCC(ret);
274
    if (OB_TMP_FAIL(trans_.end(is_commit))) {
275
      LOG_WARN("trans end failed", KR(ret), KR(tmp_ret), K(is_commit));
276
      ret = is_commit ? tmp_ret : ret;
277
    }
278
  }
279
  if (OB_NOT_NULL(ddl_trans_controller_) && OB_INVALID_ID != task_id_) {
280
    ddl_trans_controller_->remove_task(tenant_id_, task_id_);
281
  }
282
  RS_TRACE(end_ddl_trans);
283
  return ret;
284
}
285

286
int ObDDLHelper::execute()
287
{
288
  return OB_NOT_IMPLEMENT;
289
  /*
290
   * Implement of parallel ddl should has following actions:
291
   *
292
   * ----------------------------------------------
293
   * 1. start ddl trans:
294
   * - to be exclusive with non-parallel ddl.
295
   * - to be concurrent with other parallel ddl.
296
   *
297
   * if (OB_FAIL(start_ddl_trans_())) {
298
   *   LOG_WARN("fail to start ddl trans", KR(ret));
299
   * }
300
   *
301
   * ----------------------------------------------
302
   * 2. lock object by name/object_id
303
   * - to be exclusive with other parallel ddl which involving the same objects.
304
   * - lock object in trans
305
   * Attension:
306
   * 1) lock objects just for mutual exclusion, should check if related objects changed after acquire locks.
307
   * 2) For same object, lock object by name first. After that, lock object by id if it's neccessary.
308
   *
309
   * ----------------------------------------------
310
   * 3. fetch & generate schema:
311
   * - fetch the latest schemas from inner table.
312
   * - generate schema with arg and the latests schemas.
313
   *
314
   * ----------------------------------------------
315
   * 4. register task id & generate schema versions:
316
   * - generate an appropriate number of schema versions for this DDL and register task id.
317
   * - concurrent DDL trans will be committed in descending order of version later.
318
   *
319
   * if (FAILEDx(gen_task_id_and_schema_versions_())) {
320
   *   LOG_WARN("fail to gen task id and schema versions", KR(ret));
321
   * }
322
   *
323
   * ----------------------------------------------
324
   * 5. create schema:
325
   * - persist schema in inner table.
326
   *
327
   * ----------------------------------------------
328
   * 6. [optional] serialize increment data dictionary:
329
   * - if table/database/tenant schema changed, records changed schemas in log and commits with DDL trans.
330
   *
331
   * if (FAILEDx(serialize_inc_schema_dict_())) {
332
   *   LOG_WARN("fail to serialize inc schema dict", KR(ret));
333
   * }
334
   *
335
   * ----------------------------------------------
336
   * 7. wait concurrent ddl trans ended:
337
   * - wait concurrent DDL trans with smallest schema version ended.
338
   *
339
   * if (FAILEDx(wait_ddl_trans_())) {
340
   *   LOG_WARN(fail to wait ddl trans, KR(ret));
341
   * }
342
   *
343
   * ----------------------------------------------
344
   * 8. end ddl trans:
345
   * - abort/commit ddl trans.
346
   *
347
   * if (OB_FAIL(end_ddl_trans_(ret))) { // won't overwrite ret
348
   *   LOG_WARN("fail to end ddl trans", KR(ret));
349
   * }
350
   */
351
}
352

353
int ObDDLHelper::add_lock_object_to_map_(
354
    const uint64_t lock_obj_id,
355
    const transaction::tablelock::ObTableLockMode lock_mode,
356
    ObjectLockMap &lock_map)
357
{
358

359
  int ret = OB_SUCCESS;
360
  if (OB_FAIL(check_inner_stat_())) {
361
    LOG_WARN("fail to check inner stat", KR(ret));
362
  } else if (OB_UNLIKELY(transaction::tablelock::SHARE != lock_mode
363
             && transaction::tablelock::EXCLUSIVE != lock_mode)) {
364
    ret = OB_NOT_SUPPORTED;
365
    LOG_WARN("not support lock mode to lock object by name", KR(ret), K(lock_mode));
366
  } else {
367
    bool need_update = false;
368
    transaction::tablelock::ObTableLockMode existed_lock_mode = transaction::tablelock::MAX_LOCK_MODE;
369
    if (OB_FAIL(lock_map.get_refactored(lock_obj_id, existed_lock_mode))) {
370
      if (OB_HASH_NOT_EXIST == ret) {
371
        ret = OB_SUCCESS;
372
        need_update = true;
373
      } else {
374
        LOG_WARN("fail to get lock object from map", KR(ret), K(lock_obj_id));
375
      }
376
    } else if (transaction::tablelock::SHARE == existed_lock_mode
377
               && transaction::tablelock::EXCLUSIVE == lock_mode) {
378
      // upgrade lock
379
      need_update = true;
380
    }
381

382
    if (OB_SUCC(ret) && need_update) {
383
      int overwrite = 1;
384
      if (OB_FAIL(lock_map.set_refactored(lock_obj_id, lock_mode, overwrite))) {
385
        LOG_WARN("fail to set lock object to map", KR(ret), K(lock_obj_id), K(lock_mode));
386
      }
387
    }
388
  }
389
  return ret;
390
}
391

392
int ObDDLHelper::lock_objects_in_map_(
393
    const transaction::tablelock::ObLockOBJType obj_type,
394
    ObjectLockMap &lock_map)
395
{
396
  int ret = OB_SUCCESS;
397
  ObArray<ObLockObjPair> lock_pairs;
398
  const int64_t lock_cnt = lock_map.size();
399
  ObTimeoutCtx ctx;
400
  observer::ObInnerSQLConnection *conn = NULL;
401
  if (OB_FAIL(check_inner_stat_())) {
402
    LOG_WARN("fail to check inner stat", KR(ret));
403
  } else if (OB_UNLIKELY(lock_cnt < 0)) {
404
    ret = OB_INVALID_ARGUMENT;
405
    LOG_WARN("unexpected lock cnt", KR(ret), K(lock_cnt));
406
  } else if (0 == lock_cnt) {
407
    // skip
408
  } else if (OB_FAIL(lock_pairs.reserve(lock_cnt))) {
409
    LOG_WARN("fail to reserve lock pairs", KR(ret), K(lock_cnt));
410
  } else if (OB_FAIL(ObShareUtil::set_default_timeout_ctx(ctx, GCONF.rpc_timeout))) {
411
    LOG_WARN("fail to set timeout ctx", KR(ret));
412
  } else if (OB_ISNULL(conn = dynamic_cast<observer::ObInnerSQLConnection *>
413
                       (trans_.get_connection()))) {
414
    ret = OB_ERR_UNEXPECTED;
415
    LOG_WARN("trans conn is NULL", KR(ret));
416
  } else {
417
    ObLockObjPair pair;
418
    FOREACH_X(it, lock_map, OB_SUCC(ret)) {
419
      if (OB_FAIL(pair.init(it->first, it->second))) {
420
        LOG_WARN("fail to init lock pair", KR(ret),
421
                 "obj_id", it->first, "lock_mode", it->second);
422
      } else if (OB_FAIL(lock_pairs.push_back(pair))) {
423
        LOG_WARN("fail to push back lock pair", KR(ret), K(pair));
424
      }
425
    } // end foreach
426
    if (OB_SUCC(ret)) {
427
      std::sort(lock_pairs.begin(), lock_pairs.end(), ObLockObjPair::less_than);
428
      FOREACH_X(it, lock_pairs, OB_SUCC(ret)) {
429
        const int64_t timeout = ctx.get_timeout();
430
        if (OB_UNLIKELY(timeout <= 0)) {
431
          ret = OB_TIMEOUT;
432
          LOG_WARN("already timeout", KR(ret), K(timeout));
433
        } else {
434
          transaction::tablelock::ObLockObjRequest lock_arg;
435
          lock_arg.obj_type_ = obj_type;
436
          lock_arg.owner_id_ = ObTableLockOwnerID(0);
437
          lock_arg.obj_id_ = it->get_obj_id();
438
          lock_arg.lock_mode_ = it->get_lock_mode();
439
          lock_arg.op_type_ = ObTableLockOpType::IN_TRANS_COMMON_LOCK;
440
          lock_arg.timeout_us_ = timeout;
441
          LOG_INFO("try lock object", KR(ret), K(lock_arg));
442
          if (OB_FAIL(ObInnerConnectionLockUtil::lock_obj(tenant_id_, lock_arg, conn))) {
443
            LOG_WARN("lock obj failed", KR(ret), K_(tenant_id), K(lock_arg));
444
          }
445
        }
446
      } // end foreach
447
    }
448
  }
449
  (void) lock_map.clear();
450
  return ret;
451
}
452

453
int ObDDLHelper::add_lock_object_by_database_name_(
454
    const ObString &database_name,
455
    const transaction::tablelock::ObTableLockMode lock_mode)
456
{
457
  int ret = OB_SUCCESS;
458
  if (OB_FAIL(check_inner_stat_())) {
459
    LOG_WARN("fail to check inner stat", KR(ret));
460
  } else if (OB_UNLIKELY(database_name.empty())) {
461
    ret = OB_INVALID_ARGUMENT;
462
    LOG_WARN("database_name is invalid", KR(ret), K(database_name));
463
  } else {
464
    // use OB_ORIGIN_AND_INSENSITIVE and ignore end space to make more conficts for safety.
465
    common::ObCollationType cs_type = ObSchema::get_cs_type_with_cmp_mode(OB_ORIGIN_AND_INSENSITIVE);
466
    bool calc_end_space = false;
467
    uint64_t lock_obj_id = 0;
468
    lock_obj_id = common::ObCharset::hash(
469
                  cs_type, database_name.ptr(), database_name.length(),
470
                  lock_obj_id, calc_end_space, NULL);
471
    if (OB_FAIL(add_lock_object_to_map_(lock_obj_id, lock_mode, lock_database_name_map_))) {
472
      LOG_WARN("fail to add lock object to map", KR(ret), K(lock_obj_id), K(lock_mode));
473
    }
474
    LOG_INFO("add lock object by database name", KR(ret), K(database_name), K(lock_mode), K(lock_obj_id));
475
  }
476
  return ret;
477
}
478

479
int ObDDLHelper::lock_databases_by_name_()
480
{
481
  return lock_objects_in_map_(ObLockOBJType::OBJ_TYPE_DATABASE_NAME, lock_database_name_map_);
482
}
483

484
int ObDDLHelper::add_lock_object_by_name_(
485
    const ObString &database_name,
486
    const ObString &object_name,
487
    const share::schema::ObSchemaType schema_type,
488
    const transaction::tablelock::ObTableLockMode lock_mode)
489
{
490
  int ret = OB_SUCCESS;
491
  if (OB_FAIL(check_inner_stat_())) {
492
    LOG_WARN("fail to check inner stat", KR(ret));
493
  } else if (OB_UNLIKELY(database_name.empty() || object_name.empty())) {
494
    ret = OB_INVALID_ARGUMENT;
495
    LOG_WARN("database_name/object_name is invalid", KR(ret), K(database_name), K(object_name));
496
  } else {
497
    // 1. use OB_ORIGIN_AND_INSENSITIVE and ignore end space to make more conficts for safety.
498
    // 2. encoded with database name to make less conficts between different databases/users.
499
    common::ObCollationType cs_type = ObSchema::get_cs_type_with_cmp_mode(OB_ORIGIN_AND_INSENSITIVE);
500
    bool calc_end_space = false;
501
    uint64_t lock_obj_id = 0;
502
    lock_obj_id = common::ObCharset::hash(
503
                  cs_type, database_name.ptr(), database_name.length(),
504
                  lock_obj_id, calc_end_space, NULL);
505
    lock_obj_id = common::ObCharset::hash(
506
                  cs_type, object_name.ptr(), object_name.length(),
507
                  lock_obj_id, calc_end_space, NULL);
508
    if (OB_FAIL(add_lock_object_to_map_(lock_obj_id, lock_mode, lock_object_name_map_))) {
509
      LOG_WARN("fail to add lock object to map", KR(ret), K(lock_obj_id), K(lock_mode));
510
    }
511
    LOG_INFO("add lock object by name", KR(ret), K(database_name),
512
             K(object_name), K(schema_type), K(lock_mode), K(lock_obj_id));
513
  }
514
  return ret;
515
}
516

517
int ObDDLHelper::lock_existed_objects_by_name_()
518
{
519
  return lock_objects_in_map_(ObLockOBJType::OBJ_TYPE_OBJECT_NAME, lock_object_name_map_);
520
}
521

522
int ObDDLHelper::add_lock_object_by_id_(
523
    const uint64_t lock_obj_id,
524
    const share::schema::ObSchemaType schema_type,
525
    const transaction::tablelock::ObTableLockMode lock_mode)
526
{
527
  int ret = OB_SUCCESS;
528
  if (OB_FAIL(check_inner_stat_())) {
529
    LOG_WARN("fail to check inner stat", KR(ret));
530
  } else if (OB_UNLIKELY(OB_INVALID_ID == lock_obj_id)) {
531
    ret = OB_INVALID_ARGUMENT;
532
    LOG_WARN("object_id is invalid", KR(ret), K(lock_obj_id));
533
  } else if (OB_FAIL(add_lock_object_to_map_(lock_obj_id, lock_mode, lock_object_id_map_))) {
534
    LOG_WARN("fail to add lock object to map", KR(ret), K(lock_obj_id), K(lock_mode));
535
  }
536
  LOG_INFO("add lock object by id", KR(ret), K(lock_obj_id), K(schema_type), K(lock_mode));
537
  return ret;
538
}
539

540
int ObDDLHelper::lock_existed_objects_by_id_()
541
{
542
  return lock_objects_in_map_(ObLockOBJType::OBJ_TYPE_COMMON_OBJ, lock_object_id_map_);
543
}
544

545
// 1. constraint name and foreign key name are in the same namespace in oracle tenant.
546
// 2. constraint name and foreign key name are in different namespace in mysql tenant.
547
int ObDDLHelper::check_constraint_name_exist_(
548
    const share::schema::ObTableSchema &table_schema,
549
    const common::ObString &constraint_name,
550
    const bool is_foreign_key,
551
    bool &exist)
552
{
553
  int ret = OB_SUCCESS;
554
  bool is_oracle_mode = false;
555
  uint64_t constraint_id = OB_INVALID_ID;
556
  const uint64_t database_id = table_schema.get_database_id();
557
  exist = false;
558
  if (OB_FAIL(check_inner_stat_())) {
559
    LOG_WARN("fail to check inner stat", KR(ret));
560
  } else if (OB_FAIL(table_schema.check_if_oracle_compat_mode(is_oracle_mode))) {
561
    LOG_WARN("check if oracle compat mode failed", KR(ret), K_(tenant_id));
562
  } else {
563
    const bool check_fk = (is_oracle_mode || is_foreign_key);
564
    if (OB_SUCC(ret) && check_fk) {
565
      if (OB_FAIL(latest_schema_guard_.get_foreign_key_id(
566
          database_id, constraint_name, constraint_id))) {
567
        LOG_WARN("fail to get foreign key id", KR(ret), K_(tenant_id), K(database_id), K(constraint_name));
568
      } else if (OB_INVALID_ID != constraint_id) {
569
        exist = true;
570
      }
571
    }
572
    const bool check_cst = (is_oracle_mode || !is_foreign_key);
573
    if (OB_SUCC(ret) && !exist && check_cst) {
574
      if (table_schema.is_mysql_tmp_table()) {
575
        // tmp table in mysql mode, do nothing
576
      } else if (OB_FAIL(latest_schema_guard_.get_constraint_id(
577
          database_id, constraint_name, constraint_id))) {
578
        LOG_WARN("fail to get constraint id", KR(ret), K_(tenant_id), K(database_id), K(constraint_name));
579
      } else if (OB_INVALID_ID != constraint_id) {
580
        exist = true;
581
      }
582
    }
583
  }
584
  return ret;
585
}
586

587
int ObDDLHelper::gen_object_ids_(
588
    const int64_t object_cnt,
589
    share::ObIDGenerator &id_generator)
590
{
591
  int ret = OB_SUCCESS;
592
  if (OB_FAIL(check_inner_stat_())) {
593
    LOG_WARN("fail to check inner stat", KR(ret));
594
  } else if (OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id_
595
             || object_cnt < 0)) {
596
    ret = OB_INVALID_ARGUMENT;
597
    LOG_WARN("invalid tenant_id or object_cnt", KR(ret), K_(tenant_id), K(object_cnt));
598
  } else if (0 == object_cnt) {
599
    // skip
600
  } else {
601
    uint64_t max_object_id = OB_INVALID_ID;
602
    uint64_t min_object_id = OB_INVALID_ID;
603
    share::schema::ObSchemaService *schema_service_impl = NULL;
604
    if (OB_ISNULL(schema_service_)
605
        || OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
606
      ret = OB_ERR_UNEXPECTED;
607
      LOG_WARN("ptr is null", KR(ret), KP_(schema_service));
608
    } else if (OB_FAIL(schema_service_impl->fetch_new_object_ids(tenant_id_, object_cnt, max_object_id))) {
609
      LOG_WARN("fail to fetch new object ids", KR(ret), K_(tenant_id), K(object_cnt));
610
    } else if (OB_UNLIKELY(OB_INVALID_ID == max_object_id)) {
611
      ret = OB_ERR_UNEXPECTED;
612
      LOG_WARN("object_id is invalid", KR(ret), K_(tenant_id), K(object_cnt));
613
    } else if (0 >= (min_object_id = max_object_id - object_cnt + 1)) {
614
      ret = OB_ERR_UNEXPECTED;
615
      LOG_WARN("min_object_id should be greator than 0",
616
               KR(ret), K_(tenant_id), K(min_object_id), K(max_object_id), K(object_cnt));
617
    } else if (OB_FAIL(id_generator.init(1 /*step*/, min_object_id, max_object_id))) {
618
      LOG_WARN("fail to init id generator", KR(ret), K_(tenant_id),
619
               K(min_object_id), K(max_object_id), K(object_cnt));
620
    }
621
  }
622
  return ret;
623
}
624

625
int ObDDLHelper::gen_partition_object_and_tablet_ids_(
626
    ObIArray<ObTableSchema> &table_schemas)
627
{
628
  int ret = OB_SUCCESS;
629
  if (OB_FAIL(check_inner_stat_())) {
630
    LOG_WARN("fail to check inner stat", KR(ret));
631
  } else if (OB_FAIL(ddl_service_->generate_object_id_for_partition_schemas(table_schemas))) {
632
    LOG_WARN("fail to generate object_ids", KR(ret));
633
  } else if (OB_FAIL(ddl_service_->generate_tables_tablet_id(table_schemas))) {
634
    LOG_WARN("fail to generate tablet_ids", KR(ret));
635
  }
636
  return ret;
637
}
638