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 8b0cafd9b80 [fix](arrow) correct the Flight SQL GetTables schema and
the TIMESTAMPTZ arrow reader (#66344)
8b0cafd9b80 is described below
commit 8b0cafd9b800f9b9c5152b300586632c6d1c0011
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Aug 2 08:11:13 2026 +0800
[fix](arrow) correct the Flight SQL GetTables schema and the TIMESTAMPTZ
arrow reader (#66344)
### What problem does this PR solve?
Related Issue: #65615
Problem Summary:
Two places where Doris says one thing about an Arrow value and does
another. Both were found
while reading Doris over Arrow Flight SQL, and both reproduce on master
on their own.
**1. `CommandGetTables` describes types the batches do not carry**
`FlightSqlSchemaHelper.getArrowType` is documented as mirroring
`convert_to_arrow_type` in the
backend -- the schema a client reads from `GetTables` is what it types
its columns by. Two of the
mappings had drifted from what BE actually writes:
- A `DATEV2` column was described as `Date(MILLISECOND)` -- Arrow date64
-- while BE writes
`arrow::Date32Type`, a day number. A client that trusts the schema
renders and compares the
column as a datetime and then fails on the first batch with
`not support convert to datetimev2 from arrow type: 16`.
- A complex column was described with a placeholder child: an array's
element type was the Null
type, a map's pair was a bare list, a struct had no fields at all. An
Arrow ARRAY/MAP/STRUCT
carries its element types in its children and nowhere else, so this said
"array of nothing"
about every array in the catalog while BE emitted `ListType(item)`,
`MapType(key, value)` and
`StructType(fields)` in the data.
`describeTables` already reports the whole tree --
`Column.createChildrenColumn` even names an
array's element `item` and a map's pair `key`/`value`, which is what
Arrow calls them -- so the
children are now built from it, recursively. A descriptor that reports
no children keeps the old
placeholders: a source that cannot describe its nested types is no worse
off than before.
**2. A TIMESTAMPTZ column read from Arrow was decoded by copying its
bits**
`DataTypeTimeStampTzSerDe` never overrode `read_column_from_arrow`, so
the one it inherited from
`DataTypeNumberSerDe<TYPE_TIMESTAMPTZ>` ran: its fixed-width path
`memcpy`s the array's buffer
straight into the column. Arrow hands it int64 EPOCH values; the column
stores PACKED date/time
values. Both are eight bytes wide, so the width check passes, the scan
succeeds, and every row is
silently wrong -- a Doris datetime read back over Arrow Flight SQL
rendered as `+08:05` with no
date in it.
This is reachable today through the `remote_doris` catalog. Its schema
is the remote table's
`Column` objects deserialized verbatim
(`RemoteDorisRestClient.parseColumns`), so a remote
TIMESTAMPTZ column stays TIMESTAMPTZ locally; `RemoteDorisScanNode`
plans the scan as
`FORMAT_ARROW`; and `remote_doris_reader` materializes it through this
serde. The remote side
writes the epochs with the serde's own `write_column_to_arrow`, so both
halves of that round trip
are Doris code and the mismatch is entirely internal.
The other Arrow readers -- `arrow_stream_reader` for the `arrow` file
format,
`python_udtf_function`, `paimon_cpp_reader` -- share the same serde and
are covered by the same
change; whether each of them can present a TIMESTAMPTZ column today is
not something this PR
claims either way.
The reader converts by unit into the microseconds the column stores.
NANO is divided rather than
refused, since no Doris datetime type keeps sub-microsecond digits, and
the division floors so a
pre-1970 instant does not move forward by one microsecond. The value is
read as an instant on the
UTC line, the inverse of what `write_column_to_arrow` emits, so the
round trip is exact.
Two cases are deliberate. A null slot is not converted: the bytes under
it are whatever the source
left there, and running a garbage epoch through the range check would
fail a batch whose rows are
all well-formed. An Arrow type this serde cannot decode is an error
rather than a fallback --
accepting anything eight bytes wide is how the corruption above stayed
invisible.
---
.../data_type_timestamptz_serde.cpp | 88 +++++++++
.../data_type_serde/data_type_timestamptz_serde.h | 5 +
.../data_type_serde_timestamptz_test.cpp | 85 +++++++++
.../service/arrowflight/FlightSqlSchemaHelper.java | 103 +++++++---
.../FlightSqlSchemaHelperArrowTypeTest.java | 209 +++++++++++++++++++++
5 files changed, 460 insertions(+), 30 deletions(-)
diff --git a/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
b/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
index d1ae23835c7..e9968e0bc87 100644
--- a/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
@@ -20,7 +20,9 @@
#include <arrow/builder.h>
#include <cctz/time_zone.h>
+#include "common/config.h"
#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/parquet_decode_source.h"
#include "core/data_type_serde/parquet_timestamp.h"
@@ -357,6 +359,92 @@ Status
DataTypeTimeStampTzSerDe::write_column_to_arrow(const IColumn& column,
return Status::OK();
}
+/**
+ * Reads an Arrow timestamp array into a TIMESTAMPTZ column.
+ *
+ * <p>Without this the base DataTypeNumberSerDe<TYPE_TIMESTAMPTZ> reader runs
instead, and its
+ * fixed-width path memcpy's the array's int64 epoch values straight into the
column -- whose element
+ * is a PACKED date/time value, not an epoch. Both are 8 bytes wide, so no
check catches it: the scan
+ * succeeds and every row is silently wrong. That is why an unreadable Arrow
type below is an error
+ * rather than a fallback.
+ *
+ * <p>The value is read as an instant on the UTC line, which is the inverse of
what
+ * write_column_to_arrow emits (it converts with cctz::utc_time_zone(), not
with ctz). ctz is
+ * therefore unused here: an Arrow timestamp's zone -- whether it names one or
not -- describes how
+ * to DISPLAY the instant, and TIMESTAMPTZ stores the instant itself.
+ */
+Status DataTypeTimeStampTzSerDe::read_column_from_arrow(IColumn& column,
+ const arrow::Array*
arrow_array,
+ int64_t start, int64_t
end,
+ const cctz::time_zone&
ctz) const {
+ if (config::enable_arrow_input_validation) {
+ check_arrow_no_offset(*arrow_array);
+ }
+ if (arrow_array->type()->id() != arrow::Type::TIMESTAMP) {
+ LOG(WARNING) << "not support convert to timestamptz from arrow type:"
+ << arrow_array->type()->id();
+ return Status::InternalError("not support convert to timestamptz from
arrow type: {}",
+ arrow_array->type()->id());
+ }
+ const auto* concrete_array = assert_cast<const
arrow::TimestampArray*>(arrow_array);
+ const auto type =
std::static_pointer_cast<arrow::TimestampType>(arrow_array->type());
+ // Scale each unit to the microseconds the column stores. NANO is divided
rather than refused:
+ // sub-microsecond precision is beyond what any Doris datetime type keeps,
and rejecting the
+ // column over a digit would make whole tables unreadable.
+ int64_t multiplier = 1;
+ int64_t divisor = 1;
+ switch (type->unit()) {
+ case arrow::TimeUnit::type::SECOND:
+ multiplier = 1000000;
+ break;
+ case arrow::TimeUnit::type::MILLI:
+ multiplier = 1000;
+ break;
+ case arrow::TimeUnit::type::MICRO:
+ break;
+ case arrow::TimeUnit::type::NANO:
+ divisor = 1000;
+ break;
+ default:
+ LOG(WARNING) << "not support convert to timestamptz from time_unit:"
<< type->unit();
+ return Status::InvalidArgument("not support convert to timestamptz
from time_unit: {}",
+ type->unit());
+ }
+
+ auto& col_data = assert_cast<ColumnTimeStampTz&>(column).get_data();
+ const auto* base_ptr = reinterpret_cast<const
uint8_t*>(concrete_array->raw_values());
+ const size_t element_size = sizeof(int64_t);
+ for (auto value_i = start; value_i < end; ++value_i) {
+ // One value per row including the null ones: the caller
(DataTypeNullableSerDe) has already
+ // taken the validity bitmap and hands the whole range down. The value
under a null slot is
+ // whatever the source left there, so it must not be converted -- a
garbage epoch would fail
+ // the range check below and take a well-formed batch down with it.
+ if (concrete_array->IsNull(value_i)) {
+ col_data.push_back(TimestampTzValue());
+ continue;
+ }
+ const uint8_t* raw_byte_ptr = base_ptr + value_i * element_size;
+ auto value = unaligned_load<int64_t>(raw_byte_ptr);
+ int64_t timestamp_micros = 0;
+ if (__builtin_mul_overflow(value, multiplier, ×tamp_micros)) {
+ return Status::DataQualityError(
+ "Arrow timestamp {} in unit {} overflows the microsecond
range of TIMESTAMPTZ",
+ value, static_cast<int>(type->unit()));
+ }
+ if (divisor != 1) {
+ // Floor, not truncate: C++ integer division rounds toward zero,
which would move a
+ // pre-1970 instant forward by up to one microsecond.
+ int64_t remainder = timestamp_micros % divisor;
+ timestamp_micros /= divisor;
+ if (remainder < 0) {
+ --timestamp_micros;
+ }
+ }
+ RETURN_IF_ERROR(append_timestamptz_from_utc_epoch_micros(col_data,
timestamp_micros));
+ }
+ return Status::OK();
+}
+
Status DataTypeTimeStampTzSerDe::write_column_to_orc(const std::string&
timezone,
const IColumn& column,
const NullMap* null_map,
orc::ColumnVectorBatch*
orc_col_batch,
diff --git a/be/src/core/data_type_serde/data_type_timestamptz_serde.h
b/be/src/core/data_type_serde/data_type_timestamptz_serde.h
index 28b51dd3093..ce858c31c06 100644
--- a/be/src/core/data_type_serde/data_type_timestamptz_serde.h
+++ b/be/src/core/data_type_serde/data_type_timestamptz_serde.h
@@ -68,6 +68,11 @@ public:
arrow::ArrayBuilder* array_builder, int64_t
start, int64_t end,
const cctz::time_zone& ctz) const override;
+ // Overridden rather than inherited: DataTypeNumberSerDe's reader would
memcpy Arrow's epoch
+ // integers over this column's packed values, same width and no error. See
the definition.
+ Status read_column_from_arrow(IColumn& column, const arrow::Array*
arrow_array, int64_t start,
+ int64_t end, const cctz::time_zone& ctz)
const override;
+
Status write_column_to_orc(const std::string& timezone, const IColumn&
column,
const NullMap* null_map,
orc::ColumnVectorBatch* orc_col_batch,
int64_t start, int64_t end, Arena& arena,
diff --git a/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
b/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
index b88601a02ad..584106cc635 100644
--- a/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
@@ -24,7 +24,9 @@
#include <cstddef>
#include <iostream>
+#include <limits>
#include <type_traits>
+#include <vector>
#include "core/assert_cast.h"
#include "core/column/column.h"
@@ -274,4 +276,87 @@ TEST_F(DataTypeTimeStampTzSerDeTest, binary_roundtrip) {
test_func(*serde_tz_3, column_tz_3, 3);
test_func(*serde_tz_6, column_tz_6, 6);
}
+
+// Roundtrip through Arrow. Same failure shape as binary_roundtrip above:
read_column_from_arrow
+// was inherited from DataTypeNumberSerDe, whose fixed-width path memcpy'd
Arrow's int64 EPOCH
+// values over this column's PACKED values. Both are 8 bytes, so nothing
failed -- the read
+// succeeded with every row wrong. Comparing the value that comes back against
the one that went
+// out is what makes deleting the override fail here instead of in a user's
result set.
+TEST_F(DataTypeTimeStampTzSerDeTest, ArrowRoundTrip) {
+ auto test_func = [&](const DataTypeTimeStampTzSerDe& serde,
arrow::TimeUnit::type unit,
+ uint16_t year, uint8_t month, uint8_t day, uint8_t
hour, uint8_t minute,
+ uint8_t second, uint32_t microsecond, int64_t
expected_arrow_value) {
+ auto source_column = ColumnTimeStampTz::create();
+ TimestampTzValue source_value;
+ source_value.unchecked_set_time(year, month, day, hour, minute,
second, microsecond);
+ source_column->insert_value(source_value);
+
+ arrow::TimestampBuilder builder(arrow::timestamp(unit),
arrow::default_memory_pool());
+ ASSERT_TRUE(serde.write_column_to_arrow(*source_column, nullptr,
&builder, 0,
+ source_column->size(),
cctz::utc_time_zone())
+ .ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+ // Pinned so the roundtrip cannot pass by both halves sharing one
wrong encoding: what
+ // crosses the wire is an epoch count in the unit, not the column's
packed representation.
+ ASSERT_EQ(expected_arrow_value,
+ assert_cast<const
arrow::TimestampArray*>(array.get())->Value(0));
+
+ auto dest_column = ColumnTimeStampTz::create();
+ ASSERT_TRUE(serde.read_column_from_arrow(*dest_column, array.get(), 0,
array->length(),
+ cctz::utc_time_zone())
+ .ok());
+ ASSERT_EQ(1, dest_column->size());
+ EXPECT_EQ(source_value, dest_column->get_element(0));
+ };
+
+ // Before the epoch, so a division that truncates toward zero instead of
flooring lands one
+ // unit late rather than agreeing by accident.
+ test_func(*serde_tz_6, arrow::TimeUnit::MICRO, 1969, 12, 31, 23, 59, 59,
123456, -876544);
+ test_func(*serde_tz_3, arrow::TimeUnit::MILLI, 1969, 12, 31, 23, 59, 59,
123000, -877);
+ test_func(*serde_tz_0, arrow::TimeUnit::SECOND, 1969, 12, 31, 23, 59, 59,
0, -1);
+ test_func(*serde_tz_6, arrow::TimeUnit::MICRO, 2023, 1, 2, 3, 4, 5,
123456, 1672628645123456);
+}
+
+// An Arrow type this serde cannot decode must fail the scan, not produce
values. The inherited
+// reader accepted anything 8 bytes wide, which is how the corruption above
stayed invisible.
+TEST_F(DataTypeTimeStampTzSerDeTest, ReadArrowRejectsNonTimestampType) {
+ arrow::Int64Builder builder;
+ ASSERT_TRUE(builder.Append(1672628645123456).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto dest_column = ColumnTimeStampTz::create();
+ auto st = serde_tz_6->read_column_from_arrow(*dest_column, array.get(), 0,
array->length(),
+ cctz::utc_time_zone());
+ EXPECT_FALSE(st.ok());
+ EXPECT_EQ(0, dest_column->size());
+}
+
+// A null slot's value buffer holds whatever the source left there. Converting
it would range-check
+// garbage and fail a batch whose rows are all well-formed, so nulls take a
default value instead.
+TEST_F(DataTypeTimeStampTzSerDeTest, ReadArrowIgnoresValuesUnderNulls) {
+ TimestampTzValue source_value;
+ source_value.unchecked_set_time(2023, 1, 2, 3, 4, 5, 123456);
+
+ std::vector<int64_t> values = {1672628645123456,
std::numeric_limits<int64_t>::max()};
+ // Bit 0 set, bit 1 clear: row 0 valid, row 1 null.
+ std::vector<uint8_t> validity = {0b01};
+ auto data =
arrow::ArrayData::Make(arrow::timestamp(arrow::TimeUnit::MICRO), values.size(),
+ {arrow::Buffer::Wrap(validity.data(),
validity.size()),
+ arrow::Buffer::Wrap(values.data(),
values.size())},
+ /*null_count=*/1);
+ std::shared_ptr<arrow::Array> array = arrow::MakeArray(data);
+ ASSERT_TRUE(array->IsNull(1));
+
+ auto dest_column = ColumnTimeStampTz::create();
+ ASSERT_TRUE(serde_tz_6
+ ->read_column_from_arrow(*dest_column, array.get(), 0,
array->length(),
+ cctz::utc_time_zone())
+ .ok());
+ // One value per row, nulls included: the caller took the validity bitmap
and expects the
+ // nested column to have grown by the full range.
+ ASSERT_EQ(2, dest_column->size());
+ EXPECT_EQ(source_value, dest_column->get_element(0));
+}
} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlSchemaHelper.java
b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlSchemaHelper.java
index 04389e0feee..b89abf16e83 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlSchemaHelper.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlSchemaHelper.java
@@ -121,7 +121,11 @@ public class FlightSqlSchemaHelper {
case VARIANT:
return new ArrowType.Utf8();
case DATEV2:
- return new ArrowType.Date(DateUnit.MILLISECOND);
+ // DAY, not MILLISECOND: BE writes a DATEV2 column as
arrow::Date32Type (a day number),
+ // so a MILLISECOND unit here describes the metadata as date64
while the data that
+ // follows is date32. A client that trusts this schema -- one
reading through the ADBC
+ // Flight SQL driver does -- then types the column as a
datetime and fails the read.
+ return new ArrowType.Date(DateUnit.DAY);
case DATETIMEV2:
if (scale > 3) {
return new ArrowType.Timestamp(TimeUnit.MICROSECOND,
timeZone);
@@ -285,41 +289,80 @@ public class FlightSqlSchemaHelper {
Integer tableOffset =
describeTablesResult.getTablesOffset().get(tableIndex);
for (; columnIndex < tableOffset; columnIndex++) {
TColumnDef columnDef =
describeTablesResult.getColumns().get(columnIndex);
- TColumnDesc columnDesc = columnDef.getColumnDesc();
- final ArrowType columnArrowType =
columnDescToArrowType(columnDesc);
-
- List<Field> columnArrowTypeChildren;
- // Arrow complex types may require children fields for parsing
the schema on C++
- switch (columnArrowType.getTypeID()) {
- case List:
- case LargeList:
- case FixedSizeList:
- columnArrowTypeChildren = Collections.singletonList(
-
Field.notNullable(BaseRepeatedValueVector.DATA_VECTOR_NAME,
-
ZeroVector.INSTANCE.getField().getType()));
- break;
- case Map:
- columnArrowTypeChildren = Collections.singletonList(
- Field.notNullable(MapVector.DATA_VECTOR_NAME,
new ArrowType.List()));
- break;
- case Struct:
- columnArrowTypeChildren = Collections.emptyList();
- break;
- default:
- columnArrowTypeChildren = null;
- break;
- }
-
- final Field field = new Field(columnDesc.getColumnName(),
- new FieldType(columnDesc.isIsAllowNull(),
columnArrowType, null,
- createFlightSqlColumnMetadata(dbName,
tableName, columnDesc)), columnArrowTypeChildren);
- fields.add(field);
+ fields.add(buildField(dbName, tableName,
columnDef.getColumnDesc()));
}
tableToFields.put(tableName, fields);
}
return tableToFields;
}
+ /** One column, with its nested types described down to the leaves. */
+ private static Field buildField(String dbName, String tableName,
TColumnDesc desc) {
+ ArrowType arrowType = columnDescToArrowType(desc);
+ return new Field(desc.getColumnName(),
+ new FieldType(desc.isIsAllowNull(), arrowType, null,
+ createFlightSqlColumnMetadata(dbName, tableName,
desc)),
+ arrowChildren(dbName, tableName, desc, arrowType));
+ }
+
+ /**
+ * The Arrow children of a complex column, built from the descriptor's own
children.
+ *
+ * <p>These are not decoration. An Arrow ARRAY/MAP/STRUCT type carries its
element types in its
+ * children and nowhere else, so a placeholder child says the column is an
array OF NOTHING --
+ * and BE emits the real element type in the data ({@code
convert_to_arrow_type}: ListType(item),
+ * MapType(key, value), StructType(fields)), which leaves the schema
describing one thing and the
+ * batch carrying another. A client that types its columns from this
schema (one reading through
+ * the ADBC Flight SQL driver does) then rejects the column outright.
+ *
+ * <p>{@code describeTables} already reports the tree -- {@code
Column.createChildrenColumn} names
+ * an array's element "item" and a map's pair "key"/"value", which is what
Arrow calls them too.
+ * When it reports none, the old placeholders are kept rather than an
empty child list: a source
+ * that cannot describe its nested types is no worse off than before.
+ */
+ private static List<Field> arrowChildren(String dbName, String tableName,
TColumnDesc desc,
+ ArrowType arrowType) {
+ List<TColumnDesc> children = desc.isSetChildren() ? desc.getChildren()
: Collections.emptyList();
+ switch (arrowType.getTypeID()) {
+ case List:
+ case LargeList:
+ case FixedSizeList:
+ if (children.size() != 1) {
+ return Collections.singletonList(
+
Field.notNullable(BaseRepeatedValueVector.DATA_VECTOR_NAME,
+ ZeroVector.INSTANCE.getField().getType()));
+ }
+ return Collections.singletonList(buildField(dbName, tableName,
children.get(0)));
+ case Map:
+ // Arrow spells a map as list<entries: struct<key, value>>,
with the entries struct and
+ // the key both non-nullable -- the descriptor's key
nullability is not carried over,
+ // because an Arrow map with a nullable key is not a valid
schema.
+ if (children.size() != 2) {
+ return Collections.singletonList(
+ Field.notNullable(MapVector.DATA_VECTOR_NAME, new
ArrowType.List()));
+ }
+ Field key = buildField(dbName, tableName, children.get(0));
+ Field value = buildField(dbName, tableName, children.get(1));
+ Field entries = new Field(MapVector.DATA_VECTOR_NAME,
+ new FieldType(false, new ArrowType.Struct(), null),
+ Arrays.asList(new Field(key.getName(),
+ new FieldType(false, key.getType(),
null), key.getChildren()),
+ value));
+ return Collections.singletonList(entries);
+ case Struct:
+ if (children.isEmpty()) {
+ return Collections.emptyList();
+ }
+ List<Field> structFields = new ArrayList<>(children.size());
+ for (TColumnDesc child : children) {
+ structFields.add(buildField(dbName, tableName, child));
+ }
+ return structFields;
+ default:
+ return null;
+ }
+ }
+
/**
* for FlightSqlProducer Schemas.GET_CATALOGS_SCHEMA
*/
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/FlightSqlSchemaHelperArrowTypeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/FlightSqlSchemaHelperArrowTypeTest.java
new file mode 100644
index 00000000000..147a52aa55e
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/FlightSqlSchemaHelperArrowTypeTest.java
@@ -0,0 +1,209 @@
+// 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.
+
+package org.apache.doris.service.arrowflight;
+
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.thrift.TColumnDesc;
+import org.apache.doris.thrift.TPrimitiveType;
+
+import org.apache.arrow.vector.complex.BaseRepeatedValueVector;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ReadChannel;
+import org.apache.arrow.vector.ipc.message.MessageSerializer;
+import org.apache.arrow.vector.types.DateUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.channels.Channels;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * What {@code CommandGetTables} says a column is, against what the query that
follows actually
+ * carries.
+ *
+ * <p><b>Why these assertions matter.</b> A Flight SQL client is entitled to
type its columns from
+ * the schema in {@code GetTables} and then read the batches without
re-deriving anything -- that is
+ * what the schema is for, and {@code getArrowType} is documented as mirroring
+ * {@code convert_to_arrow_type} in the backend. When the two disagree the
client does not get a
+ * degraded answer, it gets a failed read: it decodes the batch as the type
the metadata promised.
+ * So each case below pins the Arrow type BE emits, not merely "some" type.
+ *
+ * <p>The descriptors are built the way {@code
FrontendServiceImpl.getColumnDesc} builds them --
+ * a complex column carries its element types as {@link TColumnDesc} children,
named "item" for an
+ * array and "key"/"value" for a map by {@code Column.createChildrenColumn}.
+ */
+public class FlightSqlSchemaHelperArrowTypeTest {
+
+ private static final String DB = "test_db";
+ private static final String TABLE = "test_tbl";
+
+ private static TColumnDesc desc(String name, TPrimitiveType type) {
+ TColumnDesc columnDesc = new TColumnDesc(name, type);
+ // Nullable, so that a place where the mapping must force NOT NULL is
proved to force it
+ // rather than to inherit it.
+ columnDesc.setIsAllowNull(true);
+ return columnDesc;
+ }
+
+ private static TColumnDesc desc(String name, TPrimitiveType type,
TColumnDesc... children) {
+ TColumnDesc columnDesc = desc(name, type);
+ columnDesc.setChildren(Arrays.asList(children));
+ return columnDesc;
+ }
+
+ private static Field buildField(TColumnDesc columnDesc) {
+ return Deencapsulation.invoke(FlightSqlSchemaHelper.class,
"buildField", DB, TABLE, columnDesc);
+ }
+
+ /**
+ * BE writes a DATEV2 column as {@code arrow::Date32Type} -- a day number.
{@link DateUnit#MILLISECOND}
+ * is date64, a different width and a different meaning, and a client that
believes it renders and
+ * compares the column as a datetime, then fails the read on the first
batch with "not support convert
+ * to datetimev2 from arrow type: 16".
+ */
+ @Test
+ public void dateV2IsDescribedAsDate32() {
+ Assertions.assertEquals(new ArrowType.Date(DateUnit.DAY),
+ buildField(desc("d", TPrimitiveType.DATEV2)).getType());
+ }
+
+ /**
+ * An Arrow list carries its element type in its child and nowhere else,
so the placeholder child this
+ * replaced ({@code ZeroVector}'s Null type) described every array in the
catalog as an array OF
+ * NOTHING while BE emitted {@code ListType(item)} in the data.
+ */
+ @Test
+ public void arrayDescribesItsElementType() {
+ Field array = buildField(desc("a", TPrimitiveType.ARRAY, desc("item",
TPrimitiveType.INT)));
+
+ Assertions.assertEquals(ArrowType.ArrowTypeID.List,
array.getType().getTypeID());
+ Assertions.assertEquals(1, array.getChildren().size());
+ Field item = array.getChildren().get(0);
+ Assertions.assertEquals("item", item.getName());
+ Assertions.assertEquals(new ArrowType.Int(32, true), item.getType());
+ }
+
+ /**
+ * Arrow spells a map as {@code list<entries: struct<key, value>>}. Both
the entries struct and the key
+ * are non-nullable in a valid Arrow schema, so the descriptor's
nullability must not be carried over to
+ * the key even though it is carried over everywhere else.
+ */
+ @Test
+ public void mapDescribesKeyAndValue() {
+ Field map = buildField(desc("m", TPrimitiveType.MAP,
+ desc("key", TPrimitiveType.VARCHAR), desc("value",
TPrimitiveType.INT)));
+
+ Assertions.assertEquals(ArrowType.ArrowTypeID.Map,
map.getType().getTypeID());
+ Assertions.assertEquals(1, map.getChildren().size());
+
+ Field entries = map.getChildren().get(0);
+ Assertions.assertEquals(MapVector.DATA_VECTOR_NAME, entries.getName());
+ Assertions.assertEquals(ArrowType.ArrowTypeID.Struct,
entries.getType().getTypeID());
+ Assertions.assertFalse(entries.isNullable(), "an arrow map's entries
struct is never nullable");
+
+ List<Field> pair = entries.getChildren();
+ Assertions.assertEquals(2, pair.size());
+ Assertions.assertEquals("key", pair.get(0).getName());
+ Assertions.assertEquals(new ArrowType.Utf8(), pair.get(0).getType());
+ Assertions.assertFalse(pair.get(0).isNullable(), "an arrow map with a
nullable key is not a valid schema");
+ Assertions.assertEquals("value", pair.get(1).getName());
+ Assertions.assertEquals(new ArrowType.Int(32, true),
pair.get(1).getType());
+ }
+
+ /** A struct with no fields is not "a struct", it is a column the client
cannot read at all. */
+ @Test
+ public void structDescribesItsFields() {
+ Field struct = buildField(desc("s", TPrimitiveType.STRUCT,
+ desc("f1", TPrimitiveType.INT), desc("f2",
TPrimitiveType.STRING)));
+
+ Assertions.assertEquals(ArrowType.ArrowTypeID.Struct,
struct.getType().getTypeID());
+ Assertions.assertEquals(2, struct.getChildren().size());
+ Assertions.assertEquals("f1", struct.getChildren().get(0).getName());
+ Assertions.assertEquals(new ArrowType.Int(32, true),
struct.getChildren().get(0).getType());
+ Assertions.assertEquals("f2", struct.getChildren().get(1).getName());
+ Assertions.assertEquals(new ArrowType.Utf8(),
struct.getChildren().get(1).getType());
+ }
+
+ /** Nesting is where a per-column fix would have stopped: the descriptor's
tree is walked to the leaves. */
+ @Test
+ public void nestedComplexTypesAreDescribedToTheLeaves() {
+ Field outer = buildField(desc("a", TPrimitiveType.ARRAY,
+ desc("item", TPrimitiveType.MAP,
+ desc("key", TPrimitiveType.VARCHAR),
+ desc("value", TPrimitiveType.ARRAY, desc("item",
TPrimitiveType.BIGINT)))));
+
+ Field innerMap = outer.getChildren().get(0);
+ Assertions.assertEquals(ArrowType.ArrowTypeID.Map,
innerMap.getType().getTypeID());
+ List<Field> pair = innerMap.getChildren().get(0).getChildren();
+ Assertions.assertEquals(new ArrowType.Utf8(), pair.get(0).getType());
+
+ Field innerArray = pair.get(1);
+ Assertions.assertEquals(ArrowType.ArrowTypeID.List,
innerArray.getType().getTypeID());
+ Assertions.assertEquals(new ArrowType.Int(64, true),
innerArray.getChildren().get(0).getType());
+ }
+
+ /**
+ * A descriptor that reports no children keeps the placeholders rather
than an empty child list: a source
+ * that cannot describe its nested types is no worse off than it was
before this mapping existed.
+ */
+ @Test
+ public void complexColumnWithoutChildrenKeepsThePlaceholder() {
+ Field array = buildField(desc("a", TPrimitiveType.ARRAY));
+ Assertions.assertEquals(1, array.getChildren().size());
+ Assertions.assertEquals(BaseRepeatedValueVector.DATA_VECTOR_NAME,
array.getChildren().get(0).getName());
+ Assertions.assertEquals(ArrowType.ArrowTypeID.Null,
array.getChildren().get(0).getType().getTypeID());
+
+ Field map = buildField(desc("m", TPrimitiveType.MAP));
+ Assertions.assertEquals(1, map.getChildren().size());
+ Assertions.assertEquals(MapVector.DATA_VECTOR_NAME,
map.getChildren().get(0).getName());
+ Assertions.assertEquals(ArrowType.ArrowTypeID.List,
map.getChildren().get(0).getType().getTypeID());
+
+ Assertions.assertTrue(buildField(desc("s",
TPrimitiveType.STRUCT)).getChildren().isEmpty());
+ }
+
+ /** A scalar column has no children to describe, and gaining one would
change how it is read. */
+ @Test
+ public void scalarColumnHasNoChildren() {
+ Assertions.assertTrue(buildField(desc("i",
TPrimitiveType.INT)).getChildren().isEmpty());
+ }
+
+ /**
+ * The client does not see the {@link Field} objects, it sees the
serialized schema in the
+ * {@code table_schema} column of {@code GetTables}. Asserting after a
round trip through that encoding
+ * is what proves the element types actually reach it.
+ */
+ @Test
+ public void theSerializedSchemaCarriesTheChildren() throws IOException {
+ byte[] serialized =
FlightSqlSchemaHelper.getSerializedSchema(Collections.singletonList(
+ buildField(desc("a", TPrimitiveType.ARRAY, desc("item",
TPrimitiveType.INT)))));
+
+ Schema schema = MessageSerializer.deserializeSchema(
+ new ReadChannel(Channels.newChannel(new
ByteArrayInputStream(serialized))));
+
+ Field array = schema.getFields().get(0);
+ Assertions.assertEquals(ArrowType.ArrowTypeID.List,
array.getType().getTypeID());
+ Assertions.assertEquals(new ArrowType.Int(32, true),
array.getChildren().get(0).getType());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
