Function test_insert_overwrite

crates/integrations/datafusion/src/table/mod.rs:595–681

()

    #[tokio::test]
    async fn test_insert_overwrite() {
        use paimon::io::FileIOBuilder;
        use paimon::spec::{DataType, IntType, Schema as PaimonSchema, TableSchema, VarCharType};

        let file_io = FileIOBuilder::new("memory").build().unwrap();
        let table_path = "memory:/test_df_insert_overwrite";
        file_io
            .mkdirs(&format!("{table_path}/snapshot/"))
            .await
            .unwrap();
        file_io
            .mkdirs(&format!("{table_path}/manifest/"))
            .await
            .unwrap();

        let schema = PaimonSchema::builder()
            .column("pt", DataType::VarChar(VarCharType::string_type()))
            .column("id", DataType::Int(IntType::new()))
            .partition_keys(["pt"])
            .build()
            .unwrap();
        let table_schema = TableSchema::new(0, &schema);
        let table = paimon::table::Table::new(
            file_io,
            Identifier::new("default", "test_overwrite"),
            table_path.to_string(),
            table_schema,
            None,
        );

        let provider = PaimonTableProvider::try_new(table).unwrap();
        let ctx = SessionContext::new();
        ctx.register_table("t", Arc::new(provider)).unwrap();

        // Initial INSERT: partition "a" and "b"
        ctx.sql("INSERT INTO t VALUES ('a', 1), ('a', 2), ('b', 3), ('b', 4)")
            .await
            .unwrap()
            .collect()
            .await
            .unwrap();

        // INSERT OVERWRITE with only partition "a" data
        // Should overwrite partition "a" but leave partition "b" intact
        ctx.sql("INSERT OVERWRITE t VALUES ('a', 10), ('a', 20)")
            .await
            .unwrap()
            .collect()
            .await
            .unwrap();

        // Read back
        let batches = ctx
            .sql("SELECT pt, id FROM t ORDER BY pt, id")
            .await
            .unwrap()
            .collect()
            .await