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..8599b7c3850e0d 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,12 @@ 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()); + // 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& 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 +247,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 +260,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 +768,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..8d1115687df943 --- /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 +#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}))), + 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(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(*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> { + 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( + 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..73d1e7a064bc54 --- /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 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/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 00000000000000..e5933f5bbd5074 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 00000000000000..1b5fb4b08fa75e 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 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 00000000000000..0ba4ec7cafa839 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 77f23f04144fd7..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 @@ -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,8 @@ private static Type toDorisType(Field field, boolean allowNull) { 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 @@ private static Type toDorisType(Field field, boolean allowNull) { 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 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 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/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/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/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/lance/LanceTypeConverterTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTypeConverterTest.java index 9730b0af39e95c..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 @@ -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,91 @@ 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()); + } + + @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 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); + } } 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()))); + } +} 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..b3f168eda4c10a --- /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..47ba3db8443dfa --- /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", 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 + """ + // 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}" + } +}