Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
The table of contents is too big for display.
Diff view
Diff view
  •  
  •  
  •  
2 changes: 1 addition & 1 deletion be/src/cloud/cloud_cumulative_compaction_policy.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -169,7 +169,7 @@ int64_t CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets(
if (tablet->tablet_state() == TABLET_NOTREADY) {
// If tablet under alter, keep latest 10 version so that base tablet max version
// not merged in new tablet, and then we can copy data from base tablet
if (rowset->version().second < max_version - 10) {
if (rowset->version().second > max_version - 10) {
continue;
}
}
Expand Down
54 changes: 53 additions & 1 deletion be/src/core/data_type_serde/data_type_array_serde.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -404,7 +404,7 @@ Status DataTypeArraySerDe::write_column_to_orc(const std::string& timezone, cons
packed_nested_size, arena, options));
// String batches borrow their source bytes, but the packed column is local to this call;
// keep only those borrowed leaves in the write Arena until Writer::add() consumes them.
copy_orc_string_data_to_arena(cur_batch->elements.get(), arena);
orc_serde_utils::copy_orc_string_data_to_arena(cur_batch->elements.get(), arena);
cur_batch->elements->numElements = packed_nested_size;

cur_batch->numElements = end - start;
Expand Down Expand Up @@ -627,4 +627,56 @@ bool DataTypeArraySerDe::write_column_to_hive_text(const IColumn& column, Buffer
return true;
}

namespace {

Status decode_list_orc_values(const DataTypeSerDeSPtr& nested_serde, IColumn& nested_column,
const OrcDecodedColumnView& orc_view) {
const auto* orc_list = dynamic_cast<const ::orc::ListVectorBatch*>(orc_view.batch);
if (orc_list == nullptr) {
return Status::InternalError("Unexpected ORC list batch type {}",
orc_view.batch->toString());
}
DORIS_CHECK(orc_view.file_type != nullptr);
DORIS_CHECK(orc_view.selected_type != nullptr);
DORIS_CHECK(orc_view.file_type->getSubtypeCount() == 1);
DORIS_CHECK(orc_view.selected_type->getSubtypeCount() == 1);
DORIS_CHECK(orc_list->elements != nullptr);
const auto* file_element_type = orc_view.file_type->getSubtype(0);
const auto* selected_element_type = orc_view.selected_type->getSubtype(0);
DORIS_CHECK(file_element_type != nullptr);
DORIS_CHECK(selected_element_type != nullptr);

auto& array_column = assert_cast<ColumnArray&>(nested_column);
size_t element_size = 0;
std::vector<size_t> element_selection;
RETURN_IF_ERROR(orc_serde_utils::append_orc_offsets(
array_column.get_offsets(), orc_list->offsets, orc_view.rows, &element_size,
orc_view.selected_rows, &element_selection));
auto element_column = array_column.get_data_ptr()->assert_mutable();
const auto child_rows = orc_view.selected_rows == nullptr
? element_size
: static_cast<size_t>(orc_list->elements->numElements);
const auto* child_selection = orc_view.selected_rows == nullptr ? nullptr : &element_selection;
auto child_view = orc_serde_utils::make_child_orc_view(
orc_view, file_element_type, selected_element_type, orc_list->elements.get(),
child_rows, child_selection);
RETURN_IF_ERROR(
orc_serde_utils::read_orc_child_column(nested_serde, element_column, child_view));
array_column.get_data_ptr() = std::move(element_column);
return Status::OK();
}

} // namespace

Status DataTypeArraySerDe::read_column_from_orc(IColumn& column,
const OrcDecodedColumnView& view) const {
DORIS_CHECK(view.file_type != nullptr);
DORIS_CHECK(view.batch != nullptr);
DORIS_CHECK(view.file_type->getKind() == ::orc::TypeKind::LIST);
if (orc_serde_utils::orc_decode_row_count(view.rows, view.selected_rows) == 0) {
return Status::OK();
}
return decode_list_orc_values(nested_serde, column, view);
}

} // namespace doris
61 changes: 61 additions & 0 deletions be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include "core/data_type/primitive_type.h"
#include "core/data_type_serde/arrow_validation.h"
#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type_serde/orc_serde_utils.h"
#include "core/data_type_serde/parquet_decode_source.h"
#include "core/data_type_serde/parquet_timestamp.h"
#include "core/types.h"
Expand All @@ -51,6 +52,52 @@ static const int64_t micro_to_nano_second = 1000;

