Support topK using the DBSP operator

[mihaibudiu]

Is this a user-visible change (yes/no): no

[mihaibudiu]

I will work on a commit to add more tests.

[mihaibudiu]

this is wrong, I will fix the comment in the next commit

[mihaibudiu]

Here is a sample output produced:

        struct Cmpstream25;
        impl CmpFunc<Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>> for Cmpstream25 {
            fn cmp(left: &Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>, right: &Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>) -> std::cmp::Ordering {
                let ord = left.1.cmp(&right.1);
                if ord != Ordering::Equal { return ord; };
                let ord = left.0.cmp(&right.0);
                if ord != Ordering::Equal { return ord; };
                let ord = left.2.cmp(&right.2);
                if ord != Ordering::Equal { return ord; };
                let ord = left.3.cmp(&right.3);
                if ord != Ordering::Equal { return ord; };
                let ord = left.4.cmp(&right.4);
                if ord != Ordering::Equal { return ord; };
                let ord = left.5.cmp(&right.5);
                if ord != Ordering::Equal { return ord; };
                return Ordering::Equal;
            }
        }
        // DBSPIndexedTopKOperator 556
        let stream25: Stream<_, OrdIndexedZSet<(), Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>, Weight>> = stream23.topk_custom_order::<Cmpstream25>(cast_to_u_i32(1i32));

[mihaibudiu]

this code is unchanged

[ryzhyk]

Here is a sample output produced:

        struct Cmpstream25;
        impl CmpFunc<Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>> for Cmpstream25 {
            fn cmp(left: &Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>, right: &Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>) -> std::cmp::Ordering {
                let ord = left.1.cmp(&right.1);
                if ord != Ordering::Equal { return ord; };
                let ord = left.0.cmp(&right.0);
                if ord != Ordering::Equal { return ord; };
                let ord = left.2.cmp(&right.2);
                if ord != Ordering::Equal { return ord; };
                let ord = left.3.cmp(&right.3);
                if ord != Ordering::Equal { return ord; };
                let ord = left.4.cmp(&right.4);
                if ord != Ordering::Equal { return ord; };
                let ord = left.5.cmp(&right.5);
                if ord != Ordering::Equal { return ord; };
                return Ordering::Equal;
            }
        }
        // DBSPIndexedTopKOperator 556
        let stream25: Stream<_, OrdIndexedZSet<(), Tuple6<i32, F64, bool, String, Option<i32>, Option<F64>>, Weight>> = stream23.topk_custom_order::<Cmpstream25>(cast_to_u_i32(1i32));

Looks good, assuming it's all order by .. ASC in SQL.

[mihaibudiu]

Right, have to handle descending. Easy.

[ryzhyk]

BTW, I think this is a user-visible change, which requires docs + a changelog entry.

[mihaibudiu]

which part is visible? LIMIT was there in the grammar.

[ryzhyk]

which part is visible? LIMIT was there in the grammar.

Clearly, we're adding a new user-visible capability.

[mihaibudiu]

No, we are fixing a bug

[ryzhyk]

No, we are fixing a bug

So you're saying we're implementing a behavior that documentation already promised before. If the documentation is accurate, then we're good. A changelog entry documenting the bug fix is still in order.

[mihaibudiu]

This code was moved here from DBSPExecutor