This is an automated email from the ASF dual-hosted git repository.
fanjia pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new c8a682c9c9 [Fix][SQLTransform] fix the scale loss for the sql
transform (#6553)
c8a682c9c9 is described below
commit c8a682c9c995c8bd40f1c0f38f46e7f05f5fada1
Author: dailai <[email protected]>
AuthorDate: Thu Mar 21 15:12:44 2024 +0800
[Fix][SQLTransform] fix the scale loss for the sql transform (#6553)
---
.../seatunnel/transform/sql/SQLTransform.java | 1 +
.../seatunnel/transform/sql/SQLTransformTest.java | 109 +++++++++++++++++++++
2 files changed, 110 insertions(+)
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java
index bddb1c64f2..ac573f29ce 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java
@@ -167,6 +167,7 @@ public class SQLTransform extends
AbstractCatalogSupportTransform {
fieldNames[i],
fieldTypes[i],
simpleColumn.getColumnLength(),
+ simpleColumn.getScale(),
simpleColumn.isNullable(),
simpleColumn.getDefaultValue(),
simpleColumn.getComment());
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java
new file mode 100644
index 0000000000..afafa57514
--- /dev/null
+++
b/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.");
+ }
+}
