[#5201] feat(client-python): Implement expressions in python client (#…

…5646)

### What changes were proposed in this pull request?
Implement expression from java, including:
- Expression.java
- FunctionExpression.java
- NamedReference.java
- UnparsedExpression.java
- literals/

convert to python client, and add unit test for each class.

### Why are the changes needed?

We need to support the expressions in python client

Fix: #5201

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Need to pass all unit tests.

---------

Co-authored-by: Xun <[email protected]>

clients/client-python/gravitino/api/expressions/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

clients/client-python/gravitino/api/expressions/expression.py

@@ -0,0 +1,51 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from gravitino.api.expressions.named_reference import NamedReference
+
+
+class Expression(ABC):
+    """Base class of the public logical expression API."""
+
+    EMPTY_EXPRESSION: list[Expression] = []
+    """
+    `EMPTY_EXPRESSION` is only used as an input when the default `children` method builds the result.
+    """
+
+    EMPTY_NAMED_REFERENCE: list[NamedReference] = []
+    """
+    `EMPTY_NAMED_REFERENCE` is only used as an input when the default `references` method builds
+    the result array to avoid repeatedly allocating an empty array.
+    """
+
+    @abstractmethod
+    def children(self) -> list[Expression]:
+        """Returns a list of the children of this node. Children should not change."""
+        pass
+
+    def references(self) -> list[NamedReference]:
+        """Returns a list of fields or columns that are referenced by this expression."""
+
+        ref_set: set[NamedReference] = set()
+        for child in self.children():
+            ref_set.update(child.references())
+        return list(ref_set)

clients/client-python/gravitino/api/expressions/function_expression.py

@@ -0,0 +1,92 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+from abc import abstractmethod
+
+from gravitino.api.expressions.expression import Expression
+
+
+class FunctionExpression(Expression):
+    """
+    The interface of a function expression. A function expression is an expression that takes a
+    function name and a list of arguments.
+    """
+
+    @staticmethod
+    def of(function_name: str, *arguments: Expression) -> FuncExpressionImpl:
+        """
+        Creates a new FunctionExpression with the given function name.
+        If no arguments are provided, it uses an empty expression.
+
+        :param function_name: The name of the function.
+        :param arguments: The arguments to the function (optional).
+        :return: The created FunctionExpression.
+        """
+        arguments = list(arguments) if arguments else Expression.EMPTY_EXPRESSION
+        return FuncExpressionImpl(function_name, arguments)
+
+    @abstractmethod
+    def function_name(self) -> str:
+        """Returns the function name."""
+
+    @abstractmethod
+    def arguments(self) -> list[Expression]:
+        """Returns the arguments passed to the function."""
+
+    def children(self) -> list[Expression]:
+        """Returns the arguments as children."""
+        return self.arguments()
+
+
+class FuncExpressionImpl(FunctionExpression):
+    """
+    A concrete implementation of the FunctionExpression interface.
+    """
+
+    _function_name: str
+    _arguments: list[Expression]
+
+    def __init__(self, function_name: str, arguments: list[Expression]):
+        super().__init__()
+        self._function_name = function_name
+        self._arguments = arguments
+
+    def function_name(self) -> str:
+        return self._function_name
+
+    def arguments(self) -> list[Expression]:
+        return self._arguments
+
+    def __str__(self) -> str:
+        if not self._arguments:
+            return f"{self._function_name}()"
+        arguments_str = ", ".join(map(str, self._arguments))
+        return f"{self._function_name}({arguments_str})"
+
+    def __eq__(self, other: FuncExpressionImpl) -> bool:
+        if self is other:
+            return True
+        if other is None or self.__class__ is not other.__class__:
+            return False
+        return (
+            self._function_name == other.function_name()
+            and self._arguments == other.arguments()
+        )
+
+    def __hash__(self) -> int:
+        return hash((self._function_name, tuple(self._arguments)))

clients/client-python/gravitino/api/expressions/literals/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

clients/client-python/gravitino/api/expressions/literals/literal.py

@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from abc import abstractmethod
+from typing import List, TypeVar, Generic
+
+from gravitino.api.expressions.expression import Expression
+from gravitino.api.types.type import Type
+
+T = TypeVar("T")
+
+
+class Literal(Generic[T], Expression):
+    """
+    Represents a constant literal value in the public expression API.
+    """
+
+    @abstractmethod
+    def value(self) -> T:
+        """The literal value."""
+        raise NotImplementedError("Subclasses must implement the `value` method.")
+
+    @abstractmethod
+    def data_type(self) -> Type:
+        """The data type of the literal."""
+        raise NotImplementedError("Subclasses must implement the `data_type` method.")
+
+    def children(self) -> List[Expression]:
+        return Expression.EMPTY_EXPRESSION

