ClickHouse/programs/odbc-bridge/ODBCSource.cpp

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

153 lines
4.6 KiB
C++
Raw Normal View History

2024-03-24 16:21:53 +00:00
#include "ODBCSource.h"
#include <vector>
2021-03-22 11:40:29 +00:00
#include <IO/ReadBufferFromString.h>
#include <DataTypes/DataTypeNullable.h>
2021-11-22 12:21:15 +00:00
#include <DataTypes/DataTypeDateTime64.h>
#include <Common/assert_cast.h>
2018-09-05 16:20:12 +00:00
#include <IO/ReadHelpers.h>
2016-12-08 02:49:04 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int UNKNOWN_TYPE;
2016-12-08 02:49:04 +00:00
}
2021-10-11 16:11:50 +00:00
ODBCSource::ODBCSource(
2021-06-07 18:09:16 +00:00
nanodbc::ConnectionHolderPtr connection_holder, const std::string & query_str, const Block & sample_block, const UInt64 max_block_size_)
2021-10-11 16:11:50 +00:00
: ISource(sample_block)
2024-01-23 17:04:50 +00:00
, log(getLogger("ODBCSource"))
2019-08-03 11:02:40 +00:00
, max_block_size{max_block_size_}
2021-03-22 11:40:29 +00:00
, query(query_str)
2016-12-08 02:49:04 +00:00
{
description.init(sample_block);
2021-06-07 18:09:16 +00:00
result = execute<nanodbc::result>(connection_holder,
[&](nanodbc::connection & connection) { return execute(connection, query); });
2016-12-08 02:49:04 +00:00
}
2021-10-11 16:11:50 +00:00
Chunk ODBCSource::generate()
2016-12-08 02:49:04 +00:00
{
2021-10-11 16:11:50 +00:00
if (is_finished)
return {};
2021-03-22 11:40:29 +00:00
MutableColumns columns(description.sample_block.cloneEmptyColumns());
2016-12-08 02:49:04 +00:00
size_t num_rows = 0;
2021-03-22 11:40:29 +00:00
while (true)
{
if (!result.next())
2016-12-08 02:49:04 +00:00
{
2021-10-11 16:11:50 +00:00
is_finished = true;
2021-03-22 11:40:29 +00:00
break;
}
2016-12-08 02:49:04 +00:00
2021-03-22 11:40:29 +00:00
for (int idx = 0; idx < result.columns(); ++idx)
{
const auto & sample = description.sample_block.getByPosition(idx);
2021-03-22 11:40:29 +00:00
if (!result.is_null(idx))
2024-03-10 08:17:29 +00:00
insertValue(*columns[idx], removeNullable(sample.type), description.types[idx].first, result, idx);
2016-12-08 02:49:04 +00:00
else
insertDefaultValue(*columns[idx], *sample.column);
2016-12-08 02:49:04 +00:00
}
2021-03-22 11:40:29 +00:00
if (++num_rows == max_block_size)
2016-12-08 02:49:04 +00:00
break;
}
2021-10-11 16:11:50 +00:00
return Chunk(std::move(columns), num_rows);
2016-12-08 02:49:04 +00:00
}
2021-03-22 11:40:29 +00:00
2021-10-11 16:11:50 +00:00
void ODBCSource::insertValue(
IColumn & column, const DataTypePtr data_type, const ValueType type, nanodbc::result & row, size_t idx)
2021-03-22 11:40:29 +00:00
{
switch (type)
{
case ValueType::vtUInt8:
2024-03-09 07:42:33 +00:00
column.insert(row.get<uint16_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtUInt16:
2024-03-09 07:42:33 +00:00
column.insert(row.get<uint16_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtUInt32:
2024-03-09 07:42:33 +00:00
column.insert(row.get<uint32_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtUInt64:
2024-03-09 07:42:33 +00:00
column.insert(row.get<uint64_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtInt8:
2024-03-09 07:42:33 +00:00
column.insert(row.get<int16_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtInt16:
2024-03-09 07:42:33 +00:00
column.insert(row.get<int16_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtInt32:
2024-03-09 07:42:33 +00:00
column.insert(row.get<int32_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtInt64:
2024-03-09 07:42:33 +00:00
column.insert(row.get<int64_t>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtFloat32:
2024-03-09 07:42:33 +00:00
column.insert(row.get<float>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtFloat64:
2024-03-09 07:42:33 +00:00
column.insert(row.get<double>(idx));
2021-03-22 11:40:29 +00:00
break;
2024-03-10 08:02:34 +00:00
case ValueType::vtFixedString:
case ValueType::vtEnum8:
case ValueType::vtEnum16:
2021-03-22 11:40:29 +00:00
case ValueType::vtString:
2024-03-09 07:42:33 +00:00
column.insert(row.get<std::string>(idx));
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtUUID:
{
auto value = row.get<std::string>(idx);
2024-03-09 07:42:33 +00:00
column.insert(parse<UUID>(value.data(), value.size()));
2021-03-22 11:40:29 +00:00
break;
}
case ValueType::vtDate:
2024-03-09 07:42:33 +00:00
column.insert(UInt16{LocalDate{row.get<std::string>(idx)}.getDayNum()});
2021-03-22 11:40:29 +00:00
break;
case ValueType::vtDateTime:
{
auto value = row.get<std::string>(idx);
ReadBufferFromString in(value);
2021-03-22 11:40:29 +00:00
time_t time = 0;
2024-03-10 08:16:21 +00:00
const DataTypeDateTime & datetime_type = assert_cast<const DataTypeDateTime &>(*data_type);
readDateTimeText(time, in, datetime_type.getTimeZone());
2021-03-22 11:40:29 +00:00
if (time < 0)
time = 0;
2024-03-09 07:42:33 +00:00
column.insert(static_cast<UInt32>(time));
2021-03-22 11:40:29 +00:00
break;
}
2021-11-22 12:21:15 +00:00
case ValueType::vtDateTime64:
{
auto value = row.get<std::string>(idx);
ReadBufferFromString in(value);
DateTime64 time = 0;
2024-03-10 08:16:21 +00:00
const DataTypeDateTime64 & datetime_type = assert_cast<const DataTypeDateTime64 &>(*data_type);
readDateTime64Text(time, datetime_type.getScale(), in, datetime_type.getTimeZone());
2024-03-09 07:42:33 +00:00
column.insert(time);
2021-11-22 12:21:15 +00:00
break;
}
2024-03-10 08:02:34 +00:00
case ValueType::vtDecimal32:
case ValueType::vtDecimal64:
case ValueType::vtDecimal128:
case ValueType::vtDecimal256:
{
auto value = row.get<std::string>(idx);
ReadBufferFromString istr(value);
2021-03-31 08:42:57 +00:00
data_type->getDefaultSerialization()->deserializeWholeText(column, istr, FormatSettings{});
break;
}
2021-03-22 11:40:29 +00:00
default:
throw Exception(ErrorCodes::UNKNOWN_TYPE, "Unsupported value type");
2021-03-22 11:40:29 +00:00
}
}
2016-12-08 02:49:04 +00:00
}