Query columns of the DataManager with a boolean expression.

        Args:
            expr: Boolean expression to query the columns with.

        Returns:
            DataManager containing the rows where the boolean expression is satisfied.

def query(self, expr, **kwargs):
        """Query columns of the DataManager with a boolean expression.

        Args:
            expr: Boolean expression to query the columns with.

        Returns:
            DataManager containing the rows where the boolean expression is satisfied.
        """

        def gen_table_expr(table, expr):
            resolver = {
                name: FakeSeries(dtype.to_pandas_dtype())
                for name, dtype in zip(table.schema.names, table.schema.types)
            }
            scope = Scope(level=0, resolvers=(resolver,))
            return Expr(expr=expr, env=scope)

        import pyarrow.gandiva as gandiva

        unary_ops = {"~": "not"}
        math_calls = {"log": "log", "exp": "exp", "log10": "log10", "cbrt": "cbrt"}
        bin_ops = {
            "+": "add",
            "-": "subtract",
            "*": "multiply",
            "/": "divide",
            "**": "power",
        }
        cmp_ops = {
            "==": "equal",
            "!=": "not_equal",
            ">": "greater_than",
            "<": "less_than",
            "<=": "less_than_or_equal_to",
            ">": "greater_than",
            ">=": "greater_than_or_equal_to",
            "like": "like",
        }

        def build_node(table, terms, builder):
            if isinstance(terms, Constant):
                return builder.make_literal(
                    terms.value, (pa.from_numpy_dtype(terms.return_type))
                )

            if isinstance(terms, Term):
                return builder.make_field(table.schema.field_by_name(terms.name))

            if isinstance(terms, BinOp):
                lnode = build_node(table, terms.lhs, builder)
                rnode = build_node(table, terms.rhs, builder)
                return_type = pa.from_numpy_dtype(terms.return_type)

                if terms.op == "&":
                    return builder.make_and([lnode, rnode])
                if terms.op == "|":
                    return builder.make_or([lnode, rnode])
                if terms.op in cmp_ops:
                    assert return_type == pa.bool_()
                    return builder.make_function(
                        cmp_ops[terms.op], [lnode, rnode], return_type
                    )
                if terms.op in bin_ops:
                    return builder.make_function(
                        bin_ops[terms.op], [lnode, rnode], return_type
                    )

            if isinstance(terms, UnaryOp):
                return_type = pa.from_numpy_dtype(terms.return_type)
                return builder.make_function(
                    unary_ops[terms.op],
                    [build_node(table, terms.operand, builder)],
                    return_type,
                )

            if isinstance(terms, MathCall):
                return_type = pa.from_numpy_dtype(terms.return_type)
                childern = [
                    build_node(table, child, builder) for child in terms.operands
                ]
                return builder.make_function(
                    math_calls[terms.op], childern, return_type
                )

            raise TypeError("Unsupported term type: %s" % terms)

        def can_be_condition(expr):
            if isinstance(expr.terms, BinOp):
                if expr.terms.op in cmp_ops or expr.terms.op in ("&", "|"):
                    return True
            elif isinstance(expr.terms, UnaryOp):
                if expr.terms.op == "~":
                    return True
            return False

        def filter_with_selection_vector(table, s):
            record_batch = table.to_batches()[0]
            indices = s.to_array()  # .to_numpy()
            new_columns = [
                pa.array(c.to_numpy()[indices]) for c in record_batch.columns
            ]
            return pa.Table.from_arrays(new_columns, record_batch.schema.names)

        def gandiva_query(table, query):
            expr = gen_table_expr(table, query)
            if not can_be_condition(expr):
                raise ValueError("Root operation should be a filter.")
            builder = gandiva.TreeExprBuilder()
            root = build_node(table, expr.terms, builder)
            cond = builder.make_condition(root)
            filt = gandiva.make_filter(table.schema, cond)
            sel_vec = filt.evaluate(table.to_batches()[0], pa.default_memory_pool())
            result = filter_with_selection_vector(table, sel_vec)
            return result

        def gandiva_query2(table, query):
            expr = gen_table_expr(table, query)
            if not can_be_condition(expr):
                raise ValueError("Root operation should be a filter.")
            builder = gandiva.TreeExprBuilder()
            root = build_node(table, expr.terms, builder)
            cond = builder.make_condition(root)
            filt = gandiva.make_filter(table.schema, cond)
            return filt

        def query_builder(arrow_table, **kwargs):
            return gandiva_query(arrow_table, kwargs.get("expr", ""))

        kwargs["expr"] = expr
        func = self._prepare_method(query_builder, **kwargs)
        new_data = self._map_across_full_axis(1, func)
        # Query removes rows, so we need to update the index
        new_index = self.compute_index(0, new_data, False)
        return self.__constructor__(
            new_data, new_index, self.columns, self._dtype_cache
        )