[Fix][SQLTransform] fix the scale loss for the sql transform

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java

@@ -167,6 +167,7 @@ protected TableSchema transformTableSchema() {
                                 fieldNames[i],
                                 fieldTypes[i],
                                 simpleColumn.getColumnLength(),
+                                simpleColumn.getScale(),
                                 simpleColumn.isNullable(),
                                 simpleColumn.getDefaultValue(),
                                 simpleColumn.getComment());

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java

@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.transform.sql;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Objects;
+
+public class SQLTransformTest {
+
+    private static final String TEST_NAME = "test";
+    private static final String TIMESTAMP_FILEDNAME = "create_time";
+    private static final String[] FILED_NAMES =
+            new String[] {"id", "name", "age", TIMESTAMP_FILEDNAME};
+    private static final String GENERATE_PARTITION_KEY = "dt";
+    private static final ReadonlyConfig READONLY_CONFIG =
+            ReadonlyConfig.fromMap(
+                    new HashMap() {
+                        {
+                            put(
+                                    "query",
+                                    "select *,FORMATDATETIME(create_time,'yyyy-MM-dd HH:mm') as dt from test");
+                        }
+                    });
+
+    @Test
+    public void testScaleSupport() {
+        SQLTransform sqlTransform = new SQLTransform(READONLY_CONFIG, getCatalogTable());
+        TableSchema tableSchema = sqlTransform.transformTableSchema();
+        tableSchema
+                .getColumns()
+                .forEach(
+                        column -> {
+                            if (column.getName().equals(TIMESTAMP_FILEDNAME)) {
+                                Assertions.assertEquals(9, column.getScale());
+                            } else if (column.getName().equals(GENERATE_PARTITION_KEY)) {
+                                Assertions.assertTrue(Objects.isNull(column.getScale()));
+                            } else {
+                                Assertions.assertEquals(3, column.getColumnLength());
+                            }
+                        });
+    }
+
+    private CatalogTable getCatalogTable() {
+        SeaTunnelRowType rowType =
+                new SeaTunnelRowType(
+                        FILED_NAMES,
+                        new SeaTunnelDataType[] {
+                            BasicType.INT_TYPE,
+                            BasicType.STRING_TYPE,
+                            BasicType.INT_TYPE,
+                            LocalTimeType.LOCAL_DATE_TIME_TYPE
+                        });
+        TableSchema.Builder schemaBuilder = TableSchema.builder();
+        for (int i = 0; i < rowType.getTotalFields(); i++) {
+            Integer scale = null;
+            Long columnLength = null;
+            if (rowType.getFieldName(i).equals(TIMESTAMP_FILEDNAME)) {
+                scale = 9;
+            } else {
+                columnLength = 3L;
+            }
+            PhysicalColumn column =
+                    PhysicalColumn.of(
+                            rowType.getFieldName(i),
+                            rowType.getFieldType(i),
+                            columnLength,
+                            scale,
+                            true,
+                            null,
+                            null);
+            schemaBuilder.column(column);
+        }
+        return CatalogTable.of(
+                TableIdentifier.of(TEST_NAME, TEST_NAME, null, TEST_NAME),
+                schemaBuilder.build(),
+                new HashMap<>(),
+                new ArrayList<>(),
+                "It has column information.");
+    }
+}