Синхронный QueryClient с более удобным API #342

Gazizonoki · 2024-11-18T21:20:24Z

Gazizonoki
Nov 18, 2024
Maintainer

Дизайн TQueryClient

Текущий Query Client - это низкоуровневый интерфейс, который дает пользователю TSession и TTxControl для прямого использования, вводит обработку ошибок через TStatus. Это приводит к boilerplate'у у пользователя. Также при синхронном использовании асинхронного API TQueryClient получается колбаса из вызовов функций.

Такой интерфейс может быть актуален в очень высоконагруженных приложениях, где нужен ручной контроль, но большинству пользователей он создает только лишние сложности. Поэтому предлагается:

Сделать синхронный клиент, более дружелюбный к пользователю для решения большинства задач пользователя: NYdb::NQuerySync::TQueryClient.
Порефакторить текущий API асинхронного клиента, чтобы он тоже был ближе к API синхронного клиента.

В синхронном клиенте:

На серверные ошибки кидается исключение TDbRequestError.
Не даем пользователю сессии напрямую, используем обертки, которые нельзя копировать и перемещать.
Прокси объект для ретраев - TRetryHandle
Вместо TTxControl транзакция создается с объектом TTransaction и действует на протяжении его лайфтайма. (Как в libpqxx)

В aсинхронном клиенте:

Объявить TTxControl deprecated
Объявить RetryQuery с клиентом в качестве объекта ретрая deprecated
В вызовах ExecuteQuery() на клиенте использовать явную сессию из пула
Поддержать ExecuteQuery() на объекте транзакции

{% list tabs %}

Старый клиент

class TQueryClient {
public:
    TQueryClient(const TDriver& driver, const TClientSettings& settings = TClientSettings());

    TAsyncExecuteQueryResult ExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncExecuteQueryResult ExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TParams& params, const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncExecuteQueryIterator StreamExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncExecuteQueryIterator StreamExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TParams& params, const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncExecuteQueryResult RetryQuery(TQueryResultFunc&& queryFunc, TRetryOperationSettings settings = TRetryOperationSettings());

    TAsyncStatus RetryQuery(TQueryFunc&& queryFunc, TRetryOperationSettings settings = TRetryOperationSettings());

    TAsyncStatus RetryQuery(TQueryWithoutSessionFunc&& queryFunc, TRetryOperationSettings settings = TRetryOperationSettings());

    TStatus RetryQuerySync(const TQuerySyncFunc& queryFunc, TRetryOperationSettings settings = TRetryOperationSettings());

    TStatus RetryQuerySync(const TQueryWithoutSessionSyncFunc& queryFunc, TRetryOperationSettings settings = TRetryOperationSettings());

    TAsyncExecuteQueryResult RetryQuery(const std::string& query, const TTxControl& txControl,
        TDuration timeout, bool isIndempotent);

    NThreading::TFuture<TScriptExecutionOperation> ExecuteScript(const std::string& script,
        const TExecuteScriptSettings& settings = TExecuteScriptSettings());

    NThreading::TFuture<TScriptExecutionOperation> ExecuteScript(const std::string& script,
        const TParams& params, const TExecuteScriptSettings& settings = TExecuteScriptSettings());

    TAsyncFetchScriptResultsResult FetchScriptResults(const NKikimr::NOperationId::TOperationId& operationId, int64_t resultSetIndex,
        const TFetchScriptResultsSettings& settings = TFetchScriptResultsSettings());

    TAsyncCreateSessionResult GetSession(const TCreateSessionSettings& settings = TCreateSessionSettings());

    //! Returns number of active sessions given via session pool
    int64_t GetActiveSessionCount() const;

    //! Returns the maximum number of sessions in session pool
    int64_t GetActiveSessionsLimit() const;

    //! Returns the size of session pool
    int64_t GetCurrentPoolSize() const;
};

class TSession {
public:
    const std::string& GetId() const;

    TAsyncExecuteQueryResult ExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncExecuteQueryResult ExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TParams& params, const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncExecuteQueryIterator StreamExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncExecuteQueryIterator StreamExecuteQuery(const std::string& query, const TTxControl& txControl,
        const TParams& params, const TExecuteQuerySettings& settings = TExecuteQuerySettings());

