hanzo-ds/cpp
1
0
Fork 0
mirror of https://github.com/hanzo-ds/cpp.git synced 2026-10-30 20:31:26 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
cpp/ut/client_ut.cpp
T
zeekay 286f257aae De-brand the client: clickhouse-cpp -> datastore-cpp 
Pure rename applied on top of upstream tag v2.6.2 (8d2415d6), not HEAD.
The fork was cut at 737145d = v2.6.2-57, which still declared version 2.6.2
in version.h; renaming there would have shipped 57 unreleased commits wearing
a 2.6.2 label. Reset to the tag first so this commit changes names only.

  clickhouse/            -> datastore/            (public include root)
  namespace clickhouse   -> namespace datastore
  CLICKHOUSE_CPP_VERSION -> DATASTORE_CPP_VERSION
  clickhouse-cpp-lib/-ut -> datastore-cpp-lib/-ut (CMake, Bazel)
  CH_MAP_BOOL_TO_UINT8   -> DS_MAP_BOOL_TO_UINT8
  CLICKHOUSE_*           -> DATASTORE_*           (test env vars)
  doc links              -> docs.hanzo.ai/datastore

This client speaks the NATIVE binary protocol only - there is no HTTP path
and no X-ClickHouse-* header anywhere in it, so the header rename that the
HTTP clients required does not apply and nothing here was broken.

The one string that reaches the wire is CLIENT_NAME, sent by both SendHello
and SendQuery's ClientInfo from a single macro, and it lands in
system.query_log.client_name. Both sites still read the one macro, so they
stay consistent and the server's Hello/Query consistency check is satisfied.
Old rows keep the old name; that is a reporting discontinuity, not a
migration, since nothing keys on it.

CI is repointed at our own server: ghcr.io/hanzoai/datastore:26.2.3.2, the
image production runs. Its entrypoint honours DATASTORE_SKIP_USER_SETUP, not
the CLICKHOUSE_ spelling the compose file used, and its config lives in
/etc/hanzo-datastore-server. ci/docker-compose/config.xml is now layered in
as a config.d drop-in so the image keeps its own white-labelling. That file
and users.xml deliberately keep a <clickhouse> root: it must match the roots
of the drop-ins the image ships, and a mismatch fails silently - the server
logs "Merging bypassed" and starts with every drop-in discarded.

The remote-TLS suites used to default to the third-party host
play.clickhouse.com. They now read DATASTORE_SECURE*_HOST and skip when it is
unset rather than point at a host we do not run.

Dropped as upstream infrastructure with no meaning here: AI_POLICY.md
(upstream's policy about their codebase), .travis.yml (dead CI referencing a
third-party apt repo), cross-repo-bug-relay.yml (relays our issues into an
upstream triage repo). LICENSE keeps its copyright notice, as Apache-2.0
requires.

Verified: cmake configure + 115-target build clean; the full suite is green
against ghcr.io/hanzoai/datastore:26.2.3.2 (894 passed, 0 failed); the
pristine v2.6.2 build run against that same server passes the identical set,
and both logged the same 3914 queries under their respective client names.
2026-07-26 11:44:39 -07:00

1950 lines
69 KiB
C++
Raw Permalink Blame History

