feat (database): array_slice formula (baserow#5062)

* feat: add array_slice() database formula function

* fix(formula): handle NaN arguments in array_slice to prevent DB crash

Wrap start/count args with handle_arg_being_nan before truncating to
int, so expressions like tonumber('x') no longer cause
"cannot convert NaN to integer" at the PostgreSQL level.
NaN is treated as 0 for both arguments.

Author: silvestrid

backend/src/baserow/contrib/database/formula/ast/function_defs.py

@@ -85,6 +85,7 @@
     GreaterThanOrEqualExpr,
     IsNullExpr,
     JSONBArrayJoinValues,
+    JSONBArraySlice,
     JSONBArrayUniqueByValue,
     LessThanEqualOrExpr,
     LessThanExpr,
@@ -263,6 +264,7 @@ def register_formula_functions(registry):
     registry.register(BaserowArrayUnique())
     registry.register(BaserowArrayLength())
     registry.register(BaserowArrayJoinValues())
+    registry.register(BaserowArraySlice())
     # ManyToMany functions
     registry.register(BaserowStringAggManyToManyValues())
     registry.register(BaserowManyToManyCount())
@@ -2498,6 +2500,113 @@ def to_django_expression(self, arg: Expression) -> Expression:
         return JSONBArrayUniqueByValue(arg)
 
 
+class BaserowArraySlice(ThreeArgumentBaserowFunction):
+    type = "array_slice"
+    arg1_type = [BaserowFormulaValidType]
+    arg2_type = [BaserowFormulaNumberType]
+    arg3_type = [BaserowFormulaNumberType]
+
+    def type_function(
+        self,
+        func_call: BaserowFunctionCall[UnTyped],
+        arg1: BaserowExpression[BaserowFormulaValidType],
+        arg2: BaserowExpression[BaserowFormulaNumberType],
+        arg3: BaserowExpression[BaserowFormulaNumberType],
+    ) -> BaserowExpression[BaserowFormulaType]:
+        if arg1.many:
+            arg1 = arg1.expression_type.collapse_many(arg1)
+
+        if not isinstance(arg1.expression_type, BaserowFormulaArrayType):
+            return func_call.with_invalid_type("array_slice requires an array input.")
+
+        return func_call.with_args([arg1, arg2, arg3]).with_valid_type(
+            arg1.expression_type
+        )
+
+    def to_django_expression(
+        self, arg1: Expression, arg2: Expression, arg3: Expression
+    ) -> Expression:
+        either_nan = EqualsExpr(
+            arg2, Value(Decimal("NaN")), output_field=fields.BooleanField()
+        ) | EqualsExpr(arg3, Value(Decimal("NaN")), output_field=fields.BooleanField())
+
+        start_int = trunc_numeric_to_int(arg2)
+        count_int = trunc_numeric_to_int(arg3)
+        abs_count = Func(count_int, function="ABS", output_field=fields.IntegerField())
+
+        is_reverse = LessThanExpr(
+            count_int, Value(0), output_field=fields.BooleanField()
+        )
+
+        array_len = Func(
+            arg1, function="jsonb_array_length", output_field=fields.IntegerField()
+        )
+
+        # Resolve negative start to a 0-based position
+        resolved_start = Case(
+            When(
+                condition=GreaterThanOrEqualExpr(
+                    start_int, Value(0), output_field=fields.BooleanField()
+                ),
+                then=start_int,
+            ),
+            default=Greatest(
+                ExpressionWrapper(
+                    array_len + start_int, output_field=fields.IntegerField()
+                ),
+                Value(0),
+            ),
+            output_field=fields.IntegerField(),
+        )
+
+        # Forward: offset = resolved_start
+        # Backward: offset = max(0, resolved_start - abs_count + 1)
+        offset_expr = Case(
+            When(
+                condition=is_reverse,
+                then=Greatest(
+                    ExpressionWrapper(
+                        resolved_start - abs_count + Value(1),
+                        output_field=fields.IntegerField(),
+                    ),
+                    Value(0),
+                ),
+            ),
+            default=resolved_start,
+            output_field=fields.IntegerField(),
+        )
+
+        # Forward: 0 → NULL (all remaining), else count
+        # Backward: abs(count) — but clamped to (resolved_start + 1)
+        #   so we don't go past the beginning
+        limit_expr = Case(
+            When(
+                condition=is_reverse,
+                then=Least(
+                    abs_count,
+                    ExpressionWrapper(
+                        resolved_start + Value(1),
+                        output_field=fields.IntegerField(),
+                    ),
+                ),
+            ),
+            When(
+                condition=EqualsExpr(
+                    count_int, Value(0), output_field=fields.BooleanField()
+                ),
+                then=Value(None),
+            ),
+            default=count_int,
+            output_field=fields.IntegerField(),
+        )
+
+        return Case(
+            When(condition=either_nan, then=Value([], output_field=JSONField())),
+            default=JSONBArraySlice(arg1, offset_expr, limit_expr, is_reverse),
+            output_field=JSONField(),
+        )
+
+
 class BaserowArrayLength(OneArgumentBaserowFunction):
     type = "array_length"
     arg_type = [BaserowFormulaArrayType]

backend/src/baserow/contrib/database/formula/expression_generator/django_expressions.py

@@ -163,6 +163,65 @@ def as_sql(self, compiler, connection, **extra_context):
         return sql, (*separator_params, *params)
 
 
+class JSONBArraySlice(Expression):
+    """
+    Slice a JSONB array with offset, limit, and optional reverse.
+
+    All parameters should be pre-computed Django expressions:
+    - offset_expr / limit_expr: the forward window (limit may be NULL for "all")
+    - reverse_expr: boolean — when true, output order is reversed
+    """
+
+    def __init__(self, array_expr, offset_expr, limit_expr, reverse_expr):
+        super().__init__(output_field=JSONField())
+        self.array_expr = array_expr
+        self.offset_expr = offset_expr
+        self.limit_expr = limit_expr
+        self.reverse_expr = reverse_expr
+
+    def resolve_expression(
+        self, query=None, allow_joins=True, reuse=None, summarize=False, for_save=False
+    ):
+        c = self.copy()
+        c.is_summary = summarize
+        c.array_expr = self.array_expr.resolve_expression(
+            query, allow_joins, reuse, summarize, for_save
+        )
+        c.offset_expr = self.offset_expr.resolve_expression(
+            query, allow_joins, reuse, summarize, for_save
+        )
+        c.limit_expr = self.limit_expr.resolve_expression(
+            query, allow_joins, reuse, summarize, for_save
+        )
+        c.reverse_expr = self.reverse_expr.resolve_expression(
+            query, allow_joins, reuse, summarize, for_save
+        )
+        return c
+
+    def as_sql(self, compiler, connection):
+        array_sql, array_params = compiler.compile(self.array_expr)
+        offset_sql, offset_params = compiler.compile(self.offset_expr)
+        limit_sql, limit_params = compiler.compile(self.limit_expr)
+        reverse_sql, reverse_params = compiler.compile(self.reverse_expr)
+
+        sql = (
+            "(SELECT COALESCE(jsonb_agg(sub.elem ORDER BY "  # noqa: S608
+            f"CASE WHEN {reverse_sql} THEN -sub.rn ELSE sub.rn END"
+            "), '[]'::jsonb) "
+            "FROM (SELECT t.elem, t.rn "
+            f"FROM jsonb_array_elements({array_sql}) WITH ORDINALITY AS t(elem, rn) "
+            "ORDER BY t.rn "
+            f"OFFSET {offset_sql} "
+            f"LIMIT {limit_sql}) sub)"
+        )
+        return sql, (
+            list(reverse_params)
+            + list(array_params)
+            + list(offset_params)
+            + list(limit_params)
+        )
+
+
 class BaserowFilterExpression(Expression):
     """
     Baserow expression that works with field_name and value