duckdb: ( df1.sql.select("*,lag(a) over() col1") .select("*,sum(coalesce((col1!=a)::int,0)) over(order by index) col2") .select("*,row_number() over(partition by col2) col3") .filter("col3<=3") .order("index") )