#include <datastore/client.h>
#include <datastore/columns/bool.h>
#include "datastore/base/socket.h"
#include "datastore/version.h"
#include "datastore/error_codes.h"
#include "readonly_client_test.h"
#include "connection_failed_client_test.h"
#include "ut/utils_comparison.h"
#include "utils.h"
#include "ut/roundtrip_column.h"
#include "ut/value_generators.h"
#include <gtest/gtest.h>
#include <memory>
#include <optional>
#include <ostream>
#include <string_view>
#include <thread>
#include <chrono>
using namespace datastore;
#if DS_MAP_BOOL_TO_UINT8
using ClientBoolColumn = ColumnUInt8;
using ClientBoolValue = uint8_t;
#else
using ClientBoolColumn = ColumnBool;
using ClientBoolValue = bool;
#endif
ClientBoolValue MakeClientBoolValue(bool value) {
#if DS_MAP_BOOL_TO_UINT8
return static_cast<uint8_t>(value);
#else
return value;
#endif
}
template <typename T>
std::shared_ptr<T> createTableWithOneColumn(Client & client, const std::string & table_name, const std::string & column_name)
{
auto col = std::make_shared<T>();
const auto type_name = col->GetType().GetName();
client.Execute("DROP TEMPORARY TABLE IF EXISTS " + table_name + ";");
client.Execute("CREATE TEMPORARY TABLE IF NOT EXISTS " + table_name + "( " + column_name + " " + type_name + " )");
return col;
}
// Use value-parameterized tests to run same tests with different client
// options.
class ClientCase : public testing::TestWithParam<ClientOptions> {
protected:
void SetUp() override {
client_ = std::make_unique<Client>(GetParam());
}
void TearDown() override {
client_.reset();
}
template <typename T>
std::shared_ptr<T> createTableWithOneColumn(Block & block)
{
auto col = ::createTableWithOneColumn<T>(*client_, table_name, column_name);
block.AppendColumn(column_name, col);
return col;
}
std::string getOneColumnSelectQuery() const
{
return "SELECT " + column_name + " FROM " + table_name;
}
void FlushLogs() {
try {
client_->Execute("SYSTEM FLUSH LOGS");
} catch (const std::exception & e) {
std::cerr << "Got error while flushing logs: " << e.what() << std::endl;
const auto wait_for_flush = []() {
// Insufficient privileges, the only safe way is to wait long enough for system
// to flush the logs automaticaly. Usually it takes 7.5 seconds, so just in case,
// wait 3 times that to ensure that all previously executed queries are in the logs now.
const auto wait_duration = std::chrono::seconds(23);
std::cerr << "Now we wait " << wait_duration << "..." << std::endl;
std::this_thread::sleep_for(wait_duration);
};
// DB::Exception: datastore_cpp_cicd: Not enough privileges. To execute this query it's necessary to have grant SYSTEM FLUSH LOGS ON
if (std::string(e.what()).find("To execute this query it's necessary to have grant SYSTEM FLUSH LOGS ON") != std::string::npos) {
wait_for_flush();
}
// DB::Exception: datastore_cpp_cicd: Cannot execute query in readonly mode
if (std::string(e.what()).find("Cannot execute query in readonly mode") != std::string::npos) {
wait_for_flush();
}
}
}
std::unique_ptr<Client> client_;
const std::string table_name = "test_datastore_cpp_test_ut_table";
const std::string column_name = "test_column";
};
TEST_P(ClientCase, Version) {
auto version = client_->GetVersion();
EXPECT_NE(0, DATASTORE_CPP_VERSION);
EXPECT_GE(2, DATASTORE_CPP_VERSION_MAJOR);
EXPECT_LE(0, DATASTORE_CPP_VERSION_MINOR);
EXPECT_LE(0, DATASTORE_CPP_VERSION_PATCH);
EXPECT_EQ(DATASTORE_CPP_VERSION_MAJOR, version.major);
EXPECT_EQ(DATASTORE_CPP_VERSION_MINOR, version.minor);
EXPECT_EQ(DATASTORE_CPP_VERSION_PATCH, version.patch);
EXPECT_EQ(DATASTORE_CPP_VERSION_BUILD, version.build);
EXPECT_EQ(DATASTORE_CPP_VERSION_PATCH, version.patch);
}
TEST_P(ClientCase, Array) {
Block b;
/// Create a table.
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_array (arr Array(UInt64)) ");
/// Insert some values.
{
auto arr = std::make_shared<ColumnArray>(std::make_shared<ColumnUInt64>());
auto id = std::make_shared<ColumnUInt64>();
id->Append(1);
arr->AppendAsColumn(id);
id->Append(3);
arr->AppendAsColumn(id);
id->Append(7);
arr->AppendAsColumn(id);
id->Append(9);
arr->AppendAsColumn(id);
b.AppendColumn("arr", arr);
client_->Insert("test_datastore_cpp_array", b);
}
const uint64_t ARR_SIZE[] = { 1, 2, 3, 4 };
const uint64_t VALUE[] = { 1, 3, 7, 9 };
size_t row = 0;
client_->Select("SELECT arr FROM test_datastore_cpp_array",
[ARR_SIZE, VALUE, &row](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
EXPECT_EQ(1U, block.GetColumnCount());
for (size_t c = 0; c < block.GetRowCount(); ++c, ++row) {
auto col = block[0]->As<ColumnArray>()->GetAsColumn(c);
EXPECT_EQ(ARR_SIZE[row], col->Size());
for (size_t i = 0; i < col->Size(); ++i) {
EXPECT_EQ(VALUE[i], (*col->As<ColumnUInt64>())[i]);
}
}
}
);
EXPECT_EQ(4U, row);
}
TEST_P(ClientCase, Time) {
Block b;
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_time (t Time) "
"ENGINE = Memory "
"SETTINGS enable_time_time64_type = 1");
auto t = std::make_shared<ColumnTime>();
int32_t ts = 3600 * 15 + 60 * 4 + 5;
t->Append(ts);
b.AppendColumn("t", t);
client_->Insert("test_datastore_cpp_time", b);
size_t total_rows = 0;
client_->Select("SELECT t, toString(t) FROM test_datastore_cpp_time", [ts, &total_rows](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
EXPECT_EQ(ts, block[0]->As<ColumnTime>()->At(0));
EXPECT_EQ("15:04:05", block[1]->As<ColumnString>()->At(0));
total_rows += block.GetRowCount();
});
EXPECT_EQ(total_rows, 1UL);
}
TEST_P(ClientCase, Time64) {
Block b;
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_time64 "
"(t0 Time64(0), t3 Time64(3), t6 Time64(6)) "
"ENGINE = Memory "
"SETTINGS enable_time_time64_type = 1");
auto t0 = std::make_shared<ColumnTime64>(0);
auto t3 = std::make_shared<ColumnTime64>(3);
auto t6 = std::make_shared<ColumnTime64>(6);
int64_t ts0 = 3600 * 15 + 60 * 4 + 5;
int64_t ts3 = ts0 * 1000 + 123;
int64_t ts6 = ts0 * 1000000 + 123456;
t0->Append(ts0);
t3->Append(ts3);
t6->Append(ts6);
b.AppendColumn("t0", t0);
b.AppendColumn("t3", t3);
b.AppendColumn("t6", t6);
client_->Insert("test_datastore_cpp_time64", b);
size_t total_rows = 0;
client_->Select("SELECT "
"t0, toString(t0), "
"t3, toString(t3), "
"t6, toString(t6) "
"FROM test_datastore_cpp_time64",
[ts0, ts3, ts6, &total_rows](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
EXPECT_EQ(ts0, block[0]->As<ColumnTime64>()->At(0));
EXPECT_EQ(0UL, block[0]->As<ColumnTime64>()->GetPrecision());
EXPECT_EQ("15:04:05", block[1]->As<ColumnString>()->At(0));
EXPECT_EQ(ts3, block[2]->As<ColumnTime64>()->At(0));
EXPECT_EQ(3UL, block[2]->As<ColumnTime64>()->GetPrecision());
EXPECT_EQ("15:04:05.123", block[3]->As<ColumnString>()->At(0));
EXPECT_EQ(ts6, block[4]->As<ColumnTime64>()->At(0));
EXPECT_EQ(6UL, block[4]->As<ColumnTime64>()->GetPrecision());
EXPECT_EQ("15:04:05.123456", block[5]->As<ColumnString>()->At(0));
total_rows += block.GetRowCount();
});
EXPECT_EQ(total_rows, 1UL);
}
TEST_P(ClientCase, Date) {
Block b;
/// Create a table.
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_date (d DateTime('UTC')) ");
auto d = std::make_shared<ColumnDateTime>();
auto const now = std::time(nullptr);
d->Append(now);
b.AppendColumn("d", d);
client_->Insert("test_datastore_cpp_date", b);
client_->Select("SELECT d FROM test_datastore_cpp_date", [&now](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
EXPECT_EQ(1U, block.GetRowCount());
EXPECT_EQ(1U, block.GetColumnCount());
for (size_t c = 0; c < block.GetRowCount(); ++c) {
auto col = block[0]->As<ColumnDateTime>();
std::time_t t = col->As<ColumnDateTime>()->At(c);
EXPECT_EQ(now, t);
EXPECT_EQ(col->Timezone(), "UTC");
}
}
);
}
TEST_P(ClientCase, LowCardinality) {
Block block;
auto lc = createTableWithOneColumn<ColumnLowCardinalityT<ColumnString>>(block);
const std::vector<std::string> data{{"FooBar", "1", "2", "Foo", "4", "Bar", "Foo", "7", "8", "Foo"}};
lc->AppendMany(data);
block.RefreshRowCount();
client_->Insert(table_name, block);
size_t total_rows = 0;
client_->Select(getOneColumnSelectQuery(),
[&total_rows, &data](const Block& block) {
total_rows += block.GetRowCount();
if (block.GetRowCount() == 0) {
return;
}
ASSERT_EQ(1U, block.GetColumnCount());
if (auto col = block[0]->As<ColumnLowCardinalityT<ColumnString>>()) {
ASSERT_EQ(data.size(), col->Size());
for (size_t i = 0; i < col->Size(); ++i) {
EXPECT_EQ(data[i], (*col)[i]) << " at index: " << i;
}
}
}
);
ASSERT_EQ(total_rows, data.size());
}
TEST_P(ClientCase, LowCardinality_InsertAfterClear) {
// User can successfully insert values after invoking Clear() on LC column.
Block block;
auto lc = createTableWithOneColumn<ColumnLowCardinalityT<ColumnString>>(block);
// Add some data, but don't care about it much.
lc->AppendMany(std::vector<std::string_view>{"abc", "def", "123", "abc", "123", "def", "ghi"});
EXPECT_GT(lc->Size(), 0u);
EXPECT_GT(lc->GetDictionarySize(), 0u);
lc->Clear();
// Now ensure that all data appended after Clear() is inserted properly
const std::vector<std::string> data{{"FooBar", "1", "2", "Foo", "4", "Bar", "Foo", "7", "8", "Foo"}};
lc->AppendMany(data);
block.RefreshRowCount();
client_->Insert(table_name, block);
// Now validate that data was properly inserted
size_t total_rows = 0;
client_->Select(getOneColumnSelectQuery(),
[&total_rows, &data](const Block& block) {
total_rows += block.GetRowCount();
if (block.GetRowCount() == 0) {
return;
}
ASSERT_EQ(1U, block.GetColumnCount());
if (auto col = block[0]->As<ColumnLowCardinalityT<ColumnString>>()) {
ASSERT_EQ(data.size(), col->Size());
for (size_t i = 0; i < col->Size(); ++i) {
EXPECT_EQ(data[i], (*col)[i]) << " at index: " << i;
}
}
}
);
ASSERT_EQ(total_rows, data.size());
}
TEST_P(ClientCase, LowCardinalityString_AsString) {
// Validate that LowCardinality(String) column values can be INSERTed from client as ColumnString
// and also read on client (enabled by special option) as ColumnString.
ClientOptions options = GetParam();
options.SetBakcwardCompatibilityFeatureLowCardinalityAsWrappedColumn(true);
client_ = std::make_unique<Client>(GetParam());
// client_->Execute("CREATE DATABASE IF NOT EXISTS test_datastore_cpp");
Block block;
auto col = std::make_shared<ColumnString>();
client_->Execute("DROP TEMPORARY TABLE IF EXISTS " + table_name + ";");
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS " + table_name + "( " + column_name + " LowCardinality(String) )");
block.AppendColumn("test_column", col);
const std::vector<std::string> data{{"FooBar", "1", "2", "Foo", "4", "Bar", "Foo", "7", "8", "Foo"}};
for (const auto & v : data)
col->Append(v);
block.RefreshRowCount();
client_->Insert(table_name, block);
// Now that we can access data via ColumnString instead of ColumnLowCardinalityT<ColumnString>
size_t total_rows = 0;
client_->Select(getOneColumnSelectQuery(),
[&total_rows, &data](const Block& block) {
total_rows += block.GetRowCount();
if (block.GetRowCount() == 0) {
return;
}
ASSERT_EQ(1U, block.GetColumnCount());
if (auto col = block[0]->As<ColumnString>()) {
ASSERT_EQ(data.size(), col->Size());
for (size_t i = 0; i < col->Size(); ++i) {
EXPECT_EQ(data[i], (*col)[i]) << " at index: " << i;
}
}
}
);
ASSERT_EQ(total_rows, data.size());
}
TEST_P(ClientCase, Generic) {
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_client (id UInt64, name String, f Bool) ");
const struct {
uint64_t id;
std::string name;
bool f;
} TEST_DATA[] = {
{ 1, "id", true },
{ 3, "foo", false },
{ 5, "bar", true },
{ 7, "name", false },
};
/// Insert some values.
{
Block block;
auto id = std::make_shared<ColumnUInt64>();
auto name = std::make_shared<ColumnString>();
auto f = std::make_shared<ClientBoolColumn>();
for (auto const& td : TEST_DATA) {
id->Append(td.id);
name->Append(td.name);
f->Append(MakeClientBoolValue(td.f));
}
block.AppendColumn("id" , id);
block.AppendColumn("name", name);
block.AppendColumn("f", f);
client_->Insert("test_datastore_cpp_client", block);
}
/// Select values inserted in the previous step.
size_t row = 0;
client_->Select("SELECT id, name, f FROM test_datastore_cpp_client", [TEST_DATA, &row](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
EXPECT_EQ("id", block.GetColumnName(0));
EXPECT_EQ("name", block.GetColumnName(1));
for (size_t c = 0; c < block.GetRowCount(); ++c, ++row) {
EXPECT_EQ(TEST_DATA[row].id, (*block[0]->As<ColumnUInt64>())[c]);
EXPECT_EQ(TEST_DATA[row].name, (*block[1]->As<ColumnString>())[c]);
EXPECT_EQ(MakeClientBoolValue(TEST_DATA[row].f), (*block[2]->As<ClientBoolColumn>())[c]);
}
}
);
EXPECT_EQ(sizeof(TEST_DATA)/sizeof(TEST_DATA[0]), row);
}
TEST_P(ClientCase, InsertData) {
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_insert (id UInt64, name String, f Bool)");
const struct {
uint64_t id;
std::string name;
bool f;
} TEST_DATA[] = {
{ 1, "id", true },
{ 3, "foo", false },
{ 5, "bar", true },
{ 7, "name", false },
};
const struct {
uint64_t id;
std::string name;
bool f;
} TEST_DATA2[] = {
{ 2, "holden", true },
{ 4, "naomi", false },
{ 6, "amos", true },
{ 8, "alex", false },
};
/// Insert some values.
{
// Prepare the insert.
auto block = client_->BeginInsert("INSERT INTO test_datastore_cpp_insert VALUES");
EXPECT_EQ(size_t(3), block.GetColumnCount());
// Fetch the derived columns.
auto id = block[0]->As<ColumnUInt64>();
auto name = block[1]->As<ColumnString>();
auto f = block[2]->As<ClientBoolColumn>();
// Insert some values.
for (auto const& td : TEST_DATA) {
id->Append(td.id);
name->Append(td.name);
f->Append(MakeClientBoolValue(td.f));
}
block.RefreshRowCount();
client_->SendInsertBlock(block);
block.Clear();
// Insert some more values.
for (auto const& td : TEST_DATA2) {
id->Append(td.id);
name->Append(td.name);
f->Append(MakeClientBoolValue(td.f));
}
block.RefreshRowCount();
client_->SendInsertBlock(block);
block.Clear();
client_->EndInsert();
// Second call to EndInsert should be no-op.
client_->EndInsert();
}
/// Select values inserted in the previous steps.
size_t row = 0;
client_->Select("SELECT id, name, f FROM test_datastore_cpp_insert", [TEST_DATA, TEST_DATA2, &row](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
EXPECT_EQ("id", block.GetColumnName(0));
EXPECT_EQ("name", block.GetColumnName(1));
size_t block_two_row_num = sizeof(TEST_DATA)/sizeof(TEST_DATA[0]);
if (row < block_two_row_num) {
for (size_t c = 0; c < block.GetRowCount(); ++c, ++row) {
EXPECT_EQ(TEST_DATA[row].id, (*block[0]->As<ColumnUInt64>())[c]);
EXPECT_EQ(TEST_DATA[row].name, (*block[1]->As<ColumnString>())[c]);
EXPECT_EQ(MakeClientBoolValue(TEST_DATA[row].f), (*block[2]->As<ClientBoolColumn>())[c]);
}
} else {
for (size_t c = 0; c < block.GetRowCount(); ++c, ++row) {
EXPECT_EQ(TEST_DATA2[row-block_two_row_num].id, (*block[0]->As<ColumnUInt64>())[c]);
EXPECT_EQ(TEST_DATA2[row-block_two_row_num].name, (*block[1]->As<ColumnString>())[c]);
EXPECT_EQ(MakeClientBoolValue(TEST_DATA2[row-block_two_row_num].f), (*block[2]->As<ClientBoolColumn>())[c]);
}
}
}
);
auto exp = sizeof(TEST_DATA)/sizeof(TEST_DATA[0]) + sizeof(TEST_DATA2)/sizeof(TEST_DATA2[0]);
EXPECT_EQ(exp, row);
}
TEST_P(ClientCase, Nullable) {
/// Create a table.
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_nullable (id Nullable(UInt64), date Nullable(Date)) ");
// Round std::time_t to start of date.
const std::time_t cur_date = std::time(nullptr) / 86400 * 86400;
const struct {
uint64_t id;
uint8_t id_null;
std::time_t date;
uint8_t date_null;
} TEST_DATA[] = {
{ 1, 0, cur_date - 2 * 86400, 0 },
{ 2, 0, cur_date - 1 * 86400, 1 },
{ 3, 1, cur_date + 1 * 86400, 0 },
{ 4, 1, cur_date + 2 * 86400, 1 },
};
/// Insert some values.
{
Block block;
{
auto id = std::make_shared<ColumnUInt64>();
auto nulls = std::make_shared<ColumnUInt8>();
for (auto const& td : TEST_DATA) {
id->Append(td.id);
nulls->Append(td.id_null);
}
block.AppendColumn("id", std::make_shared<ColumnNullable>(id, nulls));
}
{
auto date = std::make_shared<ColumnDate>();
auto nulls = std::make_shared<ColumnUInt8>();
for (auto const& td : TEST_DATA) {
date->Append(td.date);
nulls->Append(td.date_null);
}
block.AppendColumn("date", std::make_shared<ColumnNullable>(date, nulls));
}
client_->Insert("test_datastore_cpp_nullable", block);
}
/// Select values inserted in the previous step.
size_t row = 0;
client_->Select("SELECT id, date FROM test_datastore_cpp_nullable",
[TEST_DATA, &row](const Block& block)
{
for (size_t c = 0; c < block.GetRowCount(); ++c, ++row) {
auto col_id = block[0]->As<ColumnNullable>();
auto col_date = block[1]->As<ColumnNullable>();
EXPECT_EQ(static_cast<bool>(TEST_DATA[row].id_null),
col_id->IsNull(c));
if (!col_id->IsNull(c)) {
EXPECT_EQ(TEST_DATA[row].id,
col_id->Nested()->As<ColumnUInt64>()->At(c));
}
EXPECT_EQ(static_cast<bool>(TEST_DATA[row].date_null),
col_date->IsNull(c));
if (!col_date->IsNull(c)) {
// Because date column type is Date instead of
// DateTime, round to start second of date for test.
EXPECT_EQ(TEST_DATA[row].date,
col_date->Nested()->As<ColumnDate>()->At(c));
}
}
}
);
EXPECT_EQ(sizeof(TEST_DATA) / sizeof(TEST_DATA[0]), row);
}
TEST_P(ClientCase, Nothing) {
size_t total_row_count = 0;
client_->Select("SELECT NULL", [&total_row_count](const Block & block)
{
total_row_count += block.GetRowCount();
for (size_t i = 0; i < block.GetRowCount(); ++i) {
EXPECT_TRUE(block[0]->AsStrict<ColumnNullable>()->IsNull(i));
auto column = ColumnNullableT<ColumnNothing>::Wrap(block[0]->AsStrict<ColumnNullable>());
EXPECT_FALSE(column->At(i).has_value());
}
});
ASSERT_EQ(total_row_count, 1UL);
}
TEST_P(ClientCase, Numbers) {
try {
size_t num = 0;
client_->Select("SELECT number, number FROM system.numbers LIMIT 1000", [&num](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
auto col = block[0]->As<ColumnUInt64>();
for (size_t i = 0; i < col->Size(); ++i, ++num) {
EXPECT_EQ(num, col->At(i));
}
}
);
EXPECT_EQ(1000U, num);
}
catch (const datastore::ServerError & e) {
if (e.GetCode() == ErrorCodes::ACCESS_DENIED)
GTEST_SKIP() << e.what() << " : " << GetParam();
else
throw;
}
}
TEST_P(ClientCase, SimpleAggregateFunction) {
const auto & server_info = client_->GetServerInfo();
if (versionNumber(server_info) < versionNumber(19, 9)) {
GTEST_SKIP() << "Test is skipped since server '" << server_info << "' does not support SimpleAggregateFunction" << std::endl;
}
client_->Execute("DROP TEMPORARY TABLE IF EXISTS test_datastore_cpp_SimpleAggregateFunction");
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_SimpleAggregateFunction (saf SimpleAggregateFunction(sum, UInt64))");
constexpr size_t EXPECTED_ROWS = 10;
client_->Execute("INSERT INTO test_datastore_cpp_SimpleAggregateFunction (saf) VALUES (0),(1),(2),(3),(4),(5),(6),(7),(8),(9)");
size_t total_rows = 0;
client_->Select("Select * FROM test_datastore_cpp_SimpleAggregateFunction", [&total_rows](const Block & block) {
if (block.GetRowCount() == 0)
return;
total_rows += block.GetRowCount();
auto col = block[0]->As<ColumnUInt64>();
ASSERT_NE(nullptr, col);
for (size_t r = 0; r < col->Size(); ++r) {
EXPECT_EQ(r, col->At(r));
}
EXPECT_EQ(total_rows, col->Size());
});
EXPECT_EQ(EXPECTED_ROWS, total_rows);
}
TEST_P(ClientCase, Cancellable) {
/// Create a table.
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_cancel (x UInt64) ");
/// Insert a few blocks. In order to make cancel have effect, we have to
/// insert a relative larger amount of data.
const int kBlock = 10;
const int kRowEachBlock = 1000000;
for (unsigned j = 0; j < kBlock; j++) {
Block b;
auto x = std::make_shared<ColumnUInt64>();
for (uint64_t i = 0; i < kRowEachBlock; i++) {
x->Append(i);
}
b.AppendColumn("x", x);
client_->Insert("test_datastore_cpp_cancel", b);
}
/// Send a query which is canceled after receiving the first blockr.
int row_cnt = 0;
EXPECT_NO_THROW(
client_->SelectCancelable("SELECT * FROM test_datastore_cpp_cancel",
[&row_cnt](const Block& block)
{
row_cnt += block.GetRowCount();
return false;
}
);
);
/// It's easier to get query cancelled for compress enabled client.
EXPECT_LE(row_cnt, kBlock * kRowEachBlock);
}
TEST_P(ClientCase, Exception) {
/// Create a table.
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_exceptions (id UInt64, name String) ");
/// Expect failing on table creation.
EXPECT_THROW(
client_->Execute(
"CREATE TEMPORARY TABLE test_datastore_cpp_exceptions (id UInt64, name String) "),
ServerException);
}
TEST_P(ClientCase, Enum) {
/// Create a table.
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_cpp_enums (id UInt64, e Enum8('One' = 1, 'Two' = 2)) ");
const struct {
uint64_t id;
int8_t eval;
std::string ename;
} TEST_DATA[] = {
{ 1, 1, "One" },
{ 2, 2, "Two" },
{ 3, 2, "Two" },
{ 4, 1, "One", },
};
/// Insert some values.
{
Block block;
auto id = std::make_shared<ColumnUInt64>();
auto e = std::make_shared<ColumnEnum8>(Type::CreateEnum8({{"One", 1}, {"Two", 2}}));
int i = 0;
for (auto const& td : TEST_DATA) {
id->Append(td.id);
if (++i % 2) {
e->Append(td.eval);
} else {
e->Append(td.ename);
}
}
block.AppendColumn("id", id);
block.AppendColumn("e", e);
client_->Insert("test_datastore_cpp_enums", block);
}
/// Select values inserted in the previous step.
size_t row = 0;
client_->Select("SELECT id, e FROM test_datastore_cpp_enums", [&row, TEST_DATA](const Block& block)
{
if (block.GetRowCount() == 0) {
return;
}
EXPECT_EQ("id", block.GetColumnName(0));
EXPECT_EQ("e", block.GetColumnName(1));
for (size_t i = 0; i < block.GetRowCount(); ++i, ++row) {
EXPECT_EQ(TEST_DATA[row].id, (*block[0]->As<ColumnUInt64>())[i]);
EXPECT_EQ(TEST_DATA[row].eval, (*block[1]->As<ColumnEnum8>()).At(i));
EXPECT_EQ(TEST_DATA[row].ename, (*block[1]->As<ColumnEnum8>()).NameAt(i));
}
}
);
EXPECT_EQ(sizeof(TEST_DATA)/sizeof(TEST_DATA[0]), row);
}
TEST_P(ClientCase, Decimal) {
client_->Execute(
"CREATE TEMPORARY TABLE IF NOT EXISTS "
"test_datastore_cpp_decimal (id UInt64, d1 Decimal(9, 4), d2 Decimal(18, 9), d3 Decimal(38, 19), "
" d4 Decimal32(4), d5 Decimal64(9), d6 Decimal128(19)) ");
{
Block b;
auto id = std::make_shared<ColumnUInt64>();
auto d1 = std::make_shared<ColumnDecimal>(9, 4);
auto d2 = std::make_shared<ColumnDecimal>(18, 9);
auto d3 = std::make_shared<ColumnDecimal>(38, 19);
auto d4 = std::make_shared<ColumnDecimal>(9, 4);
auto d5 = std::make_shared<ColumnDecimal>(18, 9);
auto d6 = std::make_shared<ColumnDecimal>(38, 19);
EXPECT_THROW(
d1->Append("1234567890123456789012345678901234567890"),
std::runtime_error
);
EXPECT_THROW(
d1->Append("123456789012345678901234567890123456.7890"),
std::runtime_error
);
EXPECT_THROW(
d1->Append("-1234567890123456789012345678901234567890"),
std::runtime_error
);
EXPECT_THROW(
d1->Append("12345678901234567890123456789012345678a"),
std::runtime_error
);
EXPECT_THROW(
d1->Append("12345678901234567890123456789012345678-"),
std::runtime_error
);
EXPECT_THROW(
d1->Append("1234.12.1234"),
std::runtime_error
);
id->Append(1);
d1->Append(123456789);
d2->Append(123456789012345678);
d3->Append(1234567890123456789);
d4->Append(123456789);
d5->Append(123456789012345678);
d6->Append(1234567890123456789);
id->Append(2);
d1->Append(999999999);
d2->Append(999999999999999999);
d3->Append(999999999999999999);
d4->Append(999999999);
d5->Append(999999999999999999);
d6->Append(999999999999999999);
id->Append(3);
d1->Append(-999999999);
d2->Append(-999999999999999999);
d3->Append(-999999999999999999);
d4->Append(-999999999);
d5->Append(-999999999999999999);
d6->Append(-999999999999999999);
// Check strings with decimal point
id->Append(4);
d1->Append("12345.6789");
d2->Append("123456789.012345678");
d3->Append("1234567890123456789.0123456789012345678");
d4->Append("12345.6789");
d5->Append("123456789.012345678");
d6->Append("1234567890123456789.0123456789012345678");
// Check strings with minus sign and without decimal point
id->Append(5);
d1->Append("-12345.6789");
d2->Append("-123456789012345678");
d3->Append("-12345678901234567890123456789012345678");
d4->Append("-12345.6789");
d5->Append("-123456789012345678");
d6->Append("-12345678901234567890123456789012345678");
id->Append(6);
d1->Append("12345.678");
d2->Append("123456789.0123456789");
d3->Append("1234567890123456789.0123456789012345678");
d4->Append("12345.6789");
d5->Append("123456789.012345678");
d6->Append("1234567890123456789.0123456789012345678");
b.AppendColumn("id", id);
b.AppendColumn("d1", d1);
b.AppendColumn("d2", d2);
b.AppendColumn("d3", d3);
b.AppendColumn("d4", d4);
b.AppendColumn("d5", d5);
b.AppendColumn("d6", d6);
client_->Insert("test_datastore_cpp_decimal", b);
}
client_->Select("SELECT id, d1, d2, d3, d4, d5, d6 FROM test_datastore_cpp_decimal ORDER BY id", [](const Block& b) {
if (b.GetRowCount() == 0) {
return;
}
ASSERT_EQ(6u, b.GetRowCount());
auto int128_to_string = [](Int128 value) {
std::string result;
const bool sign = value >= 0;
if (!sign) {
value = -value;
}
while (value) {
result += static_cast<char>(value % 10) + '0';
value /= 10;
}
if (result.empty()) {
result = "0";
} else if (!sign) {
result.push_back('-');
}
std::reverse(result.begin(), result.end());
return result;
};
auto decimal = [&b](size_t column, size_t row) {
return b[column]->As<ColumnDecimal>()->At(row);
};
EXPECT_EQ(1u, b[0]->As<ColumnUInt64>()->At(0));
EXPECT_EQ("123456789", int128_to_string(decimal(1, 0)));
EXPECT_EQ("123456789012345678", int128_to_string(decimal(2, 0)));
EXPECT_EQ("1234567890123456789", int128_to_string(decimal(3, 0)));
EXPECT_EQ("123456789", int128_to_string(decimal(4, 0)));
EXPECT_EQ("123456789012345678", int128_to_string(decimal(5, 0)));
EXPECT_EQ("1234567890123456789", int128_to_string(decimal(6, 0)));
EXPECT_EQ(2u, b[0]->As<ColumnUInt64>()->At(1));
EXPECT_EQ("999999999", int128_to_string(decimal(1, 1)));
EXPECT_EQ("999999999999999999", int128_to_string(decimal(2, 1)));
EXPECT_EQ("999999999999999999", int128_to_string(decimal(3, 1)));
EXPECT_EQ("999999999", int128_to_string(decimal(4, 1)));
EXPECT_EQ("999999999999999999", int128_to_string(decimal(5, 1)));
EXPECT_EQ("999999999999999999", int128_to_string(decimal(6, 1)));
EXPECT_EQ(3u, b[0]->As<ColumnUInt64>()->At(2));
EXPECT_EQ("-999999999", int128_to_string(decimal(1, 2)));
EXPECT_EQ("-999999999999999999", int128_to_string(decimal(2, 2)));
EXPECT_EQ("-999999999999999999", int128_to_string(decimal(3, 2)));
EXPECT_EQ("-999999999", int128_to_string(decimal(4, 2)));
EXPECT_EQ("-999999999999999999", int128_to_string(decimal(5, 2)));
EXPECT_EQ("-999999999999999999", int128_to_string(decimal(6, 2)));
EXPECT_EQ(4u, b[0]->As<ColumnUInt64>()->At(3));
EXPECT_EQ("123456789", int128_to_string(decimal(1, 3)));
EXPECT_EQ("123456789012345678", int128_to_string(decimal(2, 3)));
EXPECT_EQ("12345678901234567890123456789012345678", int128_to_string(decimal(3, 3)));
EXPECT_EQ("123456789", int128_to_string(decimal(4, 3)));
EXPECT_EQ("123456789012345678", int128_to_string(decimal(5, 3)));
EXPECT_EQ("12345678901234567890123456789012345678", int128_to_string(decimal(6, 3)));
EXPECT_EQ(5u, b[0]->As<ColumnUInt64>()->At(4));
EXPECT_EQ("-123456789", int128_to_string(decimal(1, 4)));
EXPECT_EQ("-123456789012345678", int128_to_string(decimal(2, 4)));
EXPECT_EQ("-12345678901234567890123456789012345678", int128_to_string(decimal(3, 4)));
EXPECT_EQ("-123456789", int128_to_string(decimal(4, 4)));
EXPECT_EQ("-123456789012345678", int128_to_string(decimal(5, 4)));
EXPECT_EQ("-12345678901234567890123456789012345678", int128_to_string(decimal(6, 4)));
EXPECT_EQ(6u, b[0]->As<ColumnUInt64>()->At(5));
EXPECT_EQ("123456780", int128_to_string(decimal(1, 5)));
EXPECT_EQ("123456789012345678", int128_to_string(decimal(2, 5)));
EXPECT_EQ("12345678901234567890123456789012345678", int128_to_string(decimal(3, 5)));
EXPECT_EQ("123456789", int128_to_string(decimal(4, 5)));
EXPECT_EQ("123456789012345678", int128_to_string(decimal(5, 5)));
EXPECT_EQ("12345678901234567890123456789012345678", int128_to_string(decimal(6, 5)));
});
}
// Test special chars in names
TEST_P(ClientCase, ColEscapeNameTest) {
client_->Execute(R"sql(DROP TEMPORARY TABLE IF EXISTS "test_datastore_cpp_col_escape_""name_test";)sql");
client_->Execute(R"sql(CREATE TEMPORARY TABLE IF NOT EXISTS "test_datastore_cpp_col_escape_""name_test" ("test space" UInt64, "test "" quote" UInt64, "test ""`'[]&_\ all" UInt64))sql");
auto col1 = std::make_shared<ColumnUInt64>();
col1->Append(1);
col1->Append(2);
auto col2 = std::make_shared<ColumnUInt64>();
col2->Append(4);
col2->Append(8);
auto col3 = std::make_shared<ColumnUInt64>();
col3->Append(16);
col3->Append(32);
static const std::string column_names[] = {
"test space",
R"sql(test " quote)sql",
R"sql(test "`'[]&_\ all)sql"
};
static const auto columns_count = sizeof(column_names)/sizeof(column_names[0]);
Block block;
block.AppendColumn(column_names[0], col1);
block.AppendColumn(column_names[1], col2);
block.AppendColumn(column_names[2], col3);
client_->Insert(R"sql("test_datastore_cpp_col_escape_""name_test")sql", block);
client_->Select(R"sql(SELECT * FROM "test_datastore_cpp_col_escape_""name_test")sql", [] (const Block& sblock)
{
int row = sblock.GetRowCount();
if (row <= 0) {return;}
ASSERT_EQ(columns_count, sblock.GetColumnCount());
for (size_t i = 0; i < columns_count; ++i) {
EXPECT_EQ(column_names[i], sblock.GetColumnName(i));
}
EXPECT_EQ(row, 2);
EXPECT_EQ(sblock[0]->As<ColumnUInt64>()->At(0), 1u);
EXPECT_EQ(sblock[0]->As<ColumnUInt64>()->At(1), 2u);
EXPECT_EQ(sblock[1]->As<ColumnUInt64>()->At(0), 4u);
EXPECT_EQ(sblock[1]->As<ColumnUInt64>()->At(1), 8u);
EXPECT_EQ(sblock[2]->As<ColumnUInt64>()->At(0), 16u);
EXPECT_EQ(sblock[2]->As<ColumnUInt64>()->At(1), 32u);
});
}
// Test roundtrip of DateTime64 values
TEST_P(ClientCase, DateTime64) {
const auto & server_info = client_->GetServerInfo();
if (versionNumber(server_info) < versionNumber(20, 1)) {
GTEST_SKIP() << "Test is skipped since server '" << server_info << "' does not support DateTime64" << std::endl;
}
Block block;
client_->Execute("DROP TEMPORARY TABLE IF EXISTS test_datastore_cpp_datetime64;");
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS "
"test_datastore_cpp_datetime64 (dt DateTime64(6)) ");
auto col_dt64 = std::make_shared<ColumnDateTime64>(6);
block.AppendColumn("dt", col_dt64);
// Empty INSERT and SELECT
client_->Insert("test_datastore_cpp_datetime64", block);
client_->Select("SELECT dt FROM test_datastore_cpp_datetime64",
[](const Block& block) {
ASSERT_EQ(0U, block.GetRowCount());
}
);
const std::vector<Int64> data{
-1'234'567'890'123'456'7ll, // approx year 1578
-1'234'567'890'123ll, // 1969-12-17T17:03:52.890123
-1'234'567ll, // 1969-12-31T23:59:58.234567
0, // epoch
1'234'567ll, // 1970-01-01T00:00:01.234567
1'234'567'890'123ll, // 1970-01-15T06:56:07.890123
1'234'567'890'123'456'7ll // 2361-03-21T19:15:01.234567
};
for (const auto & d : data) {
col_dt64->Append(d);
}
block.RefreshRowCount();
// Non-empty INSERT and SELECT
client_->Insert("test_datastore_cpp_datetime64", block);
size_t total_rows = 0;
client_->Select("SELECT dt FROM test_datastore_cpp_datetime64",
[&total_rows, &data](const Block& block) {
total_rows += block.GetRowCount();
if (block.GetRowCount() == 0) {
return;
}
const auto offset = total_rows - block.GetRowCount();
ASSERT_EQ(1U, block.GetColumnCount());
if (auto col = block[0]->As<ColumnDateTime64>()) {
for (size_t i = 0; i < col->Size(); ++i) {
EXPECT_EQ(data[offset + i], col->At(i)) << " at index: " << i;
}
}
}
);
ASSERT_EQ(total_rows, data.size());
}
TEST_P(ClientCase, Query_ID) {
const auto server_info = client_->GetServerInfo();
std::srand(std::time(nullptr) + reinterpret_cast<int64_t>(&server_info));
const auto * test_info = ::testing::UnitTest::GetInstance()->current_test_info();
const std::string query_id = std::to_string(std::rand()) + "-" + test_info->test_suite_name() + "/" + test_info->name();
SCOPED_TRACE(query_id);
const std::string table_name = "test_datastore_cpp_query_id_test";
client_->Execute(Query("CREATE TEMPORARY TABLE IF NOT EXISTS " + table_name + " (a Int64)", query_id));
{
Block b;
b.AppendColumn("a", std::make_shared<ColumnInt64>(std::vector<int64_t>{1, 2, 3}));
client_->Insert(table_name, query_id, b);
}
client_->Select("SELECT 'a', count(*) FROM " + table_name, query_id, [](const Block &) {});
client_->SelectCancelable("SELECT 'b', count(*) FROM " + table_name, query_id, [](const Block &) { return true; });
client_->Execute(Query("TRUNCATE TABLE " + table_name, query_id));
FlushLogs();
size_t total_count = 0;
client_->Select("SELECT type, query_kind, query_id, query "
" FROM system.query_log "
" WHERE type = 'QueryStart' AND query_id == '" + query_id +"'",
[&total_count](const Block & block) {
total_count += block.GetRowCount();
// std::cerr << PrettyPrintBlock{block} << std::endl;
});
// We've executed 5 queries with explicit query_id, hence we expect to see 5 entries in logs.
EXPECT_EQ(5u, total_count);
}
// Spontaneously fails on INSERT int data.
TEST_P(ClientCase, DISABLED_ArrayArrayUInt64) {
// Based on https://github.com/hanzo-ds/cpp/issues/43
std::cerr << "Connected to: " << client_->GetServerInfo() << std::endl;
std::cerr << "DROPPING TABLE" << std::endl;
client_->Execute("DROP TEMPORARY TABLE IF EXISTS multiarray");
std::cerr << "CREATING TABLE" << std::endl;
client_->Execute(Query(R"sql(CREATE TEMPORARY TABLE IF NOT EXISTS multiarray
(
`arr` Array(Array(UInt64))
);
)sql"));
std::cerr << "INSERTING VALUES" << std::endl;
client_->Execute(Query(R"sql(INSERT INTO multiarray VALUES ([[0,1,2,3,4,5], [100, 200], [10,20, 50, 70]]), ([[456, 789], [1011, 1213], [], [14]]), ([[]]);)sql"));
std::cerr << "INSERTED" << std::endl;
auto result = std::make_shared<ColumnArray>(std::make_shared<ColumnArray>(std::make_shared<ColumnUInt64>()));
ASSERT_EQ(0u, result->Size());
std::cerr << "SELECTING VALUES" << std::endl;
client_->Select("SELECT arr FROM multiarray", [&result](const Block& block) {
std::cerr << "GOT BLOCK: " << block.GetRowCount() << std::endl;
if (block.GetRowCount() == 0)
return;
result->Append(block[0]);
});
std::cerr << "DONE SELECTING VALUES" << std::endl;
client_.reset();
ASSERT_EQ(3u, result->Size());
{
// ([[0,1,2,3,4,5], [100, 200], [10,20, 50, 70]])
const std::vector<std::vector<uint64_t>> expected_vals = {
{0, 1, 2, 3, 4, 5},
{100, 200},
{10, 20, 50, 70}
};
auto row = result->GetAsColumnTyped<ColumnArray>(0);
ASSERT_EQ(3u, row->Size());
EXPECT_TRUE(CompareRecursive(expected_vals[0], *row->GetAsColumnTyped<ColumnUInt64>(0)));
EXPECT_TRUE(CompareRecursive(expected_vals[1], *row->GetAsColumnTyped<ColumnUInt64>(1)));
EXPECT_TRUE(CompareRecursive(expected_vals[2], *row->GetAsColumnTyped<ColumnUInt64>(2)));
}
{
// ([[456, 789], [1011, 1213], [], [14]])
const std::vector<std::vector<uint64_t>> expected_vals = {
{456, 789},
{1011, 1213},
{},
{14}
};
auto row = result->GetAsColumnTyped<ColumnArray>(1);
ASSERT_EQ(4u, row->Size());
EXPECT_TRUE(CompareRecursive(expected_vals[0], *row->GetAsColumnTyped<ColumnUInt64>(0)));
EXPECT_TRUE(CompareRecursive(expected_vals[1], *row->GetAsColumnTyped<ColumnUInt64>(1)));
EXPECT_TRUE(CompareRecursive(expected_vals[2], *row->GetAsColumnTyped<ColumnUInt64>(2)));
EXPECT_TRUE(CompareRecursive(expected_vals[3], *row->GetAsColumnTyped<ColumnUInt64>(3)));
}
{
// ([[]])
auto row = result->GetAsColumnTyped<ColumnArray>(2);
ASSERT_EQ(1u, row->Size());
EXPECT_TRUE(CompareRecursive(std::vector<uint64_t>{}, *row->GetAsColumnTyped<ColumnUInt64>(0)));
}
}
TEST_P(ClientCase, OnProgress) {
Block block;
createTableWithOneColumn<ColumnString>(block);
std::optional<Progress> received_progress;
Query query("INSERT INTO " + table_name + " (*) VALUES (\'Foo\'), (\'Bar\')" );
query.OnProgress([&](const Progress& progress) {
received_progress = progress;
});
client_->Execute(query);
ASSERT_TRUE(received_progress.has_value());
EXPECT_GE(received_progress->rows, 0u);
EXPECT_LE(received_progress->rows, 2u);
EXPECT_GE(received_progress->bytes, 0u);
EXPECT_LE(received_progress->bytes, 10000u);
EXPECT_GE(received_progress->total_rows, 0u);
EXPECT_LE(received_progress->total_rows, 2u);
EXPECT_GE(received_progress->written_rows, 0u);
EXPECT_LE(received_progress->written_rows, 2u);
EXPECT_GE(received_progress->written_bytes, 0u);
EXPECT_LE(received_progress->written_bytes, 10000u);
}
TEST_P(ClientCase, QuerySettings) {
client_->Execute("DROP TEMPORARY TABLE IF EXISTS test_datastore_query_settings_table_1;");
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_query_settings_table_1 ( id Int64 )");
client_->Execute("DROP TEMPORARY TABLE IF EXISTS test_datastore_query_settings_table_2;");
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_datastore_query_settings_table_2 ( id Int64, value Int64 )");
client_->Execute("INSERT INTO test_datastore_query_settings_table_1 (*) VALUES (1)");
Query query("SELECT value "
"FROM test_datastore_query_settings_table_1 "
"LEFT OUTER JOIN test_datastore_query_settings_table_2 "
"ON test_datastore_query_settings_table_1.id = test_datastore_query_settings_table_2.id");
bool checked = false;
query.SetSetting("join_use_nulls", {"1"});
query.OnData(
[&](const Block& block) {
if (block.GetRowCount() == 0)
return;
ASSERT_EQ(1U, block.GetColumnCount());
ASSERT_EQ(1U, block.GetRowCount());
ASSERT_TRUE(block[0]->GetType().IsEqual(Type::CreateNullable(Type::CreateSimple<int64_t>())));
auto cl = block[0]->As<ColumnNullable>();
EXPECT_TRUE(cl->IsNull(0));
checked = true;
});
client_->Execute(query);
EXPECT_TRUE(checked);
query.SetSetting("join_use_nulls", {"0"});
query.OnData(
[&](const Block& block) {
if (block.GetRowCount() == 0)
return;
ASSERT_EQ(1U, block.GetColumnCount());
ASSERT_EQ(1U, block.GetRowCount());
ASSERT_TRUE(block[0]->GetType().IsEqual(Type::CreateSimple<int64_t>()));
auto cl = block[0]->As<ColumnInt64>();
EXPECT_EQ(cl->At(0), 0);
checked = true;
}
);
checked = false;
client_->Execute(query);
EXPECT_TRUE(checked);
query.SetSetting("wrong_setting_name", {"0", QuerySettingsField::IMPORTANT});
EXPECT_THROW(client_->Execute(query), ServerException);
}
TEST_P(ClientCase, ServerLogs) {
Block block;
createTableWithOneColumn<ColumnString>(block);
size_t received_row_count = 0;
Query query("INSERT INTO " + table_name + " (*) VALUES (\'Foo\'), (\'Bar\')" );
query.SetSetting("send_logs_level", {"trace"});
query.OnServerLog([&](const Block& block) {
received_row_count += block.GetRowCount();
return true;
});
client_->Execute(query);
EXPECT_GT(received_row_count, 0U);
}
TEST_P(ClientCase, TracingContext) {
Block block;
createTableWithOneColumn<ColumnString>(block);
Query query("INSERT INTO " + table_name + " (*) VALUES (\'Foo\'), (\'Bar\')" );
open_telemetry::TracingContext tracing_context;
std::srand(std::time(0));
tracing_context.trace_id = {std::rand(), std::rand()};
query.SetTracingContext(tracing_context);
client_->Execute(query);
FlushLogs();
size_t received_rows = 0;
client_->Select("SELECT trace_id, toString(trace_id), operation_name "
"FROM system.opentelemetry_span_log "
"WHERE trace_id = toUUID(\'" + ToString(tracing_context.trace_id) + "\');",
[&](const Block& block) {
// std::cerr << PrettyPrintBlock{block} << std::endl;
received_rows += block.GetRowCount();
});
EXPECT_GT(received_rows, 0u);
}
TEST_P(ClientCase, OnProfileEvents) {
Block block;
createTableWithOneColumn<ColumnString>(block);
client_->Execute("INSERT INTO " + table_name + " (*) VALUES (\'Foo\'), (\'Bar\')");
size_t received_row_count = 0;
Query query("SELECT * FROM " + table_name);
query.OnProfileEvents([&](const Block& block) {
received_row_count += block.GetRowCount();
return true;
});
client_->Execute(query);
const int DBMS_MIN_REVISION_WITH_INCREMENTAL_PROFILE_EVENTS = 54451;
if (client_->GetServerInfo().revision >= DBMS_MIN_REVISION_WITH_INCREMENTAL_PROFILE_EVENTS) {
EXPECT_GT(received_row_count, 0U);
}
}
TEST_P(ClientCase, OnProfile) {
try {
Query query("SELECT * FROM system.numbers LIMIT 10;");
std::optional<Profile> profile;
query.OnProfile([&profile](const Profile & new_profile) {
profile = new_profile;
});
client_->Execute(query);
// Make sure that profile event came through
ASSERT_NE(profile, std::nullopt);
EXPECT_GE(profile->rows, 10u);
EXPECT_GE(profile->blocks, 1u);
EXPECT_GT(profile->bytes, 1u);
EXPECT_GE(profile->rows_before_limit, 10u);
EXPECT_EQ(profile->applied_limit, true);
EXPECT_EQ(profile->calculated_rows_before_limit, true);
} catch (const datastore::ServerError & e) {
if (e.GetCode() == ErrorCodes::ACCESS_DENIED)
GTEST_SKIP() << e.what() << " : " << GetParam();
else
throw;
}
}
TEST_P(ClientCase, SelectAggregateFunction) {
// Verifies that perofing SELECT value of type AggregateFunction(...) doesn't crash the client.
// For details: https://github.com/hanzo-ds/cpp/issues/266
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS tableplus_crash_example (col AggregateFunction(argMax, Int32, DateTime(3))) engine = Memory");
client_->Execute("insert into tableplus_crash_example values (unhex('010000000001089170A883010000'))");
client_->Select("select version()",
[&](const Block& block) {
std::cerr << PrettyPrintBlock{block} << std::endl;
});
// Column type `AggregateFunction` is not supported.
EXPECT_THROW(client_->Select("select toTypeName(col), col from tableplus_crash_example",
[&](const Block& block) {
std::cerr << PrettyPrintBlock{block} << std::endl;
}), datastore::UnimplementedError);
}
// ============================================================================
// Interactive (pull-based) Select API tests
// ============================================================================
TEST_P(ClientCase, InteractiveSelect_Basic) {
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_interactive_basic (id UInt64)");
{
Block block;
auto id = std::make_shared<ColumnUInt64>();
id->Append(1);
id->Append(2);
id->Append(3);
block.AppendColumn("id", id);
client_->Insert("test_interactive_basic", block);
}
// Initially not selecting.
EXPECT_FALSE(client_->IsSelecting());
client_->BeginSelect("SELECT id FROM test_interactive_basic ORDER BY id");
EXPECT_TRUE(client_->IsSelecting());
std::vector<uint64_t> values;
while (auto block = client_->NextBlock()) {
if (block->GetRowCount() == 0) continue;
auto col = block->At(0)->AsStrict<ColumnUInt64>();
for (size_t i = 0; i < block->GetRowCount(); ++i) {
values.push_back(col->At(i));
}
}
EXPECT_FALSE(client_->IsSelecting());
ASSERT_EQ(3u, values.size());
EXPECT_EQ(1u, values[0]);
EXPECT_EQ(2u, values[1]);
EXPECT_EQ(3u, values[2]);
// Client is reusable after drain.
client_->BeginSelect("SELECT CAST(42, 'UInt64') AS val");
{
uint64_t val = 0;
size_t rows = 0;
while (auto b = client_->NextBlock()) {
for (size_t i = 0; i < b->GetRowCount(); ++i) {
val = b->At(0)->AsStrict<ColumnUInt64>()->At(i);
rows++;
}
}
EXPECT_EQ(1u, rows);
EXPECT_EQ(42u, val);
}
EXPECT_FALSE(client_->IsSelecting());
}
TEST_P(ClientCase, InteractiveSelect_MultipleBlocks) {
Query query("SELECT number FROM system.numbers LIMIT 10");
query.SetSetting("max_block_size", {"2"});
client_->BeginSelect(query);
size_t block_count = 0;
std::vector<uint64_t> values;
while (auto block = client_->NextBlock()) {
if (block->GetRowCount() == 0) continue;
EXPECT_LE(block->GetRowCount(), 2u);
block_count++;
auto col = block->At(0)->AsStrict<ColumnUInt64>();
for (size_t i = 0; i < block->GetRowCount(); ++i) {
values.push_back(col->At(i));
}
}
EXPECT_EQ(10u, values.size());
EXPECT_GE(block_count, 2u);
for (size_t i = 0; i < values.size(); ++i) {
EXPECT_EQ(i, values[i]);
}
}
TEST_P(ClientCase, InteractiveSelect_Cancel) {
// Cancel mid-stream on a large result set with small blocks to ensure
// we don't receive all data in a single block.
Query query("SELECT number FROM system.numbers LIMIT 100000");
query.SetSetting("max_block_size", {"100"});
client_->BeginSelect(query);
// Consume one block of data, skipping any blocks with 0 rows.
size_t rows_before_cancel = 0;
while (auto b = client_->NextBlock()) {
if (b->GetRowCount() == 0) continue;
rows_before_cancel = b->GetRowCount();
break;
}
ASSERT_GT(rows_before_cancel, 0u);
ASSERT_LT(rows_before_cancel, 100000u);
EXPECT_TRUE(client_->IsSelecting());
EXPECT_NO_THROW(client_->Cancel());
EXPECT_FALSE(client_->IsSelecting());
// Client is reusable after cancel.
client_->BeginSelect("SELECT 99 AS val");
{
size_t rows = 0;
while (auto b = client_->NextBlock()) {
for (size_t i = 0; i < b->GetRowCount(); ++i) {
EXPECT_EQ(99u, b->At(0)->AsStrict<ColumnUInt8>()->At(i));
rows++;
}
}
EXPECT_EQ(1u, rows);
}
}
TEST_P(ClientCase, InteractiveSelect_StateErrors) {
// NextBlock while idle.
EXPECT_THROW(client_->NextBlock(), ValidationError);
// Cancel while idle.
EXPECT_THROW(client_->Cancel(), ValidationError);
// Double BeginSelect.
client_->BeginSelect("SELECT 1");
EXPECT_THROW(client_->BeginSelect("SELECT 2"), ValidationError);
client_->Cancel();
// Server error (non-existent table) resets state back to Idle.
client_->BeginSelect("SELECT * FROM non_existent_table_xyz_datastore_cpp_test");
EXPECT_THROW(client_->NextBlock(), ServerException);
EXPECT_FALSE(client_->IsSelecting());
// Client is usable after server error.
client_->BeginSelect("SELECT 1 AS val");
{
size_t rows = 0;
while (auto b = client_->NextBlock()) {
for (size_t i = 0; i < b->GetRowCount(); ++i) {
EXPECT_EQ(1u, b->At(0)->AsStrict<ColumnUInt8>()->At(i));
rows++;
}
}
EXPECT_EQ(1u, rows);
}
}
TEST_P(ClientCase, InteractiveSelect_WithCallbacks) {
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_interactive_cb (id UInt64)");
{
Block block;
auto id = std::make_shared<ColumnUInt64>();
for (uint64_t i = 0; i < 100; ++i) {
id->Append(i);
}
block.AppendColumn("id", id);
client_->Insert("test_interactive_cb", block);
}
size_t callback_rows = 0;
bool progress_received = false;
Query query("SELECT id FROM test_interactive_cb");
query.OnData([&](const Block& block) {
callback_rows += block.GetRowCount();
});
query.OnProgress([&](const Progress&) {
progress_received = true;
});
client_->BeginSelect(query);
size_t nextblock_rows = 0;
while (auto block = client_->NextBlock()) {
nextblock_rows += block->GetRowCount();
}
EXPECT_EQ(100u, nextblock_rows);
// OnData fires for non-empty blocks too — both methods see the data.
EXPECT_GT(callback_rows, 0u);
EXPECT_TRUE(progress_received);
}
const auto LocalHostEndpoint = ClientOptions()
.SetHost( getEnvOrDefault("DATASTORE_HOST", "localhost"))
.SetPort( getEnvOrDefault<size_t>("DATASTORE_PORT", "9000"))
.SetUser( getEnvOrDefault("DATASTORE_USER", "default"))
.SetPassword( getEnvOrDefault("DATASTORE_PASSWORD", ""))
.SetDefaultDatabase(getEnvOrDefault("DATASTORE_DB", "default"));
INSTANTIATE_TEST_SUITE_P(
Client, ClientCase,
::testing::Values(
ClientOptions(LocalHostEndpoint)
.SetPingBeforeQuery(true),
ClientOptions(LocalHostEndpoint)
.SetPingBeforeQuery(false)
.SetCompressionMethod(CompressionMethod::LZ4)
));
namespace {
using namespace datastore;
const auto QUERIES = std::vector<std::string>{
"SELECT version()",
"SELECT fqdn()",
"SELECT blockSize()",
"SELECT uptime()",
"SELECT now()"
};
}
INSTANTIATE_TEST_SUITE_P(ClientLocalReadonly, ReadonlyClientTest,
::testing::Values(ReadonlyClientTest::ParamType{
ClientOptions(LocalHostEndpoint)
.SetSendRetries(1)
.SetPingBeforeQuery(true)
.SetCompressionMethod(CompressionMethod::None),
QUERIES
}
));
INSTANTIATE_TEST_SUITE_P(ClientLocalFailed, ConnectionFailedClientTest,
::testing::Values(ConnectionFailedClientTest::ParamType{
ClientOptions()
.SetHost( getEnvOrDefault("DATASTORE_HOST", "localhost"))
.SetPort( getEnvOrDefault<size_t>("DATASTORE_PORT", "9000"))
.SetUser("non_existing_user_datastore_cpp_test")
.SetPassword("wrongpwd")
.SetDefaultDatabase(getEnvOrDefault("DATASTORE_DB", "default"))
.SetSendRetries(1)
.SetPingBeforeQuery(true)
.SetCompressionMethod(CompressionMethod::None),
ExpectingException{"Authentication failed: password is incorrect"}
}
));
class ConnectionSuccessTestCase : public testing::TestWithParam<ClientOptions> {};
TEST_P(ConnectionSuccessTestCase, SuccessConnectionEstablished) {
const auto & client_options = GetParam();
std::unique_ptr<Client> client;
try {
client = std::make_unique<Client>(client_options);
auto endpoint = client->GetCurrentEndpoint().value();
ASSERT_EQ("localhost", endpoint.host);
ASSERT_EQ(9000u, endpoint.port);
SUCCEED();
} catch (const std::exception & e) {
FAIL() << "Got an unexpected exception : " << e.what();
}
}
INSTANTIATE_TEST_SUITE_P(ClientMultipleEndpoints, ConnectionSuccessTestCase,
::testing::Values(ClientCase::ParamType{
ClientOptions()
.SetEndpoints({
{"somedeadhost", 9000}
, {"deadaginghost", 1245}
, {"localhost", 9000}
, {"noalocalhost", 6784}
})
.SetUser( getEnvOrDefault("DATASTORE_USER", "default"))
.SetPassword( getEnvOrDefault("DATASTORE_PASSWORD", ""))
.SetDefaultDatabase(getEnvOrDefault("DATASTORE_DB", "default"))
.SetPingBeforeQuery(true)
.SetConnectionConnectTimeout(std::chrono::milliseconds(200))
.SetRetryTimeout(std::chrono::seconds(1)),
}
));
INSTANTIATE_TEST_SUITE_P(ClientMultipleEndpointsWithDefaultPort, ConnectionSuccessTestCase,
::testing::Values(ClientCase::ParamType{
ClientOptions()
.SetEndpoints({
{"somedeadhost"}
, {"deadaginghost", 1245}
, {"localhost"}
, {"noalocalhost", 6784}
})
.SetUser( getEnvOrDefault("DATASTORE_USER", "default"))
.SetPassword( getEnvOrDefault("DATASTORE_PASSWORD", ""))
.SetDefaultDatabase(getEnvOrDefault("DATASTORE_DB", "default"))
.SetPingBeforeQuery(true)
.SetConnectionConnectTimeout(std::chrono::milliseconds(200))
.SetRetryTimeout(std::chrono::seconds(1)),
}
));
INSTANTIATE_TEST_SUITE_P(MultipleEndpointsFailed, ConnectionFailedClientTest,
::testing::Values(ConnectionFailedClientTest::ParamType{
ClientOptions()
.SetEndpoints({
{"deadaginghost", 9000}
,{"somedeadhost", 1245}
,{"noalocalhost", 6784}
})
.SetUser( getEnvOrDefault("DATASTORE_USER", "default"))
.SetPassword( getEnvOrDefault("DATASTORE_PASSWORD", ""))
.SetDefaultDatabase(getEnvOrDefault("DATASTORE_DB", "default"))
.SetPingBeforeQuery(true)
.SetConnectionConnectTimeout(std::chrono::milliseconds(200))
.SetRetryTimeout(std::chrono::seconds(1)),
ExpectingException{""}
}
));
class ResetConnectionTestCase : public testing::TestWithParam<ClientOptions> {};
TEST_P(ResetConnectionTestCase, ResetConnectionEndpointTest) {
const auto & client_options = GetParam();
std::unique_ptr<Client> client;
try {
client = std::make_unique<Client>(client_options);
auto endpoint = client->GetCurrentEndpoint().value();
ASSERT_EQ("localhost", endpoint.host);
ASSERT_EQ(9000u, endpoint.port);
client->ResetConnectionEndpoint();
endpoint = client->GetCurrentEndpoint().value();
ASSERT_EQ("127.0.0.1", endpoint.host);
ASSERT_EQ(9000u, endpoint.port);
client->ResetConnectionEndpoint();
endpoint = client->GetCurrentEndpoint().value();
ASSERT_EQ("localhost", endpoint.host);
ASSERT_EQ(9000u, endpoint.port);
SUCCEED();
} catch (const std::exception & e) {
FAIL() << "Got an unexpected exception : " << e.what();
}
}
TEST_P(ResetConnectionTestCase, ResetConnectionTest) {
const auto & client_options = GetParam();
std::unique_ptr<Client> client;
try {
client = std::make_unique<Client>(client_options);
auto endpoint = client->GetCurrentEndpoint().value();
ASSERT_EQ("localhost", endpoint.host);
ASSERT_EQ(9000u, endpoint.port);
client->ResetConnection();
endpoint = client->GetCurrentEndpoint().value();
ASSERT_EQ("localhost", endpoint.host);
ASSERT_EQ(9000u, endpoint.port);
SUCCEED();
} catch (const std::exception & e) {
FAIL() << "Got an unexpected exception : " << e.what();
}
}
INSTANTIATE_TEST_SUITE_P(ResetConnectionClientTest, ResetConnectionTestCase,
::testing::Values(ResetConnectionTestCase::ParamType {
ClientOptions()
.SetEndpoints({
{"localhost", 9000}
,{"somedeadhost", 1245}
,{"noalocalhost", 6784}
,{"127.0.0.1", 9000}
})
.SetUser( getEnvOrDefault("DATASTORE_USER", "default"))
.SetPassword( getEnvOrDefault("DATASTORE_PASSWORD", ""))
.SetDefaultDatabase(getEnvOrDefault("DATASTORE_DB", "default"))
.SetPingBeforeQuery(true)
.SetConnectionConnectTimeout(std::chrono::milliseconds(200))
.SetRetryTimeout(std::chrono::seconds(1))
}
));
struct CountingSocketFactoryAdapter : public SocketFactory {
using ConnectRequests = std::vector<std::pair<ClientOptions, Endpoint>>;
SocketFactory & socket_factory;
ConnectRequests & connect_requests;
CountingSocketFactoryAdapter(SocketFactory & socket_factory, ConnectRequests& connect_requests)
: socket_factory(socket_factory)
, connect_requests(connect_requests)
{}
std::unique_ptr<SocketBase> connect(const ClientOptions& opts, const Endpoint& endpoint) {
connect_requests.emplace_back(opts, endpoint);
return socket_factory.connect(opts, endpoint);
}
void sleepFor(const std::chrono::milliseconds& duration) {
return socket_factory.sleepFor(duration);
}
size_t GetConnectRequestsCount() const {
return connect_requests.size();
}
};
TEST(SimpleClientTest, issue_335) {
// Make sure Client connects to server even with ClientOptions.SetSendRetries(0)
auto vals = MakeStrings();
auto col = std::make_shared<ColumnString>(vals);
CountingSocketFactoryAdapter::ConnectRequests connect_requests;
std::unique_ptr<SocketFactory> wrapped_socket_factory = std::make_unique<NonSecureSocketFactory>();
std::unique_ptr<SocketFactory> socket_factory = std::make_unique<CountingSocketFactoryAdapter>(*wrapped_socket_factory, connect_requests);
Client client(ClientOptions(LocalHostEndpoint)
.SetSendRetries(0), // <<=== crucial for reproducing https://github.com/hanzo-ds/cpp/issues/335
std::move(socket_factory));
EXPECT_EQ(1u, connect_requests.size());
EXPECT_TRUE(CompareRecursive(vals, *RoundtripColumnValuesTyped(client, col)));
connect_requests.clear();
client.ResetConnection();
EXPECT_EQ(1u, connect_requests.size());
EXPECT_TRUE(CompareRecursive(vals, *RoundtripColumnValuesTyped(client, col)));
connect_requests.clear();
client.ResetConnectionEndpoint();
EXPECT_EQ(1u, connect_requests.size());
EXPECT_TRUE(CompareRecursive(vals, *RoundtripColumnValuesTyped(client, col)));
}
TEST(SimpleClientTest, issue_335_reconnects_count) {
// Make sure that client attempts to connect to each endpoint at least once.
CountingSocketFactoryAdapter::ConnectRequests connect_requests;
std::unique_ptr<SocketFactory> wrapped_socket_factory = std::make_unique<NonSecureSocketFactory>();
std::unique_ptr<SocketFactory> socket_factory = std::make_unique<CountingSocketFactoryAdapter>(*wrapped_socket_factory, connect_requests);
const std::vector<Endpoint> endpoints = {
Endpoint{"foo-invalid-hostname", 1234},
Endpoint{"bar-invalid-hostname", 4567},
};
EXPECT_ANY_THROW(
Client(ClientOptions()
.SetEndpoints(endpoints)
.SetSendRetries(0), // <<=== crucial for reproducing https://github.com/hanzo-ds/cpp/issues/335
std::move(socket_factory));
);
EXPECT_EQ(endpoints.size(), connect_requests.size());
// make sure there was an attempt to connect to each endpoint at least once.
for (const auto & endpoint : endpoints)
{
auto p = std::find_if(connect_requests.begin(), connect_requests.end(), [&endpoint](const auto & connect_request) {
return connect_request.second == endpoint;
});
EXPECT_TRUE(connect_requests.end() != p)
<< "\tThere was no attempt to connect to endpoint " << endpoint;
}
}
TEST_P(ClientCase, QueryParameters) {
const auto & server_info = client_->GetServerInfo();
if (versionNumber(server_info) < versionNumber(24, 7)) {
GTEST_SKIP() << "Test is skipped since server '" << server_info << "' does not support query parameters" << std::endl;
}
const std::string table_name = "test_datastore_cpp_query_parameter";
client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS " + table_name + " (id UInt64, name String)");
{
Query query("insert into " + table_name + " values ({id: UInt64}, {name: String})");
query.SetParam("id", "1").SetParam("name", "NAME");
client_->Execute(query);
query.SetParam("id", "123").SetParam("name", "FromParam");
client_->Execute(query);
const char FirstPrintable = ' ';
char test_str1[FirstPrintable * 2 + 1];
for (unsigned int i = 0; i < FirstPrintable; i++) {
test_str1[i * 2] = 'A';
test_str1[i * 2 + 1] = i;
}
test_str1[int(FirstPrintable * 2)] = 'A';
query.SetParam("id", "333").SetParam("name", std::string(test_str1, FirstPrintable * 2 + 1));
client_->Execute(query);
const char LastPrintable = 127;
unsigned char big_string[LastPrintable - FirstPrintable];
for (unsigned int i = 0; i < sizeof(big_string); i++) big_string[i] = i + FirstPrintable;
query.SetParam("id", "444").SetParam("name", std::string((char*)big_string, sizeof(big_string)));
client_->Execute(query);
query.SetParam("id", "555").SetParam("name", "utf8Русский");
client_->Execute(query);
}
Query query("SELECT id, name, length(name) FROM " + table_name + " where id > {a: Int32}");
query.SetParam("a", "4");
size_t total_count = 0;
SelectCallback cb([&total_count](const Block& block) {
total_count += block.GetRowCount();
//std::cout << PrettyPrintBlock{block} << std::endl;
});
query.OnData(cb);
client_->Select(query);
EXPECT_EQ(4u, total_count);
client_->Execute("DROP TEMPORARY TABLE " + table_name);
}
TEST_P(ClientCase, ClientName) {
const auto server_info = client_->GetServerInfo();
std::srand(std::time(nullptr) + reinterpret_cast<int64_t>(&server_info));
const auto * test_info = ::testing::UnitTest::GetInstance()->current_test_info();
const std::string query_id = std::to_string(std::rand()) + "-" + test_info->test_suite_name() + "/" + test_info->name();
SCOPED_TRACE(query_id);
client_->Select("SELECT 1", query_id, [](const Block&) { /* make sure the data is delivered in full */ });
FlushLogs();
std::string query_log_query
= "SELECT CAST(client_name, 'String') FROM system.query_log WHERE query_id = '" + query_id + "'";
size_t total_rows = 0;
client_->Select(query_log_query, [&total_rows](const Block& block) {
const auto row_count = block.GetRowCount();
total_rows += row_count;
for (size_t i = 0; i < row_count; ++i) {
ASSERT_EQ(block[0]->AsStrict<ColumnString>()->At(i), "datastore-cpp");
}
});
ASSERT_GT(total_rows, 0UL) << "Query with query_id " << query_id << " is not found";
}
Reference in New Issue View Git Blame Copy Permalink