Function test_blob_write_and_commit

crates/paimon/src/table/table_write.rs:850–913

()

    #[tokio::test]
    async fn test_blob_write_and_commit() {
        let file_io = test_file_io();
        let table_path = "memory:/test_blob_write";
        setup_dirs(&file_io, table_path).await;

        let table = Table::new(
            file_io.clone(),
            Identifier::new("default", "test_blob_table"),
            table_path.to_string(),
            test_blob_table_schema(),
            None,
        );

        let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap();

        let schema = Arc::new(ArrowSchema::new(vec![
            ArrowField::new("id", ArrowDataType::Int32, false),
            ArrowField::new("payload", ArrowDataType::Binary, true),
        ]));
        let batch = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(Int32Array::from(vec![1, 2, 3])),
                Arc::new(arrow_array::BinaryArray::from(vec![
                    Some(b"hello" as &[u8]),
                    None,
                    Some(b"world"),
                ])),
            ],
        )
        .unwrap();

        table_write.write_arrow_batch(&batch).await.unwrap();
        let messages = table_write.prepare_commit().await.unwrap();

        assert_eq!(messages.len(), 1);
        // Should have 2 files: 1 parquet (normal cols) + 1 blob (payload)
        assert_eq!(messages[0].new_files.len(), 2);

        let parquet_files: Vec<_> = messages[0]
            .new_files
            .iter()
            .filter(|f| f.file_name.ends_with(".parquet"))
            .collect();
        let blob_files: Vec<_> = messages[0]
            .new_files
            .iter()
            .filter(|f| f.file_name.ends_with(".blob"))
            .collect();
        assert_eq!(parquet_files.len(), 1);
        assert_eq!(blob_files.len(), 1);

        assert_eq!(parquet_files[0].row_count, 3);
        assert_eq!(blob_files[0].row_count, 3);
        assert_eq!(blob_files[0].write_cols, Some(vec!["payload".to_string()]));

        // Commit and verify snapshot
        let commit = TableCommit::new(table, "test-user".to_string());