    TAsyncBeginTransactionResult BeginTransaction(const TTxSettings& txSettings,
        const TBeginTxSettings& settings = TBeginTxSettings());
};

class TTransaction {
public:
    const std::string& GetId() const;

    bool IsActive() const;

    TAsyncCommitTransactionResult Commit(const TCommitTxSettings& settings = TCommitTxSettings());
    TAsyncStatus Rollback(const TRollbackTxSettings& settings = TRollbackTxSettings());

    TSession GetSession() const;
};

Новый синхронный клиент

namespace NYdb::NQuerySync {

struct TQueryResult;
struct TStreamQueryResult;

class TTransaction {
public:
    TQueryResult ExecuteQuery(const std::string& query, const TExecuteQueryInTxSettings& settings = {});
    TStreamQueryResult StreamExecuteQuery(const std::string& query, const TExecuteQueryInTxSettings& settings = {});

    void Commit(const TCommitTxSettings& settings = {});
    void Rollback(const TRollbackTxSettings& settings = {});

    ~TTransaction(); // Commit
};

class TRetryHandle {
public:
    TTransaction BeginTransaction(const TBeginTxSettings& settings = {});

    TQueryResult ExecuteQuery(const std::string& query,
                              const TParams& params = {},
                              const TTxSettings& settings = {},
                              const TExecuteQuerySettings& settings = {});

    TStreamQueryResult StreamExecuteQuery(const std::string& query,
                                          const TParams& params = {},
                                          const TTxSettings& settings = {},
                                          const TExecuteQuerySettings& settings = {});
};

class TQueryClient {
public:
    TQueryClient(const TDriver& driver, const TClientSettings& settings = TClientSettings());

    void RetryQuery(std::function<void(TRetryHandle& handle)>, const NRetry::TRetryOperationSettings& settings = {});
    void RetryTransaction(std::function<void(TTransaction& tx)>, const TTxSettings& txSettings = {}, const NRetry::TRetryOperationSettings& settings = {});

    TTransaction BeginTransaction(const TTxSettings& settings = {});

    TQueryResult ExecuteQuery(const std::string& query,
                              const TParams& params = {},
                              const TTxSettings& settings = {},
                              const TExecuteQuerySettings& settings = {});

    TStreamQueryResult StreamExecuteQuery(const std::string& query,
                                          const TParams& params = {},
                                          const TTxSettings& settings = {},
                                          const TExecuteQuerySettings& settings = {});
};

}

Новый асинхронный клиент

class TQueryClient {
public:
    TAsyncBeginTransactionResult BeginTransaction(const TTxSettings& txSettings,
        const TBeginTxSettings& settings = TBeginTxSettings());

    TAsyncCreateSessionResult GetSession(const TCreateSessionSettings& settings = {});

    TAsyncStatus RetryQuery(std::function<TAsyncStatus(TSession)>&& queryFunc,
                            const TRetryOperationSettings& settings = {});

    TAsyncStatus RetryTransaction(std::function<TAsyncStatus(TTransaction)>&& txFunc,
                                  const TTxSettings& txSettings = {},
                                  const TRetryOperationSettings& settings = {});

    // DEPRECATED METHODS
    // ...

    TAsyncExecuteQueryResult ExecuteQuery(const std::string& query,
        const TParams& params = {}, const TTxSettings& txSettings = {}, const TExecuteQuerySettings& settings = {});

    TAsyncExecuteQueryIterator StreamExecuteQuery(const std::string& query,
        const TParams& params = {}, const TTxSettings& txSettings = {}, const TExecuteQuerySettings& settings = {});
};

class TSession {
public:
    TAsyncBeginTransactionResult BeginTransaction(const TTxSettings& txSettings,
        const TBeginTxSettings& settings = {});

    // DEPRECATED METHODS
    // ...

    TAsyncExecuteQueryResult ExecuteQuery(const std::string& query,
        const TParams& params = {}, const TTxSettings& txSettings = {}, const TExecuteQuerySettings& settings = {});

    TAsyncExecuteQueryIterator StreamExecuteQuery(const std::string& query,
        const TParams& params = {}, const TTxSettings& txSettings = {}, const TExecuteQuerySettings& settings = {});
};

class TTransaction {
public:
    TAsyncExecuteQueryResult ExecuteQuery(const std::string& query,
        const TParams& params = {}, const TExecuteQueryInTxSettings& settings = {});