clients/client-python/gravitino/api/expressions/literals/literals.py

@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import decimal
+from typing import TypeVar
+from datetime import date, time, datetime
+
+from gravitino.api.expressions.literals.literal import Literal
+from gravitino.api.types.type import Type
+from gravitino.api.types.types import Types
+
+T = TypeVar("T")
+
+
+class LiteralImpl(Literal[T]):
+    """Creates a literal with the given type value."""
+
+    _value: T
+    _data_type: Type
+
+    def __init__(self, value: T, data_type: Type):
+        self._value = value
+        self._data_type = data_type
+
+    def value(self) -> T:
+        return self._value
+
+    def data_type(self) -> Type:
+        return self._data_type
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, LiteralImpl):
+            return False
+        return (self._value == other._value) and (self._data_type == other._data_type)
+
+    def __hash__(self):
+        return hash((self._value, self._data_type))
+
+    def __str__(self):
+        return f"LiteralImpl(value={self._value}, data_type={self._data_type})"
+
+
+class Literals:
+    """The helper class to create literals to pass into Apache Gravitino."""
+
+    NULL = LiteralImpl(None, Types.NullType.get())
+
+    @staticmethod
+    def of(value: T, data_type: Type) -> Literal[T]:
+        return LiteralImpl(value, data_type)
+
+    @staticmethod
+    def boolean_literal(value: bool) -> LiteralImpl[bool]:
+        return LiteralImpl(value, Types.BooleanType.get())
+
+    @staticmethod
+    def byte_literal(value: str) -> LiteralImpl[str]:
+        return LiteralImpl(value, Types.ByteType.get())
+
+    @staticmethod
+    def unsigned_byte_literal(value: str) -> LiteralImpl[str]:
+        return LiteralImpl(value, Types.ByteType.unsigned())
+
+    @staticmethod
+    def short_literal(value: int) -> LiteralImpl[int]:
+        return LiteralImpl(value, Types.ShortType.get())
+
+    @staticmethod
+    def unsigned_short_literal(value: int) -> LiteralImpl[int]:
+        return LiteralImpl(value, Types.ShortType.unsigned())
+
+    @staticmethod
+    def integer_literal(value: int) -> LiteralImpl[int]:
+        return LiteralImpl(value, Types.IntegerType.get())
+
+    @staticmethod
+    def unsigned_integer_literal(value: int) -> LiteralImpl[int]:
+        return LiteralImpl(value, Types.IntegerType.unsigned())
+
+    @staticmethod
+    def long_literal(value: int) -> LiteralImpl[int]:
+        return LiteralImpl(value, Types.LongType.get())
+
+    @staticmethod
+    def unsigned_long_literal(value: int) -> LiteralImpl[int]:
+        return LiteralImpl(value, Types.LongType.unsigned())
+
+    @staticmethod
+    def float_literal(value: float) -> LiteralImpl[float]:
+        return LiteralImpl(value, Types.FloatType.get())
+
+    @staticmethod
+    def double_literal(value: float) -> LiteralImpl[float]:
+        return LiteralImpl(value, Types.DoubleType.get())
+
+    @staticmethod
+    def decimal_literal(value: decimal.Decimal) -> LiteralImpl[decimal.Decimal]:
+        precision: int = len(value.as_tuple().digits)
+        scale: int = -value.as_tuple().exponent
+        return LiteralImpl(value, Types.DecimalType.of(max(precision, scale), scale))
+
+    @staticmethod
+    def date_literal(value: date) -> Literal[date]:
+        return LiteralImpl(value, Types.DateType.get())
+
+    @staticmethod
+    def time_literal(value: time) -> Literal[time]:
+        return Literals.of(value, Types.TimeType.get())
+
+    @staticmethod
+    def timestamp_literal(value: datetime) -> Literal[datetime]:
+        return Literals.of(value, Types.TimestampType.without_time_zone())
+
+    @staticmethod
+    def timestamp_literal_from_string(value: str) -> Literal[datetime]:
+        return Literals.timestamp_literal(datetime.fromisoformat(value))
+
+    @staticmethod
+    def string_literal(value: str) -> Literal[str]:
+        return LiteralImpl(value, Types.StringType.get())
+
+    @staticmethod
+    def varchar_literal(length: int, value: str) -> Literal[str]:
+        return LiteralImpl(value, Types.VarCharType.of(length))