This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 2db07c83e76 [test](external) Port the branch-4.x Arrow convertor and
Parquet timestamp tests (#68963)
2db07c83e76 is described below
commit 2db07c83e76114af4648e1a25461b529bddc2789
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Tue Oct 13 23:09:46 2026 +0800
[test](external) Port the branch-4.x Arrow convertor and Parquet timestamp
tests (#68963)
### What problem does this PR solve?
Issue Number: None
Related PR: #68381, #68441, #68396, #68883, #68516
Problem Summary:
**In short.** This ports the tests of branch-4.x #68381 (explicit,
schema-owned Arrow block convertors) and #68441 (aligned Parquet and ORC
timestamp semantics) that did not come to master with their behavior.
master has the production code through #68396 and #68883, and through
#65446 and #68516; a hunk-by-hunk comparison found no gap there. This PR
changes no production code.
**Background.**
- #68381 moved format-specific Arrow writing into per-format convertors.
For Variant, Iceberg writes `arrow.parquet.variant` extension arrays
with a `metadata, value` storage struct, and Paimon writes its own
`value, metadata` struct. #68883 brought the Variant hooks of both
convertors to master.
- #68441 made the Parquet and ORC timestamp semantics explicit: Paimon
marks every timestamp field with `timestamp_is_adjusted_to_utc` (INT96
alone cannot tell a wall clock from an instant), a Hive write always
sends `hive.parquet.time-zone` (empty means wall clock), a Hudi MOR scan
says up front whether a slice may be read as native Parquet, ALTER
CATALOG validates the time zone, and only connectors that use the Hive
INT96 time zone receive it.
**The problem, and what it cost.** These behaviors were not pinned on
master:
| Behavior | Tested on master before this PR |
|---|---|
| Iceberg Variant through `IcebergArrowBlockConvertor`: extension array,
encoded bytes unchanged, outer NULLs, Parquet Variant logical type and
field ids, at the top level and nested in LIST, MAP and STRUCT | no;
only a one-value round trip of the SerDe hook |
| `PaimonArrowBlockConvertor`: bytes and outer NULLs for encoded and
typed Variants, `ARRAY<VARIANT>` through the Paimon SerDe, rejection of
an unsupported primitive at the Paimon hook | only through the generic
Arrow path, or one row without NULLs and without byte checks |
| ALTER CATALOG rejects an invalid IANA `hive.parquet.time-zone` and
accepts `+05:45` | only the rejection of `CST` |
| A Hive write without `hive.parquet.time-zone` still sends an empty
time zone | only an explicit `""` |
| Only a Parquet Hudi MOR table advertises native Parquet slices |
Parquet only; no ORC or HFile case |
| Paimon sets `timestamp_is_adjusted_to_utc`, not only its value,
including on nested timestamps | the value only, at the top level; the
wall-clock check also passes when the flag is missing |
| A connector that does not use the Hive INT96 time zone gets none from
an HMS catalog that sets one | only the opposite case |
**How this PR fixes it.**
- `DataTypeSerDeArrowTest`: `IcebergVariantExtensionAndParquetSchema`
and `NestedIcebergVariantExtensionsAndParquetSchema`, as in branch-4.x.
They came with #66620 (Iceberg Variant writes), but they build the Arrow
schema themselves and test only code master already has.
- `PaimonArrowBlockConvertorTest` (in
`data_type_variant_v2_serde_output_test.cpp`):
`BinaryStructPreservesEncodedAndTypedBytesAndOuterNulls`,
`NestedArrayUsesPaimonSerdeRecursively` and
`BinaryStructRejectsUnsupportedPaimonPrimitive`, with master's error
text for the last one. master's generic-path versions stay.
-
`HMSExternalCatalogParquetTimeZoneTest#testAlterValidatesCandidateTimezoneBeforePublishing`,
`HiveWritePlanProviderTest` (absent property, `UTC`, `+05:45`),
`HudiForceJniTest` (ORC and HFile MOR tables),
`PaimonScanPlanProviderTest#buildSchemaInfoCarriesTimestampSemantics`
(set flag, nested timestamp, other columns unflagged) and
`PluginDrivenScanNodeScanProviderSelectionTest#connectorWithoutHiveInt96SemanticsIgnoresCatalogParquetTimezone`.
Not ported: tests of branch-4.x-only code
(`NonLocalizableInt32Predicate` from #66802, the `{1,2,3,4}` expectation
master changed on purpose in #66399, VARBINARY mapping of file TVFs),
and the Iceberg Variant writer tests that need the Iceberg Variant write
port (`VParquetWriterTest` Variant cases and the Variant assertions of
`IcebergBuildsItsOwnSchemaAndMetadata`).
**Results.** The BE suites pass (42 tests, 5 of them new) and the FE
classes pass. Each new FE assertion fails when the check it guards is
removed:
| Change to the code under test | Test that fails |
|---|---|
| ALTER CATALOG skips the time zone check of the candidate properties |
`testAlterValidatesCandidateTimezoneBeforePublishing` |
| A Hive write sets the time zone only when the property is present |
`parquetTimezoneDistinguishesExplicitWallClockFromOldCoordinator`
(absent property) |
| Hudi advertises native Parquet for any input format |
`lazyMorScanAdvertisesNativeParquetBeforeRangesArrive` (ORC) |
| Paimon leaves `timestamp_is_adjusted_to_utc` unset for wall-clock
timestamps | `buildSchemaInfoCarriesTimestampSemantics` |
| The scan node passes the catalog's Hive time zone without asking the
provider |
`connectorWithoutHiveInt96SemanticsIgnoresCatalogParquetTimezone` |
Co-authored-by: Gabriel <[email protected]>
Co-authored-by: daidai <[email protected]>
Co-authored-by: Socrates <[email protected]>
---
.../data_type_serde/data_type_serde_arrow_test.cpp | 255 +++++++++++++++++++++
.../data_type_variant_v2_serde_output_test.cpp | 118 ++++++++++
.../connector/hive/HiveWritePlanProviderTest.java | 14 +-
.../doris/connector/hudi/HudiForceJniTest.java | 27 ++-
.../paimon/PaimonScanPlanProviderTest.java | 27 ++-
.../HMSExternalCatalogParquetTimeZoneTest.java | 15 ++
...ginDrivenScanNodeScanProviderSelectionTest.java | 28 +++
7 files changed, 468 insertions(+), 16 deletions(-)
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 b3b0d0fd4ab..c32d552788d 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
@@ -22,6 +22,9 @@
#include <arrow/array/builder_decimal.h>
#include <arrow/array/builder_nested.h>
#include <arrow/array/builder_primitive.h>
+#include <arrow/extension/parquet_variant.h>
+#include <arrow/extension_type.h>
+#include <arrow/io/memory.h>
#include <arrow/record_batch.h>
#include <arrow/status.h>
#include <arrow/type.h>
@@ -35,6 +38,8 @@
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
#include <gtest/gtest.h>
+#include <parquet/api/reader.h>
+#include <parquet/arrow/writer.h>
#include <array>
#include <chrono>
@@ -59,6 +64,7 @@
#include "core/column/column_struct.h"
#include "core/column/column_varbinary.h"
#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2.h"
#include "core/data_type/common_data_type_serder_test.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_array.h"
@@ -88,6 +94,7 @@
#include "core/value/vdatetime_value.h"
#include "exec/common/arrow_column_to_doris_column.h"
#include "exprs/function/cast/cast_to_datetimev2_impl.hpp"
+#include "exprs/function/parse/variant_string_parse.h"
#include "format/arrow/arrow_block_convertor.h"
#include "format/arrow/arrow_row_batch.h"
#include "format/table/iceberg/iceberg_arrow_block_convertor.h"
@@ -948,6 +955,254 @@ TEST(DataTypeSerDeArrowTest,
PlainArrowConverterDoesNotInferIcebergUuid) {
EXPECT_NE(std::string::npos, status.to_string().find("Plain Arrow writer
is not bound"));
}
+TEST(DataTypeSerDeArrowTest, IcebergVariantExtensionAndParquetSchema) {
+ auto make_variant_column = []() {
+ JsonStringToVariantEncoder encoder({.max_json_key_length = 1024,
+ .throw_on_invalid_json = true,
+ .check_duplicate_json_path =
false});
+ for (std::string_view json :
+ {std::string_view {R"({"name":"doris","n":1})"}, std::string_view
{"null"},
+ std::string_view {R"([1,true,"x"])"}}) {
+ encoder.add_json({json.data(), json.size()});
+ }
+ auto column = ColumnVariantV2::create();
+ column->insert_encoded_batch(encoder.finish_batch());
+ return column;
+ };
+
+ auto variant_column = make_variant_column();
+ std::vector<std::string> expected_metadata;
+ std::vector<std::string> expected_values;
+ const auto source_view = variant_column->read_view();
+ for (size_t row = 0; row < variant_column->size(); ++row) {
+ const VariantRef value = source_view.value_at(row);
+ expected_metadata.emplace_back(value.metadata.data,
value.metadata.size);
+ expected_values.emplace_back(value.value.data, value.value.size);
+ }
+
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign({0, 0, 1});
+ auto nullable_variant = ColumnNullable::create(std::move(variant_column),
std::move(null_map));
+ DataTypePtr variant_type = std::make_shared<DataTypeVariantV2>();
+ DataTypePtr nullable_variant_type = make_nullable(variant_type);
+
+ Block block;
+ block.insert(
+ ColumnWithTypeAndName(nullable_variant->get_ptr(),
nullable_variant_type, "payload"));
+
+ auto variant_storage = arrow::struct_({
+ arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false),
+ });
+ auto arrow_variant = arrow::extension::variant(variant_storage);
+ auto field_id_21 = arrow::KeyValueMetadata::Make({"PARQUET:field_id"},
{"21"});
+ auto schema = arrow::schema({
+ arrow::field("payload", arrow_variant, true, field_id_21),
+ });
+
+ std::shared_ptr<arrow::RecordBatch> record_batch;
+ cctz::time_zone default_timezone;
+ Status status = convert_to_arrow_batch_for_test(
+ block, schema, arrow::default_memory_pool(), &record_batch,
default_timezone, 0,
+ block.rows(),
make_arrow_convertor<iceberg::IcebergArrowBlockConvertor>);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_NE(nullptr, record_batch);
+ ASSERT_TRUE(record_batch->ValidateFull().ok());
+
+ auto extension_array =
+
std::dynamic_pointer_cast<arrow::ExtensionArray>(record_batch->column(0));
+ ASSERT_NE(nullptr, extension_array);
+ EXPECT_EQ("arrow.parquet.variant",
extension_array->extension_type()->extension_name());
+ EXPECT_FALSE(extension_array->IsNull(0));
+ EXPECT_FALSE(extension_array->IsNull(1));
+ EXPECT_TRUE(extension_array->IsNull(2));
+ auto storage_array =
std::static_pointer_cast<arrow::StructArray>(extension_array->storage());
+ auto metadata_array =
std::static_pointer_cast<arrow::BinaryArray>(storage_array->field(0));
+ auto value_array =
std::static_pointer_cast<arrow::BinaryArray>(storage_array->field(1));
+ for (int64_t row = 0; row < 2; ++row) {
+ EXPECT_EQ(expected_metadata[row], metadata_array->GetView(row));
+ EXPECT_EQ(expected_values[row], value_array->GetView(row));
+ }
+
+ auto sink_result = arrow::io::BufferOutputStream::Create();
+ ASSERT_TRUE(sink_result.ok()) << sink_result.status();
+ auto sink = std::move(sink_result).ValueUnsafe();
+ auto writer_result =
+ ::parquet::arrow::FileWriter::Open(*schema,
arrow::default_memory_pool(), sink);
+ ASSERT_TRUE(writer_result.ok()) << writer_result.status();
+ auto writer = std::move(writer_result).ValueUnsafe();
+ ASSERT_TRUE(writer->WriteRecordBatch(*record_batch).ok());
+ ASSERT_TRUE(writer->Close().ok());
+ auto buffer_result = sink->Finish();
+ ASSERT_TRUE(buffer_result.ok()) << buffer_result.status();
+
+ auto reader = ::parquet::ParquetFileReader::Open(
+
std::make_shared<arrow::io::BufferReader>(std::move(buffer_result).ValueUnsafe()));
+ const auto* root = reader->metadata()->schema()->group_node();
+ ASSERT_EQ(1, root->field_count());
+ const auto& payload_node = root->field(0);
+ ASSERT_NE(nullptr, payload_node->logical_type());
+ EXPECT_TRUE(payload_node->logical_type()->is_variant());
+ EXPECT_EQ(21, payload_node->field_id());
+ const auto& payload_group = static_cast<const
::parquet::schema::GroupNode&>(*payload_node);
+ ASSERT_EQ(2, payload_group.field_count());
+ EXPECT_EQ("metadata", payload_group.field(0)->name());
+ EXPECT_EQ(-1, payload_group.field(0)->field_id());
+ EXPECT_EQ("value", payload_group.field(1)->name());
+ EXPECT_EQ(-1, payload_group.field(1)->field_id());
+}
+
+TEST(DataTypeSerDeArrowTest, NestedIcebergVariantExtensionsAndParquetSchema) {
+ auto make_variant_column = [](std::initializer_list<std::string_view>
json_values) {
+ JsonStringToVariantEncoder encoder({.max_json_key_length = 1024,
+ .throw_on_invalid_json = true,
+ .check_duplicate_json_path =
false});
+ for (std::string_view json : json_values) {
+ encoder.add_json({json.data(), json.size()});
+ }
+ auto column = ColumnVariantV2::create();
+ column->insert_encoded_batch(encoder.finish_batch());
+ return column;
+ };
+ auto make_null_map = [](std::initializer_list<uint8_t> values) {
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign(values);
+ return null_map;
+ };
+
+ DataTypePtr variant_type = std::make_shared<DataTypeVariantV2>();
+ DataTypePtr nullable_variant_type = make_nullable(variant_type);
+
+ auto array_values =
+ ColumnNullable::create(make_variant_column({R"({"kind":"array"})",
"null", "{}", "7"}),
+ make_null_map({0, 0, 1, 0}));
+ auto array_offsets = ColumnArray::ColumnOffsets::create();
+ array_offsets->get_data().assign({3, 4});
+ auto array_column = ColumnArray::create(std::move(array_values),
std::move(array_offsets));
+ DataTypePtr array_type = std::make_shared<DataTypeArray>(variant_type);
+
+ auto map_keys_data = ColumnString::create();
+ for (std::string_view key : {std::string_view {"object"}, std::string_view
{"json_null"},
+ std::string_view {"sql_null"}}) {
+ map_keys_data->insert_data(key.data(), key.size());
+ }
+ auto map_keys = ColumnNullable::create(std::move(map_keys_data),
make_null_map({0, 0, 0}));
+ auto map_values = ColumnNullable::create(
+ make_variant_column({R"({"kind":"map"})", "null", "{}"}),
make_null_map({0, 0, 1}));
+ auto map_offsets = ColumnArray::ColumnOffsets::create();
+ map_offsets->get_data().assign({3, 3});
+ auto map_column =
+ ColumnMap::create(std::move(map_keys), std::move(map_values),
std::move(map_offsets));
+ DataTypePtr map_type = std::make_shared<DataTypeMap>(
+ make_nullable(std::make_shared<DataTypeString>()),
nullable_variant_type);
+
+ auto labels = ColumnString::create();
+ labels->insert_data("first", 5);
+ labels->insert_data("second", 6);
+ auto struct_payloads = ColumnNullable::create(
+ make_variant_column({R"({"kind":"struct"})", "{}"}),
make_null_map({0, 1}));
+ MutableColumns struct_children;
+ struct_children.emplace_back(std::move(labels));
+ struct_children.emplace_back(std::move(struct_payloads));
+ auto struct_column = ColumnStruct::create(std::move(struct_children));
+ DataTypePtr struct_type = std::make_shared<DataTypeStruct>(
+ DataTypes {std::make_shared<DataTypeString>(),
nullable_variant_type},
+ Strings {"label", "payload"});
+
+ Block block;
+ block.insert(ColumnWithTypeAndName(std::move(array_column), array_type,
"events"));
+ block.insert(ColumnWithTypeAndName(std::move(map_column), map_type,
"attrs"));
+ block.insert(ColumnWithTypeAndName(std::move(struct_column), struct_type,
"info"));
+
+ auto field_id = [](int id) {
+ return arrow::KeyValueMetadata::Make({"PARQUET:field_id"},
{std::to_string(id)});
+ };
+ auto variant_storage = arrow::struct_({
+ arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false),
+ });
+ auto arrow_variant = arrow::extension::variant(variant_storage);
+ auto schema = arrow::schema({
+ arrow::field("events",
+ arrow::list(arrow::field("element", arrow_variant,
true, field_id(31))),
+ true, field_id(30)),
+ arrow::field("attrs",
+ std::make_shared<arrow::MapType>(
+ arrow::field("key", arrow::utf8(), false,
field_id(41)),
+ arrow::field("value", arrow_variant, true,
field_id(42))),
+ true, field_id(40)),
+ arrow::field("info",
+ arrow::struct_({
+ arrow::field("label", arrow::utf8(), true,
field_id(51)),
+ arrow::field("payload", arrow_variant, true,
field_id(52)),
+ }),
+ true, field_id(50)),
+ });
+
+ std::shared_ptr<arrow::RecordBatch> record_batch;
+ cctz::time_zone default_timezone;
+ Status status = convert_to_arrow_batch_for_test(
+ block, schema, arrow::default_memory_pool(), &record_batch,
default_timezone, 0,
+ block.rows(),
make_arrow_convertor<iceberg::IcebergArrowBlockConvertor>);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_NE(nullptr, record_batch);
+ ASSERT_TRUE(record_batch->ValidateFull().ok()) <<
record_batch->ValidateFull();
+
+ auto events =
std::static_pointer_cast<arrow::ListArray>(record_batch->column(0));
+ auto event_values =
std::dynamic_pointer_cast<arrow::ExtensionArray>(events->values());
+ ASSERT_NE(nullptr, event_values);
+ EXPECT_EQ("arrow.parquet.variant",
event_values->extension_type()->extension_name());
+ EXPECT_TRUE(event_values->IsNull(2));
+
+ auto attrs =
std::static_pointer_cast<arrow::MapArray>(record_batch->column(1));
+ auto attr_values =
std::dynamic_pointer_cast<arrow::ExtensionArray>(attrs->items());
+ ASSERT_NE(nullptr, attr_values);
+ EXPECT_EQ("arrow.parquet.variant",
attr_values->extension_type()->extension_name());
+ EXPECT_TRUE(attr_values->IsNull(2));
+
+ auto info =
std::static_pointer_cast<arrow::StructArray>(record_batch->column(2));
+ auto info_payload =
std::dynamic_pointer_cast<arrow::ExtensionArray>(info->field(1));
+ ASSERT_NE(nullptr, info_payload);
+ EXPECT_EQ("arrow.parquet.variant",
info_payload->extension_type()->extension_name());
+ EXPECT_TRUE(info_payload->IsNull(1));
+
+ auto sink_result = arrow::io::BufferOutputStream::Create();
+ ASSERT_TRUE(sink_result.ok()) << sink_result.status();
+ auto sink = std::move(sink_result).ValueUnsafe();
+ auto writer_result =
+ ::parquet::arrow::FileWriter::Open(*schema,
arrow::default_memory_pool(), sink);
+ ASSERT_TRUE(writer_result.ok()) << writer_result.status();
+ auto writer = std::move(writer_result).ValueUnsafe();
+ ASSERT_TRUE(writer->WriteRecordBatch(*record_batch).ok());
+ ASSERT_TRUE(writer->Close().ok());
+ auto buffer_result = sink->Finish();
+ ASSERT_TRUE(buffer_result.ok()) << buffer_result.status();
+
+ auto reader = ::parquet::ParquetFileReader::Open(
+
std::make_shared<arrow::io::BufferReader>(std::move(buffer_result).ValueUnsafe()));
+ const auto* root = reader->metadata()->schema()->group_node();
+ const auto& events_group = static_cast<const
::parquet::schema::GroupNode&>(*root->field(0));
+ const auto& event_list =
+ static_cast<const
::parquet::schema::GroupNode&>(*events_group.field(0));
+ const auto& event_variant = event_list.field(0);
+ ASSERT_NE(nullptr, event_variant->logical_type());
+ EXPECT_TRUE(event_variant->logical_type()->is_variant());
+ EXPECT_EQ(31, event_variant->field_id());
+
+ const auto& attrs_group = static_cast<const
::parquet::schema::GroupNode&>(*root->field(1));
+ const auto& key_value = static_cast<const
::parquet::schema::GroupNode&>(*attrs_group.field(0));
+ const auto& map_variant = key_value.field(1);
+ ASSERT_NE(nullptr, map_variant->logical_type());
+ EXPECT_TRUE(map_variant->logical_type()->is_variant());
+ EXPECT_EQ(42, map_variant->field_id());
+
+ const auto& info_group = static_cast<const
::parquet::schema::GroupNode&>(*root->field(2));
+ const auto& struct_variant = info_group.field(1);
+ ASSERT_NE(nullptr, struct_variant->logical_type());
+ EXPECT_TRUE(struct_variant->logical_type()->is_variant());
+ EXPECT_EQ(52, struct_variant->field_id());
+}
+
TEST(DataTypeSerDeArrowTest, NestedIcebergUuidStringToFixedSizeBinary) {
auto block = std::make_shared<Block>();
DataTypePtr data_type = std::make_shared<DataTypeStruct>(
diff --git
a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
index 7f7177a86b9..10697c9645c 100644
--- a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
+++ b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
@@ -32,19 +32,23 @@
#include "core/arena.h"
#include "core/assert_cast.h"
+#include "core/column/column_array.h"
#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_date_or_datetime_v2.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_variant_v2.h"
#include "core/data_type_serde/data_type_nullable_serde.h"
#include "core/data_type_serde/data_type_variant_v2_serde.h"
#include "core/string_buffer.hpp"
#include "core/value/variant/variant_parquet_encoding.h"
#include "exprs/function/parse/variant_string_parse.h"
+#include "format/table/paimon/paimon_arrow_block_convertor.h"
#include "util/mysql_row_buffer.h"
namespace doris {
@@ -244,6 +248,41 @@ void expect_binary_variant_bytes(const
DataTypeVariantV2SerDe& serde, const ICol
}
}
+void expect_paimon_variant_bytes(const IColumn& column, const ColumnVariantV2&
encoded,
+ const NullMap* null_map = nullptr) {
+ auto builder = binary_variant_arrow_builder();
+ DataTypePtr type = std::make_shared<DataTypeVariantV2>();
+ const auto serde = type->get_serde();
+ const auto field = arrow::field("payload", binary_variant_arrow_type(),
true);
+ const Status status =
+ paimon::PaimonArrowBlockConvertor(arrow::schema({field}),
cctz::utc_time_zone())
+ .write_column(type, *serde, column, null_map, field,
builder.get(), 0,
+ column.size(), cctz::utc_time_zone());
+ ASSERT_TRUE(status.ok()) << status;
+
+ std::shared_ptr<arrow::Array> output;
+ ASSERT_TRUE(builder->Finish(&output).ok());
+ const auto& array = assert_cast<const arrow::StructArray&>(*output);
+ const auto& values = assert_cast<const
arrow::BinaryArray&>(*array.field(0));
+ const auto& metadata = assert_cast<const
arrow::BinaryArray&>(*array.field(1));
+ ASSERT_EQ(array.length(), static_cast<int64_t>(column.size()));
+ const auto view = encoded.read_view();
+ for (size_t row = 0; row < column.size(); ++row) {
+ const bool expected_null = null_map != nullptr && (*null_map)[row] !=
0;
+ EXPECT_EQ(array.IsNull(row), expected_null);
+ if (expected_null) {
+ continue;
+ }
+ const VariantRef expected = view.value_at(row);
+ const auto actual_value = values.GetView(row);
+ const auto actual_metadata = metadata.GetView(row);
+ EXPECT_EQ(std::string_view(actual_value.data(), actual_value.size()),
+ std::string_view(expected.value.data, expected.value.size));
+ EXPECT_EQ(std::string_view(actual_metadata.data(),
actual_metadata.size()),
+ std::string_view(expected.metadata.data,
expected.metadata.size));
+ }
+}
+
// NOLINTNEXTLINE(readability-function-cognitive-complexity) -- GTest macros
inflate the matrix.
void expect_text_surfaces(const DataTypeVariantV2SerDe& serde, const IColumn&
encoded,
const ColumnVariantV2& typed,
@@ -489,4 +528,83 @@ TEST(DataTypeVariantV2SerdeOutputTest,
BinaryStructRejectsUnsupportedPrimitive)
EXPECT_EQ(arrow_builder->length(), 0);
}
+TEST(PaimonArrowBlockConvertorTest,
BinaryStructPreservesEncodedAndTypedBytesAndOuterNulls) {
+ auto documents = encoded_json({R"({"a":[1,null,"x"]})",
R"({"hidden":true})", "null"});
+ NullMap mask {0, 1, 0};
+ expect_paimon_variant_bytes(*documents, *documents, &mask);
+
+ auto typed = typed_strings(
+ {std::string_view("plain"), std::nullopt,
std::string_view(R"({"text":"value"})")});
+ ColumnPtr encoded = encoded_copy(*typed);
+ expect_paimon_variant_bytes(*typed, assert_cast<const
ColumnVariantV2&>(*encoded));
+ EXPECT_TRUE(typed->is_typed());
+}
+
+TEST(PaimonArrowBlockConvertorTest, NestedArrayUsesPaimonSerdeRecursively) {
+ auto variants = encoded_json({R"({"id":1})", R"({"hidden":true})",
R"([true,"x"])", "null"});
+ const auto expected = variants->read_view();
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->get_data().assign({2, 4});
+ // DataTypeArray always wraps its element type in Nullable, including
VARIANT.
+ auto nulls = ColumnUInt8::create(4, 0);
+ nulls->get_data()[1] = 1;
+ auto elements = ColumnNullable::create(std::move(variants),
std::move(nulls));
+ auto array = ColumnArray::create(std::move(elements), std::move(offsets));
+
+ DataTypePtr variant_type = std::make_shared<DataTypeVariantV2>();
+ DataTypePtr array_type = std::make_shared<DataTypeArray>(variant_type);
+ auto arrow_type = arrow::list(binary_variant_arrow_type());
+ auto field = arrow::field("payloads", arrow_type, false);
+ std::unique_ptr<arrow::ArrayBuilder> builder;
+ ASSERT_TRUE(arrow::MakeBuilder(arrow::default_memory_pool(), arrow_type,
&builder).ok());
+
+ const auto serde = array_type->get_serde();
+ Status status = paimon::PaimonArrowBlockConvertor(arrow::schema({field}),
cctz::utc_time_zone())
+ .write_column(array_type, *serde, *array, nullptr,
field, builder.get(),
+ 0, array->size(),
cctz::utc_time_zone());
+ ASSERT_TRUE(status.ok()) << status;
+
+ std::shared_ptr<arrow::Array> output;
+ ASSERT_TRUE(builder->Finish(&output).ok());
+ const auto& lists = assert_cast<const arrow::ListArray&>(*output);
+ const auto& structs = assert_cast<const
arrow::StructArray&>(*lists.values());
+ const auto& values = assert_cast<const
arrow::BinaryArray&>(*structs.field(0));
+ const auto& metadata = assert_cast<const
arrow::BinaryArray&>(*structs.field(1));
+ ASSERT_TRUE(output->ValidateFull().ok());
+ ASSERT_EQ(4, structs.length());
+ EXPECT_TRUE(structs.IsNull(1));
+ EXPECT_FALSE(structs.IsNull(3));
+ for (size_t row : {0, 2, 3}) {
+ const VariantRef value = expected.value_at(row);
+ const auto actual_value = values.GetView(row);
+ const auto actual_metadata = metadata.GetView(row);
+ EXPECT_EQ(std::string_view(actual_value.data(), actual_value.size()),
+ std::string_view(value.value.data, value.value.size));
+ EXPECT_EQ(std::string_view(actual_metadata.data(),
actual_metadata.size()),
+ std::string_view(value.metadata.data, value.metadata.size));
+ }
+}
+
+TEST(PaimonArrowBlockConvertorTest,
BinaryStructRejectsUnsupportedPaimonPrimitive) {
+ VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows = 1});
+ auto row = builder.begin_row();
+ row.add_time_ntz_micros(1'500'000);
+ row.finish();
+ auto encoded = ColumnVariantV2::create();
+ encoded->insert_encoded_batch(builder.finish_batch());
+ auto arrow_builder = binary_variant_arrow_builder();
+ DataTypePtr type = std::make_shared<DataTypeVariantV2>();
+ const auto serde = type->get_serde();
+ const auto field = arrow::field("payload", binary_variant_arrow_type(),
true);
+ const Status status =
+ paimon::PaimonArrowBlockConvertor(arrow::schema({field}),
cctz::utc_time_zone())
+ .write_column(type, *serde, *encoded, nullptr, field,
arrow_builder.get(), 0,
+ encoded->size(), cctz::utc_time_zone());
+ EXPECT_EQ(status.code(), ErrorCode::NOT_IMPLEMENTED_ERROR);
+ EXPECT_NE(status.to_string().find(
+ "Binary Variant V2 Arrow encoding does not support
primitive id 17"),
+ std::string::npos);
+ EXPECT_EQ(arrow_builder->length(), 0);
+}
+
} // namespace doris
diff --git
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
index e4594916d66..b98697a085e 100644
---
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveWritePlanProviderTest.java
@@ -135,18 +135,24 @@ public class HiveWritePlanProviderTest {
@Test
public void
parquetTimezoneDistinguishesExplicitWallClockFromOldCoordinator() {
- for (String zone : Arrays.asList("", "8:00", "Asia/Shanghai")) {
+ // null leaves the property out of the catalog entirely.
+ for (String zone : Arrays.asList(null, "", "8:00", "UTC",
"Asia/Shanghai", "+05:45")) {
RecordingHmsClient client = new RecordingHmsClient();
client.table =
tableBuilder().inputFormat(PARQUET_INPUT_FORMAT).build();
RecordingConnectorContext ctx = new RecordingConnectorContext();
Map<String, String> properties = new
HashMap<>(HiveTestProperties.minimal().getRaw());
- properties.put("hive.parquet.time-zone", zone);
+ if (zone != null) {
+ properties.put("hive.parquet.time-zone", zone);
+ }
HiveWritePlanProvider provider = new HiveWritePlanProvider(client,
HiveCatalogProperties.of(properties), ctx);
THiveTableSink sink = provider.planWrite(sessionFor(client, ctx,
Collections.singletonMap("time_zone",
"America/Los_Angeles")), handle())
.getDataSink().getHiveTableSink();
- Assertions.assertTrue(sink.isSetHiveParquetTimeZone());
- Assertions.assertEquals(zone.equals("8:00") ? "+08:00" : zone,
sink.getHiveParquetTimeZone());
+ // An absent property means wall clock exactly like an explicit
"": the field must still be set, or
+ // a backend falls back to the session time zone that an older
coordinator implied.
+ Assertions.assertTrue(sink.isSetHiveParquetTimeZone(),
String.valueOf(zone));
+ String expected = zone == null ? "" : zone.equals("8:00") ?
"+08:00" : zone;
+ Assertions.assertEquals(expected, sink.getHiveParquetTimeZone(),
String.valueOf(zone));
Assertions.assertTrue(sink.isSupportsDeferredAzureMultipart());
}
}
diff --git
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiForceJniTest.java
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiForceJniTest.java
index fed1a9cbb76..194b3203395 100644
---
a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiForceJniTest.java
+++
b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiForceJniTest.java
@@ -41,15 +41,24 @@ public class HudiForceJniTest {
@Test
public void lazyMorScanAdvertisesNativeParquetBeforeRangesArrive() {
HudiScanPlanProvider provider = new
HudiScanPlanProvider(Collections.emptyMap(), null);
- HudiTableHandle table = new HudiTableHandle.Builder("db", "t",
"s3://bucket/t", "MERGE_ON_READ")
-
.inputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat").build();
- for (boolean forceJni : Arrays.asList(false, true)) {
- Map<String, String> properties = provider.getScanNodeProperties(
-
sessionWithProps(Collections.singletonMap("force_jni_scanner",
Boolean.toString(forceJni))),
- table, Collections.emptyList(), Optional.empty());
- TFileScanRangeParams params = new TFileScanRangeParams();
- provider.populateScanLevelParams(params, properties);
- Assertions.assertEquals(!forceJni,
params.isContainsNativeParquet());
+ // Only a Parquet table can hold a log-free native Parquet slice; ORC
and HFile MOR tables must not
+ // advertise the contract (branch-4.x #68441).
+ for (String inputFormat :
Arrays.asList("org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
+
"org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat",
+ "org.apache.hudi.hadoop.realtime.HoodieOrcRealtimeInputFormat",
+
"org.apache.hudi.hadoop.realtime.HoodieHFileRealtimeInputFormat")) {
+ HudiTableHandle table = new HudiTableHandle.Builder("db", "t",
"s3://bucket/t", "MERGE_ON_READ")
+ .inputFormat(inputFormat).build();
+ boolean parquet = inputFormat.contains("Parquet");
+ for (boolean forceJni : Arrays.asList(false, true)) {
+ Map<String, String> properties =
provider.getScanNodeProperties(
+
sessionWithProps(Collections.singletonMap("force_jni_scanner",
Boolean.toString(forceJni))),
+ table, Collections.emptyList(), Optional.empty());
+ TFileScanRangeParams params = new TFileScanRangeParams();
+ provider.populateScanLevelParams(params, properties);
+ Assertions.assertEquals(!forceJni && parquet,
params.isContainsNativeParquet(),
+ inputFormat + ", force_jni_scanner=" + forceJni);
+ }
}
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
index 9223305efbb..3bda0c5eb50 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
@@ -3796,13 +3796,34 @@ public class PaimonScanPlanProviderTest {
public void buildSchemaInfoCarriesTimestampSemantics() {
List<DataField> fields = Arrays.asList(
new DataField(7, "ts", DataTypes.TIMESTAMP(9)),
- new DataField(8, "ts_ltz",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(9)));
+ new DataField(8, "ts_ltz",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(9)),
+ new DataField(9, "nested", DataTypes.ROW(DataTypes.FIELD(10,
"events",
+ DataTypes.ARRAY(DataTypes.MAP(DataTypes.STRING(),
DataTypes.ROW(
+ DataTypes.FIELD(11, "at",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(9)))))))),
+ new DataField(12, "id", DataTypes.INT()));
List<TFieldPtr> top = PaimonScanPlanProvider.buildSchemaInfo(3L,
fields, false)
.getRootField().getFields();
-
Assertions.assertFalse(top.get(0).getFieldPtr().isTimestampIsAdjustedToUtc());
-
Assertions.assertTrue(top.get(1).getFieldPtr().isTimestampIsAdjustedToUtc());
+ // Paimon writes high-precision TIMESTAMP and TIMESTAMP_LTZ with the
same unannotated INT96 encoding, so the
+ // flag must be set, false for the wall clock and true for the
instant; a wall-clock check alone also passes
+ // when the flag is missing.
+ TField wallClock = top.get(0).getFieldPtr();
+ Assertions.assertTrue(wallClock.isSetTimestampIsAdjustedToUtc());
+ Assertions.assertFalse(wallClock.isTimestampIsAdjustedToUtc());
+ TField instant = top.get(1).getFieldPtr();
+ Assertions.assertTrue(instant.isSetTimestampIsAdjustedToUtc());
+ Assertions.assertTrue(instant.isTimestampIsAdjustedToUtc());
+ // The same holds for a timestamp nested in ROW, ARRAY, MAP and ROW
(branch-4.x #68441).
+ TField nested =
top.get(2).getFieldPtr().getNestedField().getStructField().getFields().get(0).getFieldPtr()
+ .getNestedField().getArrayField().getItemField().getFieldPtr()
+ .getNestedField().getMapField().getValueField().getFieldPtr()
+
.getNestedField().getStructField().getFields().get(0).getFieldPtr();
+ Assertions.assertEquals("at", nested.getName());
+ Assertions.assertTrue(nested.isSetTimestampIsAdjustedToUtc());
+ Assertions.assertTrue(nested.isTimestampIsAdjustedToUtc());
+ // Other columns carry no flag.
+
Assertions.assertFalse(top.get(3).getFieldPtr().isSetTimestampIsAdjustedToUtc());
}
@Test
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSExternalCatalogParquetTimeZoneTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSExternalCatalogParquetTimeZoneTest.java
index 02f35a51590..b15e4fa8019 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSExternalCatalogParquetTimeZoneTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSExternalCatalogParquetTimeZoneTest.java
@@ -91,4 +91,19 @@ public class HMSExternalCatalogParquetTimeZoneTest {
Assertions.assertTrue(exception.getMessage().contains(FileFormatConstants.PROP_HIVE_PARQUET_TIME_ZONE));
}
}
+
+ @Test
+ public void testAlterValidatesCandidateTimezoneBeforePublishing() throws
Exception {
+ Map<String, String> current = baseProperties();
+ current.put(FileFormatConstants.PROP_HIVE_PARQUET_TIME_ZONE, "UTC");
+ PluginDrivenExternalCatalog catalog = catalog(current);
+ Map<String, String> updated = new HashMap<>();
+ updated.put(FileFormatConstants.PROP_HIVE_PARQUET_TIME_ZONE,
"Not/A_Timezone");
+ // ALTER must validate its candidate map, not the still-valid live
catalog properties.
+ Assertions.assertThrows(DdlException.class,
+ () -> catalog.validatePropertiesBeforeUpdate(current,
updated));
+ Assertions.assertEquals("UTC",
catalog.getProperties().get(FileFormatConstants.PROP_HIVE_PARQUET_TIME_ZONE));
+ updated.put(FileFormatConstants.PROP_HIVE_PARQUET_TIME_ZONE, "+05:45");
+ Assertions.assertTrue(catalog.validatePropertiesBeforeUpdate(current,
updated));
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java
index 6345c869de1..99b07330767 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProviderSelectionTest.java
@@ -174,4 +174,32 @@ public class PluginDrivenScanNodeScanProviderSelectionTest
{
Assertions.assertEquals("Asia/Shanghai",
Deencapsulation.invoke(node, "getHiveParquetTimeZone"));
}
+
+ @Test
+ public void
connectorWithoutHiveInt96SemanticsIgnoresCatalogParquetTimezone() throws
Exception {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(FileFormatConstants.PROP_HIVE_PARQUET_TIME_ZONE,
"Asia/Shanghai");
+ PluginDrivenExternalCatalog catalog =
Mockito.mock(PluginDrivenExternalCatalog.class);
+ Mockito.when(catalog.getType()).thenReturn("hms");
+ Mockito.when(catalog.getProperties()).thenReturn(properties);
+ PluginDrivenExternalTable table =
+ Mockito.mock(PluginDrivenExternalTable.class,
Mockito.CALLS_REAL_METHODS);
+ Deencapsulation.setField(table, "catalog", catalog);
+
+ ConnectorTableHandle handle = Mockito.mock(ConnectorTableHandle.class);
+ ConnectorScanPlanProvider provider =
Mockito.mock(ConnectorScanPlanProvider.class);
+
Mockito.when(provider.usesHiveParquetInt96TimeZone()).thenReturn(false);
+ Connector connector = Mockito.mock(Connector.class);
+
Mockito.when(connector.getScanPlanProvider(handle)).thenReturn(provider);
+
+ TupleDescriptor descriptor = new TupleDescriptor(new TupleId(0));
+ descriptor.setTable(table);
+ PluginDrivenScanNode node = new PluginDrivenScanNode(
+ new PlanNodeId(0), descriptor, false, new SessionVariable(),
ScanContext.EMPTY,
+ connector, Mockito.mock(ConnectorSession.class), handle);
+
+ // A Paimon or Iceberg table behind an HMS catalog keeps its own
timestamp semantics: the catalog's Hive
+ // INT96 timezone must not reach a provider that does not use it
(branch-4.x #68441).
+ Assertions.assertEquals("", Deencapsulation.invoke(node,
"getHiveParquetTimeZone"));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