    TAsyncExecuteQueryIterator StreamExecuteQuery(const std::string& query,
        const TParams& params = {}, const TExecuteQueryInTxSettings& settings = {});

    TAsyncCommitTransactionResult Commit(const TCommitTxSettings& settings = {});
    TAsyncStatus Rollback(const TRollbackTxSettings& settings = {});
};

{% endlist %}

Один запрос в транзакции (на явной сессии)

При выполнении запроса на синхронном клиенте используется явная сессия из пула, а не посылается запрос с пустым session id (неявная сессия). Такой запрос исполняется в одной транзакции, настройки транзакции (TTxSettings) - опциональный параметр.

{% list tabs %}

Старый клиент

TDriver driver{"grpc://localhost:2136/?database=local"};
TQueryClient client{driver};
auto getSessRes = client.GetSession().GetValueSync();
if (!getSessRes.IsSuccess()) {
    std::cerr << ToString(getSessRes);
}
auto session = getSessRes.GetSession();
auto result = session.ExecuteQuery("SELECT 1;", TTxControl::BeginTx().CommitTx()).GetValueSync();
if (!result.IsSuccess()) {
    std::cerr << ToString(result);
}
// parse result

Новый синхронный клиент

TDriver driver{"grpc://localhost:2136/local"};
TQueryClient client{driver};
try {
    auto result = client.ExecuteQuery("SELECT 1;");
    // parse result
} catch (NYdb::TDbRequestError& e) {
    std::cerr << e;
}

{% endlist %}

Асинхронный запрос

При выполнении запроса на асинхронном клиенте теперь также используется явная сессия из пула. Поэтому этап получения сессии можно пропустить. Запрос будет исполнен в 1 транзакции, настройки можно передать через опциональный параметр.

{% list tabs %}

Старый клиент

TDriver driver{"grpc://localhost:2136/?database=local"};
TQueryClient client{driver};
auto asyncStatus = client.GetSession().Apply([](auto result) -> TStatus {
    if (!result.GetValue().IsSuccess()) {
        return result;
    }

    auto session = result.GetValue().GetSession();

    return session.ExecuteQuery("SELECT 1;", TTxControl::BeginTx().CommitTx()).Apply([](auto result) -> TStatus {
        if (!result.GetValue().IsSuccess()) {
            return result;
        }
    
        // parse result
    });
});

////////////////////////////////////////////////////////

auto status = asyncStatus.GetValueSync();
if (!status.IsSuccess()) {
    std::cerr << ToString(status);
}

Новый aсинхронный клиент

TDriver driver{"grpc://localhost:2136/local"};
TQueryClient client{driver};
auto asyncStatus = client.ExecuteQuery("SELECT 1;").Apply([](auto result) -> TStatus {
    if (!result.GetValue().IsSuccess()) {
        return result;
    }

    // parse result
});

////////////////////////////////////////////////////////
  
auto status = asyncStatus.GetValueSync();
if (!status.IsSuccess()) {
    std::cerr << ToString(status);
}

{% endlist %}

Ретрай запроса

Раньше ретраить можно было либо на клиенте (неявные сессии), либо на сессии.
В синхронном клиенте теперь ретраи можно делать только через TRetryHandle. Это некопируемый и неперемещаемый объект, инициализируется сессией и все запросы перенаправляет в нее.

{% list tabs %}

Старый клиент

TDriver driver{"grpc://localhost:2136/?database=local"};
TQueryClient client{driver};
auto result = client.RetryQuery([](TSession session) {
    return session.ExecuteQuery("SELECT 1;", TTxControl::BeginTx().CommitTx());
}).GetValueSync();

if (!result.IsSuccess()) {
    std::cerr << ToString(result);
}
// parse result

Новый синхронный клиент

TDriver driver{"grpc://localhost:2136/local"};
TQueryClient client{driver};
try {
    client.RetryQuery([](TRetryHandle& handle) {
        auto result = handle.ExecuteQuery("SELECT 1;");
        // parse result
    });
} catch (NYdb::TDbRequestError& e) {
    std::cerr << e;
}

{% endlist %}

Интерактивная транзакция

Для интерактивной транзакции используется объект TTransaction. Он некопируемый и неперемещаемый. Семантика у него такая же, как у транзакции в libpqxx, транзакция активна в переделах скоупа этого объекта. В деструкторе вызывается Rollback(). Если Rollback() завершился ошибкой, сессия выкидывается из пула.

