fix(io): normalize the FixedSizeList type of VECTOR batches

src/paimon/core/io/vector_file_batch_reader.cpp

@@ -168,7 +168,13 @@ Result<std::shared_ptr<arrow::Array>> ConvertToReadType(
                                                        vector_type.ToString()));
                 }
                 PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(*array));
-                return array;
+                // Writers disagree on the element field, for example `element: float not null`
+                // for Paimon Rust against the `item: float` of a Paimon schema. Restore the
+                // requested type so that files storing VECTOR as LIST and files storing it as
+                // FixedSizeList produce batches of one type.
+                std::shared_ptr<arrow::ArrayData> data = array->data()->Copy();
+                data->type = read_type;
+                return arrow::MakeArray(data);
             }
             return CastListToVector(
                 array, checked_pointer_cast<arrow::FixedSizeListType>(read_type), pool);

src/paimon/core/io/vector_file_batch_reader_test.cpp

@@ -112,6 +112,46 @@ TEST(VectorFileBatchReaderTest, KeepFixedSizeListFileSchema) {
     ASSERT_TRUE(logical_array->Equals(std::move(actual_result).ValueOrDie()));
 }
 
+// Paimon Rust names the element field of a VECTOR column `element` and marks it non-nullable,
+// while a Paimon schema names it `item`. Batches must carry the requested type either way,
+// otherwise they cannot be combined with batches read from a file storing VECTOR as LIST.
+TEST(VectorFileBatchReaderTest, NormalizeFixedSizeListElementField) {
+    auto file_vector =
+        arrow::fixed_size_list(arrow::field("element", arrow::float32(), /*nullable=*/false), 3);
+    auto logical_vector = arrow::fixed_size_list(arrow::float32(), 3);
+    auto file_type = AsStructType(arrow::struct_({
+        arrow::field("embedding", file_vector),
+        arrow::field("history", arrow::list(file_vector)),
+    }));
+    auto logical_type = AsStructType(arrow::struct_({
+        arrow::field("embedding", logical_vector),
+        arrow::field("history", arrow::list(logical_vector)),
+    }));
+    const std::string json = R"([
+        [[1.0, 2.0, 3.0], [[4.0, 5.0, 6.0]]],
+        [null, []]
+    ])";
+    auto file_array = arrow::ipc::internal::json::ArrayFromJSON(file_type, json).ValueOrDie();
+    auto mock_reader =
+        std::make_unique<MockFileBatchReader>(file_array, file_type, /*batch_size=*/10);
+    mock_reader->EnableRandomizeBatchSize(false);
+    VectorFileBatchReader reader(std::move(mock_reader), GetDefaultPool());
+
+    ArrowSchema c_read_schema;
+    ASSERT_TRUE(arrow::ExportSchema(*arrow::schema(logical_type->fields()), &c_read_schema).ok());
+    ASSERT_OK(reader.SetReadSchema(&c_read_schema, /*predicate=*/nullptr,
+                                   /*selection_bitmap=*/std::nullopt));
+
+    ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, reader.NextBatch());
+    arrow::Result<std::shared_ptr<arrow::Array>> actual_result =
+        arrow::ImportArray(batch.first.get(), batch.second.get());
+    ASSERT_TRUE(actual_result.ok()) << actual_result.status().ToString();
+    std::shared_ptr<arrow::Array> actual = std::move(actual_result).ValueOrDie();
+    ASSERT_TRUE(actual->type()->Equals(logical_type)) << actual->type()->ToString();
+    auto expected = arrow::ipc::internal::json::ArrayFromJSON(logical_type, json).ValueOrDie();
+    ASSERT_TRUE(expected->Equals(actual)) << actual->ToString();
+}
+
 TEST(VectorFileBatchReaderTest, ConvertNestedVectorsWithBitmap) {
     auto logical_vector =
         arrow::fixed_size_list(arrow::field("item", arrow::float64(), /*nullable=*/false), 2);

src/paimon/format/parquet/parquet_vector_io_test.cpp

@@ -368,6 +368,45 @@ TEST_F(ParquetVectorIoTest, ReadNullableJavaFixture) {
                         {{{1.0f, 2.0f, 3.0f}}, std::nullopt, {{4.0f, 5.0f, 6.0f}}});
 }
 
+// A table can hold files from several writers, and Paimon Java stores VECTOR as Parquet LIST
+// while Paimon Rust stores it as FixedSizeList. Reading both with the table schema must produce
+// batches of one Arrow type, otherwise they cannot be combined into a single result.
+TEST_F(ParquetVectorIoTest, ReadMixedListAndFixedSizeListFixtures) {
+    // The Arrow type a Paimon schema builds for `id INT, embedding VECTOR<FLOAT, 3>`. The Rust
+    // fixture instead names the element field `element` and marks it non-nullable.
+    auto logical_schema =
+        arrow::schema({arrow::field("id", arrow::int32()),
+                       arrow::field("embedding", arrow::fixed_size_list(arrow::float32(), 3))});
+    std::shared_ptr<arrow::DataType> logical_type = arrow::struct_(logical_schema->fields());
+
+    arrow::ArrayVector chunks;
+    for (const std::string& file_name : {"java_vector_nullable.parquet", "rust_vector.parquet"}) {
+        std::string file_path =
+            paimon::test::GetDataDir() + "/parquet/vector_compatibility/" + file_name;
+        std::unique_ptr<FileBatchReader> reader;
+        CreateVectorReader(file_path, logical_schema, /*predicate=*/nullptr, /*options=*/{},
+                           /*batch_size=*/10, &reader);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual,
+                             paimon::test::ReadResultCollector::CollectResult(reader.get()));
+        ASSERT_TRUE(actual->type()->Equals(logical_type))
+            << file_name << ": " << actual->type()->ToString();
+        chunks.insert(chunks.end(), actual->chunks().begin(), actual->chunks().end());
+    }
+
+    arrow::Result<std::shared_ptr<arrow::ChunkedArray>> merged_result =
+        arrow::ChunkedArray::Make(chunks);
+    ASSERT_TRUE(merged_result.ok()) << merged_result.status().ToString();
+    arrow::Result<std::shared_ptr<arrow::Array>> expected_result =
+        arrow::ipc::internal::json::ArrayFromJSON(
+            logical_type, R"([[1, [1.0, 2.0, 3.0]], [2, null], [3, [4.0, 5.0, 6.0]],
+                              [1, [1.0, 2.0, 3.0]], [2, [7.0, 8.0, 9.0]], [3, [4.0, 5.0, 6.0]]])");
+    ASSERT_TRUE(expected_result.ok()) << expected_result.status().ToString();
+    std::shared_ptr<arrow::ChunkedArray> merged = std::move(merged_result).ValueOrDie();
+    ASSERT_TRUE(std::make_shared<arrow::ChunkedArray>(std::move(expected_result).ValueOrDie())
+                    ->Equals(merged))
+        << merged->ToString();
+}
+
 // A writer that stores the Arrow schema, such as Paimon Rust or Python, exposes the VECTOR column
 // as FixedSizeList. Arrow 17 cannot read a null value from such a column: Parquet stores a null
 // list slot with no values, while FixedSizeListReader::AssembleArray in