| 376 | |
| 377 | |
| 378 | String DatabaseReplicatedDDLWorker::enqueueQueryImpl(const ZooKeeperPtr & zookeeper, DDLLogEntry & entry, |
| 379 | DatabaseReplicated * const database, bool committed, Coordination::Requests additional_checks) |
| 380 | { |
| 381 | const String query_path_prefix = database->zookeeper_path + "/log/query-"; |
| 382 | |
| 383 | /// We cannot create sequential node and it's ephemeral child in a single transaction, so allocate sequential number another way |
| 384 | String counter_prefix = database->zookeeper_path + "/counter/cnt-"; |
| 385 | String counter_lock_path = database->zookeeper_path + "/counter_lock"; |
| 386 | |
| 387 | String counter_path; |
| 388 | size_t iters = 1000; |
| 389 | while (--iters) |
| 390 | { |
| 391 | Coordination::Requests ops; |
| 392 | ops.emplace_back(zkutil::makeCreateRequest(counter_lock_path, database->getFullReplicaName(), zkutil::CreateMode::Ephemeral)); |
| 393 | ops.emplace_back(zkutil::makeCreateRequest(counter_prefix, "", zkutil::CreateMode::EphemeralSequential)); |
| 394 | ops.insert(ops.end(), additional_checks.begin(), additional_checks.end()); |
| 395 | Coordination::Responses res; |
| 396 | |
| 397 | Coordination::Error code = zookeeper->tryMulti(ops, res); |
| 398 | if (code == Coordination::Error::ZOK) |
| 399 | { |
| 400 | counter_path = dynamic_cast<const Coordination::CreateResponse &>(*res[1]).path_created; |
| 401 | break; |
| 402 | } |
| 403 | if (res[0]->error != Coordination::Error::ZNODEEXISTS) |
| 404 | zkutil::KeeperMultiException::check(code, ops, res); |
| 405 | |
| 406 | sleepForMilliseconds(50); |
| 407 | } |
| 408 | |
| 409 | if (counter_path.empty()) |
| 410 | throw Exception(ErrorCodes::UNFINISHED, |
| 411 | "Cannot enqueue query, because some replica are trying to enqueue another query. " |
| 412 | "It may happen on high queries rate or, in rare cases, after connection loss. Client should retry."); |
| 413 | |
| 414 | String node_path = query_path_prefix + counter_path.substr(counter_prefix.size()); |
| 415 | |
| 416 | /// Now create task in queue |
| 417 | Coordination::Requests ops; |
| 418 | /// Query is not committed yet, but we have to write it into log to avoid reordering |
| 419 | ops.emplace_back(zkutil::makeCreateRequest(node_path, entry.toString(), zkutil::CreateMode::Persistent)); |
| 420 | /// '/try' will be replaced with '/committed' or will be removed due to expired session or other error |
| 421 | if (committed) |
| 422 | ops.emplace_back(zkutil::makeCreateRequest(node_path + "/committed", database->getFullReplicaName(), zkutil::CreateMode::Persistent)); |
| 423 | else |
| 424 | ops.emplace_back(zkutil::makeCreateRequest(node_path + "/try", database->getFullReplicaName(), zkutil::CreateMode::Ephemeral)); |
| 425 | /// We don't need it anymore |
| 426 | ops.emplace_back(zkutil::makeRemoveRequest(counter_path, -1)); |
| 427 | /// Unlock counters |
| 428 | ops.emplace_back(zkutil::makeRemoveRequest(counter_lock_path, -1)); |
| 429 | /// Create status dirs |
| 430 | ops.emplace_back(zkutil::makeCreateRequest(node_path + "/active", "", zkutil::CreateMode::Persistent)); |
| 431 | ops.emplace_back(zkutil::makeCreateRequest(node_path + "/finished", "", zkutil::CreateMode::Persistent)); |
| 432 | ops.emplace_back(zkutil::makeCreateRequest(node_path + "/synced", "", zkutil::CreateMode::Persistent)); |
| 433 | zookeeper->multi(ops); |
| 434 | |
| 435 |
nothing calls this directly
no test coverage detected