Сделать коммит можно 2 способами - явный вызов Commit() или передав Commit флаг в настройках. Во втором случае в запросе будет передан TTxControl::Tx().Commit().

Для удобства объект транзакции можно создать вызовом метода BeginTransaction() прямо из клиента и TRetryHandle. Для BeginTransaction() можно выставить режим неявного начала транзакции. Тогда в первом запросе ExecuteQuery() будет передан TTxControl::BeginTx().

BeginTransaction() можно сделать сразу на клиенте. Тогда будет использована сессия из пула.

{% list tabs %}

Старый клиент

TDriver driver{"grpc://localhost:2136/?database=local"};
TQueryClient client{driver};
auto getSessRes = client.GetSession().GetValueSync();
if (!getSessRes.IsSuccess()) {
    std::cerr << ToString(getSessRes);
}
auto session = getSessRes.GetSession();
auto result1 = session.ExecuteQuery("SELECT 1;", TTxControl::BeginTx()).GetValueSync();
if (!result1.IsSuccess()) {
    std::cerr << ToString(result1);
}
// parse result
auto result2 = session.ExecuteQuery("SELECT 2;", TTxControl::Tx(result.GetTransaction()).CommitTx()).GetValueSync();
if (!result2.IsSuccess()) {
    std::cerr << ToString(result2);
}
// parse result

Новый синхронный клиент

TDriver driver{"grpc://localhost:2136/local"};
TQueryClient client{driver};
try {
    auto tx = client.BeginTransaction();
    auto result1 = tx.ExecuteQuery("SELECT 1;");
    // parse result
    auto result2 = tx.ExecuteQuery("SELECT 2;", {}, ExecuteInTxSettings().Commit());
    // parse result
} catch (NYdb::TDbRequestError& e) {
    std::cerr << e;
}

{% endlist %}

Асинхронная интерактивная транзакция

В отличие от старого клиента, новый будет создавать транзакции только через BeginTransaction(). В нем можно выставить режим неявного начала транзакции. Тогда в первом запросе ExecuteQuery() будет передан TTxControl::BeginTx(). Запросы выполняются через объект транзакции.

Сделать коммит можно 2 способами - явный вызов Commit() или передав Commit флаг в настройках. Во втором случае в запросе будет передан TTxControl::Tx().Commit().

BeginTransaction() можно сделать сразу на клиенте. Тогда будет использована сессия из пула.

{% list tabs %}

Старый клиент

TDriver driver{"grpc://localhost:2136/?database=local"};
TQueryClient client{driver};
auto asyncStatus = client.GetSession().Apply([](auto result) -> TStatus {
    if (!result.GetValue().IsSuccess()) {
        return result;
    }

    auto session = result.GetValue().GetSession();

    return session.ExecuteQuery("SELECT 1;", TTxControl::Tx(tx)).Apply([session, tx](auto result) -> TStatus {
        if (!result.GetValue().IsSuccess()) {
            return result;
        }
    
        // parse result

        return session.ExecuteQuery("SELECT 2;", TTxControl::Tx(result.GetValue().GetTransaction()).CommitTx()).Apply([](auto result) -> TStatus {
            if (!result.GetValue().IsSuccess()) {
                return result;
            }

            // parse result
        });
    });
});

////////////////////////////////////////////////////////
      
auto status = asyncStatus.GetValueSync();
if (!status.IsSuccess()) {
    std::cerr << ToString(status);
}

Новый асинхронный клиент

TDriver driver{"grpc://localhost:2136/local"};
TQueryClient client{driver};
auto asyncStatus = client.BeginTransaction().Apply([](auto result) {
    auto tx = result.GetValue().GetTransaction();

    return tx.ExecuteQuery("SELECT 1;").Apply([session, tx](auto result) -> TStatus {
        if (!result.GetValue().IsSuccess()) {
            return result;
        }
    
        // parse result

        return tx.ExecuteQuery("SELECT 2;", {}, TExecuteQueryInTxSettings().Commit()).Apply([](auto result) -> TStatus {
            if (!result.GetValue().IsSuccess()) {
                return result;
            }

            // parse result
        });
    });
});

////////////////////////////////////////////////////////
    
