add nullType for null literal value

api/src/main/java/com/datastrato/gravitino/rel/expressions/literals/Literals.java

@@ -13,6 +13,8 @@
 
 /** The helper class to create literals to pass into Gravitino. */
 public class Literals {
+  /** Used to represent a null literal. */
+  public static final Literal<Types.NullType> NULL = new LiteralImpl<>(null, Types.NullType.get());
 
   /**
    * Creates a literal with the given value and data type.

api/src/main/java/com/datastrato/gravitino/rel/types/Type.java

@@ -36,7 +36,8 @@ enum Name {
     STRUCT,
     LIST,
     MAP,
-    UNION
+    UNION,
+    NULL
   }
 
   /** The base type of all primitive types. */

api/src/main/java/com/datastrato/gravitino/rel/types/Types.java

@@ -12,6 +12,28 @@
 /** The helper class for {@link Type}. */
 public class Types {
 
+  /** The data type representing `NULL` values. */
+  public static class NullType implements Type {
+    private static final NullType INSTANCE = new NullType();
+
+    /** @return The singleton instance of {@link NullType}. */
+    public static NullType get() {
+      return INSTANCE;
+    }
+
+    private NullType() {}
+
+    @Override
+    public Name name() {
+      return Name.NULL;
+    }
+
+    @Override
+    public String simpleString() {
+      return "null";
+    }
+  }
+
   /** The boolean type in Gravitino. */
   public static class BooleanType extends Type.PrimitiveType {
     private static final BooleanType INSTANCE = new BooleanType();

api/src/test/java/com/datastrato/gravitino/rel/TestLiteral.java

@@ -17,6 +17,7 @@
 import static com.datastrato.gravitino.rel.expressions.literals.Literals.timestamp;
 
 import com.datastrato.gravitino.rel.expressions.literals.Literal;
+import com.datastrato.gravitino.rel.expressions.literals.Literals;
 import com.datastrato.gravitino.rel.types.Types;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -71,5 +72,7 @@ public void testLiterals() {
     literal = string("hello");
     Assertions.assertEquals(literal.value(), "hello");
     Assertions.assertEquals(literal.dataType(), Types.StringType.get());
+
+    Assertions.assertEquals(Literals.of(null, Types.NullType.get()), Literals.NULL);
   }
 }

api/src/test/java/com/datastrato/gravitino/rel/TestTypes.java

@@ -19,6 +19,11 @@ public void testPrimitiveTypes() {
     Assertions.assertSame(booleanType, Types.BooleanType.get());
     Assertions.assertEquals("boolean", booleanType.simpleString());
 
+    Types.NullType nullType = Types.NullType.get();
+    Assertions.assertEquals(Type.Name.NULL, nullType.name());
+    Assertions.assertSame(nullType, Types.NullType.get());
+    Assertions.assertEquals("null", nullType.simpleString());
+
     Types.ByteType byteType = Types.ByteType.get();
     Assertions.assertEquals(Type.Name.BYTE, byteType.name());
     Assertions.assertSame(byteType, Types.ByteType.get());

common/src/main/java/com/datastrato/gravitino/dto/util/DTOConverters.java

@@ -193,8 +193,8 @@ public static FunctionArg toFunctionArg(Expression expression) {
 
     if (expression instanceof Literal) {
       return new LiteralDTO.Builder()
-          .withValue(((Literal<String>) expression).value())
-          .withDataType(((Literal<String>) expression).dataType())
+          .withValue(((Literal<?>) expression).value().toString())
+          .withDataType(((Literal<?>) expression).dataType())
           .build();
     } else if (expression instanceof NamedReference.FieldReference) {
       return new FieldReferenceDTO.Builder()
@@ -222,14 +222,14 @@ public static FunctionArg[] toFunctionArg(Expression[] expressions) {
 
   public static SortOrderDTO[] toDTOs(SortOrder[] sortOrders) {
     if (ArrayUtils.isEmpty(sortOrders)) {
-      return new SortOrderDTO[0];
+      return SortOrderDTO.EMPTY_SORT;
     }
     return Arrays.stream(sortOrders).map(DTOConverters::toDTO).toArray(SortOrderDTO[]::new);
   }
 
   public static Partitioning[] toDTOs(Transform[] transforms) {
     if (ArrayUtils.isEmpty(transforms)) {
-      return new Partitioning[0];
+      return Partitioning.EMPTY_PARTITIONING;
     }
     return Arrays.stream(transforms).map(DTOConverters::toDTO).toArray(Partitioning[]::new);
   }
@@ -242,10 +242,10 @@ public static Distribution fromDTO(DistributionDTO distributionDTO) {
     return Distributions.of(
         distributionDTO.strategy(),
         distributionDTO.number(),
-        fromFunctionArg(distributionDTO.args()));
+        fromFunctionArgs(distributionDTO.args()));
   }
 
-  public static Expression[] fromFunctionArg(FunctionArg[] args) {
+  public static Expression[] fromFunctionArgs(FunctionArg[] args) {
     if (ArrayUtils.isEmpty(args)) {
       return Expression.EMPTY_EXPRESSION;
     }
@@ -255,13 +255,16 @@ public static Expression[] fromFunctionArg(FunctionArg[] args) {
   public static Expression fromFunctionArg(FunctionArg arg) {
     switch (arg.argType()) {
       case LITERAL:
+        if (((LiteralDTO) arg).value() == null) {
+          return Literals.NULL;
+        }
         return Literals.of(((LiteralDTO) arg).value(), ((LiteralDTO) arg).dataType());
       case FIELD:
         return NamedReference.field(((FieldReferenceDTO) arg).fieldName());
       case FUNCTION:
         return FunctionExpression.of(
             ((FuncExpressionDTO) arg).functionName(),
-            fromFunctionArg(((FuncExpressionDTO) arg).args()));
+            fromFunctionArgs(((FuncExpressionDTO) arg).args()));
       default:
         throw new IllegalArgumentException("Unsupported expression type: " + arg.getClass());
     }
@@ -317,7 +320,7 @@ public static Transform fromDTO(Partitioning partitioning) {
       case FUNCTION:
         return Transforms.apply(
             ((FunctionPartitioningDTO) partitioning).functionName(),
-            fromFunctionArg(((FunctionPartitioningDTO) partitioning).args()));
+            fromFunctionArgs(((FunctionPartitioningDTO) partitioning).args()));
       default:
         throw new IllegalArgumentException("Unsupported partitioning: " + partitioning.strategy());
     }