Support int_range(s) in streaming mode

[kalocide]

Description

Using int_range or int_ranges in streaming mode (LazyFrame.sink_*) causes a polars.exceptions.InvalidOperationError. That makes some operations (e.g. prefix explodes) impossible within streaming mode.

Example:

import polars as pl

# imagine a larger-than-ram dataset
lf = pl.scan_parquet("hf://datasets/allenai/tulu-3-sft-mixture/data/*.parquet").head(100)

# get indices
lf = lf.with_columns(
    #indices=pl.lit([0, 1, 2, 3]) # works
    indices=pl.int_ranges(0, pl.col("messages").list.len()) # doesn't work
).explode("indices")

# prefix explode
lf = lf.select(
    messages=pl.col("messages").list.slice(0, pl.col("indices"))
)

print(lf.explain(streaming=True))

# fails here
lf.sink_parquet("output.parquet")

# but this works
#lf.collect().write_parquet("output.parquet")

[kalocide]

The behavior in the example can be approximated with:

...
# get indices
lf = lf.with_columns(
    indices=pl.lit(list(range(1, 32)))
).explode("indices").filter(
    pl.col("indices") <= pl.col("messages").list.len()
)

But only if you know the maximum length of that list. Otherwise, I don't think this is possible without UDFs.

[tlongatalaya]

This has come up multiple times for me, would also find it very useful

[cmdlineluser]

It works for me on the new streaming engine: #20947

import polars as pl
import os

os.environ["POLARS_FORCE_NEW_STREAMING"] = "1"

lf = pl.LazyFrame({"messages": [[1, 2], [3, 4, 5], [4], [78, 9, 10]]})

(
    lf.with_columns(
        indices=pl.int_ranges(pl.col("messages").list.len()) 
    )
    .explode("indices")
    .select(
        messages=pl.col("messages").list.slice(0, pl.col("indices"))
    )
    .sink_parquet("20015.parquet")
)

[tlongatalaya]

It works for me on the new streaming engine: #20947

import polars as pl
import os

os.environ["POLARS_FORCE_NEW_STREAMING"] = "1"

lf = pl.LazyFrame({"messages": [[1, 2], [3, 4, 5], [4], [78, 9, 10]]})

(
lf.with_columns(
indices=pl.int_ranges(pl.col("messages").list.len())
)
.explode("indices")
.select(
messages=pl.col("messages").list.slice(0, pl.col("indices"))
)
.sink_parquet("20015.parquet")
)

That's good news - unfortunately I also need some of the other functionality not yet implemented like .over but will be great once new streaming engine is fully featured.