auto status = asyncStatus.GetValueSync();
if (!status.IsSuccess()) {
    std::cerr << ToString(status);
}

{% endlist %}

Ретрай интерактивной транзакции

Для ретрая транзакции используется метод RetryTransaction(). В нем пользователю передается в использование сразу TTransaction. В настройках можно выставить режим неявного начала транзакции. Тогда в первом запросе ExecuteQuery() будет передан TTxControl::BeginTx().

{% list tabs %}

Старый клиент

TDriver driver{"grpc://localhost:2136/?database=local"};
TQueryClient client{driver};
auto status = client.RetryQuery([](TSession session) {
    auto result1 = session.ExecuteQuery("SELECT 1;", TTxControl::BeginTx()).GetValueSync();
    if (!result1.IsSuccess()) {
        std::cerr << ToString(result1);
        return result1;
    }
    // parse result
    auto result2 = session.ExecuteQuery("SELECT 2;", TTxControl::Tx(result.GetTransaction()).CommitTx()).GetValueSync();
    if (!result2.IsSuccess()) {
        std::cerr << ToString(result2);
        return result2;
    }
    // parse result
}).GetValueSync();
if (!status.IsSuccess()) {
    std::cerr << ToString(status);
}

Новый синхронный клиент

TDriver driver{"grpc://localhost:2136/local"};
TQueryClient client{driver};
try {
    client.RetryTransaction([](TTransaction& tx) {
        auto result1 = tx.ExecuteQuery("SELECT 1;");
        // parse result
        auto result2 = tx.ExecuteQuery("SELECT 2;");
        // parse result
    });
} catch (NYdb::TDbRequestError& e) {
    std::cerr << e;
}

{% endlist %}

Асинхронный ретрай интерактивной транзакции

Для ретрая транзакции используется метод RetryTransaction(). В нем пользователю передается в использование сразу TTransaction. В настройках можно выставить режим неявного начала транзакции. Тогда в первом запросе ExecuteQuery() будет передан TTxControl::BeginTx().

{% list tabs %}

Старый клиент

TDriver driver{"grpc://localhost:2136/?database=local"};
TQueryClient client{driver};
auto asyncStatus = client.RetryQuery([](TSession session) {
    return session.ExecuteQuery("SELECT 1;", TTxControl::BeginTx()).Apply([session](auto result) -> TStatus {
        if (!result.GetValue().IsSuccess()) {
            return result;
        }

        // parse result

        return session.ExecuteQuery("SELECT 2;", TTxControl::Tx(result.GetValue().GetTransaction()).CommitTx()).Apply([](auto result) -> TStatus {
            if (!result.GetValue().IsSuccess()) {
                return result;
            }

            // parse result
        });
    });
});

////////////////////////////////////////////////////////
      
auto status = asyncStatus.GetValueSync();
if (!status.IsSuccess()) {
    std::cerr << ToString(status);
}

Новый асинхронный клиент

TDriver driver{"grpc://localhost:2136/local"};
TQueryClient client{driver};
auto asyncStatus = client.RetryTransaction([](TTransaction tx) -> TStatus {
    return tx.ExecuteQuery("SELECT 1;").Apply([tx](auto result) -> TStatus {
        if (!result.GetValue().IsSuccess()) {
            return result;
        }
  
        // parse result

        return tx.ExecuteQuery("SELECT 2;", {}, TExecuteQueryInTxSettings().Commit()).Apply([](auto result) -> TStatus {
            if (!result.GetValue().IsSuccess()) {
                return result;
            }
      
            // parse result
        });
    });
});

////////////////////////////////////////////////////////
    
auto status = asyncStatus.GetValueSync();
if (!status.IsSuccess()) {
    std::cerr << ToString(status);
}

{% endlist %}

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Синхронный QueryClient с более удобным API #342

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Replies: 0 comments

Select a reply

Uh oh!

Синхронный QueryClient с более удобным API #342

Uh oh!

Uh oh!

Gazizonoki Nov 18, 2024 Maintainer

Дизайн TQueryClient

Один запрос в транзакции (на явной сессии)

Асинхронный запрос

Ретрай запроса

Интерактивная транзакция

Асинхронная интерактивная транзакция

Ретрай интерактивной транзакции

Асинхронный ретрай интерактивной транзакции

Replies: 0 comments

Gazizonoki
Nov 18, 2024
Maintainer