Rust WASM 中函数返回时创建线程

Question

我正在 wasm 环境中使用 Polars。

我注意到 LazyFrame.collect 操作不一致，在处理某些数据集时有时会创建线程。

这是与该问题相关的代码

#[wasm_bindgen]
pub fn start(buff: &[u8],
    item_id:&str, 
    order_id:&str,
    item_name:&str) -> JsValue{

    let cursor = Cursor::new(buff);
    let lf = CsvReader::new(cursor).with_ignore_parser_errors(true).finish().unwrap().lazy();


    let df = lf.groupby([col(order_id)]);
    let df = df.agg([col(item_id),col(item_name)]);
    
    // Error occurs here
    let df = df.collect().unwrap();

} 

使用特定数据集给我带来了错误：

panicked at 'failed to spawn thread: Error { kind: Unsupported, message: "operation not supported on this platform" }'

因为它正在尝试在 WASM 环境中生成线程。

然而，对于其他数据集，这个过程将完美执行。并且它不会尝试创建线程。由于使用各种数据集进行测试，问题似乎不是文件大小。

我想知道 Lazyframe.collect 操作的哪一部分会造成这种不一致以及如何避免它。

工作.csv

Order ID,Product ID,Product Name
InvoiceNo0,Product ID0,Product Name0
InvoiceNo0,Product ID1,Product Name1
InvoiceNo0,Product ID2,Product Name2
InvoiceNo0,Product ID3,Product Name3
InvoiceNo0,Product ID4,Product Name4
InvoiceNo0,Product ID5,Product Name5

不工作.csv

Order ID,Product ID,Product Name
B0000001,P0001,Product - 0001
B0000001,P0002,Product - 0002
B0000001,P0003,Product - 0003
B0000001,P0004,Product - 0004
B0000001,P0005,Product - 0005
B0000002,P0006,Product - 0006

允许 wasm 的 Polars 分叉由https://github.com/universalmind303/polars/tree/wasm提供

您可以在此处查看完整的项目以及两个 CSV 文件： https: //github.com/KivalM/lazyframe-min-test

编辑：describe_plan() 的输出

工作数据集

    [col("Product ID"), col("Product Name")] BY [col("Order ID")] FROM DATAFRAME(in-memory): ["Order ID", "Product ID", "Product Name"];
    project */3 columns |   details: None;
    selection: "None"

数据集不工作

    [col("Product ID"), col("Product Name")] BY [col("Order ID")] FROM DATAFRAME(in-memory): ["Order ID", "Product ID", "Product Name"];
    project */3 columns |   details: None;
    selection: "None"

schema() 的输出

工作数据集

name: Order ID, data type: Utf8
name: Product ID, data type: Utf8
name: Product Name, data type: Utf8

数据集不工作

name: Order ID, data type: Utf8
name: Product ID, data type: Utf8
name: Product Name, data type: Utf8

输出describe_optimized_plan()：

    [col("Product ID"), col("Product Name")] BY [col("Order ID")] FROM DATAFRAME(in-memory): ["Product ID", "Product Name", "Order ID"];
    project 3/3 columns |   details: Some([col("Product ID"), col("Product Name"), col("Order ID")]);
    selection: "None"

编辑：仔细查看源代码后。这个问题似乎不是直接来自任何 Polars 代码。我已将问题追溯到polars-lazy/src/physical_plan/executors/groupby.rs功能

panicked at 'failed to spawn thread: Error { kind: Unsupported, message: "operation not supported on this platform" }'

然后返回一个值

Order ID,Product ID,Product Name
InvoiceNo0,Product ID0,Product Name0
InvoiceNo0,Product ID1,Product Name1
InvoiceNo0,Product ID2,Product Name2
InvoiceNo0,Product ID3,Product Name3
InvoiceNo0,Product ID4,Product Name4
InvoiceNo0,Product ID5,Product Name5

但是，该groupby_helper函数运行完成，并且数据帧已成功创建。groupby_helper当数据帧从to返回时，会出现错误fn execute。奇怪的是，只有当该函数返回时才尝试创建线程。RUST WASM 中是否存在可能导致此类行为的内容？

Answer 1

所以看起来std::thread我在创建分支时错过了 groupby 的操作。

impl Drop for GroupsIdx {
    fn drop(&mut self) {
        let v = std::mem::take(&mut self.all);
        // ~65k took approximately 1ms on local machine, so from that point we drop on other thread
        // to stop query from being blocked
        if v.len() > 1 << 16 {
            std::thread::spawn(move || drop(v));
        } else {
            drop(v);
        }
    }
}

数据集大小决定线程生成。

任何大于(~65k)的组都会生成一个线程。1 << 16

标记为impl仅在非 wasm 目标上编译的功能应该可以解决您的问题。