github-actions[bot] commented on code in PR #68396: URL: https://github.com/apache/doris/pull/68396#discussion_r4074355633
########## be/src/format/table/paimon/paimon_arrow_block_convertor.cpp: ########## @@ -0,0 +1,56 @@ +// 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. + +#include "format/table/paimon/paimon_arrow_block_convertor.h" + +#include <arrow/array/builder_base.h> +#include <arrow/io/memory.h> +#include <arrow/ipc/reader.h> +#include <arrow/type.h> + +namespace doris::paimon { + +Status PaimonArrowBlockConvertor::init() { + if (_arrow_schema != nullptr) { + return Status::OK(); + } + // Decode the pinned table schema here; rebuilding it from Doris types would lose + // nested nullability, timestamp precision and Paimon's physical Variant layout. + auto input = std::make_shared<arrow::io::BufferReader>( + arrow::Buffer::FromString(_serialized_schema)); + auto reader = arrow::ipc::RecordBatchStreamReader::Open(input); + if (!reader.ok()) { + return Status::InvalidArgument("Failed to deserialize Paimon Arrow schema: {}", + reader.status().ToString()); + } + _arrow_schema = reader.ValueOrDie()->schema(); + _serialized_schema.clear(); + return Status::OK(); +} + +Status PaimonArrowBlockConvertor::write_column(const std::shared_ptr<const IDataType>& type, + const DataTypeSerDe& serde, const IColumn& column, + const NullMap* null_map, + const std::shared_ptr<arrow::Field>& field, + arrow::ArrayBuilder* array_builder, int64_t start, + int64_t end, const cctz::time_zone& ctz) const { + return serde.write_column_to_paimon_arrow(type, column, null_map, Review Comment: [P2] Bind timestamps to the Paimon target unit. This new protocol is not wired to master's Paimon backend yet, but its advertised target-schema contract is already incorrect: Paimon 1.3 maps TIMESTAMP precision 7-9 to an Arrow nanosecond field, while Doris maps those columns to scale 6. The default hook derives a microsecond count only from the Doris scale and ignores the builder's NANO unit, so the int64 append succeeds but a value such as `2023-04-20 00:00:00.123456` is 1,000x too small; nested/nullable timestamps take the same path. Please scale from the target `TimestampType::unit()` (including negative epochs), or reject the binding, and add a precision-9 target test before this interface is consumed. ########## regression-test/suites/pythonudf_p0/test_python_arrow_convertor_timezone.groovy: ########## @@ -0,0 +1,88 @@ +// 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. + +suite("test_python_arrow_convertor_timezone") { + def runtimeVersion = getPythonUdfRuntimeVersion() + def originalTimezone = sql("select @@time_zone")[0][0] + def scalar = "python_arrow_convertor_scalar" + def tableFunction = "python_arrow_convertor_rows" + def aggregate = "python_arrow_convertor_max" + try { + sql "DROP TABLE IF EXISTS python_arrow_convertor_values" + sql """CREATE TABLE python_arrow_convertor_values (id INT, ts DATETIME(6)) + DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES("replication_num"="1")""" + sql """INSERT INTO python_arrow_convertor_values VALUES + (1, '1969-12-31 23:59:59.999999'), + (2, '2023-04-20 00:00:00.123456'), (3, NULL)""" + [scalar, tableFunction, aggregate].each { name -> + sql "DROP FUNCTION IF EXISTS ${name}(DATETIME(6))" + } + sql """CREATE FUNCTION ${scalar}(DATETIME(6)) RETURNS STRING + PROPERTIES("type"="PYTHON_UDF", "symbol"="evaluate", + "runtime_version"="${runtimeVersion}") AS \$\$ +def evaluate(value): + return None if value is None else value.strftime('%Y-%m-%d %H:%M:%S.%f') +\$\$""" + // Declare the yielded string as the array element; ARRAY<STRUCT<...>> emits a struct. + sql """CREATE TABLES FUNCTION ${tableFunction}(DATETIME(6)) + RETURNS ARRAY<STRING> + PROPERTIES("type"="PYTHON_UDF", "symbol"="evaluate", + "runtime_version"="${runtimeVersion}") AS \$\$ +def evaluate(value): + yield (None if value is None else value.strftime('%Y-%m-%d %H:%M:%S.%f'),) +\$\$""" + sql """CREATE AGGREGATE FUNCTION ${aggregate}(DATETIME(6)) RETURNS DATETIME(6) + PROPERTIES("type"="PYTHON_UDF", "symbol"="Maximum", + "runtime_version"="${runtimeVersion}") AS \$\$ +class Maximum: + def __init__(self): + self.value = None + @property + def aggregate_state(self): + return self.value + def accumulate(self, value): + if value is not None and (self.value is None or value > self.value): + self.value = value + def merge(self, value): + self.accumulate(value) + def finish(self): + return self.value +\$\$""" + // The Python protocol uses its declared default offset regardless of the session zone. + // Check values observed inside Python as well as returned values to detect offset cancellation. + ["UTC", "+05:45", "-03:30", "Asia/Shanghai"].each { zone -> + sql "SET time_zone = '${zone}'" + assertEquals([[true], [true], [true]], sql(""" Review Comment: [P2] Use a generated query-test result for these fixed expectations. This new regression compares deterministic rows with `assertEquals`, but the repository's regression-test rules require determined results to use `qt_`/`order_qt_` and commit the runner-generated `.out` snapshot. Please express the zone/function cases as named query tests and generate the corresponding result file. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