namespace {

Status decode_timestamp_orc_values(IColumn& nested_column, const OrcDecodedColumnView& orc_view,
const cctz::time_zone& timezone) {
const auto* orc_batch = dynamic_cast<const ::orc::TimestampVectorBatch*>(orc_view.batch);
if (orc_batch == nullptr) {
return Status::InternalError("Unexpected ORC timestamp batch type {}",
orc_view.batch->toString());
}
auto& data = assert_cast<ColumnDateTimeV2&>(nested_column).get_data();
const size_t old_data_size = data.size();
const auto output_rows =
orc_serde_utils::orc_decode_row_count(orc_view.rows, orc_view.selected_rows);
data.resize(old_data_size + output_rows);
for (size_t row = 0; row < output_rows; ++row) {
const auto source_row = orc_serde_utils::orc_source_row_at(row, orc_view.selected_rows);
if (orc_serde_utils::orc_row_is_null(*orc_view.batch, source_row)) {
data[old_data_size + row] = DateV2Value<DateTimeV2ValueType> {};
continue;
}
auto& value =
reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(data[old_data_size + row]);
orc_serde_utils::RoundedOrcTimestamp timestamp;
auto status = orc_serde_utils::round_orc_timestamp_to_microseconds(
orc_batch->data[source_row], orc_batch->nanoseconds[source_row], &timestamp);
if (!status.ok()) {
data.resize(old_data_size);
return status;
}
value.from_unixtime(orc_batch->data[source_row], timezone);
if (!value.is_valid_date()) {
data.resize(old_data_size);
return Status::DataQualityError(
"Decoded ORC timestamp is outside the target timezone range");
}
value.set_microsecond(timestamp.microseconds);
// Plain ORC TIMESTAMP is a civil value. Carry after timezone conversion so a fractional
// round does not jump backward or skip an hour at a daylight-saving transition.
if (timestamp.carry &&
!value.date_add_interval<TimeUnit::SECOND>(TimeInterval {TimeUnit::SECOND, 1, false})) {
data.resize(old_data_size);
return Status::DataQualityError(
"Decoded ORC timestamp is outside the target timezone range");
}
}
return Status::OK();
}

Status append_datetimev2_from_epoch_micros(ColumnDateTimeV2::Container& data,
int64_t timestamp_micros) {
static constexpr int64_t MICROS_PER_SECOND = 1000000;
Expand Down Expand Up @@ -926,4 +973,18 @@ template Status DataTypeDateTimeV2SerDe::from_decimal_strict_mode_batch<DataType
const DataTypeDecimal128::ColumnType& decimal_col, IColumn& target_col) const;
template Status DataTypeDateTimeV2SerDe::from_decimal_strict_mode_batch<DataTypeDecimal256>(
const DataTypeDecimal256::ColumnType& decimal_col, IColumn& target_col) const;

Status DataTypeDateTimeV2SerDe::read_column_from_orc(IColumn& column,
const OrcDecodedColumnView& view) const {
DORIS_CHECK(view.file_type != nullptr);
DORIS_CHECK(view.batch != nullptr);
const auto kind = view.file_type->getKind();
DORIS_CHECK(kind == ::orc::TypeKind::TIMESTAMP || kind == ::orc::TypeKind::TIMESTAMP_INSTANT);
DORIS_CHECK(view.timezone != nullptr);
if (orc_serde_utils::orc_decode_row_count(view.rows, view.selected_rows) == 0) {
return Status::OK();
}
return decode_timestamp_orc_values(column, view, *view.timezone);
}

} // namespace doris
43 changes: 43 additions & 0 deletions be/src/core/data_type_serde/data_type_datev2_serde.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
#include <fmt/core.h>

#include <cstdint>
#include <vector>

#include "common/config.h"
#include "core/column/column_const.h"
Expand All @@ -30,11 +31,13 @@
#include "core/data_type/define_primitive_type.h"
#include "core/data_type_serde/arrow_validation.h"
#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type_serde/orc_serde_utils.h"
#include "core/data_type_serde/parquet_decode_source.h"
#include "core/types.h"
#include "core/value/vdatetime_value.h"
#include "exprs/function/cast/cast_to_datev2_impl.hpp"
#include "exprs/function/cast/cast_to_string.h"
#include "storage/olap_common.h"

