diff --git a/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp b/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp index 44402bb5b531cd..e4fed917b8abb3 100644 --- a/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp +++ b/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp @@ -497,6 +497,11 @@ Status DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column, if (arrow_array->type()->id() == arrow::Type::TIMESTAMP) { const auto* concrete_array = dynamic_cast(arrow_array); const auto type = std::static_pointer_cast(arrow_array->type()); + // Mirror write_column_to_arrow: a timezone-naive Arrow timestamp (empty timezone) + // stores its value as an as-if-UTC epoch, so it must be read back with UTC. + // Interpreting it with the session timezone would shift the wall-clock value for + // cross-cluster (type=doris) reads. See apache/doris#65741. + const cctz::time_zone& real_ctz = type->timezone().empty() ? cctz::utc_time_zone() : ctz; switch (type->unit()) { case arrow::TimeUnit::type::SECOND: { divisor = DIVISOR_FOR_SECOND; @@ -525,16 +530,22 @@ Status DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column, for (auto value_i = start; value_i < end; ++value_i) { const uint8_t* raw_byte_ptr = base_ptr + value_i * element_size; auto date_value = unaligned_load(raw_byte_ptr); - auto utc_epoch = static_cast(date_value); + // Keep the epoch signed: a pre-1970 DATETIMEV2 (e.g. datetime(3) + // '1969-12-31 23:59:59.500' -> raw -500) must not be cast to unsigned, and the + // sub-second remainder has to be floored toward -inf, mirroring + // append_datetimev2_from_utc_epoch_micros() used by the INT96 path above. + int64_t seconds = date_value / divisor; + int64_t sub_second = date_value % divisor; + if (sub_second < 0) { + sub_second += divisor; + --seconds; + } DateV2Value v; - // convert second - v.from_unixtime(utc_epoch / divisor, ctz); - // get rest time - // add 0 on the right to make it 6 digits. DateTimeV2Value microsecond is 6 digits, - // the scale decides to keep the first few digits, so the valid digits should be kept at the front. - // "2022-01-01 11:11:11.111", utc_epoch = 1641035471111, divisor = 1000, set_microsecond(111000) - v.set_microsecond((utc_epoch % divisor) * DIVISOR_FOR_MICRO / divisor); + v.from_unixtime(seconds, real_ctz); + // scale the sub-second remainder into microseconds (6 digits); NANO input is truncated. + // "2022-01-01 11:11:11.111", raw = 1641035471111, divisor = 1000, set_microsecond(111000) + v.set_microsecond(static_cast(sub_second * DIVISOR_FOR_MICRO / divisor)); col_data.emplace_back(v); } } else { diff --git a/be/src/exec/operator/result_sink_operator.cpp b/be/src/exec/operator/result_sink_operator.cpp index 021ffb60983180..f4258c79741b0a 100644 --- a/be/src/exec/operator/result_sink_operator.cpp +++ b/be/src/exec/operator/result_sink_operator.cpp @@ -56,8 +56,9 @@ Status ResultSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) } else { std::shared_ptr arrow_schema; if (p._sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL) { - RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(_output_vexpr_ctxs, &arrow_schema, - state->timezone())); + RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs( + _output_vexpr_ctxs, &arrow_schema, state->timezone(), + state->query_options().enable_arrow_flight_datetime_naive)); } VLOG_DEBUG << "create sender in INIT with instance id " << fragment_instance_id; RETURN_IF_ERROR(state->exec_env()->result_mgr()->create_sender( @@ -121,8 +122,9 @@ Status ResultSinkOperatorX::prepare(RuntimeState* state) { if (state->query_options().enable_parallel_result_sink) { std::shared_ptr arrow_schema; if (_sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL) { - RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(_output_vexpr_ctxs, &arrow_schema, - state->timezone())); + RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs( + _output_vexpr_ctxs, &arrow_schema, state->timezone(), + state->query_options().enable_arrow_flight_datetime_naive)); } VLOG_DEBUG << "create sender in prepare with query id " << state->query_id(); RETURN_IF_ERROR(state->exec_env()->result_mgr()->create_sender( diff --git a/be/src/format/arrow/arrow_row_batch.cpp b/be/src/format/arrow/arrow_row_batch.cpp index 9c8e94e10c4103..5a88b531d0c4e5 100644 --- a/be/src/format/arrow/arrow_row_batch.cpp +++ b/be/src/format/arrow/arrow_row_batch.cpp @@ -49,8 +49,8 @@ namespace doris { Status convert_to_arrow_type(const DataTypePtr& origin_type, - std::shared_ptr* result, - const std::string& timezone) { + std::shared_ptr* result, const std::string& timezone, + bool datetime_naive) { auto type = get_serialized_type(origin_type); switch (type->get_primitive_type()) { case TYPE_NULL: @@ -97,17 +97,25 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type, case TYPE_DATEV2: *result = std::make_shared(); break; - // TODO: maybe need to distinguish TYPE_DATETIME and TYPE_TIMESTAMPTZ - case TYPE_TIMESTAMPTZ: + // TIMESTAMPTZ has instant semantics and always keeps the timezone. DATETIMEV2 is a + // timezone-naive wall-clock type: only the Arrow Flight result path (datetime_naive) maps + // it WITHOUT timezone, so ADBC/pyarrow clients do not treat it as an instant and render a + // spurious "+00:00" (apache/doris#65741). Other callers (Parquet export, including forced + // INT96, and Python UDF) keep the timezone to preserve their existing round-trip semantics. case TYPE_DATETIMEV2: + case TYPE_TIMESTAMPTZ: { + const std::string empty_timezone; + const bool naive = type->get_primitive_type() == TYPE_DATETIMEV2 && datetime_naive; + const std::string& ts_timezone = naive ? empty_timezone : timezone; if (type->get_scale() > 3) { - *result = std::make_shared(arrow::TimeUnit::MICRO, timezone); + *result = std::make_shared(arrow::TimeUnit::MICRO, ts_timezone); } else if (type->get_scale() > 0) { - *result = std::make_shared(arrow::TimeUnit::MILLI, timezone); + *result = std::make_shared(arrow::TimeUnit::MILLI, ts_timezone); } else { - *result = std::make_shared(arrow::TimeUnit::SECOND, timezone); + *result = std::make_shared(arrow::TimeUnit::SECOND, ts_timezone); } break; + } case TYPE_DECIMALV2: case TYPE_DECIMAL32: case TYPE_DECIMAL64: @@ -123,7 +131,8 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type, case TYPE_ARRAY: { const auto* type_arr = assert_cast(remove_nullable(type).get()); std::shared_ptr item_type; - RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type, timezone)); + RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type, timezone, + datetime_naive)); *result = std::make_shared(item_type); break; } @@ -131,8 +140,10 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type, const auto* type_map = assert_cast(remove_nullable(type).get()); std::shared_ptr key_type; std::shared_ptr val_type; - RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type, timezone)); - RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type, timezone)); + RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type, timezone, + datetime_naive)); + RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type, timezone, + datetime_naive)); *result = std::make_shared(key_type, val_type); break; } @@ -141,8 +152,8 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type, std::vector> fields; for (size_t i = 0; i < type_struct->get_elements().size(); i++) { std::shared_ptr field_type; - RETURN_IF_ERROR( - convert_to_arrow_type(type_struct->get_element(i), &field_type, timezone)); + RETURN_IF_ERROR(convert_to_arrow_type(type_struct->get_element(i), &field_type, + timezone, datetime_naive)); fields.push_back( std::make_shared(type_struct->get_element_name(i), field_type, type_struct->get_element(i)->is_nullable())); @@ -206,12 +217,16 @@ Status get_arrow_schema_from_block(const Block& block, std::shared_ptr* result, - const std::string& timezone) { + const std::string& timezone, bool datetime_naive) { std::vector> fields; for (int i = 0; i < output_vexpr_ctxs.size(); i++) { std::shared_ptr arrow_type; auto root_expr = output_vexpr_ctxs.at(i)->root(); - RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type, timezone)); + // The Arrow Flight result path emits timezone-naive DATETIMEV2 when datetime_naive is set + // (session variable enable_arrow_flight_datetime_naive), so ADBC clients receive + // wall-clock values instead of timezone-aware instants (apache/doris#65741). + RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type, timezone, + datetime_naive)); auto field_name = root_expr->is_slot_ref() && !root_expr->expr_label().empty() ? root_expr->expr_label() : fmt::format("{}_{}", root_expr->data_type()->get_name(), i); diff --git a/be/src/format/arrow/arrow_row_batch.h b/be/src/format/arrow/arrow_row_batch.h index e7b77ed707b772..bb0a7f71cb6e98 100644 --- a/be/src/format/arrow/arrow_row_batch.h +++ b/be/src/format/arrow/arrow_row_batch.h @@ -43,8 +43,12 @@ constexpr size_t MAX_ARROW_UTF8 = (1ULL << 31); // 2G class RowDescriptor; +// When datetime_naive is true, TYPE_DATETIMEV2 maps to a timezone-naive Arrow timestamp. +// This is used only by the Arrow Flight result path; other callers (Parquet export, Python +// UDF, ...) keep the timezone-aware mapping to preserve their existing semantics. +// See apache/doris#65741. Status convert_to_arrow_type(const DataTypePtr& type, std::shared_ptr* result, - const std::string& timezone); + const std::string& timezone, bool datetime_naive = false); std::shared_ptr create_arrow_field_with_metadata( const std::string& field_name, const std::shared_ptr& arrow_type, @@ -55,7 +59,7 @@ Status get_arrow_schema_from_block(const Block& block, std::shared_ptr* result, - const std::string& timezone); + const std::string& timezone, bool datetime_naive = false); Status serialize_record_batch(const arrow::RecordBatch& record_batch, std::string* result); diff --git a/be/src/format/arrow/arrow_utils.cpp b/be/src/format/arrow/arrow_utils.cpp index d9c29eceed01ef..43d0723ffc11bd 100644 --- a/be/src/format/arrow/arrow_utils.cpp +++ b/be/src/format/arrow/arrow_utils.cpp @@ -17,8 +17,12 @@ #include "format/arrow/arrow_utils.h" +#include #include #include +#include + +#include "util/timezone_utils.h" namespace doris { @@ -30,6 +34,23 @@ Status to_doris_status(const arrow::Status& status) { } } +cctz::time_zone resolve_arrow_reader_timezone(const arrow::Array& array, + const cctz::time_zone& default_ctz) { + if (array.type()->id() != arrow::Type::TIMESTAMP) { + return default_ctz; + } + const std::string& timezone = + std::static_pointer_cast(array.type())->timezone(); + if (timezone.empty()) { + return default_ctz; + } + cctz::time_zone parsed; + if (TimezoneUtils::find_cctz_time_zone(timezone, parsed)) { + return parsed; + } + return default_ctz; +} + arrow::Status to_arrow_status(const Status& status) { if (LIKELY(status.ok())) { return arrow::Status::OK(); diff --git a/be/src/format/arrow/arrow_utils.h b/be/src/format/arrow/arrow_utils.h index 7794906b384475..869130d6612931 100644 --- a/be/src/format/arrow/arrow_utils.h +++ b/be/src/format/arrow/arrow_utils.h @@ -18,6 +18,7 @@ #pragma once #include +#include #include @@ -75,6 +76,14 @@ Status arrow_pretty_print(const arrow::Array& rb, std::ostream* os); Status to_doris_status(const arrow::Status& status); arrow::Status to_arrow_status(const Status& status); +// Returns the timezone to decode an Arrow timestamp `array` with. A timezone-aware Arrow timestamp +// is decoded in its own advertised timezone, so a remote DATETIMEV2 wall-clock round-trips even +// when the producer session timezone differs from the reader's default (apache/doris#65741). A +// non-timestamp or timezone-naive array returns `default_ctz` (a naive timestamp is then read as +// UTC inside the serde). Falls back to `default_ctz` when the advertised timezone cannot be parsed. +cctz::time_zone resolve_arrow_reader_timezone(const arrow::Array& array, + const cctz::time_zone& default_ctz); + template inline void assign_from_result(T& output, const arrow::Result& result) { output = *result; diff --git a/be/src/format/table/remote_doris_reader.cpp b/be/src/format/table/remote_doris_reader.cpp index 0e2184d65b62f5..6a8ed616f17c33 100644 --- a/be/src/format/table/remote_doris_reader.cpp +++ b/be/src/format/table/remote_doris_reader.cpp @@ -86,10 +86,14 @@ Status RemoteDorisReader::_do_get_next_block(Block* block, size_t* read_rows, bo try { auto block_pos = (*_col_name_to_block_idx)[column_name]; + // Decode a timezone-aware Arrow timestamp in its own advertised timezone so a remote + // DATETIMEV2 wall-clock round-trips even when the producer session timezone differs + // from this reader's default (apache/doris#65741). + const cctz::time_zone col_ctz = resolve_arrow_reader_timezone(*column, _ctzz); RETURN_IF_ERROR(columns_guard.get_datatype_by_position(block_pos) ->get_serde() ->read_column_from_arrow(*columns[block_pos], column, 0, - num_rows, _ctzz)); + num_rows, col_ctz)); } catch (Exception& e) { return Status::InternalError( "Failed to convert from arrow to block, column_name: {}, e: {}", column_name, diff --git a/be/src/format_v2/table/remote_doris_reader.cpp b/be/src/format_v2/table/remote_doris_reader.cpp index c67cece6e05b8c..ca1df684db132c 100644 --- a/be/src/format_v2/table/remote_doris_reader.cpp +++ b/be/src/format_v2/table/remote_doris_reader.cpp @@ -312,12 +312,17 @@ Status RemoteDorisFileReader::_materialize_arrow_column(const arrow::RecordBatch const auto column_name = batch.schema()->field(arrow_column_idx)->name(); auto columns_guard = file_block->mutate_columns_scoped(); auto& columns = columns_guard.mutable_columns(); + // Decode a timezone-aware Arrow timestamp in its own advertised timezone so a remote + // DATETIMEV2 wall-clock round-trips even when the producer session timezone differs from this + // reader's default (apache/doris#65741). + const auto& arrow_column = batch.column(arrow_column_idx); + const cctz::time_zone col_ctz = resolve_arrow_reader_timezone(*arrow_column, _ctz); try { RETURN_IF_ERROR(columns_guard.get_datatype_by_position(block_position.value()) ->get_serde() ->read_column_from_arrow(*columns[block_position.value()], - batch.column(arrow_column_idx).get(), 0, - batch.num_rows(), _ctz)); + arrow_column.get(), 0, batch.num_rows(), + col_ctz)); } catch (const Exception& e) { return Status::InternalError( "Failed to convert Remote Doris Arrow column '{}' (file_column_id={}) to Doris " diff --git a/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp b/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp index 7b20c2f82d0c65..f8d178582ce359 100644 --- a/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp +++ b/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp @@ -70,6 +70,7 @@ #include "core/data_type/data_type_quantilestate.h" #include "core/data_type/data_type_string.h" #include "core/data_type/data_type_struct.h" +#include "core/data_type/data_type_timestamptz.h" #include "core/data_type/define_primitive_type.h" #include "core/field.h" #include "core/types.h" @@ -637,4 +638,133 @@ TEST(DataTypeSerDeArrowTest, BlockConverterTest) { block_converter_test(cols, 7, false); } +// apache/doris#65741: DATETIMEV2 is a timezone-naive wall-clock type. Only the Arrow Flight +// result path (datetime_naive=true) maps it to a timezone-naive Arrow timestamp, so ADBC/pyarrow +// clients do not render a spurious "+00:00". Every other caller (Parquet export, Python UDF, ...) +// keeps the timezone-aware mapping to preserve its existing round-trip semantics. +TEST(DataTypeSerDeArrowTest, DateTimeV2ArrowTimezoneDependsOnNaiveFlag) { + const std::string session_tz = "Asia/Shanghai"; + for (int scale : {0, 3, 6}) { + DataTypePtr type = std::make_shared(scale); + + // Arrow Flight path: timezone-naive. + std::shared_ptr naive_type; + ASSERT_TRUE( + convert_to_arrow_type(type, &naive_type, session_tz, /*datetime_naive=*/true).ok()) + << "scale=" << scale; + ASSERT_EQ(naive_type->id(), arrow::Type::TIMESTAMP) << "scale=" << scale; + EXPECT_TRUE(std::static_pointer_cast(naive_type)->timezone().empty()) + << "DATETIMEV2 must be timezone-naive on the Flight path at scale=" << scale; + + // Default (Parquet export / Python UDF / ...): timezone preserved. + std::shared_ptr tz_type; + ASSERT_TRUE(convert_to_arrow_type(type, &tz_type, session_tz).ok()) << "scale=" << scale; + EXPECT_EQ(std::static_pointer_cast(tz_type)->timezone(), session_tz) + << "DATETIMEV2 must keep the timezone by default at scale=" << scale; + } +} + +// TIMESTAMPTZ has instant semantics: it always keeps the timezone, regardless of datetime_naive. +TEST(DataTypeSerDeArrowTest, TimestampTzAlwaysKeepsArrowTimezone) { + const std::string session_tz = "Asia/Shanghai"; + for (int scale : {0, 3, 6}) { + DataTypePtr type = std::make_shared(scale); + for (bool naive : {false, true}) { + std::shared_ptr arrow_type; + ASSERT_TRUE(convert_to_arrow_type(type, &arrow_type, session_tz, naive).ok()) + << "scale=" << scale << " naive=" << naive; + ASSERT_EQ(arrow_type->id(), arrow::Type::TIMESTAMP); + EXPECT_EQ(std::static_pointer_cast(arrow_type)->timezone(), + session_tz) + << "scale=" << scale << " naive=" << naive; + } + } +} + +// apache/doris#65741: a timezone-naive DATETIMEV2 value (the Arrow Flight encoding) must +// round-trip through Arrow unchanged even when the reader uses a non-UTC session timezone (the +// type=doris federation read path hard-codes the +08:00 default). Before the read-side fix, +// reading a naive Arrow timestamp with a non-UTC timezone shifted the wall-clock value. +TEST(DataTypeSerDeArrowTest, DateTimeV2NaiveRoundTripWithNonUtcReader) { + std::shared_ptr source_block = create_test_block({TYPE_DATETIMEV2}, 7, false); + const auto& col = source_block->get_by_position(0); + + // Build the timezone-naive Arrow schema that the Flight path produces. + std::shared_ptr naive_type; + ASSERT_TRUE( + convert_to_arrow_type(col.type, &naive_type, "Asia/Shanghai", /*datetime_naive=*/true) + .ok()); + ASSERT_TRUE(std::static_pointer_cast(naive_type)->timezone().empty()); + auto schema = arrow::schema({arrow::field(col.name, naive_type, false)}); + + cctz::time_zone shanghai; + TimezoneUtils::find_cctz_time_zone("Asia/Shanghai", shanghai); + + std::shared_ptr record_batch; + ASSERT_TRUE(convert_to_arrow_batch(*source_block, schema, arrow::default_memory_pool(), + &record_batch, shanghai) + .ok()); + + auto target_block = std::make_shared(source_block->clone_empty()); + DataTypes source_data_types = source_block->get_data_types(); + // Read back with a non-UTC session timezone; the naive value must not shift. + ASSERT_TRUE(convert_from_arrow_batch(record_batch, source_data_types, &*target_block, shanghai) + .ok()); + CommonDataTypeSerdeTest::compare_two_blocks(source_block, target_block); +} + +// apache/doris#65741: a pre-1970 DATETIMEV2 encodes as a NEGATIVE Arrow epoch on the naive Flight +// path (e.g. datetime(3) '1969-12-31 23:59:59.500' -> raw -500). The read side must keep the epoch +// signed and floor the sub-second remainder; a prior static_cast mangled it (decoding +// '.500' as '.116000' and a garbage out-of-range date). Round-trip fractional pre-epoch values. +TEST(DataTypeSerDeArrowTest, DateTimeV2NaivePre1970RoundTrip) { + struct Case { + int scale; + std::string literal; + }; + const std::vector cases = {{3, "1969-12-31 23:59:59.500"}, + {6, "1969-12-31 23:59:59.999999"}, + {0, "1969-12-31 23:59:59"}, + {6, "1930-05-06 12:34:56.654321"}}; + + cctz::time_zone utc; + TimezoneUtils::find_cctz_time_zone("UTC", utc); + + for (const auto& c : cases) { + // One-row source block holding the exact pre-1970 value. + auto column = ColumnVector::create(); + DateV2Value value; + { + CastParameters p; + ASSERT_TRUE(CastToDatetimeV2::from_string_strict_mode( + {c.literal.c_str(), c.literal.size()}, value, &utc, c.scale, p)) + << c.literal; + } + column->insert(Field::create_field(value)); + DataTypePtr type = std::make_shared(c.scale); + auto source_block = std::make_shared(); + source_block->insert(ColumnWithTypeAndName(column->get_ptr(), type, "dt")); + + // Encode with the timezone-naive Flight schema (empty Arrow timezone -> as-if-UTC epoch). + std::shared_ptr naive_type; + ASSERT_TRUE(convert_to_arrow_type(type, &naive_type, "UTC", /*datetime_naive=*/true).ok()) + << c.literal; + ASSERT_TRUE(std::static_pointer_cast(naive_type)->timezone().empty()); + auto schema = arrow::schema({arrow::field("dt", naive_type, false)}); + + std::shared_ptr record_batch; + ASSERT_TRUE(convert_to_arrow_batch(*source_block, schema, arrow::default_memory_pool(), + &record_batch, utc) + .ok()) + << c.literal; + + auto target_block = std::make_shared(source_block->clone_empty()); + DataTypes source_data_types = source_block->get_data_types(); + ASSERT_TRUE( + convert_from_arrow_batch(record_batch, source_data_types, &*target_block, utc).ok()) + << c.literal; + CommonDataTypeSerdeTest::compare_two_blocks(source_block, target_block); + } +} + } // namespace doris diff --git a/be/test/format_v2/table/remote_doris_reader_test.cpp b/be/test/format_v2/table/remote_doris_reader_test.cpp index b17f82f505c2c9..8d236813421bf7 100644 --- a/be/test/format_v2/table/remote_doris_reader_test.cpp +++ b/be/test/format_v2/table/remote_doris_reader_test.cpp @@ -18,6 +18,7 @@ #include "format_v2/table/remote_doris_reader.h" #include +#include #include #include @@ -41,6 +42,7 @@ #include "core/data_type/data_type_struct.h" #include "exprs/vexpr.h" #include "exprs/vexpr_context.h" +#include "format/arrow/arrow_utils.h" #include "format_v2/file_reader.h" #include "gen_cpp/PlanNodes_types.h" #include "io/file_factory.h" @@ -48,6 +50,7 @@ #include "runtime/runtime_profile.h" #include "runtime/runtime_state.h" #include "testutil/desc_tbl_builder.h" +#include "util/timezone_utils.h" namespace doris::format::remote_doris { namespace { @@ -314,6 +317,44 @@ TEST(RemoteDorisV2ReaderTest, BuildsSchemaFromSlotsAndProjectsRequestedColumns) EXPECT_EQ(*close_count, 1); } +// apache/doris#65741: a timezone-aware Arrow timestamp must be decoded in its OWN advertised +// timezone, so a remote DATETIMEV2 wall-clock round-trips even when the producer session timezone +// differs from this reader's hard-coded +08:00 default. Before the fix both federation readers +// passed the +08:00 default, shifting a UTC-produced value by 8 hours. +TEST(RemoteDorisV2ReaderTest, ResolvesAwareArrowTimestampTimezone) { + cctz::time_zone default_ctz; + ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone("+08:00", default_ctz)); + cctz::time_zone utc; + ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone("UTC", utc)); + + // Timezone-aware timestamp -> its own advertised timezone (UTC), not the +08:00 default. + { + auto ts_type = arrow::timestamp(arrow::TimeUnit::SECOND, "UTC"); + arrow::TimestampBuilder builder(ts_type, arrow::default_memory_pool()); + ASSERT_TRUE(builder.Append(1782956182).ok()); + std::shared_ptr array; + ASSERT_TRUE(builder.Finish(&array).ok()); + EXPECT_TRUE(resolve_arrow_reader_timezone(*array, default_ctz) == utc); + } + // Timezone-naive timestamp -> default (the serde then reads it as UTC). + { + auto ts_type = arrow::timestamp(arrow::TimeUnit::SECOND); + arrow::TimestampBuilder builder(ts_type, arrow::default_memory_pool()); + ASSERT_TRUE(builder.Append(1782956182).ok()); + std::shared_ptr array; + ASSERT_TRUE(builder.Finish(&array).ok()); + EXPECT_TRUE(resolve_arrow_reader_timezone(*array, default_ctz) == default_ctz); + } + // Non-timestamp column -> default. + { + arrow::Int32Builder builder; + ASSERT_TRUE(builder.Append(1).ok()); + std::shared_ptr array; + ASSERT_TRUE(builder.Finish(&array).ok()); + EXPECT_TRUE(resolve_arrow_reader_timezone(*array, default_ctz) == default_ctz); + } +} + TEST(RemoteDorisV2ReaderTest, BuildsComplexSchemaChildrenFromSlots) { ObjectPool pool; DescriptorTbl* desc_tbl = nullptr; diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index aa70a4f6ea6052..bf98814d33a632 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -97,6 +97,7 @@ public class SessionVariable implements Serializable, Writable { public static final String MAX_SCANNERS_CONCURRENCY = "max_scanners_concurrency"; public static final String MAX_FILE_SCANNERS_CONCURRENCY = "max_file_scanners_concurrency"; public static final String ENABLE_FILE_SCANNER_V2 = "enable_file_scanner_v2"; + public static final String ENABLE_ARROW_FLIGHT_DATETIME_NAIVE = "enable_arrow_flight_datetime_naive"; public static final String MIN_SCANNERS_CONCURRENCY = "min_scanners_concurrency"; public static final String MIN_FILE_SCANNERS_CONCURRENCY = "min_file_scanners_concurrency"; public static final String MIN_SCAN_SCHEDULER_CONCURRENCY = "min_scan_scheduler_concurrency"; @@ -1171,6 +1172,16 @@ public static double getHotValueThreshold() { "When enabled, FileScanNode uses FileScannerV2 for supported query scans. Enabled by default."}) public boolean enableFileScannerV2 = true; + @VarAttrDef.VarAttr(name = ENABLE_ARROW_FLIGHT_DATETIME_NAIVE, needForward = true, description = { + "开启后，Arrow Flight/ADBC 查询返回的 DATETIMEV2 映射为无时区(timezone-naive)的 " + + "Arrow timestamp，值为墙钟本身；默认关闭，保持带会话时区的旧行为以兼容现有客户端。" + + "(旧版 DATETIME v1 本就以字符串返回，不受影响。)", + "When enabled, DATETIMEV2 returned over Arrow Flight/ADBC is mapped to a " + + "timezone-naive Arrow timestamp carrying the wall-clock value. Disabled by default, " + + "keeping the previous timezone-aware behavior for client compatibility. " + + "(Legacy DATETIME v1 is already returned as a string and is unaffected.)"}) + public boolean enableArrowFlightDatetimeNaive = false; + @VarAttrDef.VarAttr(name = LOCAL_EXCHANGE_FREE_BLOCKS_LIMIT) public int localExchangeFreeBlocksLimit = 4; @@ -5628,6 +5639,7 @@ public TQueryOptions toThrift() { tResult.setMaxScannersConcurrency(maxScannersConcurrency); tResult.setMaxFileScannersConcurrency(maxFileScannersConcurrency); tResult.setEnableFileScannerV2(enableFileScannerV2); + tResult.setEnableArrowFlightDatetimeNaive(enableArrowFlightDatetimeNaive); tResult.setMaxColumnReaderNum(maxColumnReaderNum); tResult.setParallelPrepareThreshold(parallelPrepareThreshold); tResult.setMinScannersConcurrency(minScannersConcurrency); diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift index 578ff2fca799fb..16bd50a369f3e1 100644 --- a/gensrc/thrift/PaloInternalService.thrift +++ b/gensrc/thrift/PaloInternalService.thrift @@ -515,6 +515,12 @@ struct TQueryOptions { 1002: optional bool enable_file_scanner_v2 = false 1003: optional bool enable_topn_lazy_mat_phase2_no_write_file_cache = false 1004: optional i64 file_cache_query_limit_bytes = -1 + + // When true, DATETIMEV2 returned over Arrow Flight is mapped to a timezone-naive Arrow + // timestamp (carrying the wall-clock value); when false (default), the previous timezone-aware + // mapping is kept. Legacy DATETIME v1 is returned as a string and is unaffected. + // See apache/doris#65741. + 1005: optional bool enable_arrow_flight_datetime_naive = false } diff --git a/regression-test/data/arrow_flight_sql_p0/test_select.out b/regression-test/data/arrow_flight_sql_p0/test_select.out index 62888cd3dfcabd..439fbeabf228aa 100644 --- a/regression-test/data/arrow_flight_sql_p0/test_select.out +++ b/regression-test/data/arrow_flight_sql_p0/test_select.out @@ -7,6 +7,9 @@ 222 plsql222 2024-07-20 12:00:00.123456 2024-07-20 12:00:00.0 111 plsql111 2024-07-19 12:00:00.123456 2024-07-19 12:00:00.0 +-- !arrow_flight_sql_datetime_cast -- +2026-07-02 01:36:22.069504 + -- !arrow_flight_sql_jsonb -- 1 {"k1":1,"k2":"v2"} 2 [1,2,{"nested":true}] diff --git a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_datetime_naive_tz.out b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_datetime_naive_tz.out new file mode 100644 index 00000000000000..68b4330daf5dec --- /dev/null +++ b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_datetime_naive_tz.out @@ -0,0 +1,5 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !sql -- +1 2026-07-02T01:36:22 2026-07-02T01:36:22.069 2026-07-02T01:36:22.069504 +2 2026-07-02T23:30:45 2026-07-02T23:30:45.654 2026-07-02T23:30:45.654321 + diff --git a/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_datetime_naive.groovy b/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_datetime_naive.groovy new file mode 100644 index 00000000000000..7cd2c74c539286 --- /dev/null +++ b/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_datetime_naive.groovy @@ -0,0 +1,76 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// apache/doris#65741: session variable enable_arrow_flight_datetime_naive. +// When enabled, DATETIME/DATETIMEV2 is returned over Arrow Flight as a timezone-naive Arrow +// timestamp carrying the wall-clock value. Read as a plain string (getString(), like pyarrow), +// the value is exact and independent of the client timezone. Note that reading it as +// java.sql.Timestamp (getObject()) anchors the naive value with the client JVM timezone, so that +// path is timezone-dependent by nature -- see the getObject on/off difference check below. +// The BE-side schema mapping for both switch states is asserted in the BE unit test +// DataTypeSerDeArrowTest.DateTimeV2ArrowTimezoneDependsOnNaiveFlag. +suite("test_arrow_flight_datetime_naive", "arrow_flight_sql") { + def q = "select cast('2024-07-19 12:00:00.123456' as datetime(6)) as ts" + // Use one Arrow Flight connection so the SET applies to the same session as the query. + def conn = context.getArrowFlightSqlConnection() + + def setNaive = { boolean on -> + conn.createStatement().withCloseable { st -> + st.execute("set enable_arrow_flight_datetime_naive = " + on) + } + } + def readString = { + conn.createStatement().withCloseable { st -> + def rs = st.executeQuery(q) + assertTrue(rs.next()) + return rs.getString(1) + } + } + def readObject = { + conn.createStatement().withCloseable { st -> + def rs = st.executeQuery(q) + assertTrue(rs.next()) + return String.valueOf(rs.getObject(1)) + } + } + + // Exercise both the parallel and the serial Arrow Flight result-sink code paths, since the + // naive schema is built independently in each (result_sink_operator.cpp prepare/init). + for (boolean parallelSink : [true, false]) { + conn.createStatement().withCloseable { st -> + st.execute("set enable_parallel_result_sink = " + parallelSink) + } + + // With the switch ON, getString() returns the wall-clock value unchanged, in any client tz. + setNaive(true) + assertEquals("2024-07-19 12:00:00.123456", readString()) + + // The switch changes the wire encoding. On a non-UTC client, the timezone-aware (off) and + // timezone-naive (on) values differ when read as java.sql.Timestamp via getObject(). This + // also confirms the SET actually took effect over Arrow Flight. On a UTC client this cannot + // be observed via JDBC, so the wire encoding itself is asserted in the BE unit test + // DataTypeSerDeArrowTest.DateTimeV2ArrowTimezoneDependsOnNaiveFlag. + if (java.util.TimeZone.getDefault().getRawOffset() != 0) { + setNaive(false) + def off = readObject() + setNaive(true) + def on = readObject() + logger.info("arrow flight datetime getObject (parallelSink=${parallelSink}): off=${off}, on=${on}".toString()) + assertTrue(off != on) + } + } +} diff --git a/regression-test/suites/arrow_flight_sql_p0/test_select.groovy b/regression-test/suites/arrow_flight_sql_p0/test_select.groovy index 85f119fc2c3a9b..22f47c02b1cbab 100644 --- a/regression-test/suites/arrow_flight_sql_p0/test_select.groovy +++ b/regression-test/suites/arrow_flight_sql_p0/test_select.groovy @@ -41,6 +41,12 @@ suite("test_select", "arrow_flight_sql") { qt_arrow_flight_sql_datetime "select * from ${tableName} order by id desc" + // apache/doris#65741: by default (session variable enable_arrow_flight_datetime_naive = false) + // DATETIME(V2) keeps the timezone-aware Arrow mapping, so the wall-clock value round-trips + // unchanged over Arrow Flight. The opt-in timezone-naive mapping is verified in the BE unit + // test DataTypeSerDeArrowTest.DateTimeV2ArrowTimezoneDependsOnNaiveFlag. + qt_arrow_flight_sql_datetime_cast "select cast('2026-07-02 01:36:22.069504' as datetime(6)) as ts" + tableName = "test_select_jsonb" sql "DROP TABLE IF EXISTS ${tableName}" sql """ diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_datetime_naive_tz.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_datetime_naive_tz.groovy new file mode 100644 index 00000000000000..1e6914827aab43 --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_datetime_naive_tz.groovy @@ -0,0 +1,92 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Regression for apache/doris#65741 (federation datetime correctness). +// When a DATETIMEV2 value is read back through a type=doris catalog with use_arrow_flight=true, +// the wall-clock value must round-trip unchanged with no timezone shift. The producer Flight query +// runs with the DEFAULT (timezone-aware) encoding -- RemoteDorisScanNode does not enable +// enable_arrow_flight_datetime_naive on the remote session -- so this suite validates the aware +// federation round-trip. The naive read path (empty Arrow timezone -> UTC) is covered by the BE +// unit test DataTypeSerDeArrowTest.DateTimeV2NaiveRoundTripWithNonUtcReader; exercising naive +// end-to-end through federation needs the catalog to propagate the session variable to the +// producer, which is tracked as a follow-up. +suite("test_remote_doris_datetime_naive_tz", "p0,external") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + String remote_doris_thrift_port = context.config.otherConfigs.get("extFeThriftPort") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + remote_doris_thrift_port = showres[0][5] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}, thrift: ${remote_doris_thrift_port}") + + sql """DROP DATABASE IF EXISTS test_remote_doris_datetime_naive_tz_db""" + sql """CREATE DATABASE IF NOT EXISTS test_remote_doris_datetime_naive_tz_db""" + + sql """ + CREATE TABLE `test_remote_doris_datetime_naive_tz_db`.`t` ( + `k` int NOT NULL, + `dt0` datetime(0) NULL, + `dt3` datetime(3) NULL, + `dt6` datetime(6) NULL + ) ENGINE=OLAP + DUPLICATE KEY(`k`) + DISTRIBUTED BY HASH(`k`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + // Row 1 = the exact value from the issue report. + sql """ + INSERT INTO `test_remote_doris_datetime_naive_tz_db`.`t` values + (1, '2026-07-02 01:36:22', '2026-07-02 01:36:22.069', '2026-07-02 01:36:22.069504') + """ + // Row 2 = a late-night value: a session-timezone shift (e.g. +08:00) would push it to the + // next day (07:30), so any regression on the receive side is glaringly visible. + sql """ + INSERT INTO `test_remote_doris_datetime_naive_tz_db`.`t` values + (2, '2026-07-02 23:30:45', '2026-07-02 23:30:45.654', '2026-07-02 23:30:45.654321') + """ + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_datetime_naive_tz_catalog` + """ + + sql """ + CREATE CATALOG `test_remote_doris_datetime_naive_tz_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'fe_thrift_hosts' = '${remote_doris_host}:${remote_doris_thrift_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}', + 'use_arrow_flight' = 'true' + ); + """ + + qt_sql """ + select k, dt0, dt3, dt6 from `test_remote_doris_datetime_naive_tz_catalog`.`test_remote_doris_datetime_naive_tz_db`.`t` order by k + """ + + // Test objects are intentionally left in place after the assertions for failure + // investigation; the pre-setup DROP ... IF EXISTS above keeps reruns deterministic. +}