Skip to content

[feature](function) Support nanosecond precision for TIMEV2 #67909

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jacktengg wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[feature](function) Support nanosecond precision for TIMEV2 #67909

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion be/src/core/data_type/data_type_time.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class IColumn;

class DataTypeTimeV2 final : public DataTypeNumberBase<PrimitiveType::TYPE_TIMEV2> {
public:
static constexpr UInt32 MAX_SCALE = 6;
static constexpr UInt32 MAX_SCALE = 9;

DataTypeTimeV2(int scale = 0) : _scale(scale) {
if (UNLIKELY(scale > static_cast<int>(MAX_SCALE))) {
Expand Down
89 changes: 51 additions & 38 deletions be/src/core/data_type_serde/data_type_time_serde.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,9 @@
#include <arrow/array.h>
#include <arrow/type.h>

#include <algorithm>
#include <limits>
#include <utility>

#include "common/config.h"
#include "core/data_type/data_type_decimal.h"
Expand All @@ -30,47 +32,48 @@
#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type_serde/parquet_decode_source.h"
#include "core/value/time_value.h"
#include "exec/common/int_exp.h"
#include "exprs/function/cast/cast_base.h"
#include "exprs/function/cast/cast_to_time_impl.hpp"
#include "util/unaligned.h"

namespace doris {
namespace {

TimeValue::TimeType read_time_decoded_value(const DecodedColumnView& view, int64_t row) {
int64_t micros = 0;
TimeValue::TimeType time_from_nanoseconds(int64_t nanoseconds, int scale) {
const int effective_scale = std::max(scale, static_cast<int>(TimeValue::MICROS_SCALE));
const int64_t divisor = common::exp10_i64(TimeValue::NANOS_SCALE - effective_scale);
return TimeValue::from_nanoseconds(nanoseconds / divisor * divisor);
}

TimeValue::TimeType read_time_decoded_value(const DecodedColumnView& view, int64_t row, int scale) {
if (view.value_kind == DecodedValueKind::INT32) {
const auto* values = reinterpret_cast<const int32_t*>(view.values);
micros = static_cast<int64_t>(values[row]) * 1000;
} else {
const auto* values = reinterpret_cast<const int64_t*>(view.values);
micros = values[row];
if (view.time_unit == DecodedTimeUnit::MILLIS) {
micros *= 1000;
} else if (view.time_unit == DecodedTimeUnit::NANOS) {
micros /= 1000;
}
return static_cast<TimeValue::TimeType>(values[row]) * 1000;
}
const bool negative = micros < 0;
const int64_t abs_micros = std::abs(micros);
return TimeValue::make_time(
abs_micros / TimeValue::ONE_HOUR_MICROSECONDS,
(abs_micros % TimeValue::ONE_HOUR_MICROSECONDS) / TimeValue::ONE_MINUTE_MICROSECONDS,
(abs_micros % TimeValue::ONE_MINUTE_MICROSECONDS) / TimeValue::ONE_SECOND_MICROSECONDS,
abs_micros % TimeValue::ONE_SECOND_MICROSECONDS, negative);
const auto* values = reinterpret_cast<const int64_t*>(view.values);
if (view.time_unit == DecodedTimeUnit::MILLIS) {
return static_cast<TimeValue::TimeType>(values[row]) * 1000;
}
if (view.time_unit == DecodedTimeUnit::NANOS) {
return time_from_nanoseconds(values[row], scale);
}
return static_cast<TimeValue::TimeType>(values[row]);
}

class TimeV2ParquetConsumer final : public ParquetFixedValueConsumer {
public:
TimeV2ParquetConsumer(IColumn& column, const ParquetDecodeContext& context,
TimeV2ParquetConsumer(IColumn& column, int scale, const ParquetDecodeContext& context,
ParquetMaterializationState* state = nullptr)
: _data(assert_cast<ColumnTimeV2&>(column).get_data()),
_scale(scale),
_context(context),
_state(state) {}

TimeV2ParquetConsumer(ColumnTimeV2::Container& data, const ParquetDecodeContext& context,
TimeV2ParquetConsumer(ColumnTimeV2::Container& data, int scale,
const ParquetDecodeContext& context,
ParquetMaterializationState* state = nullptr)
: _data(data), _context(context), _state(state) {}
: _data(data), _scale(scale), _context(context), _state(state) {}

Status consume(const uint8_t* values, size_t num_values, size_t value_width) override {
const size_t old_size = _data.size();
Expand Down Expand Up @@ -106,21 +109,20 @@ class TimeV2ParquetConsumer final : public ParquetFixedValueConsumer {
return Status::DataQualityError(
"Parquet TIME value {} is outside the one-day domain", raw_value);
}
int64_t micros = raw_value;
TimeValue::TimeType time = raw_value;
if (_context.time_unit == ParquetTimeUnit::MILLIS) {
micros *= 1000;
time *= 1000;
} else if (_context.time_unit == ParquetTimeUnit::NANOS) {
micros /= 1000;
time = time_from_nanoseconds(raw_value, _scale);
}
// Doris TIMEV2 stores signed microseconds in a double. Splitting into calendar fields
// and immediately recombining them is an identity operation with several divisions.
_data[old_size + row] = static_cast<TimeValue::TimeType>(micros);
_data[old_size + row] = time;
}
return Status::OK();
}

private:
ColumnTimeV2::Container& _data;
const int _scale;
const ParquetDecodeContext& _context;
ParquetMaterializationState* _state;
};
Expand All @@ -134,11 +136,12 @@ class RejectTimeV2BinaryConsumer final : public ParquetBinaryValueConsumer {

class TimeV2PredicateParquetConsumer final : public ParquetFixedValueConsumer {
public:
TimeV2PredicateParquetConsumer(const ParquetDecodeContext& context, bool enable_strict_mode,
ParquetLogicalValueConsumer& consumer,
TimeV2PredicateParquetConsumer(const ParquetDecodeContext& context, int scale,
bool enable_strict_mode, ParquetLogicalValueConsumer& consumer,
ColumnTimeV2::Container& logical_values,
IColumn::Filter& conversion_nulls)
: _context(context),
_scale(scale),
_enable_strict_mode(enable_strict_mode),
_consumer(consumer),
_logical_values(logical_values),
Expand All @@ -151,14 +154,15 @@ class TimeV2PredicateParquetConsumer final : public ParquetFixedValueConsumer {
ParquetMaterializationState state;
state.enable_strict_mode = _enable_strict_mode;
state.conversion_failure_null_map = &_conversion_nulls;
TimeV2ParquetConsumer converter(_logical_values, _context, &state);
TimeV2ParquetConsumer converter(_logical_values, _scale, _context, &state);
RETURN_IF_ERROR(converter.consume(values, num_values, value_width));
return _consumer.consume(reinterpret_cast<const uint8_t*>(_logical_values.data()),
num_values, sizeof(TimeValue::TimeType), _conversion_nulls.data());
}

private:
const ParquetDecodeContext& _context;
const int _scale;
bool _enable_strict_mode;
ParquetLogicalValueConsumer& _consumer;
ColumnTimeV2::Container& _logical_values;
Expand All @@ -173,7 +177,14 @@ Status DataTypeTimeV2SerDe::write_column_to_mysql_binary(const IColumn& column,
const FormatOptions& options) const {
const auto& data = assert_cast<const ColumnTimeV2&>(column).get_data();
const auto col_index = index_check_const(row_idx, col_const);
if (UNLIKELY(0 != result.push_timev2(data[col_index], _scale))) {
int push_result;
if (std::cmp_greater(_scale, TimeValue::MICROS_SCALE)) {
const auto value = TimeValue::to_string(data[col_index], _scale);
push_result = result.push_string(value.data(), value.size());
} else {
push_result = result.push_timev2(data[col_index], _scale);
}
if (UNLIKELY(push_result != 0)) {
return Status::InternalError("pack mysql buffer failed.");
}
return Status::OK();
Expand Down Expand Up @@ -365,8 +376,10 @@ Status DataTypeTimeV2SerDe::read_column_from_arrow(IColumn& column, const arrow:
"Arrow Time64 value is outside the time-of-day range: row={}, value={}",
row, value);
}
const int64_t micros = type->unit() == arrow::TimeUnit::NANO ? value / 1000 : value;
data.emplace_back(static_cast<TimeValue::TimeType>(micros));
const auto time = type->unit() == arrow::TimeUnit::NANO
? time_from_nanoseconds(value, _scale)
: static_cast<TimeValue::TimeType>(value);
data.emplace_back(time);
}
return Status::OK();
}
Expand All @@ -391,14 +404,14 @@ Status DataTypeTimeV2SerDe::read_column_from_decoded_values(IColumn& column,
data.push_back(TimeValue::TimeType());
continue;
}
data.push_back(read_time_decoded_value(view, row));
data.push_back(read_time_decoded_value(view, row, _scale));
}
return Status::OK();
}

Status DataTypeTimeV2SerDe::read_parquet_dictionary(IColumn& column, ParquetDecodeSource& source,
const ParquetDecodeContext& context) const {
TimeV2ParquetConsumer consumer(column, context);
TimeV2ParquetConsumer consumer(column, _scale, context);
RejectTimeV2BinaryConsumer binary_consumer;
return source.decode_dictionary(consumer, binary_consumer);
}
Expand All @@ -412,14 +425,14 @@ Status DataTypeTimeV2SerDe::read_column_from_parquet(IColumn& column, ParquetDec
context.logical_type != ParquetLogicalType::TIME) {
return Status::NotSupported("TIMEV2 expects Parquet TIME stored as INT32 or INT64");
}
TimeV2ParquetConsumer consumer(column, context, &state);
TimeV2ParquetConsumer consumer(column, _scale, context, &state);
if (context.encoding != ParquetValueEncoding::DICTIONARY) {
return source.decode_fixed_values(num_values, consumer);
}
if (state.dictionary_generation != source.dictionary_generation()) {
state.typed_dictionary = column.clone_empty();
auto* output_null_map = state.begin_dictionary_conversion(source.dictionary_size());
TimeV2ParquetConsumer dictionary_consumer(*state.typed_dictionary, context, &state);
TimeV2ParquetConsumer dictionary_consumer(*state.typed_dictionary, _scale, context, &state);
RejectTimeV2BinaryConsumer binary_consumer;
const Status dictionary_status =
source.decode_dictionary(dictionary_consumer, binary_consumer);
Expand All @@ -445,7 +458,7 @@ Status DataTypeTimeV2SerDe::read_parquet_raw_predicate(
if (!supports_parquet_raw_predicate(context)) {
return Status::NotSupported("Unsupported Parquet raw predicate conversion for TIMEV2");
}
TimeV2PredicateParquetConsumer predicate_consumer(context, enable_strict_mode, consumer,
TimeV2PredicateParquetConsumer predicate_consumer(context, _scale, enable_strict_mode, consumer,
_parquet_predicate_values,
_parquet_predicate_nulls);
return source.decode_fixed_values(num_values, predicate_consumer);
Expand Down
91 changes: 72 additions & 19 deletions be/src/core/value/time_value.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,13 @@ class TimeValue {
constexpr static int64_t ONE_SECOND_MICROSECONDS = 1000000;
constexpr static int64_t ONE_MINUTE_MICROSECONDS = 60 * ONE_SECOND_MICROSECONDS;
constexpr static int64_t ONE_HOUR_MICROSECONDS = 60 * ONE_MINUTE_MICROSECONDS;
constexpr static int64_t NANOS_PER_MICROSECOND = 1000;
constexpr static int64_t ONE_SECOND_NANOSECONDS =
ONE_SECOND_MICROSECONDS * NANOS_PER_MICROSECOND;
constexpr static int64_t ONE_MINUTE_SECONDS = 60;
constexpr static int64_t ONE_HOUR_SECONDS = 60 * ONE_MINUTE_SECONDS;
constexpr static uint32_t MICROS_SCALE = 6;
constexpr static uint32_t NANOS_SCALE = 9;
constexpr static int64_t MAX_TIME = 838 * ONE_HOUR_MICROSECONDS + 59 * ONE_MINUTE_MICROSECONDS +
59 * ONE_SECOND_MICROSECONDS; // 838:59:59.000000

Expand All @@ -50,13 +54,13 @@ class TimeValue {
using ColumnTimeV2 = typename PrimitiveTypeTraits<TYPE_TIMEV2>::ColumnType;

#include "common/compile_check_avoid_begin.h"
static int64_t round_time(TimeType value, uint32_t scale) {
int64_t time = value;
DCHECK(scale <= MICROS_SCALE);
int64_t factor = std::pow(10, 6 - scale);
int64_t roundedValue = (time >= 0) ? (time + factor / 2) / factor * factor
: (time - factor / 2) / factor * factor;
return roundedValue;
static TimeType round_time(TimeType value, uint32_t scale) {
DCHECK(scale <= NANOS_SCALE);
int64_t time = to_nanoseconds(value);
int64_t factor = std::pow(10, NANOS_SCALE - scale);
int64_t rounded_value = (time >= 0) ? (time + factor / 2) / factor * factor
: (time - factor / 2) / factor * factor;
return from_nanoseconds(rounded_value);
}

// Construct time based on hour/minute/second/microsecond
Expand All @@ -80,6 +84,27 @@ class TimeValue {
return static_cast<TimeType>(negative ? -value : value);
}

// Construct time based on hour/minute/second/nanosecond.
template <bool CHECK = false>
static TimeType make_time_from_nanoseconds(int64_t hour, int64_t minute, int64_t second,
int64_t nanosecond = 0, bool negative = false) {
if constexpr (CHECK) {
if (std::abs(hour) > 838 || std::abs(minute) >= 60 || std::abs(second) >= 60 ||
std::abs(nanosecond) >= ONE_SECOND_NANOSECONDS) [[unlikely]] {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"Invalid time value: hour={}, minute={}, second={}, nanosecond={}",
hour, minute, second, nanosecond);
}
}
DCHECK(hour >= 0 && minute >= 0 && second >= 0 && nanosecond >= 0)
<< "Hour, minute, second and nanosecond must be non-negative but got " << hour
<< ":" << minute << ":" << second << "." << nanosecond;
const int64_t value = ((hour * ONE_HOUR_SECONDS) + (minute * ONE_MINUTE_SECONDS) + second) *
ONE_SECOND_NANOSECONDS +
nanosecond;
return from_nanoseconds(negative ? -value : value);
}

// if time is negative, ms should be negative too. in existing scenario, we ensure microsecond's bound by caller.
static TimeType init_microsecond(TimeType time, int32_t microsecond) {
DCHECK(std::signbit(time) == std::signbit(microsecond) || !time || !microsecond)
Expand All @@ -88,6 +113,14 @@ class TimeValue {

return static_cast<TimeType>(time + microsecond);
}

// If time is negative, nanosecond should be negative too. Callers validate its bound.
static TimeType init_nanosecond(TimeType time, int64_t nanosecond) {
DCHECK(std::signbit(time) == std::signbit(nanosecond) || !time || !nanosecond)
<< "Time and nanosecond must have the same sign but got " << time << " and "
<< nanosecond;
return time + static_cast<TimeType>(nanosecond) / NANOS_PER_MICROSECOND;
}
#include "common/compile_check_avoid_end.h"

// in existing scenario, we ensure microsecond's bound by caller.
Expand All @@ -107,25 +140,45 @@ class TimeValue {

/// Return the hour/minute/second part of the time, ignoring the sign
static int32_t hour(TimeType time) {
return (int32_t)std::abs(
static_cast<int64_t>(limit_with_bound(time) / ONE_HOUR_MICROSECONDS));
return static_cast<int32_t>(
std::abs(to_nanoseconds(time) / (ONE_HOUR_MICROSECONDS * NANOS_PER_MICROSECOND)));
}

static int32_t minute(TimeType time) {
return (int32_t)std::abs(
(static_cast<int64_t>(limit_with_bound(time)) % ONE_HOUR_MICROSECONDS) /
ONE_MINUTE_MICROSECONDS);
return static_cast<int32_t>(
std::abs((to_nanoseconds(time) % (ONE_HOUR_MICROSECONDS * NANOS_PER_MICROSECOND)) /
(ONE_MINUTE_MICROSECONDS * NANOS_PER_MICROSECOND)));
}

static int32_t second(TimeType time) {
return (int32_t)std::abs(
(static_cast<int64_t>(limit_with_bound(time)) / ONE_SECOND_MICROSECONDS) %
ONE_MINUTE_SECONDS);
return static_cast<int32_t>(
std::abs((to_nanoseconds(time) / ONE_SECOND_NANOSECONDS) % ONE_MINUTE_SECONDS));
}

static int32_t microsecond(TimeType time) {
return (int32_t)std::abs(static_cast<int64_t>(limit_with_bound(time)) %
ONE_SECOND_MICROSECONDS);
static int32_t microsecond(TimeType time) { return nanosecond(time) / NANOS_PER_MICROSECOND; }

static int32_t nanosecond(TimeType time) {
return static_cast<int32_t>(std::abs(to_nanoseconds(time) % ONE_SECOND_NANOSECONDS));
}

static int64_t to_nanoseconds(TimeType time) {
return std::llround(limit_with_bound(time) * NANOS_PER_MICROSECOND);
}

static TimeType from_nanoseconds(int64_t nanoseconds) {
return static_cast<TimeType>(nanoseconds) / NANOS_PER_MICROSECOND;
}

static TimeType from_nanoseconds_with_limit(__int128 nanoseconds) {
constexpr __int128 max_time_nanoseconds =
static_cast<__int128>(MAX_TIME) * NANOS_PER_MICROSECOND;
if (nanoseconds > max_time_nanoseconds) {
return MAX_TIME;
}
if (nanoseconds < -max_time_nanoseconds) {
return -MAX_TIME;
}
return from_nanoseconds(static_cast<int64_t>(nanoseconds));
}

static int8_t sign(TimeType time) { return (time < 0) ? -1 : 1; }
Expand Down Expand Up @@ -167,7 +220,7 @@ class TimeValue {
return DatetimeValueUtil::to_format_string_without_check<true>(
format, len, to, max_valid_length, 0, 0, 0, TimeValue::hour(time),
TimeValue::minute(time), TimeValue::second(time), TimeValue::microsecond(time),
nanosecond);
nanosecond < 0 ? TimeValue::nanosecond(time) : nanosecond);
}
};
} // namespace doris
Loading
Loading