namespace doris {

Expand All @@ -43,6 +46,35 @@ static constexpr int32_t date_threshold = 719528;

namespace {

constexpr int32_t DORIS_DATE_EPOCH_DAYNR = 719528;

Status decode_date_orc_values(const DataTypeSerDe& serde, IColumn& column,
const OrcDecodedColumnView& orc_view) {
const auto* orc_batch = dynamic_cast<const ::orc::LongVectorBatch*>(orc_view.batch);
if (orc_batch == nullptr) {
return Status::InternalError("Unexpected ORC date batch type {}",
orc_view.batch->toString());
}
auto view = orc_serde_utils::make_orc_decoded_view(orc_view, DecodedValueKind::INT32);
NullMap null_map;
orc_serde_utils::fill_orc_decoded_null_map(*orc_view.batch, orc_view.rows,
orc_view.selected_rows, &null_map);
view.null_map = null_map.empty() ? nullptr : null_map.data();
const auto output_rows =
orc_serde_utils::orc_decode_row_count(orc_view.rows, orc_view.selected_rows);
std::vector<int32_t> date_values;
date_values.resize(output_rows);
auto& date_dict = date_day_offset_dict::get();
for (size_t row = 0; row < output_rows; ++row) {
const auto source_row = orc_serde_utils::orc_source_row_at(row, orc_view.selected_rows);
const auto date = date_dict[cast_set<int>(orc_batch->data[source_row])];
date_values[row] = cast_set<int32_t>(date.daynr() - DORIS_DATE_EPOCH_DAYNR);
}
view.values = reinterpret_cast<const uint8_t*>(date_values.data());
RETURN_IF_ERROR(orc_serde_utils::read_decoded_values(serde, column, &view));
return Status::OK();
}

Status decode_parquet_date(int32_t encoded_date, DateV2Value<DateV2ValueType>* value) {
DORIS_CHECK(value != nullptr);
const int64_t day_number = static_cast<int64_t>(encoded_date) + date_threshold;
Expand Down Expand Up @@ -696,4 +728,15 @@ template Status DataTypeDateV2SerDe::from_decimal_strict_mode_batch<DataTypeDeci
template Status DataTypeDateV2SerDe::from_decimal_strict_mode_batch<DataTypeDecimal256>(
const DataTypeDecimal256::ColumnType& decimal_col, IColumn& target_col) const;

Status DataTypeDateV2SerDe::read_column_from_orc(IColumn& column,
const OrcDecodedColumnView& view) const {
DORIS_CHECK(view.file_type != nullptr);
DORIS_CHECK(view.batch != nullptr);
DORIS_CHECK(view.file_type->getKind() == ::orc::TypeKind::DATE);
if (orc_serde_utils::orc_decode_row_count(view.rows, view.selected_rows) == 0) {
return Status::OK();
}
return decode_date_orc_values(*this, column, view);
}

} // namespace doris
122 changes: 122 additions & 0 deletions be/src/core/data_type_serde/data_type_decimal_serde.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
#include "core/data_type/storage_field_type.h"
#include "core/data_type_serde/arrow_validation.h"
#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type_serde/orc_serde_utils.h"
#include "core/data_type_serde/parquet_decode_source.h"
#include "core/types.h"
#include "exec/common/arithmetic_overflow.h"
Expand All @@ -52,6 +53,115 @@
namespace doris {
namespace {

constexpr int DECIMAL_PRECISION_FOR_HIVE11 = BeConsts::MAX_DECIMAL128_PRECISION;

Int128 to_int128(::orc::Int128 value) {
const auto high_bits = static_cast<__uint128_t>(static_cast<uint64_t>(value.getHighBits()));
const auto low_bits = static_cast<__uint128_t>(value.getLowBits());
return static_cast<Int128>((high_bits << 64) | low_bits);
}

::orc::Int128 to_orc_int128(Int128 value) {
const auto unsigned_value = static_cast<__uint128_t>(value);
return ::orc::Int128(static_cast<int64_t>(static_cast<uint64_t>(unsigned_value >> 64)),
static_cast<uint64_t>(unsigned_value));
}

Status scale_decimal_value(Int128 value, int32_t source_scale, int32_t target_scale,
Int128* scaled_value) {
DORIS_CHECK(scaled_value != nullptr);
if (source_scale == target_scale) {
*scaled_value = value;
return Status::OK();
}
if (source_scale < target_scale) {
bool overflow = false;
const auto scaled = ::orc::scaleUpInt128ByPowerOfTen(to_orc_int128(value),
target_scale - source_scale, overflow);
if (overflow) {
return Status::DataQualityError(
"ORC decimal value overflows when scaling from {} to {}", source_scale,
target_scale);
}
*scaled_value = to_int128(scaled);
return Status::OK();
}
*scaled_value = to_int128(
::orc::scaleDownInt128ByPowerOfTen(to_orc_int128(value), source_scale - target_scale));
return Status::OK();
}

void fill_decimal_big_endian_value(Int128 value, std::array<uint8_t, sizeof(Int128)>* bytes) {
DORIS_CHECK(bytes != nullptr);
const auto unsigned_value = static_cast<__uint128_t>(value);
for (size_t byte_idx = 0; byte_idx < bytes->size(); ++byte_idx) {
const auto shift = (bytes->size() - byte_idx - 1) * 8;
(*bytes)[byte_idx] = static_cast<uint8_t>(unsigned_value >> shift);
}
}

Status decode_decimal_orc_values(const DataTypeSerDe& serde, IColumn& column,
const OrcDecodedColumnView& orc_view, int32_t target_scale) {
DORIS_CHECK(orc_view.file_type != nullptr);
auto view = orc_serde_utils::make_orc_decoded_view(orc_view, DecodedValueKind::FIXED_BINARY);
view.decimal_precision = orc_view.file_type->getPrecision() == 0
? DECIMAL_PRECISION_FOR_HIVE11
: cast_set<int>(orc_view.file_type->getPrecision());
NullMap null_map;
orc_serde_utils::fill_orc_decoded_null_map(*orc_view.batch, orc_view.rows,
orc_view.selected_rows, &null_map);
view.null_map = null_map.empty() ? nullptr : null_map.data();
view.fixed_length = sizeof(Int128);

std::vector<StringRef> binary_values;
std::vector<std::array<uint8_t, sizeof(Int128)>> decimal_values;
const auto output_rows =
orc_serde_utils::orc_decode_row_count(orc_view.rows, orc_view.selected_rows);
decimal_values.resize(output_rows);
binary_values.reserve(output_rows);
if (const auto* decimal64_batch =
dynamic_cast<const ::orc::Decimal64VectorBatch*>(orc_view.batch);
decimal64_batch != nullptr) {
view.decimal_scale = decimal64_batch->scale;
for (size_t row = 0; row < output_rows; ++row) {
Int128 value = 0;
const auto source_row = orc_serde_utils::orc_source_row_at(row, orc_view.selected_rows);
if (!orc_serde_utils::orc_row_is_null(*orc_view.batch, source_row)) {
RETURN_IF_ERROR(scale_decimal_value(decimal64_batch->values[source_row],
decimal64_batch->scale, target_scale, &value));
}
fill_decimal_big_endian_value(value, &decimal_values[row]);
binary_values.emplace_back(reinterpret_cast<const char*>(decimal_values[row].data()),
decimal_values[row].size());
}
view.binary_values = &binary_values;
RETURN_IF_ERROR(orc_serde_utils::read_decoded_values(serde, column, &view));
return Status::OK();
}

const auto* decimal128_batch =
dynamic_cast<const ::orc::Decimal128VectorBatch*>(orc_view.batch);
if (decimal128_batch == nullptr) {
return Status::InternalError("Unexpected ORC decimal batch type {}",
orc_view.batch->toString());
}
view.decimal_scale = decimal128_batch->scale;
for (size_t row = 0; row < output_rows; ++row) {
Int128 value = 0;
const auto source_row = orc_serde_utils::orc_source_row_at(row, orc_view.selected_rows);
if (!orc_serde_utils::orc_row_is_null(*orc_view.batch, source_row)) {
RETURN_IF_ERROR(scale_decimal_value(to_int128(decimal128_batch->values[source_row]),
decimal128_batch->scale, target_scale, &value));
}
fill_decimal_big_endian_value(value, &decimal_values[row]);
binary_values.emplace_back(reinterpret_cast<const char*>(decimal_values[row].data()),
decimal_values[row].size());
}
view.binary_values = &binary_values;
RETURN_IF_ERROR(orc_serde_utils::read_decoded_values(serde, column, &view));
return Status::OK();
}

template <typename NativeType>
NativeType decode_big_endian_signed_integer(const uint8_t* data, int length) {
if constexpr (std::is_same_v<NativeType, wide::Int256>) {
Expand Down Expand Up @@ -1360,6 +1470,18 @@ const uint8_t* DataTypeDecimalSerDe<T>::deserialize_binary_to_field(const uint8_
return data;
}

template <PrimitiveType T>
Status DataTypeDecimalSerDe<T>::read_column_from_orc(IColumn& column,
const OrcDecodedColumnView& view) const {
DORIS_CHECK(view.file_type != nullptr);
DORIS_CHECK(view.batch != nullptr);
DORIS_CHECK(view.file_type->getKind() == ::orc::TypeKind::DECIMAL);
if (orc_serde_utils::orc_decode_row_count(view.rows, view.selected_rows) == 0) {
return Status::OK();
}
return decode_decimal_orc_values(*this, column, view, cast_set<int32_t>(scale));
}

template class DataTypeDecimalSerDe<TYPE_DECIMAL32>;
template class DataTypeDecimalSerDe<TYPE_DECIMAL64>;
template class DataTypeDecimalSerDe<TYPE_DECIMAL128I>;
Expand Down
Loading
Loading