Bind Short/BigDecimal values in Java and Spark table sinks

SqlTypeUtils now maps Short->SMALLINT and BigDecimal->NUMERIC for the generated
DDL, but the sinks still bound those values via the String fallback. Complete the
write path:
- JavaTableSink: bind via setShort / setBigDecimal
- SparkTableSink: map Short->ShortType and BigDecimal->DecimalType(38,18) so the
  Spark DataFrame writes SMALLINT / NUMERIC columns instead of text

Author: gepa

wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaTableSink.java

@@ -59,6 +59,10 @@ private void setRecordValue(PreparedStatement ps, int index, Object value) throw
             ps.setDouble(index, (Double) value);
         } else if (value instanceof Float) {
             ps.setFloat(index, (Float) value);
+        } else if (value instanceof Short) {
+            ps.setShort(index, (Short) value);
+        } else if (value instanceof java.math.BigDecimal) {
+            ps.setBigDecimal(index, (java.math.BigDecimal) value);
         } else if (value instanceof Boolean) {
             ps.setBoolean(index, (Boolean) value);
         } else if (value instanceof java.sql.Date) {

wayang-platforms/wayang-spark/src/main/java/org/apache/wayang/spark/operators/SparkTableSink.java

@@ -146,10 +146,14 @@ private org.apache.spark.sql.types.DataType getSparkDataType(Class<?> cls) {
             return DataTypes.IntegerType;
         if (cls == Long.class || cls == long.class)
             return DataTypes.LongType;
+        if (cls == Short.class || cls == short.class)
+            return DataTypes.ShortType;
         if (cls == Double.class || cls == double.class)
             return DataTypes.DoubleType;
         if (cls == Float.class || cls == float.class)
             return DataTypes.FloatType;
+        if (cls == java.math.BigDecimal.class)
+            return DataTypes.createDecimalType(38, 18);
         if (cls == Boolean.class || cls == boolean.class)
             return DataTypes.BooleanType;
         if (cls == java.sql.Date.class || cls == java.time.LocalDate.class)