From d9c26bc5c0876ff3409a1d51692b3ed6bf3a26a5 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 14 Sep 2026 10:19:01 +0800 Subject: [PATCH 1/5] [Feature](lance) Support nested Arrow Null types --- be/src/core/data_type/data_type.h | 4 +- .../data_type_nullable_serde.cpp | 12 + .../format_v2/lance/lance_reader_helper.cpp | 19 +- .../lance/lance_nested_null_test.cpp | 270 ++++++++++++++++++ be/test/format_v2/table/lance_reader_test.cpp | 19 -- .../scripts/lance_build_nested_null.py | 61 ++++ ...0-d6a5ba97-9164-4b3e-b773-15de06005c54.txn | Bin 0 -> 643 bytes .../_versions/18446744073709551614.manifest | Bin 0 -> 1363 bytes .../_versions/latest_version_hint.json | 1 + ...11010110112b5b4742fd82cf05955fa974a3.lance | Bin 0 -> 3278 bytes .../datasource/lance/LanceTypeConverter.java | 19 +- .../lance/LanceTypeConverterTest.java | 47 ++- .../lance/test_lance_nested_null.out | 11 + .../lance/test_lance_nested_null.groovy | 66 +++++ 14 files changed, 473 insertions(+), 56 deletions(-) create mode 100644 be/test/format_v2/lance/lance_nested_null_test.cpp create mode 100644 docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py create mode 100644 docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_transactions/0-d6a5ba97-9164-4b3e-b773-15de06005c54.txn create mode 100644 docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/18446744073709551614.manifest create mode 100644 docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/latest_version_hint.json create mode 100644 docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/data/0110100101011111010110112b5b4742fd82cf05955fa974a3.lance create mode 100644 regression-test/data/external_table_p0/lance/test_lance_nested_null.out create mode 100644 regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy diff --git a/be/src/core/data_type/data_type.h b/be/src/core/data_type/data_type.h index b976809bbfd6bf..552729d7d2fcca 100644 --- a/be/src/core/data_type/data_type.h +++ b/be/src/core/data_type/data_type.h @@ -170,7 +170,9 @@ class IDataType : private boost::noncopyable { auto node = ptype->add_types(); node->set_type(TTypeNodeType::SCALAR); auto scalar_type = node->mutable_scalar_type(); - scalar_type->set_type(doris::to_thrift(get_primitive_type())); + // NULL uses UInt8 internally; preserve its logical type even inside complex schemas. + scalar_type->set_type(is_null_literal() ? TPrimitiveType::NULL_TYPE + : doris::to_thrift(get_primitive_type())); to_protobuf(ptype, node, scalar_type); } #ifdef BE_TEST diff --git a/be/src/core/data_type_serde/data_type_nullable_serde.cpp b/be/src/core/data_type_serde/data_type_nullable_serde.cpp index dd84eda304c037..115b37ea2a891c 100644 --- a/be/src/core/data_type_serde/data_type_nullable_serde.cpp +++ b/be/src/core/data_type_serde/data_type_nullable_serde.cpp @@ -389,6 +389,18 @@ Status DataTypeNullableSerDe::read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start, int64_t end, const cctz::time_zone& ctz) const { + if (arrow_array->type_id() == arrow::Type::NA) { + // Arrow Null has neither a validity bitmap nor values, even when sliced. Avoid the + // physical SerDe for Doris's UInt8 placeholder and keep nested values and nulls aligned. + if (arrow_array->offset() < 0 || start < 0 || end < start || end > arrow_array->length()) { + return Status::InvalidArgument( + "Invalid Arrow Null read range: start={}, end={}, " + "length={}, offset={}", + start, end, arrow_array->length(), arrow_array->offset()); + } + column.insert_many_defaults(end - start); + return Status::OK(); + } if (config::enable_arrow_input_validation) { check_arrow_array_range(*arrow_array, start, end); check_arrow_validity_bitmap(*arrow_array); diff --git a/be/src/format_v2/lance/lance_reader_helper.cpp b/be/src/format_v2/lance/lance_reader_helper.cpp index c47e67a38a1e21..313535a70dd39c 100644 --- a/be/src/format_v2/lance/lance_reader_helper.cpp +++ b/be/src/format_v2/lance/lance_reader_helper.cpp @@ -138,9 +138,9 @@ Status get_lance_extension(const std::shared_ptr& field, *extension_name, field->name()); } -// Map an Arrow field to a Doris type, allowing Doris NULL only at the top level. +// Map Arrow fields recursively, preserving Null leaves in complex types. Status arrow_field_to_doris_type(const std::shared_ptr& field, - DataTypePtr* doris_type, bool allow_null) { + DataTypePtr* doris_type) { const auto nullable_primitive = [&](PrimitiveType type, int precision = 0, int scale = 0, int len = -1) { *doris_type = @@ -166,10 +166,7 @@ Status arrow_field_to_doris_type(const std::shared_ptr& field, switch (arrow_type->id()) { case arrow::Type::NA: - return allow_null ? nullable_primitive(TYPE_NULL) - : Status::NotSupported( - "nested Arrow null type is unsupported for Lance field '{}'", - field->name()); + return nullable_primitive(TYPE_NULL); case arrow::Type::BOOL: return nullable_primitive(TYPE_BOOLEAN); case arrow::Type::INT8: @@ -237,7 +234,7 @@ Status arrow_field_to_doris_type(const std::shared_ptr& field, case arrow::Type::FIXED_SIZE_LIST: { const auto list = std::static_pointer_cast(arrow_type); DataTypePtr value_type; - RETURN_IF_ERROR(arrow_field_to_doris_type(list->value_field(), &value_type, false)); + RETURN_IF_ERROR(arrow_field_to_doris_type(list->value_field(), &value_type)); *doris_type = make_nullable(std::make_shared(value_type)); return Status::OK(); } @@ -245,8 +242,8 @@ Status arrow_field_to_doris_type(const std::shared_ptr& field, const auto map = std::static_pointer_cast(arrow_type); DataTypePtr key_type; DataTypePtr item_type; - RETURN_IF_ERROR(arrow_field_to_doris_type(map->key_field(), &key_type, false)); - RETURN_IF_ERROR(arrow_field_to_doris_type(map->item_field(), &item_type, false)); + RETURN_IF_ERROR(arrow_field_to_doris_type(map->key_field(), &key_type)); + RETURN_IF_ERROR(arrow_field_to_doris_type(map->item_field(), &item_type)); *doris_type = make_nullable(std::make_shared(key_type, item_type)); return Status::OK(); } @@ -258,7 +255,7 @@ Status arrow_field_to_doris_type(const std::shared_ptr& field, field_names.reserve(struct_type->num_fields()); for (const auto& child : struct_type->fields()) { DataTypePtr field_type; - RETURN_IF_ERROR(arrow_field_to_doris_type(child, &field_type, false)); + RETURN_IF_ERROR(arrow_field_to_doris_type(child, &field_type)); field_types.emplace_back(std::move(field_type)); field_names.emplace_back(child->name()); } @@ -766,7 +763,7 @@ Status convert_arrow_schema_to_doris(const std::shared_ptr& arrow return Status::InvalidArgument("duplicate Lance schema column: {}", field->name()); } DataTypePtr doris_type; - const auto type_status = arrow_field_to_doris_type(field, &doris_type, true); + const auto type_status = arrow_field_to_doris_type(field, &doris_type); if (type_status.is()) { parsed_types.emplace_back(std::make_shared()); } else { diff --git a/be/test/format_v2/lance/lance_nested_null_test.cpp b/be/test/format_v2/lance/lance_nested_null_test.cpp new file mode 100644 index 00000000000000..e9fd061a299034 --- /dev/null +++ b/be/test/format_v2/lance/lance_nested_null_test.cpp @@ -0,0 +1,270 @@ +// 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 +#include + +#include "common/config.h" +#include "core/column/column_array.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_factory.hpp" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_struct.h" +#include "format_v2/lance/lance_reader_helper.h" + +namespace doris::format::lance { +namespace { + +class LanceNestedNullTest : public testing::Test { +protected: + void SetUp() override { + _old_validation = config::enable_arrow_input_validation; + config::enable_arrow_input_validation = true; + } + void TearDown() override { config::enable_arrow_input_validation = _old_validation; } + + DataTypePtr null_type() { + return DataTypeFactory::instance().create_data_type(TYPE_NULL, true); + } + + void expect_nulls(const IColumn& column, size_t count) { + const auto& nullable = assert_cast(column); + ASSERT_EQ(count, nullable.size()); + ASSERT_EQ(count, nullable.get_nested_column().size()); + for (size_t row = 0; row < count; ++row) { + EXPECT_TRUE(nullable.is_null_at(row)); + } + } + + void read(const std::shared_ptr& array, const DataTypePtr& type, + MutableColumnPtr* column) { + std::shared_ptr normalized; + ASSERT_TRUE(normalize_lance_arrow_array(arrow::field("value", array->type()), array, + arrow::default_memory_pool(), &normalized) + .ok()); + *column = type->create_column(); + ASSERT_TRUE(type->get_serde() + ->read_column_from_arrow(**column, normalized.get(), 0, + normalized->length(), cctz::time_zone {}) + .ok()); + } + +private: + bool _old_validation; +}; + +TEST_F(LanceNestedNullTest, MapsNestedNullSchemas) { + const auto null_field = arrow::field("value", arrow::null()); + const auto schema = arrow::schema({ + arrow::field("list", arrow::list(null_field)), + arrow::field("large_list", arrow::large_list(null_field)), + arrow::field("fixed_list", arrow::fixed_size_list(null_field, 2)), + arrow::field("struct", arrow::struct_({null_field})), + arrow::field("map", arrow::map(arrow::utf8(), null_field)), + arrow::field("nested", arrow::list(arrow::struct_({null_field}))), + }); + std::vector names; + std::vector types; + ASSERT_TRUE(convert_arrow_schema_to_doris(schema, &names, &types).ok()); + ASSERT_EQ(6, types.size()); + for (const auto& type : types) { + EXPECT_NE(INVALID_TYPE, type->get_primitive_type()); + } + for (size_t i = 0; i < 3; ++i) { + ASSERT_EQ(TYPE_ARRAY, types[i]->get_primitive_type()); + EXPECT_TRUE(assert_cast(*remove_nullable(types[i])) + .get_nested_type() + ->is_null_literal()); + } +} + +TEST_F(LanceNestedNullTest, PreservesNullTypeInNestedSchemaRpc) { + const auto null = null_type(); + const auto type = make_nullable(std::make_shared( + DataTypes {null, make_nullable(std::make_shared(null)), + make_nullable(std::make_shared( + DataTypeFactory::instance().create_data_type(TYPE_STRING, true), + null))}, + Strings {"scalar", "list", "map"})); + PTypeDesc descriptor; + type->to_protobuf(&descriptor); + ASSERT_EQ(7, descriptor.types_size()); + for (int index : {1, 3, 6}) { + EXPECT_EQ(TPrimitiveType::NULL_TYPE, descriptor.types(index).scalar_type().type()); + } +} + +TEST_F(LanceNestedNullTest, ReadsNullLeafRangesWithoutPhysicalBuffers) { + const auto type = null_type(); + const auto source = std::make_shared(7)->Slice(2, 4); + for (bool validation : {false, true}) { + config::enable_arrow_input_validation = validation; + auto column = type->create_column(); + column->insert_default(); + ASSERT_TRUE( + type->get_serde() + ->read_column_from_arrow(*column, source.get(), 1, 3, cctz::time_zone {}) + .ok()); + expect_nulls(*column, 3); + ASSERT_TRUE( + type->get_serde() + ->read_column_from_arrow(*column, source.get(), 4, 4, cctz::time_zone {}) + .ok()); + expect_nulls(*column, 3); + } +} + +TEST_F(LanceNestedNullTest, RejectsInvalidNullRangesBeforeAppending) { + const auto type = null_type(); + const arrow::NullArray source(3); + for (bool validation : {false, true}) { + config::enable_arrow_input_validation = validation; + auto column = type->create_column(); + column->insert_default(); + for (const auto& [start, end] : + std::vector> {{-1, 1}, {2, 1}, {0, 4}}) { + EXPECT_FALSE(type->get_serde() + ->read_column_from_arrow(*column, &source, start, end, + cctz::time_zone {}) + .ok()); + expect_nulls(*column, 1); + } + } +} + +TEST_F(LanceNestedNullTest, ReadsNullStructParents) { + const std::vector validity {0x05}; + const auto source = std::make_shared( + arrow::struct_({arrow::field("empty", arrow::null())}), 3, + arrow::ArrayVector {std::make_shared(3)}, + arrow::Buffer::Wrap(validity), 1); + const auto type = make_nullable( + std::make_shared(DataTypes {null_type()}, Strings {"empty"})); + MutableColumnPtr column; + read(source->Slice(1, 2), type, &column); + ASSERT_NE(nullptr, column.get()); + const auto& parent = assert_cast(*column); + EXPECT_EQ((NullMap {1, 0}), parent.get_null_map_data()); + const auto& structure = assert_cast(parent.get_nested_column()); + expect_nulls(structure.get_column(0), 2); +} + +TEST_F(LanceNestedNullTest, ReadsNullListElementsAndPreservesParentShape) { + const auto null = null_type(); + const auto type = make_nullable(std::make_shared(null)); + const auto values = std::make_shared(5); + const std::vector offsets {0, 2, 2, 2, 5}; + const std::vector large_offsets {0, 2, 2, 2, 5}; + const std::vector validity {0x0d}; + const std::vector> arrays { + std::make_shared(arrow::list(arrow::null()), 4, + arrow::Buffer::Wrap(offsets), values, + arrow::Buffer::Wrap(validity), 1), + std::make_shared(arrow::large_list(arrow::null()), 4, + arrow::Buffer::Wrap(large_offsets), values, + arrow::Buffer::Wrap(validity), 1), + }; + for (const auto& source : arrays) { + for (bool sliced : {false, true}) { + MutableColumnPtr column; + read(sliced ? source->Slice(1, 3) : source, type, &column); + ASSERT_NE(nullptr, column.get()); + const auto& parent = assert_cast(*column); + EXPECT_EQ((sliced ? NullMap {1, 0, 0} : NullMap {0, 1, 0, 0}), + parent.get_null_map_data()); + const auto& list = assert_cast(parent.get_nested_column()); + EXPECT_EQ((sliced ? ColumnArray::Offsets64 {0, 0, 3} + : ColumnArray::Offsets64 {2, 2, 2, 5}), + list.get_offsets()); + expect_nulls(list.get_data(), sliced ? 3 : 5); + } + } +} + +TEST_F(LanceNestedNullTest, ReadsFixedNullLists) { + const std::vector validity {0x05}; + const auto source = std::make_shared( + arrow::fixed_size_list(arrow::null(), 2), 3, std::make_shared(6), + arrow::Buffer::Wrap(validity), 1); + const auto type = make_nullable(std::make_shared(null_type())); + MutableColumnPtr column; + read(source->Slice(1, 2), type, &column); + ASSERT_NE(nullptr, column.get()); + const auto& parent = assert_cast(*column); + EXPECT_EQ((NullMap {1, 0}), parent.get_null_map_data()); + const auto& list = assert_cast(parent.get_nested_column()); + EXPECT_EQ((ColumnArray::Offsets64 {2, 4}), list.get_offsets()); + expect_nulls(list.get_data(), 4); +} + +TEST_F(LanceNestedNullTest, ReadsNullStructFieldsInsideLists) { + const auto nulls = std::make_shared(3); + const std::vector ids {10, 20, 30}; + const auto id_array = std::make_shared(3, arrow::Buffer::Wrap(ids)); + const auto struct_type = arrow::struct_( + {arrow::field("empty", arrow::null()), arrow::field("id", arrow::int32())}); + const auto structs = std::make_shared(struct_type, 3, + arrow::ArrayVector {nulls, id_array}); + const std::vector offsets {0, 1, 3}; + const auto source = std::make_shared(arrow::list(struct_type), 2, + arrow::Buffer::Wrap(offsets), structs); + const auto doris_struct = make_nullable(std::make_shared( + DataTypes {null_type(), DataTypeFactory::instance().create_data_type(TYPE_INT, true)}, + Strings {"empty", "id"})); + MutableColumnPtr column; + read(source->Slice(1, 1), make_nullable(std::make_shared(doris_struct)), + &column); + ASSERT_NE(nullptr, column.get()); + const auto& list = assert_cast( + assert_cast(*column).get_nested_column()); + const auto& result = assert_cast( + assert_cast(list.get_data()).get_nested_column()); + expect_nulls(result.get_column(0), 2); + const auto& id_column = assert_cast( + assert_cast(result.get_column(1)).get_nested_column()); + EXPECT_EQ((ColumnInt32::Container {20, 30}), id_column.get_data()); +} + +TEST_F(LanceNestedNullTest, ReadsNullMapValues) { + const std::vector keys {10, 20, 30}; + const std::vector offsets {0, 1, 1, 3}; + const std::vector validity {0x05}; + const auto source = std::make_shared( + arrow::map(arrow::int32(), arrow::null()), 3, arrow::Buffer::Wrap(offsets), + std::make_shared(3, arrow::Buffer::Wrap(keys)), + std::make_shared(3), arrow::Buffer::Wrap(validity), 1); + const auto type = make_nullable(std::make_shared( + DataTypeFactory::instance().create_data_type(TYPE_INT, true), null_type())); + MutableColumnPtr column; + read(source->Slice(1, 2), type, &column); + ASSERT_NE(nullptr, column.get()); + const auto& parent = assert_cast(*column); + EXPECT_EQ((NullMap {1, 0}), parent.get_null_map_data()); + const auto& map = assert_cast(parent.get_nested_column()); + EXPECT_EQ((ColumnArray::Offsets64 {0, 2}), map.get_offsets()); + expect_nulls(map.get_values(), 2); + const auto& key_column = assert_cast( + assert_cast(map.get_keys()).get_nested_column()); + EXPECT_EQ((ColumnInt32::Container {20, 30}), key_column.get_data()); +} + +} // namespace +} // namespace doris::format::lance diff --git a/be/test/format_v2/table/lance_reader_test.cpp b/be/test/format_v2/table/lance_reader_test.cpp index cfcb4597963246..e23d7c2ea6bd21 100644 --- a/be/test/format_v2/table/lance_reader_test.cpp +++ b/be/test/format_v2/table/lance_reader_test.cpp @@ -1795,25 +1795,6 @@ TEST(LanceTableReaderSchemaTest, RejectsMalformedKnownExtensionStorage) { } } -// Verifies nested Null fields remain unsupported. -TEST(LanceTableReaderSchemaTest, MarksNestedNullTypesAsUnsupported) { - const auto arrow_schema = arrow::schema({ - arrow::field("null_list", arrow::list(arrow::field("item", arrow::null()))), - arrow::field("null_struct", arrow::struct_({arrow::field("value", arrow::null())})), - }); - - std::vector column_names; - std::vector column_types; - ASSERT_TRUE(convert_arrow_schema_to_doris(arrow_schema, &column_names, &column_types).ok()); - - EXPECT_EQ((std::vector {"null_list", "null_struct"}), column_names); - ASSERT_EQ(2, column_types.size()); - for (const auto& column_type : column_types) { - ASSERT_NE(nullptr, column_type); - EXPECT_EQ(INVALID_TYPE, column_type->get_primitive_type()); - } -} - // Verifies values, nullability, and precision when reading the additional types. TEST(LanceTableReaderTypeTest, ReadsAdditionalArrowAndLanceTypes) { const auto json_extension_metadata = diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py new file mode 100644 index 00000000000000..810a7c57778047 --- /dev/null +++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py @@ -0,0 +1,61 @@ +# 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. + +"""Generate the nested Null regression fixture with pylance 7.0.0 and PyArrow 21.0.0.""" + +import argparse +from pathlib import Path + +import lance +import pyarrow as pa + + +def build(output: Path) -> None: + null_struct = pa.struct([("empty", pa.null()), ("value", pa.int32())]) + schema = pa.schema([ + ("id", pa.int32()), + ("null_list", pa.list_(pa.null())), + ("null_large_list", pa.large_list(pa.null())), + ("null_fixed_list", pa.list_(pa.null(), 2)), + ("null_struct", null_struct), + ("nested_list", pa.list_(null_struct)), + ("null_map", pa.map_(pa.string(), pa.null())), + ]) + rows = [ + {"id": 1, "null_list": [None, None], "null_large_list": [None], + "null_fixed_list": [None, None], "null_struct": {"empty": None, "value": 10}, + "nested_list": [{"empty": None, "value": 11}, {"empty": None, "value": 12}], + "null_map": [("a", None), ("b", None)]}, + {"id": 2, "null_list": None, "null_large_list": None, "null_fixed_list": None, + "null_struct": {"empty": None, "value": 20}, "nested_list": None, "null_map": None}, + {"id": 3, "null_list": [], "null_large_list": [], + "null_fixed_list": [None, None], "null_struct": {"empty": None, "value": 30}, + "nested_list": [], "null_map": []}, + {"id": 4, "null_list": [None], "null_large_list": [None, None, None], + "null_fixed_list": [None, None], "null_struct": {"empty": None, "value": 40}, + "nested_list": [{"empty": None, "value": 41}], "null_map": [("c", None)]}, + ] + # pylance 7 cannot encode a null parent struct with a Null child; BE tests cover that shape. + table = pa.Table.from_pylist(rows, schema=schema) + dataset = lance.write_dataset(table, str(output), data_storage_version="2.2") + assert dataset.to_table().equals(table) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("output", type=Path) + build(parser.parse_args().output) diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_transactions/0-d6a5ba97-9164-4b3e-b773-15de06005c54.txn b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_transactions/0-d6a5ba97-9164-4b3e-b773-15de06005c54.txn new file mode 100644 index 0000000000000000000000000000000000000000..e5933f5bbd50740b52ad0febc59e9c326b672a65 GIT binary patch literal 643 zcmZuuTTa6;5Ns0X-Iokhre7B72P#VCJlwRs2_FdM;ZzZ6R5dB09E*eS&uPF(nkop^ zmTZl8XLiSM7B2%D25aA1+e^>#LYG;g@4J>wV@8&Q(1>~;$`>%6;G?l3woPopH(yoM zC%-x&4L#p;l6d7r388CB6He*{?jj5Fh(V(#0$J6}{%AZkdV(kcSy43|i}xb=yu|=c zp_s<_yA5G3r+Mi*L|6&%4(j1E*v~*|ws0Db31^Ma#>5yFPnjV153No{5dW zdVLOd^f1)w5q7~I4Z30yM{3h zf|2F;F(qR{{JHOa(toJ%DS84Y#Bqp2*v+nTR8RKmMtvZ88w%V=Zyu1vn(#wt2OauVa8^#DUH$%>InzyzQ>sGY6T z^jVtZg$cw(Y8k55AfzS*U79imL$zE@0J2e^=Y^{i%Ci+Q8>ZWdD;g?Rwi>gI#v(bR zG1t&UjYB?>Cl}mdRyxl3zK>C-^3RLadR~~in5Xj~+vN?U(dB7zZnm(#wLL`I%P?J2 z(_lfN75A7V&5L>~ZBepQHp8xUS>JAMdIE8K*x=W}aM!QPf`T7LMpeQxJT(tQGlfN} zrrkmr=4KBc)IO{#yP3Dt-7@`mmWkbJ?X|TJO+D|Myri!K`b1 z`}p>|K6v;3^BeG3iiX76=ghN~C* E3*a23;s5{u literal 0 HcmV?d00001 diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/latest_version_hint.json b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/latest_version_hint.json new file mode 100644 index 00000000000000..491d734467aa3c --- /dev/null +++ b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/latest_version_hint.json @@ -0,0 +1 @@ +{"version":1} \ No newline at end of file diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/data/0110100101011111010110112b5b4742fd82cf05955fa974a3.lance b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/data/0110100101011111010110112b5b4742fd82cf05955fa974a3.lance new file mode 100644 index 0000000000000000000000000000000000000000..0ba4ec7cafa8394c23c02ae4d08d0eb14f2addfa GIT binary patch literal 3278 zcmc&$Ur!T35Wn5qyS7t=G9q3`)Kej9O)AzuK-CCENL6B*_~3&Nrou@YOCj{C;*0m- zgYg6S34Hc5+^5pS7fgSC!4zY>uA1KZkt<$tFril5!?|&;ruKTse(14yHin1HX?|MQ^2Xr^`mO`ATowUBB zG|~JK3}io!gLEEw9dw>_7UnN+Zm|62r+t(b(mGD7@qD!=J(GC;c7D|QfA{-u3o9@5xn}>2gcXfu+r3tvX0r^=VXob6wVt(_9oJ%UBOz6R4Weke&W6QpRus~D znn4LiLchlLx)XZqac^k#4*FsUTkplFs2DGrZ=AI-h7pf39v*a>Z$pNPWUPdFPhvjc z@3`CDRo61(erBkoERUn%Y;3tZR>n4awBu;JYP7nJl_h1Y6_+Wb({ZCza`BLP*^E*Y zm-;rQdIEN?xDQ|6hClg+NEzd^F&mVy(b%$%#mRKi1Wh}E8K>=TH=T|(Y#)nws4D%X zvty0e!-1;VUZ3T%)Uc`osnsnN1YE)kFa>9>v>NSIr{uI(H`mC%Q(D|?bvN2e3L^&% ztH8B+Qv)QvQ(kBOE09Qp7Nu`b~vEF`hocsgJmt`0%KiSnIQ z{AS%u!@mIB(~*gRdpt2xV3u}KOp6Mi(5g&N!}bt_ulqW3El-4Z@D|J^RcqlmRm-Y- zs;#FQr6uaJ@P!(t8N314l6VmcYMA(8rs)AO8or^X={8niHi_{ho+#=%JJL8k0n;;m z)61YP<_py0WxNN~B<8blCN@b?-8y6DESqOFJ>3WCDVTo7%X&G=p(b;9fHhc7%3%u5 z_vMg9Gr)HFuzAoG_&E>$#TDGm&{sFfX81?@qVOw)C$r(kQ~0yOZG|xxZr&*TQQ>Wc ze<=J~;SPkG_X>Yg_=Cdlw&RoQ)#KcPtEu?)rOSy_0r_zR3w BD%=17 literal 0 HcmV?d00001 diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java index 77f23f04144fd7..e806fba4d52905 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java @@ -44,11 +44,6 @@ public final class LanceTypeConverter { private LanceTypeConverter() { } - /** Converts Arrow fields exposed by Lance to Doris types. */ - public static Type toDorisType(Field field) { - return toDorisType(field, true); - } - /** Returns whether this field needs the current BE Lance materialization logic. */ public static boolean requiresCurrentBeReader(Field field) { ArrowType.ArrowTypeID typeId = field.getType().getTypeID(); @@ -71,8 +66,8 @@ public static boolean requiresCurrentBeReader(Field field) { return false; } - /** Converts an Arrow field, allowing Doris NULL only at the top level. */ - private static Type toDorisType(Field field, boolean allowNull) { + /** Converts Arrow fields, including Null leaves in complex types. */ + public static Type toDorisType(Field field) { // TODO(lance): Dataset.getSchema() currently erases the Dictionary marker, while // Dataset.getLanceSchema() fails to convert a schema containing Dictionary in the // Lance 9.1.0-beta.3 Java SDK. Reject physical Dictionary columns after that SDK @@ -89,7 +84,7 @@ private static Type toDorisType(Field field, boolean allowNull) { ArrowType arrowType = field.getType(); switch (arrowType.getTypeID()) { case Null: - return allowNull ? Type.NULL : Type.UNSUPPORTED; + return Type.NULL; case Bool: return Type.BOOLEAN; case Int: @@ -135,7 +130,7 @@ private static Type toDorisType(Field field, boolean allowNull) { case LargeList: case FixedSizeList: requireChildren(field, 1); - Type itemType = toDorisType(field.getChildren().get(0), false); + Type itemType = toDorisType(field.getChildren().get(0)); return itemType.isSupported() ? new ArrayType(itemType) : Type.UNSUPPORTED; case Map: requireChildren(field, 1); @@ -143,15 +138,15 @@ private static Type toDorisType(Field field, boolean allowNull) { requireChildren(entries, 2); Field key = entries.getChildren().get(0); Field value = entries.getChildren().get(1); - Type keyType = toDorisType(key, false); - Type valueType = toDorisType(value, false); + Type keyType = toDorisType(key); + Type valueType = toDorisType(value); return keyType.isSupported() && valueType.isSupported() ? new MapType(keyType, valueType, key.isNullable(), value.isNullable()) : Type.UNSUPPORTED; case Struct: List fields = new ArrayList<>(); for (Field child : field.getChildren()) { - Type childType = toDorisType(child, false); + Type childType = toDorisType(child); if (!childType.isSupported()) { return Type.UNSUPPORTED; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java index 9730b0af39e95c..7e5a940f00d59a 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java @@ -17,7 +17,10 @@ package org.apache.doris.datasource.lance; +import org.apache.doris.catalog.ArrayType; +import org.apache.doris.catalog.MapType; import org.apache.doris.catalog.ScalarType; +import org.apache.doris.catalog.StructType; import org.apache.doris.catalog.Type; import org.apache.arrow.vector.types.DateUnit; @@ -30,6 +33,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.Collections; public class LanceTypeConverterTest { @@ -124,21 +128,38 @@ public void testNullAndDurationMappings() { Assertions.assertTrue(LanceTypeConverter.requiresCurrentBeReader(durationList)); } - /** Verifies nested Null fields remain unsupported. */ @Test - public void testNestedNullIsUnsupported() { + public void testNestedNullMappings() { Field nullItem = Field.nullable("item", ArrowType.Null.INSTANCE); - Field nullList = new Field( - "null_list", - FieldType.nullable(ArrowType.List.INSTANCE), - Collections.singletonList(nullItem)); - Field nullStruct = new Field( - "null_struct", - FieldType.nullable(ArrowType.Struct.INSTANCE), - Collections.singletonList(Field.nullable("value", ArrowType.Null.INSTANCE))); - - Assertions.assertEquals(Type.UNSUPPORTED, LanceTypeConverter.toDorisType(nullList)); - Assertions.assertEquals(Type.UNSUPPORTED, LanceTypeConverter.toDorisType(nullStruct)); + for (ArrowType listType : Arrays.asList(ArrowType.List.INSTANCE, + ArrowType.LargeList.INSTANCE, new ArrowType.FixedSizeList(2))) { + Field list = new Field("null_list", FieldType.nullable(listType), + Collections.singletonList(nullItem)); + Type converted = LanceTypeConverter.toDorisType(list); + Assertions.assertInstanceOf(ArrayType.class, converted); + Assertions.assertEquals(Type.NULL, ((ArrayType) converted).getItemType()); + Assertions.assertTrue(LanceTypeConverter.requiresCurrentBeReader(list)); + } + + Field struct = new Field("null_struct", FieldType.nullable(ArrowType.Struct.INSTANCE), + Arrays.asList(Field.nullable("id", new ArrowType.Int(32, true)), nullItem)); + Type convertedStruct = LanceTypeConverter.toDorisType(struct); + Assertions.assertInstanceOf(StructType.class, convertedStruct); + Assertions.assertEquals(Type.NULL, ((StructType) convertedStruct).getFields().get(1).getType()); + + Field entries = new Field("entries", FieldType.notNullable(ArrowType.Struct.INSTANCE), + Arrays.asList(Field.notNullable("key", ArrowType.Utf8.INSTANCE), nullItem)); + Field map = new Field("null_map", FieldType.nullable(new ArrowType.Map(false)), + Collections.singletonList(entries)); + Type convertedMap = LanceTypeConverter.toDorisType(map); + Assertions.assertInstanceOf(MapType.class, convertedMap); + Assertions.assertEquals(Type.NULL, ((MapType) convertedMap).getValueType()); + + Field nested = new Field("nested", FieldType.nullable(ArrowType.List.INSTANCE), + Collections.singletonList(struct)); + ArrayType convertedNested = (ArrayType) LanceTypeConverter.toDorisType(nested); + Assertions.assertEquals(Type.NULL, + ((StructType) convertedNested.getItemType()).getFields().get(1).getType()); } /** Verifies known extension mappings and storage validation. */ diff --git a/regression-test/data/external_table_p0/lance/test_lance_nested_null.out b/regression-test/data/external_table_p0/lance/test_lance_nested_null.out new file mode 100644 index 00000000000000..f1eb6122f07645 --- /dev/null +++ b/regression-test/data/external_table_p0/lance/test_lance_nested_null.out @@ -0,0 +1,11 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !nested_null_values -- +1 false 2 true false 1 false 2 true 10 false 2 false 2 true +2 true -1 true true -1 true -1 true 20 true -1 true -1 true +3 false 0 true false 0 false 2 true 30 false 0 false 0 true +4 false 1 true false 3 false 2 true 40 false 1 false 1 true + +-- !nested_null_projection -- +1 [NULL, NULL] true +2 \N true +3 [] true diff --git a/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy b/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy new file mode 100644 index 00000000000000..7db6147f6dccec --- /dev/null +++ b/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy @@ -0,0 +1,66 @@ +// 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_lance_nested_null", "p0,external") { + String enabled = context.config.otherConfigs.get("enableIcebergTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable Lance S3 TVF test because the Iceberg MinIO environment is disabled.") + return + } + + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String lanceTvf = """ + s3( + "uri" = "s3://warehouse/lance/nested_null.lance", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}", + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.region" = "us-east-1", + "use_path_style" = "true", + "format" = "lance" + ) + """ + + String originalScannerV2 = sql("SHOW VARIABLES LIKE 'enable_file_scanner_v2'")[0][1] + try { + sql "SET enable_file_scanner_v2 = true" + def columns = sql "DESC FUNCTION ${lanceTvf}" + assertEquals(7, columns.size()) + assertTrue(columns.every { !it[1].toString().contains("UNSUPPORTED") }) + assertEquals("array", columns.find { it[0] == "null_list" }[1].toString()) + assertEquals("map", columns.find { it[0] == "null_map" }[1].toString()) + + qt_nested_null_values """ + SELECT id, + null_list IS NULL, COALESCE(size(null_list), -1), null_list[1] IS NULL, + null_large_list IS NULL, COALESCE(size(null_large_list), -1), + null_fixed_list IS NULL, COALESCE(size(null_fixed_list), -1), + struct_element(null_struct, 'empty') IS NULL, + struct_element(null_struct, 'value'), + nested_list IS NULL, COALESCE(size(nested_list), -1), + null_map IS NULL, COALESCE(map_size(null_map), -1), null_map['a'] IS NULL + FROM ${lanceTvf} ORDER BY id + """ + qt_nested_null_projection """ + SELECT id, null_list, struct_element(null_struct, 'empty') IS NULL + FROM ${lanceTvf} WHERE id >= 1 ORDER BY id LIMIT 3 + """ + } finally { + sql "SET enable_file_scanner_v2 = ${originalScannerV2}" + } +} From 027201d093600fc5509ab48fcd80da6eed27e683 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 14 Sep 2026 13:08:03 +0800 Subject: [PATCH 2/5] [fix](lance) validate recursive Null schemas and preserve fixture rebuilds --- .../format_v2/lance/lance_reader_helper.cpp | 5 ++ .../lance/lance_nested_null_test.cpp | 24 ++++++++- .../scripts/lance_build_nested_null.py | 19 +++++-- .../lance_build_preinstalled_catalog.py | 9 +++- .../datasource/lance/LanceTypeConverter.java | 10 ++-- .../lance/LanceTypeConverterTest.java | 53 +++++++++++++++++++ 6 files changed, 109 insertions(+), 11 deletions(-) diff --git a/be/src/format_v2/lance/lance_reader_helper.cpp b/be/src/format_v2/lance/lance_reader_helper.cpp index 313535a70dd39c..8599b7c3850e0d 100644 --- a/be/src/format_v2/lance/lance_reader_helper.cpp +++ b/be/src/format_v2/lance/lance_reader_helper.cpp @@ -166,6 +166,11 @@ Status arrow_field_to_doris_type(const std::shared_ptr& field, switch (arrow_type->id()) { case arrow::Type::NA: + // Required Null leaves can bypass NullableSerDe through FE schema reconstruction. + // Reject them before Arrow NA buffers reach the Boolean-backed physical SerDe. + if (!field->nullable()) { + return Status::NotSupported("non-nullable Lance Arrow Null field: {}", field->name()); + } return nullable_primitive(TYPE_NULL); case arrow::Type::BOOL: return nullable_primitive(TYPE_BOOLEAN); diff --git a/be/test/format_v2/lance/lance_nested_null_test.cpp b/be/test/format_v2/lance/lance_nested_null_test.cpp index e9fd061a299034..8d1115687df943 100644 --- a/be/test/format_v2/lance/lance_nested_null_test.cpp +++ b/be/test/format_v2/lance/lance_nested_null_test.cpp @@ -80,11 +80,15 @@ TEST_F(LanceNestedNullTest, MapsNestedNullSchemas) { arrow::field("struct", arrow::struct_({null_field})), arrow::field("map", arrow::map(arrow::utf8(), null_field)), arrow::field("nested", arrow::list(arrow::struct_({null_field}))), + arrow::field("list_list", arrow::list(arrow::list(null_field))), + arrow::field("struct_list", + arrow::struct_({arrow::field("list", arrow::list(null_field))})), + arrow::field("list_map", arrow::list(arrow::map(arrow::utf8(), null_field))), }); std::vector names; std::vector types; ASSERT_TRUE(convert_arrow_schema_to_doris(schema, &names, &types).ok()); - ASSERT_EQ(6, types.size()); + ASSERT_EQ(9, types.size()); for (const auto& type : types) { EXPECT_NE(INVALID_TYPE, type->get_primitive_type()); } @@ -96,6 +100,24 @@ TEST_F(LanceNestedNullTest, MapsNestedNullSchemas) { } } +TEST_F(LanceNestedNullTest, RejectsNonNullableNullLeaves) { + const auto leaf = arrow::field("item", arrow::null(), false); + for (const auto& type : std::vector> { + arrow::null(), arrow::list(leaf), arrow::large_list(leaf), + arrow::fixed_size_list(leaf, 2), arrow::struct_({leaf}), + arrow::map(arrow::null(), arrow::utf8()), arrow::map(arrow::utf8(), leaf), + arrow::list(arrow::map(arrow::utf8(), leaf))}) { + SCOPED_TRACE(type->ToString()); + std::vector names; + std::vector types; + ASSERT_TRUE(convert_arrow_schema_to_doris( + arrow::schema({arrow::field("value", type, false)}), &names, &types) + .ok()); + ASSERT_EQ(1, types.size()); + EXPECT_EQ(INVALID_TYPE, types[0]->get_primitive_type()); + } +} + TEST_F(LanceNestedNullTest, PreservesNullTypeInNestedSchemaRpc) { const auto null = null_type(); const auto type = make_nullable(std::make_shared( diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py index 810a7c57778047..73d1e7a064bc54 100644 --- a/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py +++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py @@ -15,7 +15,7 @@ # specific language governing permissions and limitations # under the License. -"""Generate the nested Null regression fixture with pylance 7.0.0 and PyArrow 21.0.0.""" +"""Generate the nested Null regression fixture using lance_fixture_requirements.txt.""" import argparse from pathlib import Path @@ -24,7 +24,7 @@ import pyarrow as pa -def build(output: Path) -> None: +def expected_table() -> pa.Table: null_struct = pa.struct([("empty", pa.null()), ("value", pa.int32())]) schema = pa.schema([ ("id", pa.int32()), @@ -50,9 +50,18 @@ def build(output: Path) -> None: "nested_list": [{"empty": None, "value": 41}], "null_map": [("c", None)]}, ] # pylance 7 cannot encode a null parent struct with a Null child; BE tests cover that shape. - table = pa.Table.from_pylist(rows, schema=schema) - dataset = lance.write_dataset(table, str(output), data_storage_version="2.2") - assert dataset.to_table().equals(table) + return pa.Table.from_pylist(rows, schema=schema) + + +def check(output: Path) -> None: + # Validate the persisted schema and values, including empty collections and null parents. + actual = lance.dataset(str(output)).to_table() + assert actual.equals(expected_table()), f"nested Null fixture differs from expected data: {output}" + + +def build(output: Path) -> None: + lance.write_dataset(expected_table(), str(output), data_storage_version="2.2") + check(output) if __name__ == "__main__": diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py index d6ea30be20373e..92220554c7b0f2 100644 --- a/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py +++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py @@ -28,6 +28,7 @@ The generated catalog contains: - __manifest Directory Namespace V2 manifest table (with its scalar indexes). - all_types.lance The pre-existing compatibility-mode root table, re-registered as-is. + - nested_null.lance Nullable Null leaves inside lists, structs, and maps. - The `doris` namespace with two full-text-search fixtures, one indexed vector table per cell of the algorithm x element type x metric matrix (hash-prefixed directories), listed in VECTOR_TABLES below; BREADTH_TABLE, one table carrying the remaining cells at plan @@ -85,6 +86,7 @@ import lance_namespace import pyarrow as pa import pyarrow.ipc as ipc +from lance_build_nested_null import build as build_nested_null, check as check_nested_null from lance_namespace_urllib3_client.models import ( CreateNamespaceRequest, CreateTableRequest, @@ -99,6 +101,7 @@ NUM_PARTITIONS = 4 NAMESPACE = "doris" ALL_TYPES_DIR = "all_types.lance" +NESTED_NULL_DIR = "nested_null.lance" MANIFEST_DIR = "__manifest" # 4-bit PQ keeps codebook training comfortable on 1024 rows. This only serves fixture @@ -918,7 +921,8 @@ def build_multi_frag(root: Path) -> None: location = str(root / MULTI_FRAG_DIR) for index in range(MULTI_FRAG_NUM_FRAGMENTS): offset = index * MULTI_FRAG_FRAGMENT_ROWS - fragment = make_fragment_table(offset, offset + MULTI_FRAG_FRAGMENT_ROWS) + # The shared builder requires a vector profile even though embedding is dropped below. + fragment = make_fragment_table(COLLINEAR, pa.float32(), offset, offset + MULTI_FRAG_FRAGMENT_ROWS) fragment = fragment.drop_columns(["embedding"]) # Match all_types.lance (data storage version 2.2) so every committed Lance data file # shares one on-disk format and the oldest reader (lance-rs 4.0.1) can open it. @@ -933,6 +937,8 @@ def build_multi_frag(root: Path) -> None: def build(root: Path, all_types_source: Path) -> None: shutil.copytree(all_types_source, root / ALL_TYPES_DIR) build_multi_frag(root) + # Recreate this fixture in staging because promotion replaces the entire catalog tree. + build_nested_null(root / NESTED_NULL_DIR) namespace = lance_namespace.connect("dir", {"root": str(root)}) namespace.register_table( RegisterTableRequest(id=["all_types"], location=ALL_TYPES_DIR) @@ -1721,6 +1727,7 @@ def check_catalog(root: Path) -> None: assert nested_path.is_dir(), f"{NESTED_TABLE} location missing: {nested.location}" check_nested_dataset(nested.location) check_multi_frag(root) + check_nested_null(root / NESTED_NULL_DIR) full_fts = namespace.describe_table(DescribeTableRequest(id=[NAMESPACE, FTS_TABLE])) check_fts_dataset( diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java index e806fba4d52905..fd9576a3c9461f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTypeConverter.java @@ -84,7 +84,8 @@ public static Type toDorisType(Field field) { ArrowType arrowType = field.getType(); switch (arrowType.getTypeID()) { case Null: - return Type.NULL; + // Required Null leaves cannot use the nullable SerDe that handles Arrow NA buffers. + return field.isNullable() ? Type.NULL : Type.UNSUPPORTED; case Bool: return Type.BOOLEAN; case Int: @@ -131,7 +132,8 @@ public static Type toDorisType(Field field) { case FixedSizeList: requireChildren(field, 1); Type itemType = toDorisType(field.getChildren().get(0)); - return itemType.isSupported() ? new ArrayType(itemType) : Type.UNSUPPORTED; + // Generic isSupported() rejects Null items even inside successfully converted composites. + return itemType.equals(Type.UNSUPPORTED) ? Type.UNSUPPORTED : new ArrayType(itemType); case Map: requireChildren(field, 1); Field entries = field.getChildren().get(0); @@ -140,14 +142,14 @@ public static Type toDorisType(Field field) { Field value = entries.getChildren().get(1); Type keyType = toDorisType(key); Type valueType = toDorisType(value); - return keyType.isSupported() && valueType.isSupported() + return !keyType.equals(Type.UNSUPPORTED) && !valueType.equals(Type.UNSUPPORTED) ? new MapType(keyType, valueType, key.isNullable(), value.isNullable()) : Type.UNSUPPORTED; case Struct: List fields = new ArrayList<>(); for (Field child : field.getChildren()) { Type childType = toDorisType(child); - if (!childType.isSupported()) { + if (childType.equals(Type.UNSUPPORTED)) { return Type.UNSUPPORTED; } fields.add(new StructField(child.getName(), childType, diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java index 7e5a940f00d59a..1e842385f5083e 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java @@ -162,6 +162,59 @@ public void testNestedNullMappings() { ((StructType) convertedNested.getItemType()).getFields().get(1).getType()); } + @Test + public void testDeepNullComposites() { + Field leaf = Field.nullable("item", ArrowType.Null.INSTANCE); + Field list = new Field("item", FieldType.nullable(ArrowType.List.INSTANCE), + Collections.singletonList(leaf)); + Field entries = new Field("entries", FieldType.notNullable(ArrowType.Struct.INSTANCE), + Arrays.asList(Field.notNullable("key", ArrowType.Utf8.INSTANCE), leaf)); + Field map = new Field("item", FieldType.nullable(new ArrowType.Map(false)), + Collections.singletonList(entries)); + for (Field child : Arrays.asList(list, map)) { + Type childType = LanceTypeConverter.toDorisType(child); + for (ArrowType outer : Arrays.asList(ArrowType.List.INSTANCE, + ArrowType.LargeList.INSTANCE, new ArrowType.FixedSizeList(2))) { + Field nested = new Field("nested", FieldType.nullable(outer), + Collections.singletonList(child)); + Assertions.assertEquals(new ArrayType(childType), LanceTypeConverter.toDorisType(nested)); + } + Field struct = new Field("nested", FieldType.nullable(ArrowType.Struct.INSTANCE), + Collections.singletonList(child)); + Type converted = LanceTypeConverter.toDorisType(struct); + Assertions.assertInstanceOf(StructType.class, converted); + Assertions.assertEquals(childType, ((StructType) converted).getFields().get(0).getType()); + Field nestedEntries = new Field("entries", FieldType.notNullable(ArrowType.Struct.INSTANCE), + Arrays.asList(Field.notNullable("key", ArrowType.Utf8.INSTANCE), child)); + Field nestedMap = new Field("nested", FieldType.nullable(new ArrowType.Map(false)), + Collections.singletonList(nestedEntries)); + Assertions.assertEquals(new MapType(Type.STRING, childType, false, true), + LanceTypeConverter.toDorisType(nestedMap)); + } + } + + @Test + public void testNonNullableNullLeavesAreUnsupported() { + Field leaf = Field.notNullable("item", ArrowType.Null.INSTANCE); + Assertions.assertEquals(Type.UNSUPPORTED, LanceTypeConverter.toDorisType(leaf)); + for (ArrowType outer : Arrays.asList(ArrowType.List.INSTANCE, + ArrowType.LargeList.INSTANCE, new ArrowType.FixedSizeList(2), ArrowType.Struct.INSTANCE)) { + Field nested = new Field("nested", FieldType.nullable(outer), Collections.singletonList(leaf)); + Assertions.assertEquals(Type.UNSUPPORTED, LanceTypeConverter.toDorisType(nested)); + } + for (boolean nullKey : Arrays.asList(false, true)) { + Field entries = new Field("entries", FieldType.notNullable(ArrowType.Struct.INSTANCE), + Arrays.asList(nullKey ? leaf : Field.notNullable("key", ArrowType.Utf8.INSTANCE), + nullKey ? Field.nullable("value", ArrowType.Utf8.INSTANCE) : leaf)); + Field map = new Field("map", FieldType.nullable(new ArrowType.Map(false)), + Collections.singletonList(entries)); + Assertions.assertEquals(Type.UNSUPPORTED, LanceTypeConverter.toDorisType(map)); + Field nested = new Field("nested", FieldType.nullable(ArrowType.List.INSTANCE), + Collections.singletonList(map)); + Assertions.assertEquals(Type.UNSUPPORTED, LanceTypeConverter.toDorisType(nested)); + } + } + /** Verifies known extension mappings and storage validation. */ @Test public void testKnownExtensionMappingsAndStorageValidation() { From 68cf7d8e140a5ad245dd1617c73ff3de0b4e8c74 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 14 Sep 2026 16:44:19 +0800 Subject: [PATCH 3/5] [fix](test) correct nested Null array output expectation --- .../data/external_table_p0/lance/test_lance_nested_null.out | 2 +- .../external_table_p0/lance/test_lance_nested_null.groovy | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/regression-test/data/external_table_p0/lance/test_lance_nested_null.out b/regression-test/data/external_table_p0/lance/test_lance_nested_null.out index f1eb6122f07645..b3f168eda4c10a 100644 --- a/regression-test/data/external_table_p0/lance/test_lance_nested_null.out +++ b/regression-test/data/external_table_p0/lance/test_lance_nested_null.out @@ -6,6 +6,6 @@ 4 false 1 true false 3 false 2 true 40 false 1 false 1 true -- !nested_null_projection -- -1 [NULL, NULL] true +1 [null, null] true 2 \N true 3 [] true diff --git a/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy b/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy index 7db6147f6dccec..47ba3db8443dfa 100644 --- a/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy +++ b/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy @@ -56,6 +56,7 @@ suite("test_lance_nested_null", "p0,external") { null_map IS NULL, COALESCE(map_size(null_map), -1), null_map['a'] IS NULL FROM ${lanceTvf} ORDER BY id """ + // Complex output uses JSON-style null; a top-level SQL NULL remains \N. qt_nested_null_projection """ SELECT id, null_list, struct_element(null_struct, 'empty') IS NULL FROM ${lanceTvf} WHERE id >= 1 ORDER BY id LIMIT 3 From 116f42eb0397f3ae7e15aea1a8315ecb2564457e Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 15 Sep 2026 09:15:46 +0800 Subject: [PATCH 4/5] [fix](lance) preserve reader safety through file TVF --- .../datasource/tvf/source/TVFScanNode.java | 14 +-- .../ExternalFileTableValuedFunction.java | 5 + .../FileTableValuedFunction.java | 12 +++ .../LocalTableValuedFunction.java | 2 +- .../tvf/source/TVFScanNodeTest.java | 95 +++++++++++++++++++ 5 files changed, 118 insertions(+), 10 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java index 2dccffa071b235..dd7b4620adab38 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java @@ -40,7 +40,6 @@ import org.apache.doris.spi.Split; import org.apache.doris.statistics.StatisticalType; import org.apache.doris.tablefunction.ExternalFileTableValuedFunction; -import org.apache.doris.tablefunction.LocalTableValuedFunction; import org.apache.doris.thrift.TBrokerFileStatus; import org.apache.doris.thrift.TFileAttributes; import org.apache.doris.thrift.TFileCompressType; @@ -82,14 +81,11 @@ public TVFScanNode(PlanNodeId id, TupleDescriptor desc, boolean needCheckColumnP @Override protected void initBackendPolicy() throws UserException { - if (tableValuedFunction instanceof LocalTableValuedFunction) { - long backendId = - ((LocalTableValuedFunction) tableValuedFunction).getBackendIdForExecution(); - if (backendId != -1) { - backendPolicy.initWithBackendId(backendId); - numNodes = backendPolicy.numBackends(); - return; - } + long backendId = tableValuedFunction.getBackendIdForExecution(); + if (backendId != -1) { + backendPolicy.initWithBackendId(backendId); + numNodes = backendPolicy.numBackends(); + return; } backendPolicy.init(); numNodes = backendPolicy.numBackends(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java index 3c4acfaab8af27..5c4f7bc3a3eb46 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java @@ -144,6 +144,11 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio private List lanceFragments = Collections.emptyList(); private Set lanceCurrentReaderColumns = Collections.emptySet(); + /** Return the only backend that may execute this TVF, or -1 when execution may be distributed. */ + public long getBackendIdForExecution() { + return -1; + } + public abstract TFileType getTFileType(); public abstract String getFilePath(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java index 6ec9f12a81cb3f..46e29ff25e5e58 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java @@ -95,6 +95,18 @@ public boolean isLanceFormat() { return delegateTvf.isLanceFormat(); } + @Override + public boolean requiresCurrentLanceReader(String columnName) { + // Schema discovery records reader requirements on the delegate, not this wrapper. + return delegateTvf.requiresCurrentLanceReader(columnName); + } + + @Override + public long getBackendIdForExecution() { + // Local Lance must run on its schema backend, including through the generic file() entry point. + return delegateTvf.getBackendIdForExecution(); + } + @Override public long getLanceDatasetVersion() { return delegateTvf.getLanceDatasetVersion(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java index 3e04d614aaca86..4e4bf163145cf9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java @@ -159,7 +159,7 @@ public Long getBackendId() { return backendId; } - /** Return the only backend that may execute this TVF, or -1 when execution may be distributed. */ + @Override public long getBackendIdForExecution() { return isLanceFormat() ? backendIdForRequest : backendId; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java index ef2dbb8f041dad..8c3087962d1766 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java @@ -21,6 +21,7 @@ import org.apache.doris.analysis.SlotId; import org.apache.doris.analysis.TupleDescriptor; import org.apache.doris.analysis.TupleId; +import org.apache.doris.catalog.ArrayType; import org.apache.doris.catalog.Column; import org.apache.doris.catalog.FunctionGenTable; import org.apache.doris.catalog.Type; @@ -37,7 +38,9 @@ import org.apache.doris.spi.Split; import org.apache.doris.system.Backend; import org.apache.doris.tablefunction.ExternalFileTableValuedFunction; +import org.apache.doris.tablefunction.FileTableValuedFunction; import org.apache.doris.tablefunction.LocalTableValuedFunction; +import org.apache.doris.tablefunction.S3TableValuedFunction; import org.apache.doris.thrift.TBrokerFileStatus; import org.apache.doris.thrift.TFileFormatType; import org.apache.doris.thrift.TFileRangeDesc; @@ -252,6 +255,7 @@ public void testS3LanceAdditionalTypesRejectSmoothUpgradeSource() throws Excepti Mockito.when(table.getTvf()).thenReturn(tvf); Mockito.when(tvf.isLanceFormat()).thenReturn(true); Mockito.when(tvf.requiresCurrentLanceReader("json_value")).thenReturn(true); + Mockito.when(tvf.getBackendIdForExecution()).thenCallRealMethod(); desc.setTable(table); Backend smoothUpgradeSource = Mockito.mock(Backend.class); @@ -270,4 +274,95 @@ public void testS3LanceAdditionalTypesRejectSmoothUpgradeSource() throws Excepti Assert.assertTrue(exception.getMessage().contains("102")); } + + @Test + public void testFileS3LanceNestedNullRejectsSmoothUpgradeSource() throws Exception { + S3TableValuedFunction delegate = Mockito.mock(S3TableValuedFunction.class); + Mockito.when(delegate.isLanceFormat()).thenReturn(true); + Mockito.when(delegate.getBackendIdForExecution()).thenCallRealMethod(); + Mockito.when(delegate.requiresCurrentLanceReader("nested_null")).thenReturn(true); + FileTableValuedFunction tvf = wrapFileTvf(delegate); + TupleDescriptor desc = new TupleDescriptor(new TupleId(0)); + SlotDescriptor slot = new SlotDescriptor(new SlotId(1), desc); + slot.setColumn(new Column("nested_null", ArrayType.create(Type.NULL, true))); + desc.addSlot(slot); + FunctionGenTable table = Mockito.mock(FunctionGenTable.class); + Mockito.when(table.getTvf()).thenReturn(tvf); + desc.setTable(table); + + Backend source = Mockito.mock(Backend.class); + Mockito.when(source.isSmoothUpgradeSrc()).thenReturn(true); + Mockito.when(source.getId()).thenReturn(102L); + FederationBackendPolicy policy = Mockito.mock(FederationBackendPolicy.class); + Mockito.when(policy.getBackends()).thenReturn(Collections.singletonList(source)); + TVFScanNode node = new TVFScanNode( + new PlanNodeId(0), desc, false, new SessionVariable(), ScanContext.EMPTY); + setBackendPolicy(node, policy); + + UserException exception = Assert.assertThrows(UserException.class, node::initBackendPolicy); + Assert.assertTrue(exception.getMessage().contains("102")); + + // An ordinary projection must remain usable while an unrelated root needs the new reader. + slot.setColumn(new Column("ordinary", Type.INT)); + node.initBackendPolicy(); + Mockito.verify(policy, Mockito.times(2)).init(); + } + + @Test + public void testFileLocalLancePinsExecutionToSchemaBackendDuringUpgrade() throws Exception { + LocalTableValuedFunction delegate = Mockito.mock(LocalTableValuedFunction.class); + Mockito.when(delegate.isLanceFormat()).thenReturn(true); + Mockito.when(delegate.getBackendIdForExecution()).thenCallRealMethod(); + Field backendId = LocalTableValuedFunction.class.getDeclaredField("backendId"); + backendId.setAccessible(true); + backendId.set(delegate, -1L); + Field requestBackend = LocalTableValuedFunction.class.getDeclaredField("backendIdForRequest"); + requestBackend.setAccessible(true); + requestBackend.set(delegate, 101L); + Field sharedStorage = LocalTableValuedFunction.class.getDeclaredField("sharedStorage"); + sharedStorage.setAccessible(true); + sharedStorage.set(delegate, true); + FileTableValuedFunction tvf = wrapFileTvf(delegate); + TupleDescriptor desc = new TupleDescriptor(new TupleId(0)); + FunctionGenTable table = Mockito.mock(FunctionGenTable.class); + Mockito.when(table.getTvf()).thenReturn(tvf); + desc.setTable(table); + + Backend source = Mockito.mock(Backend.class); + Mockito.when(source.isSmoothUpgradeSrc()).thenReturn(true); + FederationBackendPolicy policy = Mockito.mock(FederationBackendPolicy.class); + Mockito.when(policy.getBackends()).thenReturn(Collections.singletonList(source)); + Mockito.when(policy.numBackends()).thenReturn(1); + TVFScanNode node = new TVFScanNode( + new PlanNodeId(0), desc, false, new SessionVariable(), ScanContext.EMPTY); + setBackendPolicy(node, policy); + + node.initBackendPolicy(); + Mockito.verify(policy).initWithBackendId(101L); + Mockito.verify(policy, Mockito.never()).init(); + + // Shared non-Lance files still allow distributed execution through the same wrapper. + Mockito.when(delegate.isLanceFormat()).thenReturn(false); + node.initBackendPolicy(); + Mockito.verify(policy).init(); + } + + private static FileTableValuedFunction wrapFileTvf(ExternalFileTableValuedFunction delegate) + throws Exception { + // Avoid storage discovery while exercising the real wrapper methods used by FunctionGenTable. + FileTableValuedFunction wrapper = Mockito.mock(FileTableValuedFunction.class, Mockito.CALLS_REAL_METHODS); + Field delegateField = FileTableValuedFunction.class.getDeclaredField("delegateTvf"); + delegateField.setAccessible(true); + delegateField.set(wrapper, delegate); + Field columns = ExternalFileTableValuedFunction.class.getDeclaredField("lanceCurrentReaderColumns"); + columns.setAccessible(true); + columns.set(wrapper, Collections.emptySet()); + return wrapper; + } + + private static void setBackendPolicy(TVFScanNode node, FederationBackendPolicy policy) throws Exception { + Field field = ExternalScanNode.class.getDeclaredField("backendPolicy"); + field.setAccessible(true); + field.set(node, policy); + } } From 713052e7b4f61bd8c2d4fc5b6d7eac3ae59cecfd Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 15 Sep 2026 12:48:26 +0800 Subject: [PATCH 5/5] [fix](lance) retain reader requirements for lazy search columns --- .../lance/source/LanceScanNode.java | 11 +- .../translator/PhysicalPlanTranslator.java | 35 +--- .../PhysicalLazyMaterializeTVFScan.java | 6 +- .../LanceLazyMaterializationUpgradeTest.java | 159 ++++++++++++++++++ 4 files changed, 180 insertions(+), 31 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/LanceLazyMaterializationUpgradeTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java index 4bbde106e8168e..5b079ab4e851af 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java @@ -97,6 +97,7 @@ private enum SearchKind { private final SearchKind searchKind; private byte[] lanceSubstraitFilter = new byte[0]; private String lancePushdownPredicate = ""; + private final Set lazyMaterializedColumns = new HashSet<>(); private long plannedVersion = -1; private int plannedFragments; private int plannedUnindexedFragments; @@ -166,9 +167,15 @@ protected void doInitialize() throws UserException { } } - /** Checks whether any projected Lance column requires the current BE reader. */ + public void addLazyMaterializedColumn(String columnName) { + lazyMaterializedColumns.add(columnName.toLowerCase(Locale.ROOT)); + } + + /** Checks columns read in either phase of a Lance scan. */ private boolean projectsCurrentReaderType() { - Set projectedColumns = new HashSet<>(); + // Global row IDs route the second-phase take back to the first-phase BE, so lazy pruning + // must not hide a column's reader requirement when checking mixed-version backends. + Set projectedColumns = new HashSet<>(lazyMaterializedColumns); for (SlotDescriptor slot : desc.getSlots()) { if (slot.getColumn() != null) { projectedColumns.add(slot.getColumn().getName().toLowerCase(Locale.ROOT)); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 4271c7fffa3f16..be66a89526c57d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java @@ -1257,6 +1257,12 @@ public PlanFragment visitPhysicalTVFRelation(PhysicalTVFRelation tvfRelation, Pl TableValuedFunctionIf catalogFunction = tvfRelation.getFunction().getCatalogFunction(); SessionVariable sv = ConnectContext.get().getSessionVariable(); ScanNode scanNode = catalogFunction.getScanNode(context.nextPlanNodeId(), tupleDescriptor, sv); + if (scanNode instanceof LanceScanNode && tvfRelation instanceof PhysicalLazyMaterializeTVFScan) { + for (Slot slot : ((PhysicalLazyMaterializeTVFScan) tvfRelation).getLazySlots()) { + ((LanceScanNode) scanNode).addLazyMaterializedColumn( + ((SlotReference) slot).getOriginalColumn().map(Column::getName).orElse(slot.getName())); + } + } scanNode.setNereidsId(tvfRelation.getId()); context.getNereidsIdToPlanNodeIdMap().put(tvfRelation.getId(), scanNode.getId()); Utils.execWithUncheckedException(scanNode::init); @@ -3001,34 +3007,7 @@ private boolean hasNestedAccessPaths(SlotReference slotReference) { @Override public PlanFragment visitPhysicalLazyMaterializeTVFScan(PhysicalLazyMaterializeTVFScan tvfRelation, PlanTranslatorContext context) { - List slots = tvfRelation.getOutput(); - TupleDescriptor tupleDescriptor = generateTupleDesc(slots, tvfRelation.getFunction().getTable(), context); - - TableValuedFunctionIf catalogFunction = tvfRelation.getFunction().getCatalogFunction(); - SessionVariable sv = ConnectContext.get().getSessionVariable(); - ScanNode scanNode = catalogFunction.getScanNode(context.nextPlanNodeId(), tupleDescriptor, sv); - scanNode.setNereidsId(tvfRelation.getId()); - context.getNereidsIdToPlanNodeIdMap().put(tvfRelation.getId(), scanNode.getId()); - Utils.execWithUncheckedException(scanNode::init); - context.getRuntimeTranslator().ifPresent( - runtimeFilterGenerator -> runtimeFilterGenerator.getContext().getTargetListByScan(tvfRelation) - .forEach(expr -> runtimeFilterGenerator.translateRuntimeFilterTarget(expr, scanNode, context) - ) - ); - context.addScanNode(scanNode, tvfRelation); - - // TODO: it is weird update label in this way - // set label for explain - for (Slot slot : slots) { - String tableColumnName = TableValuedFunctionIf.TVF_TABLE_PREFIX + tvfRelation.getFunction().getName() - + "." + slot.getName(); - context.findSlotRef(slot.getExprId()).setLabel(tableColumnName); - } - - PlanFragment planFragment = createPlanFragment(scanNode, DataPartition.RANDOM, tvfRelation); - context.addPlanFragment(planFragment); - updateLegacyPlanIdToPhysicalPlan(planFragment.getPlanRoot(), tvfRelation); - return planFragment; + return visitPhysicalTVFRelation(tvfRelation, context); } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeTVFScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeTVFScan.java index 03787887563177..1278cfee53bde6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeTVFScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeTVFScan.java @@ -47,7 +47,11 @@ public PhysicalLazyMaterializeTVFScan(PhysicalTVFRelation scan, SlotReference ro super(scan.getRelationId(), scan.getFunction(), scan.getOperativeSlots(), scan.getLogicalProperties()); this.scan = scan; this.rowId = rowId; - this.lazySlots = lazySlots; + this.lazySlots = ImmutableList.copyOf(lazySlots); + } + + public List getLazySlots() { + return lazySlots; } @Override diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/LanceLazyMaterializationUpgradeTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/LanceLazyMaterializationUpgradeTest.java new file mode 100644 index 00000000000000..a28e4959aa0738 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/LanceLazyMaterializationUpgradeTest.java @@ -0,0 +1,159 @@ +// 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.nereids.glue.translator; + +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.catalog.ArrayType; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.FunctionGenTable; +import org.apache.doris.catalog.Table; +import org.apache.doris.catalog.Type; +import org.apache.doris.datasource.ExternalScanNode; +import org.apache.doris.datasource.FederationBackendPolicy; +import org.apache.doris.datasource.lance.LanceExternalTable; +import org.apache.doris.datasource.lance.LanceTableMetadata; +import org.apache.doris.datasource.lance.source.LanceScanNode; +import org.apache.doris.nereids.properties.DataTrait; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.functions.table.FullTextSearch; +import org.apache.doris.nereids.trees.expressions.functions.table.TableValuedFunction; +import org.apache.doris.nereids.trees.expressions.functions.table.VectorSearch; +import org.apache.doris.nereids.trees.plans.RelationId; +import org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterializeTVFScan; +import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; +import org.apache.doris.system.Backend; +import org.apache.doris.tablefunction.FullTextSearchTableValuedFunction; +import org.apache.doris.tablefunction.TableValuedFunctionIf; +import org.apache.doris.tablefunction.VectorSearchTableValuedFunction; +import org.apache.doris.thrift.TExternalSearchQuery; +import org.apache.doris.thrift.TExternalSearchRequest; +import org.apache.doris.thrift.TFullTextSearchParams; +import org.apache.doris.thrift.TVectorSearchParams; + +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.arrow.vector.types.pojo.Schema; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; + +public class LanceLazyMaterializationUpgradeTest { + @Test + public void testVectorSearchLazyNullUpgradeFence() throws Exception { + assertUpgradeFence(true); + } + + @Test + public void testFullTextSearchLazyNullUpgradeFence() throws Exception { + assertUpgradeFence(false); + } + + private void assertUpgradeFence(boolean vector) throws Exception { + ConnectContext previous = ConnectContext.get(); + ConnectContext context = new ConnectContext(); + context.setThreadLocalInfo(); + try { + RuntimeException exception = Assertions.assertThrows(RuntimeException.class, + () -> translate(vector, true, true)); + Assertions.assertTrue(exception.toString().contains("smooth upgrade source"), exception.toString()); + translate(vector, true, false); + // An unreferenced nested Null field must not fence an ordinary lazy projection. + translate(vector, false, true); + } finally { + if (previous == null) { + ConnectContext.remove(); + } else { + previous.setThreadLocalInfo(); + } + } + } + + private void translate(boolean vector, boolean lazyNull, boolean mixedVersion) throws Exception { + String functionName = vector ? "vector_search" : "full_text_search"; + String scoreName = vector ? "_distance" : "_score"; + Field nestedNull = new Field("nested_null", FieldType.nullable(ArrowType.List.INSTANCE), + Collections.singletonList(Field.nullable("item", ArrowType.Null.INSTANCE))); + LanceTableMetadata metadata = LanceTableMetadata.withoutIndexSegments( + "s3://bucket/table.lance", 42, + new Schema(Arrays.asList(nestedNull, Field.nullable("ordinary", ArrowType.Utf8.INSTANCE))), + Collections.emptyList(), Collections.emptyMap()); + Column score = new Column(scoreName, Type.DOUBLE); + Column payload = new Column("nested_null", ArrayType.create(Type.NULL, true)); + Column ordinary = new Column("ordinary", Type.STRING); + Column rowIdColumn = new Column(Column.GLOBAL_ROWID_COL + functionName, Type.STRING); + TableValuedFunctionIf catalogFunction = vector + ? Mockito.mock(VectorSearchTableValuedFunction.class) + : Mockito.mock(FullTextSearchTableValuedFunction.class); + FunctionGenTable table = new FunctionGenTable(1, functionName, Table.TableType.TABLE_VALUED_FUNCTION, + Arrays.asList(score, payload, ordinary), catalogFunction); + TableValuedFunction function = vector ? Mockito.mock(VectorSearch.class) : Mockito.mock(FullTextSearch.class); + Mockito.when(function.getName()).thenReturn(functionName); + Mockito.when(function.getTable()).thenReturn(table); + Mockito.when(function.getCatalogFunction()).thenReturn(catalogFunction); + SlotReference scoreSlot = SlotReference.fromColumn(new ExprId(1), table, score, Collections.emptyList()); + // Retain the source column identity even when the output uses an alias. + SlotReference lazySlot = SlotReference.fromColumn(new ExprId(2), table, + lazyNull ? payload : ordinary, "payload_alias", Collections.emptyList()); + SlotReference rowId = SlotReference.fromColumn(new ExprId(3), table, rowIdColumn, Collections.emptyList()); + List output = Arrays.asList(scoreSlot, lazySlot); + PhysicalTVFRelation relation = new PhysicalTVFRelation(new RelationId(1), function, + Collections.singletonList(scoreSlot), new LogicalProperties(() -> output, () -> DataTrait.EMPTY_TRAIT)); + PhysicalLazyMaterializeTVFScan lazyScan = new PhysicalLazyMaterializeTVFScan( + relation, rowId, Collections.singletonList(lazySlot)); + Assertions.assertFalse(lazyScan.getOutput().contains(lazySlot)); + + Backend current = Mockito.mock(Backend.class); + Backend old = Mockito.mock(Backend.class); + Mockito.when(old.isSmoothUpgradeSrc()).thenReturn(true); + Mockito.when(old.getId()).thenReturn(102L); + FederationBackendPolicy policy = Mockito.mock(FederationBackendPolicy.class); + Mockito.when(policy.getBackends()).thenReturn(mixedVersion + ? Arrays.asList(current, old) : Collections.singletonList(current)); + LanceExternalTable source = Mockito.mock(LanceExternalTable.class); + AtomicReference translated = new AtomicReference<>(); + Mockito.when(catalogFunction.getScanNode(Mockito.any(), Mockito.any(), Mockito.any())).thenAnswer(call -> { + TExternalSearchQuery query = vector + ? TExternalSearchQuery.vector_search(new TVectorSearchParams().setColumn("vector")) + : TExternalSearchQuery.full_text_search(new TFullTextSearchParams().setColumn("text")); + LanceScanNode node = LanceScanNode.forExternalSearch(call.getArgument(0, PlanNodeId.class), + call.getArgument(1, TupleDescriptor.class), source, metadata, 0, + new TExternalSearchRequest().setSearchQuery(query), call.getArgument(2, SessionVariable.class)); + java.lang.reflect.Field backendPolicy = ExternalScanNode.class.getDeclaredField("backendPolicy"); + backendPolicy.setAccessible(true); + backendPolicy.set(node, policy); + translated.set(node); + return node; + }); + PlanTranslatorContext translatorContext = new PlanTranslatorContext(); + lazyScan.accept(new PhysicalPlanTranslator(translatorContext), translatorContext); + Assertions.assertFalse(translated.get().getTupleDesc().getSlots().stream() + .anyMatch(slot -> "nested_null".equals(slot.getColumn().getName()))); + } +}