This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 42bf1581939f7c0b1f529322400e3280c226f1a7 Author: Gabriel <[email protected]> AuthorDate: Tue Sep 15 18:41:30 2026 +0800 [Feature](lance) Support nested Arrow Null types (#67921) ### What problem does this PR solve? Follow-up to #67325; partially addresses #66496. Lance columns containing Arrow Null inside lists, structs, or map values are currently reported as unsupported. This change maps nullable Null leaves recursively, including deeper list/map/struct combinations, materializes them without accessing nonexistent Arrow validity/value buffers, and preserves their logical `NULL_TYPE` through nested schema RPC serialization. Array offsets, parent nullability, empty collections, and sliced inputs remain intact. Unknown extensions and non-nullable Null leaves are rejected before materialization. Generic `file()` scans preserve the delegate’s reader compatibility requirements and Local schema-backend affinity during rolling upgrades. Search TVFs retain reader compatibility requirements for deferred columns across TopN lazy materialization. Both read phases are checked without adding deferred fields back to the first-phase tuple. ### Tests - Lazy search mixed-version translation and Lance scan tests: 19 passed in an isolated runner using cached dependencies, including both search TVFs, aliased deferred fields, and ordinary projections. Both new rejection tests failed before the fix. - Generic-file upgrade scheduling and external TVF tests: 13 passed in an isolated runner using cached dependencies. Both new safety tests failed before the fix. - FE type-conversion tests: 12 passed in an isolated runner using existing dependencies; the new nested mapping test failed before the change. - BE nested Null tests: 10 passed in an isolated harness using the changed routines and compatible cached dependencies. The original schema conversion, RPC type preservation, and nested read failures were reproduced. - All affected C++ translation units passed syntax checks against the current source headers and generated schemas. - FE `mvn checkstyle:check`, clang-format 16 checks, and `git diff --check` passed. - The canonical fixture rebuild generates and validates the nested-null dataset. Its full check passes against the committed fixture; a temporary rebuild passes with `--repin` (one existing HNSW discriminator changes on retraining), and the check rejects a missing nested-null dataset. - Added S3 TVF regression coverage and a reproducible seven-column Lance fixture, verified by PyLance round-trip reading. Full FE Maven tests are blocked locally by a Thrift compiler/runtime version mismatch (installed compiler 0.16 versus branch dependency 0.24). The complete BE suite and S3 regression have not been run locally and require CI. ### Release note Support Arrow Null elements and fields nested in Lance arrays, structs, and map values. --- be/src/core/data_type/data_type.h | 4 +- .../data_type_serde/data_type_nullable_serde.cpp | 12 + be/src/format_v2/lance/lance_reader_helper.cpp | 24 +- be/test/format_v2/lance/lance_nested_null_test.cpp | 292 +++++++++++++++++++++ be/test/format_v2/table/lance_reader_test.cpp | 19 -- .../iceberg/scripts/lance_build_nested_null.py | 70 +++++ .../scripts/lance_build_preinstalled_catalog.py | 9 +- .../0-d6a5ba97-9164-4b3e-b773-15de06005c54.txn | Bin 0 -> 643 bytes .../_versions/18446744073709551614.manifest | Bin 0 -> 1363 bytes .../_versions/latest_version_hint.json | 1 + ...1011111010110112b5b4742fd82cf05955fa974a3.lance | Bin 0 -> 3278 bytes .../doris/datasource/lance/LanceTypeConverter.java | 27 +- .../datasource/lance/source/LanceScanNode.java | 11 +- .../doris/datasource/tvf/source/TVFScanNode.java | 14 +- .../glue/translator/PhysicalPlanTranslator.java | 35 +-- .../physical/PhysicalLazyMaterializeTVFScan.java | 6 +- .../ExternalFileTableValuedFunction.java | 5 + .../tablefunction/FileTableValuedFunction.java | 12 + .../tablefunction/LocalTableValuedFunction.java | 2 +- .../datasource/lance/LanceTypeConverterTest.java | 100 ++++++- .../datasource/tvf/source/TVFScanNodeTest.java | 95 +++++++ .../LanceLazyMaterializationUpgradeTest.java | 159 +++++++++++ .../lance/test_lance_nested_null.out | 11 + .../lance/test_lance_nested_null.groovy | 67 +++++ 24 files changed, 874 insertions(+), 101 deletions(-) diff --git a/be/src/core/data_type/data_type.h b/be/src/core/data_type/data_type.h index b976809bbfd..552729d7d2f 100644 --- a/be/src/core/data_type/data_type.h +++ b/be/src/core/data_type/data_type.h @@ -170,7 +170,9 @@ public: 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 dd84eda304c..115b37ea2a8 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 c47e67a38a1..8599b7c3850 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<arrow::Field>& 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<arrow::Field>& 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,12 @@ Status arrow_field_to_doris_type(const std::shared_ptr<arrow::Field>& 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()); + // 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); case arrow::Type::INT8: @@ -237,7 +239,7 @@ Status arrow_field_to_doris_type(const std::shared_ptr<arrow::Field>& field, case arrow::Type::FIXED_SIZE_LIST: { const auto list = std::static_pointer_cast<arrow::BaseListType>(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<DataTypeArray>(value_type)); return Status::OK(); } @@ -245,8 +247,8 @@ Status arrow_field_to_doris_type(const std::shared_ptr<arrow::Field>& field, const auto map = std::static_pointer_cast<arrow::MapType>(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<DataTypeMap>(key_type, item_type)); return Status::OK(); } @@ -258,7 +260,7 @@ Status arrow_field_to_doris_type(const std::shared_ptr<arrow::Field>& 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 +768,7 @@ Status convert_arrow_schema_to_doris(const std::shared_ptr<arrow::Schema>& 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<ErrorCode::NOT_IMPLEMENTED_ERROR>()) { parsed_types.emplace_back(std::make_shared<DataTypeNothing>()); } 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 00000000000..8d1115687df --- /dev/null +++ b/be/test/format_v2/lance/lance_nested_null_test.cpp @@ -0,0 +1,292 @@ +// 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 <arrow/api.h> +#include <gtest/gtest.h> + +#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<const ColumnNullable&>(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<arrow::Array>& array, const DataTypePtr& type, + MutableColumnPtr* column) { + std::shared_ptr<arrow::Array> 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}))), + 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<std::string> names; + std::vector<DataTypePtr> types; + ASSERT_TRUE(convert_arrow_schema_to_doris(schema, &names, &types).ok()); + ASSERT_EQ(9, 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<const DataTypeArray&>(*remove_nullable(types[i])) + .get_nested_type() + ->is_null_literal()); + } +} + +TEST_F(LanceNestedNullTest, RejectsNonNullableNullLeaves) { + const auto leaf = arrow::field("item", arrow::null(), false); + for (const auto& type : std::vector<std::shared_ptr<arrow::DataType>> { + 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<std::string> names; + std::vector<DataTypePtr> 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<DataTypeStruct>( + DataTypes {null, make_nullable(std::make_shared<DataTypeArray>(null)), + make_nullable(std::make_shared<DataTypeMap>( + 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<arrow::NullArray>(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<std::pair<int64_t, int64_t>> {{-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<uint8_t> validity {0x05}; + const auto source = std::make_shared<arrow::StructArray>( + arrow::struct_({arrow::field("empty", arrow::null())}), 3, + arrow::ArrayVector {std::make_shared<arrow::NullArray>(3)}, + arrow::Buffer::Wrap(validity), 1); + const auto type = make_nullable( + std::make_shared<DataTypeStruct>(DataTypes {null_type()}, Strings {"empty"})); + MutableColumnPtr column; + read(source->Slice(1, 2), type, &column); + ASSERT_NE(nullptr, column.get()); + const auto& parent = assert_cast<const ColumnNullable&>(*column); + EXPECT_EQ((NullMap {1, 0}), parent.get_null_map_data()); + const auto& structure = assert_cast<const ColumnStruct&>(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<DataTypeArray>(null)); + const auto values = std::make_shared<arrow::NullArray>(5); + const std::vector<int32_t> offsets {0, 2, 2, 2, 5}; + const std::vector<int64_t> large_offsets {0, 2, 2, 2, 5}; + const std::vector<uint8_t> validity {0x0d}; + const std::vector<std::shared_ptr<arrow::Array>> arrays { + std::make_shared<arrow::ListArray>(arrow::list(arrow::null()), 4, + arrow::Buffer::Wrap(offsets), values, + arrow::Buffer::Wrap(validity), 1), + std::make_shared<arrow::LargeListArray>(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<const ColumnNullable&>(*column); + EXPECT_EQ((sliced ? NullMap {1, 0, 0} : NullMap {0, 1, 0, 0}), + parent.get_null_map_data()); + const auto& list = assert_cast<const ColumnArray&>(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<uint8_t> validity {0x05}; + const auto source = std::make_shared<arrow::FixedSizeListArray>( + arrow::fixed_size_list(arrow::null(), 2), 3, std::make_shared<arrow::NullArray>(6), + arrow::Buffer::Wrap(validity), 1); + const auto type = make_nullable(std::make_shared<DataTypeArray>(null_type())); + MutableColumnPtr column; + read(source->Slice(1, 2), type, &column); + ASSERT_NE(nullptr, column.get()); + const auto& parent = assert_cast<const ColumnNullable&>(*column); + EXPECT_EQ((NullMap {1, 0}), parent.get_null_map_data()); + const auto& list = assert_cast<const ColumnArray&>(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<arrow::NullArray>(3); + const std::vector<int32_t> ids {10, 20, 30}; + const auto id_array = std::make_shared<arrow::Int32Array>(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<arrow::StructArray>(struct_type, 3, + arrow::ArrayVector {nulls, id_array}); + const std::vector<int32_t> offsets {0, 1, 3}; + const auto source = std::make_shared<arrow::ListArray>(arrow::list(struct_type), 2, + arrow::Buffer::Wrap(offsets), structs); + const auto doris_struct = make_nullable(std::make_shared<DataTypeStruct>( + 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<DataTypeArray>(doris_struct)), + &column); + ASSERT_NE(nullptr, column.get()); + const auto& list = assert_cast<const ColumnArray&>( + assert_cast<const ColumnNullable&>(*column).get_nested_column()); + const auto& result = assert_cast<const ColumnStruct&>( + assert_cast<const ColumnNullable&>(list.get_data()).get_nested_column()); + expect_nulls(result.get_column(0), 2); + const auto& id_column = assert_cast<const ColumnInt32&>( + assert_cast<const ColumnNullable&>(result.get_column(1)).get_nested_column()); + EXPECT_EQ((ColumnInt32::Container {20, 30}), id_column.get_data()); +} + +TEST_F(LanceNestedNullTest, ReadsNullMapValues) { + const std::vector<int32_t> keys {10, 20, 30}; + const std::vector<int32_t> offsets {0, 1, 1, 3}; + const std::vector<uint8_t> validity {0x05}; + const auto source = std::make_shared<arrow::MapArray>( + arrow::map(arrow::int32(), arrow::null()), 3, arrow::Buffer::Wrap(offsets), + std::make_shared<arrow::Int32Array>(3, arrow::Buffer::Wrap(keys)), + std::make_shared<arrow::NullArray>(3), arrow::Buffer::Wrap(validity), 1); + const auto type = make_nullable(std::make_shared<DataTypeMap>( + 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<const ColumnNullable&>(*column); + EXPECT_EQ((NullMap {1, 0}), parent.get_null_map_data()); + const auto& map = assert_cast<const ColumnMap&>(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<const ColumnInt32&>( + assert_cast<const ColumnNullable&>(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 cfcb4597963..e23d7c2ea6b 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<std::string> column_names; - std::vector<DataTypePtr> column_types; - ASSERT_TRUE(convert_arrow_schema_to_doris(arrow_schema, &column_names, &column_types).ok()); - - EXPECT_EQ((std::vector<std::string> {"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 00000000000..73d1e7a064b --- /dev/null +++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_nested_null.py @@ -0,0 +1,70 @@ +# 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 using lance_fixture_requirements.txt.""" + +import argparse +from pathlib import Path + +import lance +import pyarrow as pa + + +def expected_table() -> pa.Table: + 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. + 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__": + 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/lance_build_preinstalled_catalog.py b/docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py index d6ea30be203..92220554c7b 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 @@ every "indexed" query in test_lance_vector_search silently ran a flat KNN scan. 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 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 @@ FRAGMENT_ROWS = 512 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/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 00000000000..e5933f5bbd5 Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_transactions/0-d6a5ba97-9164-4b3e-b773-15de06005c54.txn differ diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/18446744073709551614.manifest b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/18446744073709551614.manifest new file mode 100644 index 00000000000..1b5fb4b08fa Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/_versions/18446744073709551614.manifest differ 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 00000000000..491d734467a --- /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 00000000000..0ba4ec7cafa Binary files /dev/null and b/docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/lance/nested_null.lance/data/0110100101011111010110112b5b4742fd82cf05955fa974a3.lance differ 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 77f23f04144..fd9576a3c94 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 final class LanceTypeConverter { 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,8 @@ public final class LanceTypeConverter { ArrowType arrowType = field.getType(); switch (arrowType.getTypeID()) { case Null: - return allowNull ? Type.NULL : Type.UNSUPPORTED; + // 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: @@ -135,24 +131,25 @@ public final class LanceTypeConverter { case LargeList: case FixedSizeList: requireChildren(field, 1); - Type itemType = toDorisType(field.getChildren().get(0), false); - return itemType.isSupported() ? new ArrayType(itemType) : Type.UNSUPPORTED; + Type itemType = toDorisType(field.getChildren().get(0)); + // 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); 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); - return keyType.isSupported() && valueType.isSupported() + Type keyType = toDorisType(key); + Type valueType = toDorisType(value); + return !keyType.equals(Type.UNSUPPORTED) && !valueType.equals(Type.UNSUPPORTED) ? new MapType(keyType, valueType, key.isNullable(), value.isNullable()) : Type.UNSUPPORTED; case Struct: List<StructField> fields = new ArrayList<>(); for (Field child : field.getChildren()) { - Type childType = toDorisType(child, false); - if (!childType.isSupported()) { + Type childType = toDorisType(child); + if (childType.equals(Type.UNSUPPORTED)) { return Type.UNSUPPORTED; } fields.add(new StructField(child.getName(), childType, 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 4bbde106e81..5b079ab4e85 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 @@ public class LanceScanNode extends FileQueryScanNode { private final SearchKind searchKind; private byte[] lanceSubstraitFilter = new byte[0]; private String lancePushdownPredicate = ""; + private final Set<String> lazyMaterializedColumns = new HashSet<>(); private long plannedVersion = -1; private int plannedFragments; private int plannedUnindexedFragments; @@ -166,9 +167,15 @@ public class LanceScanNode extends FileQueryScanNode { } } - /** 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<String> 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<String> 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/datasource/tvf/source/TVFScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java index 2dccffa071b..dd7b4620ada 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.qe.SessionVariable; 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 class TVFScanNode extends FileQueryScanNode { @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/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 4271c7fffa3..be66a89526c 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 class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla 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 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla @Override public PlanFragment visitPhysicalLazyMaterializeTVFScan(PhysicalLazyMaterializeTVFScan tvfRelation, PlanTranslatorContext context) { - List<Slot> 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 03787887563..1278cfee53b 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 class PhysicalLazyMaterializeTVFScan extends PhysicalTVFRelation { 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<Slot> getLazySlots() { + return lazySlots; } @Override 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 3c4acfaab8a..5c4f7bc3a3e 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<LanceFragmentInfo> lanceFragments = Collections.emptyList(); private Set<String> 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 6ec9f12a81c..46e29ff25e5 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 class FileTableValuedFunction extends ExternalFileTableValuedFunction { 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 3e04d614aac..4e4bf163145 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 class LocalTableValuedFunction extends ExternalFileTableValuedFunction { 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/lance/LanceTypeConverterTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java index 9730b0af39e..1e842385f50 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.apache.arrow.vector.types.pojo.FieldType; 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,91 @@ public class LanceTypeConverterTest { 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()); + } + + @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. */ 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 ef2dbb8f041..8c3087962d1 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.SlotDescriptor; 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.qe.SessionVariable; 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 class TVFScanNodeTest { 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 class TVFScanNodeTest { 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); + } } 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 00000000000..a28e4959aa0 --- /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<Slot> 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<LanceScanNode> 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()))); + } +} 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 00000000000..b3f168eda4c --- /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 00000000000..47ba3db8443 --- /dev/null +++ b/regression-test/suites/external_table_p0/lance/test_lance_nested_null.groovy @@ -0,0 +1,67 @@ +// 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<null_type>", columns.find { it[0] == "null_list" }[1].toString()) + assertEquals("map<text,null_type>", 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 + """ + // 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 + """ + } finally { + sql "SET enable_file_scanner_v2 = ${originalScannerV2}" + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
